feat: cherry-pick upstream #85 #101 #102 — runtime injection + waiter fix + Windows compat (#54)

* fix(waiter): complete immediately when team has zero tasks

TaskWaiter.wait() would loop forever when no tasks existed because the
completion condition `total > 0 and completed == total` was never true
for total=0. Remove the `total > 0` guard so the vacuous case (0 == 0)
returns "completed" immediately, matching expected CLI/CI behavior.

Also adds 19 comprehensive tests for TaskWaiter covering zero-task
completion, normal completion, timeout, signal handling, progress
deduplication, message draining, and dead-agent recovery.

Made-with: Cursor

* feat(runtime): inject inbox messages into running tmux sessions (upstream #85)

* fix(tests): remove unused timedelta import

* fix(watcher): send runtime warnings to stderr in json mode

* fix: Windows compat — replace Path.rename() with os.replace() (upstream #102)

Path.rename() fails on Windows when target exists. os.replace() is
cross-platform safe. Applied to 5 files that still used .rename().

Also trimmed test_cli_commands.py to keep only runtime injection +
lifecycle tests that are compatible with our fork.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: xzq.xu <zhiqiang.xu@nodeskai.com>
Co-authored-by: who96 <825265100@qq.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 4 people

clawteam/cli/commands.py

@@ -789,6 +789,130 @@ def inbox_watch(
     watcher.watch()
 
 
+# ============================================================================
+# Runtime Commands
+# ============================================================================
+
+runtime_app = typer.Typer(help="Tmux-only runtime routing and live injection")
+app.add_typer(runtime_app, name="runtime")
+
+
+@runtime_app.command("inject")
+def runtime_inject(
+    team: str = typer.Argument(..., help="Team name"),
+    agent: str = typer.Argument(..., help="Target agent name"),
+    source: str = typer.Option("system", "--source", "-s", help="Runtime notification source"),
+    channel: str = typer.Option("direct", "--channel", help="Runtime notification channel"),
+    priority: str = typer.Option("medium", "--priority", help="Runtime notification priority"),
+    summary: str = typer.Option(..., "--summary", help="Summary text for the injected notification"),
+    evidence: list[str] = typer.Option([], "--evidence", "-e", help="Repeatable evidence line"),
+    recommended_next_action: Optional[str] = typer.Option(
+        None,
+        "--recommended-next-action",
+        help="Optional recommended next action",
+    ),
+):
+    """Inject a structured runtime notification into a running tmux agent."""
+    from clawteam.spawn.tmux_backend import TmuxBackend
+    from clawteam.team.routing_policy import RuntimeEnvelope
+
+    envelope = RuntimeEnvelope(
+        source=source,
+        target=agent,
+        channel=channel,
+        priority=priority,
+        message_type="manual",
+        summary=summary,
+        evidence=list(evidence),
+        recommended_next_action=recommended_next_action,
+    )
+    ok, status = TmuxBackend().inject_runtime_message(team, agent, envelope)
+    if not ok:
+        console.print(f"[red]{status}[/red]")
+        raise typer.Exit(1)
+
+    _output(
+        {"team": team, "agent": agent, "status": status},
+        lambda data: console.print(f"[green]OK[/green] {data['status']}"),
+    )
+
+
+@runtime_app.command("watch")
+def runtime_watch(
+    team: str = typer.Argument(..., help="Team name"),
+    agent: Optional[str] = typer.Option(None, "--agent", "-a", help="Agent name (default: from env)"),
+    poll_interval: float = typer.Option(1.0, "--poll-interval", "-p", help="Poll interval in seconds"),
+    exec_cmd: Optional[str] = typer.Option(
+        None,
+        "--exec",
+        "-e",
+        help="Shell command to run for each new message (msg data in env vars)",
+    ),
+):
+    """Watch an inbox and route new messages into the running tmux session."""
+    from clawteam.identity import AgentIdentity
+    from clawteam.team.mailbox import MailboxManager
+    from clawteam.team.manager import TeamManager
+    from clawteam.team.router import RuntimeRouter
+    from clawteam.team.watcher import InboxWatcher
+
+    identity = AgentIdentity.from_env()
+    agent_name = TeamManager.resolve_inbox(team, agent or identity.agent_name, identity.user)
+    mailbox = MailboxManager(team)
+    router = RuntimeRouter(
+        team_name=team,
+        agent_name=agent_name,
+        session_agent_name=agent or identity.agent_name,
+    )
+
+    if not _json_output:
+        console.print(
+            f"Watching runtime routes for '{agent_name}' in team '{team}'... (Ctrl+C to stop)"
+        )
+        if exec_cmd:
+            console.print(f"  exec: {exec_cmd}")
+
+    watcher = InboxWatcher(
+        team_name=team,
+        agent_name=agent_name,
+        mailbox=mailbox,
+        poll_interval=poll_interval,
+        json_output=_json_output,
+        exec_cmd=exec_cmd,
+        runtime_router=router,
+    )
+    watcher.watch()
+
+
+@runtime_app.command("state")
+def runtime_state(
+    team: str = typer.Argument(..., help="Team name"),
+):
+    """Show persisted Phase 1 runtime throttle and dispatch state."""
+    from clawteam.team.routing_policy import DefaultRoutingPolicy
+
+    state = DefaultRoutingPolicy(team_name=team).read_state()
+
+    def _human(data):
+        console.print(
+            f"Runtime state for '{data['team']}' (throttle={data['throttleSeconds']}s)"
+        )
+        routes = data.get("routes", {})
+        if not routes:
+            console.print("[dim]No runtime route state.[/dim]")
+            return
+        for key in sorted(routes):
+            route = routes[key]
+            console.print(
+                f"  {route.get('source', '?')} -> {route.get('target', '?')} "
+                f"pending={route.get('pendingCount', 0)} "
+                f"status={route.get('lastDispatchStatus', 'idle')} "
+                f"flushAfter={route.get('flushAfter', '') or '-'}"
+            )
+
+    _output(state, _human)
+
+
 # ============================================================================
 # Task Commands
 # ============================================================================

clawteam/spawn/sessions.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import os
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
@@ -67,7 +68,7 @@ def save(
         tmp.write_text(
             session.model_dump_json(indent=2, by_alias=True), encoding="utf-8"
         )
-        tmp.rename(path)
+        os.replace(str(tmp), str(path))
         return session
 
     def load(self, agent_name: str) -> SessionState | None:

clawteam/spawn/tmux_backend.py

@@ -10,6 +10,7 @@
 import sys
 import tempfile
 import time
+from xml.sax.saxutils import escape
 
 from clawteam.spawn.base import SpawnBackend
 from clawteam.spawn.cli_env import build_spawn_path, resolve_clawteam_executable
@@ -289,6 +290,31 @@ def list_running(self) -> list[dict[str, str]]:
             for name, target in self._agents.items()
         ]
 
+    def inject_runtime_message(self, team: str, agent_name: str, envelope) -> tuple[bool, str]:
+        """Best-effort runtime injection into an existing tmux agent pane."""
+        if not shutil.which("tmux"):
+            return False, "tmux not installed"
+
+        target = f"{self.session_name(team)}:{agent_name}"
+        probe = subprocess.run(
+            ["tmux", "list-panes", "-t", target, "-F", "#{pane_id}"],
+            capture_output=True,
+            text=True,
+        )
+        if probe.returncode != 0 or not probe.stdout.strip():
+            return False, f"tmux target '{target}' not found"
+
+        try:
+            _inject_prompt_via_buffer(
+                target,
+                agent_name,
+                _render_runtime_notification(envelope),
+            )
+        except Exception as exc:
+            return False, f"runtime injection failed for '{target}': {exc}"
+
+        return True, f"Injected runtime notification into {target}"
+
     @staticmethod
     def session_name(team_name: str) -> str:
         return f"clawteam-{team_name}"
@@ -665,3 +691,39 @@ def _inject_prompt_via_buffer(
         )
     finally:
         os.unlink(tmp_path)
+
+def _render_runtime_notification(envelope) -> str:
+    summary = str(getattr(envelope, "summary", "") or "").strip()
+    if not summary:
+        summary = "Runtime update"
+
+    evidence = getattr(envelope, "evidence", []) or []
+    if isinstance(evidence, str):
+        evidence = [evidence]
+    evidence_block = "\n".join(str(item) for item in evidence if item)
+
+    lines = [
+        '<clawteam_notification version="1"',
+        f'  source="{escape(str(getattr(envelope, "source", "system") or "system"))}"',
+        f'  target="{escape(str(getattr(envelope, "target", "") or ""))}"',
+        f'  channel="{escape(str(getattr(envelope, "channel", "direct") or "direct"))}"',
+        f'  priority="{escape(str(getattr(envelope, "priority", "medium") or "medium"))}">',
+        "<summary>",
+        escape(summary),
+        "</summary>",
+    ]
+
+    if evidence_block:
+        lines.extend(["<evidence>", escape(evidence_block), "</evidence>"])
+    recommended_next_action = str(getattr(envelope, "recommended_next_action", "") or "").strip()
+    if recommended_next_action:
+        lines.extend(
+            [
+                "<recommended_next_action>",
+                escape(recommended_next_action),
+                "</recommended_next_action>",
+            ]
+        )
+
+    lines.append("</clawteam_notification>")
+    return "\n".join(lines)

clawteam/team/mailbox.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import os
 import time
 import uuid
 
@@ -55,7 +56,7 @@ def _log_event(self, msg: TeamMessage) -> None:
             msg.model_dump_json(indent=2, by_alias=True, exclude_none=True),
             encoding="utf-8",
         )
-        tmp.rename(path)
+        os.replace(str(tmp), str(path))
 
     def get_event_log(self, limit: int = 100) -> list[TeamMessage]:
         """Read event log (newest first). Non-destructive."""

clawteam/team/manager.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import os
 import shutil
 from pathlib import Path
 
@@ -43,7 +44,7 @@ def _save_config(config: TeamConfig) -> None:
     tmp.write_text(
         config.model_dump_json(indent=2, by_alias=True), encoding="utf-8"
     )
-    tmp.rename(path)
+    os.replace(str(tmp), str(path))
 
 
 class TeamManager:

clawteam/team/router.py

@@ -0,0 +1,126 @@
+"""Thin runtime router for tmux live injection."""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime
+
+from clawteam.spawn.tmux_backend import TmuxBackend
+from clawteam.team.models import MessageType, TeamMessage
+from clawteam.team.routing_policy import DefaultRoutingPolicy, RouteDecision, RuntimeEnvelope
+
+
+class RuntimeRouter:
+    """Normalize inbox messages, ask policy for a decision, then dispatch."""
+
+    def __init__(
+        self,
+        team_name: str,
+        agent_name: str,
+        backend: TmuxBackend | None = None,
+        policy: DefaultRoutingPolicy | None = None,
+        session_agent_name: str | None = None,
+    ):
+        self.team_name = team_name
+        self.inbox_agent_name = agent_name
+        self.agent_name = session_agent_name or agent_name
+        self.backend = backend or TmuxBackend()
+        self.policy = policy or DefaultRoutingPolicy(team_name=team_name)
+
+    def normalize_message(self, message: TeamMessage) -> RuntimeEnvelope:
+        source = message.from_agent or "system"
+        # Route to the live tmux pane name when the message does not carry an explicit target.
+        target = message.to or self.agent_name
+        channel = "team" if message.type == MessageType.broadcast else "direct"
+        priority = self._priority_for_message(message)
+        evidence = []
+        if message.summary:
+            evidence.append(f"summary: {message.summary}")
+        if message.plan_file:
+            evidence.append(f"planFile: {message.plan_file}")
+        if message.status:
+            evidence.append(f"status: {message.status}")
+        if message.last_task:
+            evidence.append(f"lastTask: {message.last_task}")
+        if message.reason:
+            evidence.append(f"reason: {message.reason}")
+        if message.feedback:
+            evidence.append(f"feedback: {message.feedback}")
+        if message.request_id:
+            evidence.append(f"requestId: {message.request_id}")
+
+        summary = (message.content or "").strip() or f"{message.type.value} from {source}"
+        payload = json.loads(message.model_dump_json(by_alias=True, exclude_none=True))
+
+        return RuntimeEnvelope(
+            source=source,
+            target=target,
+            channel=channel,
+            priority=priority,
+            message_type=message.type.value,
+            summary=summary,
+            evidence=evidence,
+            recommended_next_action=self._recommended_next_action(message),
+            payload=payload,
+            dedupe_key=message.request_id or f"{source}:{target}:{message.type.value}:{message.timestamp}",
+            created_at=message.timestamp,
+        )
+
+    def route_message(
+        self,
+        message: TeamMessage,
+        *,
+        now: datetime | str | None = None,
+    ) -> RouteDecision:
+        envelope = self.normalize_message(message)
+        decision = self.policy.decide(envelope, now=now)
+        self.dispatch(decision, now=now)
+        return decision
+
+    def flush_due(self, *, now: datetime | str | None = None) -> list[RouteDecision]:
+        decisions = self.policy.flush_due(target_agent=self.agent_name, now=now)
+        for decision in decisions:
+            self.dispatch(decision, now=now)
+        return decisions
+
+    def dispatch(self, decision: RouteDecision, *, now: datetime | str | None = None) -> bool:
+        if decision.action != "inject" or not decision.envelope.requires_injection:
+            return False
+
+        if not hasattr(self.backend, "inject_runtime_message"):
+            self.policy.record_dispatch_result(
+                decision,
+                success=False,
+                now=now,
+                error="backend does not support runtime injection",
+            )
+            return False
+
+        ok, reason = self.backend.inject_runtime_message(
+            self.team_name,
+            decision.envelope.target,
+            decision.envelope,
+        )
+        self.policy.record_dispatch_result(
+            decision,
+            success=ok,
+            now=now,
+            error="" if ok else reason,
+        )
+        return ok
+
+    @staticmethod
+    def _priority_for_message(message: TeamMessage) -> str:
+        if message.type in {MessageType.shutdown_request, MessageType.shutdown_approved, MessageType.shutdown_rejected}:
+            return "high"
+        if message.type in {MessageType.idle, MessageType.plan_approval_request, MessageType.plan_rejected}:
+            return "high"
+        return "medium"
+
+    @staticmethod
+    def _recommended_next_action(message: TeamMessage) -> str | None:
+        if message.type == MessageType.plan_approval_request:
+            return "Review the plan and respond with an approval decision."
+        if message.type == MessageType.idle and message.last_task:
+            return f"Check blocker status for {message.last_task}."
+        return None