feat(harvester): add 2-turn emergency escalation follow-up to opm on persistent silence stalls

bin/response-harvester.py

@@ -1057,10 +1057,13 @@ def maybe_nudge_untagged_sidechat(agent, thread_id, thread_name, mid, text, dry_
 
     tracker = load_json_file(NUDGE_TRACKER_FILE)
     rec = tracker.get(thread_id, {})
-    if rec.get("nudged_for_mid") == mid or rec.get("nudge_count", 0) >= 1:
+    if rec.get("nudged_for_mid") == mid:
         return
 
-    # Inject strict tool directive footer
+    n_count = rec.get("nudge_count", 0)
+
+    # 1st turn: Strict rejection and tool invocation prompt
+    if n_count == 0:
         thread_url = f"https://box.muse-dev.online/thread/{thread_id}"
         nudge_msg = (
             f"[STRICT ENFORCEMENT: Conversational commentary is rejected. Work requires active execution.]\n"
@@ -1071,18 +1074,34 @@ def maybe_nudge_untagged_sidechat(agent, thread_id, thread_name, mid, text, dry_
         )
         try:
             import muse_hybrid
-        print(f"[{agent}] Injecting 1-turn nudge into {thread_name or thread_id[:8]} for job {matching_job_id}")
+            print(f"[{agent}] Injecting 1-turn strict nudge into {thread_name or thread_id[:8]} for job {matching_job_id}")
             muse_hybrid.send_message(agent, nudge_msg, thread_id=thread_id, wait=0)
             tracker[thread_id] = {
                 "ts": utcnow(),
                 "job_id": matching_job_id,
                 "nudged_for_mid": mid,
-            "nudge_count": rec.get("nudge_count", 0) + 1,
+                "nudge_count": 1,
             }
             save_json_file(NUDGE_TRACKER_FILE, tracker)
         except Exception as ne:
             sys.stderr.write(f"warning: failed to deliver conversational nudge: {ne}\n")
 
+    # 2nd turn: Trigger emergency 5m escalation follow-up to opm/coordinator
+    elif n_count == 1 and not rec.get("escalated"):
+        print(f"[{agent}] Conversational stall persisted in {thread_name or thread_id[:8]} - scheduling emergency escalation to opm")
+        esc_prompt = f"[ESCALATION] Agent {agent} stalled in thread {thread_name or thread_id[:8]} on job {matching_job_id}. Review thread or reassign."
+        try:
+            cmd = [
+                "systemd-run", "--user", "--on-active=300s",
+                sys.executable, str(BIN_DIR / "box-ctl.py"), "notify", "opm", esc_prompt, "--sender", agent
+            ]
+            subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+            tracker[thread_id]["escalated"] = True
+            tracker[thread_id]["escalated_ts"] = utcnow()
+            save_json_file(NUDGE_TRACKER_FILE, tracker)
+        except Exception as ee:
+            sys.stderr.write(f"warning: failed to schedule emergency escalation: {ee}\n")
+
 
 
 # Chain deduplication