feat(turns): watch persisted journal progress

docs/guides/custom-agent-runner-integration.md

@@ -207,6 +207,23 @@ the outer runner applies and acknowledges scheduler state correctly. Those are
 useful extension and contribution surfaces for making Turn more mature; an
 Agent process exit code or scraped transcript is not a substitute for them.
 
+### Watch a running Turn
+
+In a second terminal, inspect the same Turn while `run-once` is active:
+
+```bash
+loopx --registry <registry.json> --runtime-root <runtime-root> turn inspect-journal \
+  --goal-id <goal-id> --agent-id <agent-id> --turn-key <sha256-turn-key> \
+  --watch --watch-interval 1 --format json
+```
+
+Watch mode emits one newline-delimited JSON progress event whenever the journal
+status or completed phase list changes. The event contains only Turn identity,
+journal status, completed phases, and an empty effects list; it never includes
+host output or session content. It stops at `committed`, `stopped`, or `failed`.
+The command is read-only; use Ctrl-C to stop watching early. Start it after the
+Turn journal exists, and use the same registry and runtime root as the runner.
+
 ## Acceptance Checklist
 
 Before calling the integration autonomous, prove that:

docs/guides/custom-agent-runner-integration.zh-CN.md

@@ -183,6 +183,22 @@ effect，并且外层 runner 能正确应用和 ACK scheduler state。这些正
 成熟度的 extension / contribution surface；Agent 进程退出码或从 transcript 猜结果不能
 替代这些证明。
 
+### 观察正在运行的 Turn
+
+在第二个终端中，可以在 `run-once` 执行期间查看同一 Turn：
+
+```bash
+loopx --registry <registry.json> --runtime-root <runtime-root> turn inspect-journal \
+  --goal-id <goal-id> --agent-id <agent-id> --turn-key <sha256-turn-key> \
+  --watch --watch-interval 1 --format json
+```
+
+Watch 模式只在 journal 状态或已完成阶段列表变化时输出一条 JSON Lines 进度事件。
+事件仅包含 Turn 身份、journal 状态、已完成阶段和空 effects 列表，不包含 host 输出或
+session 内容。状态变为 `committed`、`stopped` 或 `failed` 后命令退出。该命令只读；
+提前停止可按 Ctrl-C。请在 Turn journal 创建后启动，并使用与 runner 相同的 registry 和
+runtime root。
+
 ## 验收清单
 
 在把集成称为“自主运行”前，至少证明：

loopx/cli_commands/turn_inspection.py

@@ -2,7 +2,10 @@
 
 import argparse
 from collections.abc import Callable
+import json
+import math
 from pathlib import Path
+import time
 
 from ..control_plane.runtime.status_projection_cache import (
     resolve_status_projection_cache_runtime_root,
@@ -14,6 +17,8 @@
     None,
 ]
 FormatSelector = Callable[..., str]
