fix(orchestrator): add persistent dispatch state tracking

Issue #356 requires tracking dispatch state in local SQLite or JSON.
This change adds JSON-based persistence so the orchestrator remembers
which issues have already been dispatched across restarts.

- Add DISPATCH_STATE_PATH constant (~/.hermes/orchestrator/dispatch_state.json)
- Add load_dispatch_state(), save_dispatch_state() helpers
- Add is_already_dispatched() and mark_dispatched() helpers
- dispatch_cycle now skips already-dispatched issues
- Successful dispatches are recorded persistent (dry-run does not mutate)
- Uses stdlib json only; atomic write with .tmp fallback
- Ignore state directory in .gitignore

Closes #356

.gitignore

@@ -38,3 +38,6 @@ reports/
 # Prevent test artifacts
 /test-*.txt
 .DS_Store
+
+# Orchestrator persistent state
+.hermes/orchestrator/

hermes-sovereign/orchestrator/orchestrator.py

@@ -18,6 +18,7 @@ import urllib.request
 import urllib.error
 import urllib.parse
 from datetime import datetime, timezone
+from pathlib import Path
 
 # ---------------------------------------------------------------------------
 # CONFIG
@@ -66,6 +67,8 @@ AGENTS = {
     },
 }
 
+DISPATCH_STATE_PATH = Path.home() / ".hermes" / "orchestrator" / "dispatch_state.json"
+
 # ---------------------------------------------------------------------------
 # CREDENTIALS
 # ---------------------------------------------------------------------------
@@ -432,6 +435,58 @@ def dispatch_to_gateway(agent_name, agent, issue):
         return False
 
 
+
+# ---------------------------------------------------------------------------
+
+def load_dispatch_state():
+    """Load persistent dispatch state from JSON file."""
+    if not DISPATCH_STATE_PATH.exists():
+        return {"dispatched": {}}
+    try:
+        with open(DISPATCH_STATE_PATH, "r") as f:
+            return json.load(f)
+    except (json.JSONDecodeError, IOError):
+        # Corrupt or unreadable — start fresh
+        return {"dispatched": {}}
+
+
+def save_dispatch_state(state):
+    """Save dispatch state atomically."""
+    DISPATCH_STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
+    tmp_path = DISPATCH_STATE_PATH.with_suffix(".tmp")
+    try:
+        with open(tmp_path, "w") as f:
+            json.dump(state, f, indent=2)
+        tmp_path.replace(DISPATCH_STATE_PATH)
+    except Exception:
+        # Best-effort: if atomic replace fails, still try direct write
+        try:
+            with open(DISPATCH_STATE_PATH, "w") as f:
+                json.dump(state, f, indent=2)
+        except Exception:
+            pass  # Last resort: skip persistence
+
+
+def is_already_dispatched(issue_key):
+    """Check if an issue has already been dispatched (persistent check)."""
+    state = load_dispatch_state()
+    return issue_key in state.get("dispatched", {})
+
+
+def mark_dispatched(issue_key, agent_name, dry_run=False):
+    """Record a successful dispatch in persistent state. No-op for dry-run."""
+    if dry_run:
+        return
+    state = load_dispatch_state()
+    state.setdefault("dispatched", {})[issue_key] = {
+        "agent": agent_name,
+        "dispatched_at": datetime.now(timezone.utc).isoformat(),
+    }
+    save_dispatch_state(state)
+
+
+
+
 def dispatch_cycle(backlog, agent_status, dry_run=False):
     """Run one dispatch cycle. Returns dispatch report."""
     dispatched = []
@@ -440,6 +495,11 @@ def dispatch_cycle(backlog, agent_status, dry_run=False):
 
     # Only dispatch unassigned issues (or issues not assigned to agents)
     for issue in backlog:
+        # Skip if already dispatched (persistent state)
+        issue_key = f"{issue['repo']}#{issue['number']}"
+        if is_already_dispatched(issue_key):
+            skipped.append((issue, "already dispatched in persistent state"))
+            continue
         agent_assigned = any(a.lower() in AGENT_USERNAMES for a in issue["assignees"])
 
         if agent_assigned:
@@ -475,6 +535,7 @@ def dispatch_cycle(backlog, agent_status, dry_run=False):
             if agent["type"] == "gateway":
                 dispatch_to_gateway(best_agent, agent, issue)
 
+            mark_dispatched(issue_key, best_agent, dry_run=dry_run)
             dispatched.append({
                 "agent": best_agent,
                 "repo": issue["repo"],