Something went wrong. Try again.
CLI tools and operational extensions for the void architecture on ATProtocol.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081#!/usr/bin/env python3"""Post-dispatch verification: Compares inbox actionable items to dispatched actions.Alerts if under-dispatch detected."""import osimport yamlimport jsonimport sys
def count_actionable_items(inbox_path='inbox.yaml'): """Count actionable items (mentions, replies, non-passive notifications).""" if not os.path.exists(inbox_path): return 0
try: with open(inbox_path, 'r') as f: inbox = yaml.safe_load(f)
if not inbox or not isinstance(inbox, list): return 0
actionable = 0 PASSIVE_TYPES = {'like', 'repost', 'follow'}
for item in inbox: notif_type = item.get('type', '').lower() # Count if it's not a passive type if notif_type not in PASSIVE_TYPES: actionable += 1
return actionable except Exception as e: print(f"Error reading inbox: {e}", file=sys.stderr) return 0
def count_dispatched_actions(): """Count actions in the most recent outbox archive.""" archive_dir = 'outbox_archive' if not os.path.exists(archive_dir): return 0
try: # Get most recent archive file archives = sorted([f for f in os.listdir(archive_dir) if f.endswith('.yaml')]) if not archives: return 0
latest = os.path.join(archive_dir, archives[-1]) with open(latest, 'r') as f: outbox = yaml.safe_load(f)
actions = outbox.get('actions', []) # Count non-ignore actions return len([a for a in actions if a.get('action') != 'ignore']) except Exception as e: print(f"Error reading archive: {e}", file=sys.stderr) return 0
def main(): actionable = count_actionable_items() dispatched = count_dispatched_actions()
print(f"Dispatch verification: {actionable} actionable items, {dispatched} dispatched")
if actionable > dispatched: deficit = actionable - dispatched print(f"WARNING: Under-dispatch detected. {deficit} actionable items not addressed.") # Alert level depends on deficit size if deficit >= 3: print("ALERT: Significant under-dispatch. Manual review recommended.") sys.exit(2) # Special exit code for under-dispatch else: sys.exit(1) # Minor under-dispatch else: print("Dispatch verification: PASSED") sys.exit(0)
if __name__ == '__main__': main()