.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",
    "heartbeatAgeSeconds": "<number> | 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 datetime import datetime
from pathlib import Path


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

from launch_worker import (  # noqa: E402
    LaunchError,
    parse_timestamp,
    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
HEARTBEAT_INTERVAL_SECONDS = 5 * 60
HEARTBEAT_MAX_AGE_SECONDS = HEARTBEAT_INTERVAL_SECONDS * 3


def heartbeat_age_seconds(
    heartbeat: object, now: datetime | None = None
) -> tuple[float | None, str | None]:
    """Return heartbeat age and a diagnostic for malformed/stale timestamps."""

    if heartbeat is None:
        return None, None
    try:
        heartbeat_at = parse_timestamp(heartbeat, "last_heartbeat_at")
    except LaunchError:
        return None, "invalid-heartbeat"

    current = now if now is not None else utc_now()
    age = (current - heartbeat_at).total_seconds()
    if age < -60:
        return None, "future-heartbeat"
    return max(age, 0.0), None


def _result(
    *,
    status: str,
    stall_reason: str | None,
    heartbeat: object,
    heartbeat_age: float | None,
    evidence: str,
) -> dict[str, object]:
    return {
        "status": status,
        "stallReason": stall_reason,
        "heartbeatAt": heartbeat,
        "heartbeatAgeSeconds": heartbeat_age,
        "evidence": evidence[:MAX_EVIDENCE_CHARS],
    }


def classify(
    tail: str | list[str],
    heartbeat: object,
    *,
    now: datetime | None = None,
) -> dict[str, object]:
    if isinstance(tail, list):
        tail = "\n".join(tail)
    heartbeat_age, heartbeat_error = heartbeat_age_seconds(heartbeat, now)
    for label, pattern in STALL_PATTERNS:
        if pattern.search(tail):
            return _result(
                status="stall",
                stall_reason=label,
                heartbeat=heartbeat,
                heartbeat_age=heartbeat_age,
                evidence=tail,
            )
    if heartbeat_error:
        return _result(
            status="unknown",
            stall_reason=heartbeat_error,
            heartbeat=heartbeat,
            heartbeat_age=heartbeat_age,
            evidence=tail,
        )
    if heartbeat_age is not None and heartbeat_age <= HEARTBEAT_MAX_AGE_SECONDS:
        return _result(
            status="progress",
            stall_reason=None,
            heartbeat=heartbeat,
            heartbeat_age=heartbeat_age,
            evidence=tail,
        )
    if heartbeat_age is not None:
        return _result(
            status="unknown",
            stall_reason="stale-heartbeat",
            heartbeat=heartbeat,
            heartbeat_age=heartbeat_age,
            evidence=tail,
        )
    if WORKING_PATTERN.search(tail):
        return _result(
            status="running",
            stall_reason=None,
            heartbeat=heartbeat,
            heartbeat_age=heartbeat_age,
            evidence=tail,
        )
    # 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 _result(
            status="not-started",
            stall_reason=None,
            heartbeat=heartbeat,
            heartbeat_age=heartbeat_age,
            evidence=tail,
        )
    return _result(
        status="unknown",
        stall_reason=None,
        heartbeat=heartbeat,
        heartbeat_age=heartbeat_age,
        evidence=tail,
    )


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,
                    "heartbeatAgeSeconds": 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())