mirror of
https://github.com/shareAI-lab/analysis_claude_code.git
synced 2026-08-06 03:53:37 +08:00
Cover completed background tasks in the Lead's inbox poller, and mark when teammates finish (#46)
The inbox poller added in #291 wakes the Lead on teammate inbox messages, but a completed background task still strands its result: nothing wakes the Lead for it, and collect_background_results() runs only inside the tool-use branch, so a plain-text turn never picks it up. There is also no signal once every teammate has finished. Extend the same poller instead of adding a second path. has_pending_background() reports, without consuming, whether any background task has completed. The poller now wakes on unread inbox messages or a completed background task, and the wake handler drains both the inbox and collect_background_results() before running the turn, so background results become new turns without user input, the same way teammate results already do. When the last teammate finishes and its output has been drained, a single "[all teammates done]" marker is printed.
This commit is contained in:
@@ -344,6 +344,13 @@ def collect_background_results() -> list[str]:
|
||||
return notifications
|
||||
|
||||
|
||||
def has_pending_background() -> bool:
|
||||
"""Non-destructive: True if any background task has completed and is
|
||||
waiting to be collected. The inbox poller uses this in its wake condition."""
|
||||
with background_lock:
|
||||
return any(t["status"] == "completed" for t in background_tasks.values())
|
||||
|
||||
|
||||
# ── Cron Scheduler (from s14, synced) ──
|
||||
|
||||
DURABLE_PATH = WORKDIR / ".scheduled_tasks.json"
|
||||
@@ -911,7 +918,8 @@ if __name__ == "__main__":
|
||||
history = []
|
||||
context = update_context({}, [])
|
||||
|
||||
# input() and a 1s inbox poller feed one event queue (issue #291).
|
||||
# input() and a 1s poller (teammate inbox or background results) feed one
|
||||
# event queue (issues #291, #46).
|
||||
events = queue.Queue()
|
||||
|
||||
def input_reader():
|
||||
@@ -924,17 +932,19 @@ if __name__ == "__main__":
|
||||
events.put(("user", line))
|
||||
|
||||
def inbox_poller():
|
||||
# Poll ~1s and submit the Lead's inbox as a new turn. Don't gate on
|
||||
# Poll ~1s and wake the Lead when async results are ready: teammate
|
||||
# inbox messages or completed background tasks. Don't gate on
|
||||
# active_teammates: a teammate sends its result and then removes itself,
|
||||
# so the final message can outlive its registry entry.
|
||||
while True:
|
||||
time.sleep(1)
|
||||
if BUS.peek("lead"):
|
||||
events.put(("inbox", None))
|
||||
if BUS.peek("lead") or has_pending_background():
|
||||
events.put(("wake", None))
|
||||
|
||||
threading.Thread(target=input_reader, daemon=True).start()
|
||||
threading.Thread(target=inbox_poller, daemon=True).start()
|
||||
|
||||
had_teammates = False
|
||||
while True:
|
||||
kind, payload = events.get()
|
||||
if kind == "quit":
|
||||
@@ -943,15 +953,19 @@ if __name__ == "__main__":
|
||||
if payload.strip().lower() in ("q", "exit", ""):
|
||||
break
|
||||
history.append({"role": "user", "content": payload})
|
||||
else: # "inbox": a teammate message woke the Lead
|
||||
else: # "wake": teammate inbox or background results are ready
|
||||
parts = []
|
||||
inbox = BUS.read_inbox("lead")
|
||||
if not inbox:
|
||||
if inbox:
|
||||
parts.append("[Inbox]\n" + "\n".join(
|
||||
f"From {m['from']}: {m['content'][:200]}" for m in inbox))
|
||||
bg = collect_background_results()
|
||||
parts.extend(bg)
|
||||
if not parts:
|
||||
continue # already drained by an earlier wake (idempotent)
|
||||
inbox_text = "\n".join(
|
||||
f"From {m['from']}: {m['content'][:200]}" for m in inbox)
|
||||
history.append({"role": "user",
|
||||
"content": f"[Inbox]\n{inbox_text}"})
|
||||
print(f"\n\033[33m[Inbox: {len(inbox)} messages → new turn]\033[0m")
|
||||
history.append({"role": "user", "content": "\n".join(parts)})
|
||||
print(f"\n\033[33m[wake: {len(inbox)} inbox + {len(bg)} background "
|
||||
f"-> new turn]\033[0m")
|
||||
|
||||
# One turn for whichever source woke us.
|
||||
agent_loop(history, context)
|
||||
@@ -959,4 +973,11 @@ if __name__ == "__main__":
|
||||
for block in history[-1]["content"]:
|
||||
if getattr(block, "type", None) == "text":
|
||||
print(block.text)
|
||||
|
||||
# Announce once when every teammate has finished and its output drained.
|
||||
if active_teammates:
|
||||
had_teammates = True
|
||||
elif had_teammates and not BUS.peek("lead") and not has_pending_background():
|
||||
print("\033[32m[all teammates done]\033[0m")
|
||||
had_teammates = False
|
||||
print()
|
||||
|
||||
Reference in New Issue
Block a user