+TURN_PROGRESS_EVENT_SCHEMA_VERSION = "loopx_turn_progress_event_v0"
+_TERMINAL_TURN_STATUSES = {"committed", "stopped", "failed"}
 
 
 def handle_turn_journal_inspection(
@@ -44,6 +49,99 @@ def handle_turn_journal_inspection(
                 runtime_root, goal_id=args.goal_id, turn_key=args.turn_key,
             )
             scope = str(((plan.get("session") or {}).get("context_policy") or {}).get("binding_scope") or "todo")
+        fmt = output_format(args)
+        if args.watch:
+            interval = args.watch_interval
+            if not math.isfinite(interval) or interval <= 0:
+                payload = {
+                    "ok": False,
+                    "schema_version": LOOPX_TURN_JOURNAL_INSPECTION_SCHEMA_VERSION,
+                    "error": "watch interval must be a finite positive number",
+                    "effects": [],
+                }
+                print_payload(
+                    payload,
+                    fmt,
+                    render_loopx_turn_journal_inspection_markdown,
+                )
+                return 1
+
+            previous: tuple[str, tuple[str, ...]] | None = None
+            try:
+                while True:
+                    payload = inspect_loopx_turn_journal(
+                        runtime_root,
+                        goal_id=args.goal_id,
+                        agent_id=args.agent_id,
+                        turn_key=args.turn_key,
+                        retry_failed=bool(args.retry_failed_turn),
+                        session_binding_resolver=(
+                            lambda turn_envelope: codex_cli_session_binding(
+                                runtime_root,
+                                turn_envelope,
+                                session_scope=scope,
+                            )
+                        ),
+                    )
+                    if payload.get("ok") is not True:
+                        print_payload(
+                            payload,
+                            fmt,
+                            render_loopx_turn_journal_inspection_markdown,
+                        )
+                        return 1
+
+                    if payload.get("journal_consistent") is not True:
+                        print_payload(
+                            payload,
+                            fmt,
+                            render_loopx_turn_journal_inspection_markdown,
+                        )
+                        return 1
+
+                    status = str(payload.get("journal_status") or "")
+                    phases = payload.get("completed_phases")
+                    completed_phases = (
+                        [str(phase) for phase in phases]
+                        if isinstance(phases, list)
+                        else []
+                    )
+                    snapshot = (status, tuple(completed_phases))
+                    if snapshot != previous:
+                        event: dict[str, object] = {
+                            "schema_version": TURN_PROGRESS_EVENT_SCHEMA_VERSION,
+                            "goal_id": args.goal_id,
+                            "agent_id": args.agent_id,
+                            "turn_key": args.turn_key,
+                            "journal_status": status,
+                            "completed_phases": completed_phases,
+                            "effects": [],
+                        }
+                        if fmt == "json":
+                            print(
+                                json.dumps(
+                                    event,
+                                    ensure_ascii=False,
+                                    separators=(",", ":"),
+                                ),
+                                flush=True,
+                            )
+                        else:
+                            from .turn_rendering import (
+                                render_loopx_turn_journal_progress_markdown,
+                            )
+
+                            print(
+                                render_loopx_turn_journal_progress_markdown(event),
+                                flush=True,
+                            )
+                        previous = snapshot
+                    if status in _TERMINAL_TURN_STATUSES:
+                        return 0
+                    time.sleep(interval)
+            except KeyboardInterrupt:
+                return 130
+
         payload = inspect_loopx_turn_journal(
             runtime_root,
             goal_id=args.goal_id,

loopx/cli_commands/turn_registration.py

@@ -48,6 +48,21 @@ def register_turn_commands(
             "Host Session binding check."
         ),
     )
+    inspect_journal.add_argument(
+        "--watch",
+        action="store_true",
+        help=(
+            "Poll the read-only journal projection and emit one progress event "
+            "per changed phase/status until the Turn is terminal. With --format "
+            "json, events are newline-delimited JSON."
+        ),
+    )
+    inspect_journal.add_argument(
+        "--watch-interval",
+        type=float,
+        default=1.0,
+        help="Seconds between journal reads while --watch is active (default: 1).",
+    )
 
     plan = command_sub.add_parser(
         "plan",

loopx/cli_commands/turn_rendering.py

@@ -274,6 +274,26 @@ def render_loopx_turn_journal_inspection_markdown(
     )
 
 
+def render_loopx_turn_journal_progress_markdown(
+    payload: dict[str, object],
+) -> str:
+    """Render one allowlisted progress event from a Turn journal watch."""
+    phases = payload.get("completed_phases")
+    completed = (
+        ", ".join(str(phase) for phase in phases)
+        if isinstance(phases, list) and phases
+        else "none"
+    )
+    return "\n".join(
+        [
+            "# LoopX Turn Progress",
+            f"- journal_status: {payload.get('journal_status')}",
+            f"- completed_phases: {completed}",
+            "- effects: none",
+        ]
+    )
+
+
 def render_loopx_turn_managed_step_markdown(payload: dict[str, object]) -> str:
     if not payload.get("ok"):
         error = payload.get("error") or "Turn managed step failed"