feat(orchestrator): add persistent dispatch state tracking (#356)

Track which issues have been dispatched across cycles in a local
JSON file at ~/.hermes/orchestrator/dispatch_state.json.

This prevents re-dispatching the same issue after restart and
satisfies the remaining requirement for issue #356:
"Track dispatch state in a local SQLite or JSON."

Changes:
- Add DISPATCH_STATE_PATH constant
- Add load_dispatch_state(), save_dispatch_state()
- Add is_already_dispatched() and mark_dispatched() helpers
- dispatch_cycle() now checks persistent state before dispatching
- Successful dispatches are persisted immediately
- dry-run mode does not affect persistent state

Closes #356

hermes-sovereign/orchestrator/orchestrator.py

@@ -30,6 +30,9 @@ REPOS = ["timmy-config", "the-nexus", "timmy-home"]
 TELEGRAM_CHAT_ID = "-1003664764329"
 DAEMON_INTERVAL = 900  # 15 minutes
 
+# Dispatch state tracking (persistent JSON)
+DISPATCH_STATE_PATH = os.path.expanduser("~/.hermes/orchestrator/dispatch_state.json")
+
 # Tags that mark issues we should never auto-dispatch
 FILTER_TAGS = ["[EPIC]", "[DO NOT CLOSE]", "[PERMANENT]", "[PHILOSOPHY]", "[MORNING REPORT]"]
 
@@ -144,6 +147,53 @@ def send_telegram(message):
         return False
 
 
+# ---------------------------------------------------------------------------
+# DISPATCH STATE PERSISTENCE
+# ---------------------------------------------------------------------------
+
+def load_dispatch_state():
+    """Load persistent dispatch state from JSON file."""
+    path = DISPATCH_STATE_PATH
+    if not os.path.exists(path):
+        return {"dispatched": {}}
+    try:
+        with open(path) as f:
+            data = json.load(f)
+            if "dispatched" not in data:
+                data["dispatched"] = {}
+            return data
+    except Exception as e:
+        print(f"[WARN] Failed to load dispatch state: {e}")
+        return {"dispatched": {}}
+
+
+def save_dispatch_state(state):
+    """Save dispatch state to JSON file."""
+    path = DISPATCH_STATE_PATH
+    try:
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, 'w') as f:
+            json.dump(state, f, indent=2)
+    except Exception as e:
+        print(f"[WARN] Failed to save dispatch state: {e}")
+
+
+def is_already_dispatched(state, repo, number):
+    """Check if an issue has already been dispatched."""
+    key = f"{repo}#{number}"
+    return key in state.get("dispatched", {})
+
+
+def mark_dispatched(state, repo, number, agent_name, dry_run=False):
+    """Record that an issue has been dispatched."""
+    key = f"{repo}#{number}"
+    state.setdefault("dispatched", {})[key] = {
+        "agent": agent_name,
+        "dispatched_at": datetime.now(timezone.utc).isoformat(),
+        "dry_run": dry_run,
+    }
+
+
 # ---------------------------------------------------------------------------
 # 1. BACKLOG READER
 # ---------------------------------------------------------------------------
@@ -432,7 +482,7 @@ def dispatch_to_gateway(agent_name, agent, issue):
         return False
 
 
-def dispatch_cycle(backlog, agent_status, dry_run=False):
+def dispatch_cycle(backlog, agent_status, dispatch_state, dry_run=False):
     """Run one dispatch cycle. Returns dispatch report."""
     dispatched = []
     skipped = []
@@ -446,6 +496,11 @@ def dispatch_cycle(backlog, agent_status, dry_run=False):
             skipped.append((issue, "already assigned to agent"))
             continue
 
+        # Check if already dispatched in a previous cycle (persistent state)
+        if is_already_dispatched(dispatch_state, issue["repo"], issue["number"]):
+            skipped.append((issue, "already dispatched in previous cycle"))
+            continue
+
         if issue["score"] < 5:
             skipped.append((issue, "score too low"))
             continue
@@ -483,6 +538,10 @@ def dispatch_cycle(backlog, agent_status, dry_run=False):
                 "score": issue["score"],
             })
             dispatched_count[best_agent] = dispatched_count.get(best_agent, 0) + 1
+
+            # Persist dispatch state
+            mark_dispatched(dispatch_state, issue["repo"], issue["number"], best_agent, dry_run=False)
+            save_dispatch_state(dispatch_state)
         else:
             skipped.append((issue, "assignment failed"))
 
@@ -578,8 +637,11 @@ def format_telegram_report(backlog, dispatched, skipped, agent_status, dry_run=F
 def run_cycle(dry_run=False):
     """Execute one full orchestration cycle."""
     global GITEA_TOKEN, TELEGRAM_TOKEN
-    GITEA_TOKEN = load_gitea_token()
-    TELEGRAM_TOKEN = load_telegram_token()
+    GITEA_TOKEN=load_gitea_token()
+    TELEGRAM_TOKEN=load_telegram_token()
+
+    # Load persistent dispatch state
+    dispatch_state = load_dispatch_state()
 
     print("\n[1/4] Reading backlog...")
     backlog = read_backlog()
@@ -593,7 +655,7 @@ def run_cycle(dry_run=False):
     agent_status = get_agent_status()
 
     print("\n[4/4] Dispatching...")
-    dispatched, skipped = dispatch_cycle(backlog, agent_status, dry_run=dry_run)
+    dispatched, skipped = dispatch_cycle(backlog, agent_status, dispatch_state, dry_run=dry_run)
 
     # Generate reports
     report = generate_report(backlog, dispatched, skipped, agent_status, dry_run=dry_run)