.pouch/skills/ack/scripts/worker_probe.py

#!/usr/bin/env python3
"""Probe one dispatched ACK worker's liveness and emit a single JSON status.

Read-only supervision helper for the coordinator's wait loop. It never sends
input, never mutates dispatch or terminal state, and never marks a task
outcome. The coordinator runs it between rolling ``check --wait`` windows to
detect workers that never started, stalled on an approval/choice prompt, hit a
usage limit, or lost heartbeat.

Output (single JSON document on stdout):
  {
    "probedAt": "<RFC3339>",
    "taskId": "<task-id>",
    "dispatchId": "<dispatch-id>",
    "terminal": "<handle>",
    "status": "running | progress | stall | not-started | unknown",
    "stallReason": "<label> | null",
    "heartbeatAt": "<value> | null",
    "evidence": "<bounded terminal tail>"
  }

Exit code is always 0 for a probe attempt: a failed probe is ``unknown`` for
the coordinator to reconcile, never an automatic retry trigger.
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path


sys.path.insert(0, str(Path(__file__).resolve().parent))

from launch_worker import (  # noqa: E402
    LaunchError,
    resolve_executable,
    run_json,
    utc_now,
)

# Conservative stall patterns: an interactive prompt the worker is waiting on.
# Matching only means "evidence of a stall to inspect", never a verdict alone.
STALL_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
    ("approval", re.compile(r"(?i)approv(e|al)|allow tool|permission|批准|允许")),
    ("usage-limit", re.compile(r"(?i)usage limit|rate limit|额度|quota")),
    ("model-switch", re.compile(r"(?i)switch to|keep current model|choose an action|切换")),
    ("press-enter", re.compile(r"(?i)press enter|回车|按回车")),
)
WORKING_PATTERN = re.compile(r"(?i)working|•working|running|执行中|正在")
IDLE_TAIL_PATTERN = re.compile(r"(?i)welcome to|type help|fish, the friendly|>\\s*$")

MAX_EVIDENCE_CHARS = 500


def classify(tail: str | list[str], heartbeat: object) -> dict[str, object]:
    if isinstance(tail, list):
        tail = "\n".join(tail)
    for label, pattern in STALL_PATTERNS:
        if pattern.search(tail):
            return {
                "status": "stall",
                "stallReason": label,
                "heartbeatAt": heartbeat,
                "evidence": tail[:MAX_EVIDENCE_CHARS],
            }
    if heartbeat:
        return {
            "status": "progress",
            "stallReason": None,
            "heartbeatAt": heartbeat,
            "evidence": tail[:MAX_EVIDENCE_CHARS],
        }
    if WORKING_PATTERN.search(tail):
        return {
            "status": "running",
            "stallReason": None,
            "heartbeatAt": None,
            "evidence": tail[:MAX_EVIDENCE_CHARS],
        }
    # No heartbeat and no working marker: the terminal may still be sitting at
    # a welcome/idle prompt (task never started) or have unclassified output.
    if IDLE_TAIL_PATTERN.search(tail) or not tail.strip():
        return {
            "status": "not-started",
            "stallReason": None,
            "heartbeatAt": None,
            "evidence": tail[:MAX_EVIDENCE_CHARS],
        }
    return {
        "status": "unknown",
        "stallReason": None,
        "heartbeatAt": None,
        "evidence": tail[:MAX_EVIDENCE_CHARS],
    }


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--task-id", required=True, help="Orca orchestration task ID")
    parser.add_argument("--terminal", required=True, help="worker terminal handle")
    args = parser.parse_args(argv)

    orca = resolve_executable("orca")
    try:
        show = run_json(
            [str(orca), "orchestration", "dispatch-show", "--task", args.task_id, "--json"],
            "dispatch-show",
        )
        dispatch = show["result"]["dispatch"]
        read_response = run_json(
            [str(orca), "terminal", "read", "--terminal", args.terminal, "--json"],
            "terminal read",
        )
        terminal = read_response["result"]["terminal"]
    except (LaunchError, KeyError, TypeError, IndexError) as exc:
        print(
            json.dumps(
                {
                    "probedAt": utc_now().isoformat().replace("+00:00", "Z"),
                    "taskId": args.task_id,
                    "dispatchId": None,
                    "terminal": args.terminal,
                    "status": "unknown",
                    "stallReason": None,
                    "heartbeatAt": None,
                    "evidence": f"probe failed: {type(exc).__name__}: {exc}",
                },
                ensure_ascii=False,
            )
        )
        return 0

    dispatch_id = dispatch.get("id")
    heartbeat = dispatch.get("last_heartbeat_at")
    tail = "\n".join(terminal.get("tail") or [])
    result = classify(tail, heartbeat)
    print(
        json.dumps(
            {
                "probedAt": utc_now().isoformat().replace("+00:00", "Z"),
                "taskId": args.task_id,
                "dispatchId": dispatch_id,
                "terminal": args.terminal,
                **result,
            },
            ensure_ascii=False,
        )
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())