feat(verify): independent verification pass and escalation ladder

src/crossagent/advisors.py

@@ -46,6 +46,13 @@ class Advisor:
     result_parser: str = "text"
     resume_command: tuple[str, ...] | None = None
     session_event_field: str | None = None
+    # Flag that requests a machine-checkable JSON output contract for the
+    # independent verification pass (slice S5). ``None`` means the advisor has no
+    # such contract, so a verifier built on it degrades to parsing a JSON verdict
+    # out of the answer text (D4 graceful degradation — never a hard failure).
+    # Claude exposes ``--json-schema`` (research finding [6]: the payload lands in
+    # ``structured_output``); no other built-in advisor has a verified equivalent.
+    json_schema_flag: str | None = None
     experimental: bool = False
     notes: str = ""
 
@@ -84,6 +91,7 @@ def supports_stream(self) -> bool:
         session_name_flag="--name",
         fork_flag="--fork-session",
         result_parser="claude-stream",
+        json_schema_flag="--json-schema",
     ),
     "codex": Advisor(
         name="codex",

src/crossagent/cli.py

@@ -387,6 +387,37 @@ def _parse_job_args(subcommand: str, argv: list[str]) -> argparse.Namespace:
                 "vars are withheld from the delegate."
             ),
         )
+        parser.add_argument(
+            "--verify-with",
+            dest="verify_with",
+            help=(
+                "Advisor that independently verifies the delegate's work in a "
+                "FRESH peer session, with the diff/answer supplied as user-turn "
+                "input (removes the implicit-authorship channel that weakens "
+                "self-grading). A failing verdict blocks the green path; a "
+                "prose-only or errored verifier degrades to unverified, never a "
+                "pass. Omit to leave verification off."
+            ),
+        )
+        parser.add_argument(
+            "--verify-model",
+            dest="verify_model",
+            help="Model/alias for the --verify-with advisor (advisor default if omitted).",
+        )
+        parser.add_argument(
+            "--escalate-to",
+            action="append",
+            default=None,
+            dest="escalate_to",
+            metavar="ADVISOR[:MODEL]",
+            help=(
+                "Re-dispatch a FAILED delegation (failing check, scope violation, "
+                "or failing verification) to this larger peer as a same-trace "
+                "child job. Repeatable to form an escalation ladder; each rung is "
+                "tried in turn as the previous fails. Bounded by the nesting-depth "
+                "cap. Omit to leave escalation off."
+            ),
+        )
         parser.add_argument("--json", action="store_true")
         parser.set_defaults(stream=True)
     elif subcommand == "wait":
@@ -657,23 +688,38 @@ def _cmd_result(args: argparse.Namespace) -> int:
 
 def _print_verdict(job: jobs_mod.Job, verdict: str, *, file: Any = sys.stderr) -> None:
     """Print a one-line delegation verdict to *file* (stderr by default)."""
-    check = job.check_result
     if verdict == "verified":
-        print("[crossagent] delegation verified — check passed", file=file)
+        print("[crossagent] delegation verified — all declared gates passed", file=file)
     elif verdict == "unverified":
         print(
-            "[crossagent] delegation UNVERIFIED — no --check ran; the delegate "
-            "finished but its work was not checked",
+            "[crossagent] delegation UNVERIFIED — the delegate finished but no "
+            "gate confirmed its work (no --check/--verify-with, or an "
+            "inconclusive verifier)",
             file=file,
         )
     elif verdict == "failed":
-        exit_code = check.get("exit_code") if check else None
         print(
-            f"[crossagent] delegation FAILED verification — check exited {exit_code}",
+            f"[crossagent] delegation FAILED verification — {_failed_reason(job)}",
             file=file,
         )
 
 
+def _failed_reason(job: jobs_mod.Job) -> str:
+    """Describe why a delegation failed, naming the actual failing gate."""
+    if job.status != jobs_mod.JobState.SUCCEEDED:
+        return f"delegate did not finish cleanly (status {job.status.value})"
+    scope = job.scope_result
+    if scope is not None and scope.get("status") != "ok":
+        return f"scope {scope.get('status')} ({len(scope.get('violating_paths') or [])} path(s))"
+    check = job.check_result
+    if check is not None and check.get("exit_code") != 0:
+        return f"check exited {check.get('exit_code')}"
+    verify = job.verify_result
+    if verify is not None and verify.get("verdict") == "fail":
+        return f"independent verification by {verify.get('advisor')} returned fail"
+    return "a declared gate did not pass"
+
+
 def _metrics_summary(job: jobs_mod.Job) -> str:
     """Return a one-line advisor-metrics summary, or ``""`` when nothing was
     measured. Cost/token/duration go to stderr so piped stdout stays the result.
@@ -899,6 +945,11 @@ def _write_command_info(
         # --allow-path was given (enforcement off), distinct from an empty list.
         "scope_paths": getattr(args, "allow_path", None),
         "pass_env": getattr(args, "pass_env", []),
+        # Independent verification + escalation ladder (S5). ``verify_with`` is
+        # ``None`` when off; ``escalate_to`` is the (possibly empty) rung list.
+        "verify_with": getattr(args, "verify_with", None),
+        "verify_model": getattr(args, "verify_model", None),
+        "escalate_to": getattr(args, "escalate_to", None) or [],
     }
     jobs_mod.atomic_json_write(info, job_dir / "command.json")
 

src/crossagent/escalate.py

@@ -0,0 +1,280 @@
+"""Escalate-on-failure ladder for delegated work (slice S5).
+
+When a delegation fails — a failing check, a scope violation, or a failing
+independent verification, all of which resolve to
+:func:`~crossagent.jobs.delegation_verdict` == ``"failed"`` — the caller may
+declare an *escalation ladder*: an ordered list of larger peers to re-dispatch
+the same task to. On failure the first rung is spawned; the remaining rungs are
+handed to that child so a further failure climbs the next rung.
+
+Recording (option (a) — satisfies the shipped analytics definition)
+-------------------------------------------------------------------
+``analytics.py`` (merged before this slice) computes the escalation rate as
+"a failed delegation counts as escalated if it has a **same-trace child**". So a
+re-dispatch is recorded as exactly that: a child job with the failed job as its
+``parent_job_id`` and the **same ``trace_id``**, resolved through the existing
+:func:`~crossagent.jobs.resolve_lineage` machinery. Nothing in ``analytics.py``
+changes; the escalation column starts reflecting reality the moment this ships.
+
+Runaway protection
+------------------
+An escalation ladder is a recursion source. It is bounded twice over:
+
+1. The ladder is a finite list that shrinks by one rung each hop, so it
+   self-terminates even if every rung fails.
+2. Each child is one level deeper, and lineage resolution enforces
+   :data:`~crossagent.jobs.MAX_NESTING_DEPTH`; a rung that would exceed the cap
+   raises :class:`~crossagent.jobs.LineageError`, which is caught and recorded as
+   a skipped escalation rather than crashing or looping.
+
+Security posture carried to the child
+-------------------------------------
+An escalated child is a delegate too. It goes through the ordinary worker path,
+so credential scrubbing (``credentials.py``) and the diff-scope assertion apply
+to it unchanged: the declared ``scope_paths`` allowlist and ``pass_env`` opt-ins
+are propagated so the larger peer is held to the *same* write boundary and the
+*same* credential withholding as the original delegate. The check and
+verification gates are propagated too, so each rung's output is judged the same
+way before the ladder climbs again.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Callable, Optional
+
+from . import advisors as advisors_mod
+from . import jobs as jobs_mod
+from .advisors import Advisor
+from .jobs import Job, delegation_verdict
+
+# Launches a detached worker for a child job. Injectable so escalation can be
+# unit-tested without spawning a real process.
+Launcher = Callable[[str, Path], Any]
+
+
+def parse_rungs(rungs: Optional[list[str]]) -> list[tuple[str, Optional[str]]]:
+    """Parse ``advisor[:model]`` ladder rungs into ``(advisor, model)`` pairs.
+
+    Splits on the first colon only, so a model containing no colon (the usual
+    case: ``opus``, ``gpt-5.6-sol``) is preserved intact. Blank entries are
+    dropped so a stray empty ``--escalate-to`` cannot spawn a nameless job.
+    """
+    parsed: list[tuple[str, Optional[str]]] = []
+    for raw in rungs or []:
+        entry = raw.strip()
+        if not entry:
+            continue
+        advisor_name, sep, model = entry.partition(":")
+        advisor_name = advisor_name.strip()
+        if not advisor_name:
+            continue
+        parsed.append((advisor_name, model.strip() if sep else None))
+    return parsed
+
+
+def _build_child_argv(advisor: Advisor, model: Optional[str]) -> list[str]:
+    """Build the escalated child's advisor argv — a FRESH delegation.
+
+    Mirrors the advisor-invocation core of ``crossagent start`` (default stream
+    mode) but adds no session-attachment flag: an escalation re-dispatches the
+    task to a new, larger peer, so there is no prior session to resume. The
+    child does not inherit the parent's fine-grained invocation flags
+    (``--tools``, ``--safe-mode``, ``--permission-mode``); its write boundary is
+    enforced structurally by the propagated scope allowlist instead.
+    """
+    cmd = [advisor.executable, *advisor.base_args, *advisor.invoke_args]
+    if model and advisor.model_flag:
+        cmd.extend([advisor.model_flag, model])
+    if advisor.supports_stream:
+        cmd.extend(advisor.stream_args)
+    return cmd
+
+
+def maybe_escalate(
+    failed_job: Job,
+    *,
+    prompt: str,
+    state_root: Path,
+    job_dir: Path,
+    cwd: str,
+    registry_path: str,
+    escalate_to: Optional[list[str]],
+    check: Optional[str],
+    check_timeout: float,
+    scope_paths: Optional[list[str]],
+    pass_env: list[str],
+    verify_with: Optional[str],
+    verify_model: Optional[str],
+    launcher: Optional[Launcher] = None,
+) -> Optional[str]:
+    """Re-dispatch a *failed* delegation to the next ladder rung, if any.
+
+    Returns the spawned child's job id, or ``None`` when nothing was escalated
+    (the delegation did not fail, no rungs remain, the rung advisor is unknown,
+    or the depth cap was reached). Never raises: an escalation that cannot be
+    launched is recorded and skipped, never allowed to crash the worker.
+    """
+    if delegation_verdict(failed_job) != "failed":
+        return None
+
+    rungs = parse_rungs(escalate_to)
+    if not rungs:
+        return None
+    advisor_name, model = rungs[0]
+    # ``remaining`` keeps the raw ``advisor[:model]`` strings for the child, minus
+    # the rung being spawned now — the ladder that a further failure will climb.
+    remaining = _drop_first_nonblank(escalate_to)
+
+    try:
+        advisor = advisors_mod.resolve(advisor_name)
+    except KeyError as exc:
+        _audit_skip(job_dir, reason=f"unknown escalation advisor: {exc}")
+        return None
+
+    child_id = jobs_mod.generate_job_id()
+    try:
+        parent_id, trace_id, label, depth = jobs_mod.resolve_lineage(
+            parent_flag=failed_job.job_id,
+            state_root=state_root,
+            new_job_id=child_id,
+        )
+    except jobs_mod.LineageError as exc:
+        # Depth cap reached (or a corrupt chain): the ladder stops here. This is
+        # the runaway guard doing its job, not an error.
+        _audit_skip(job_dir, reason=f"escalation halted: {exc}")
+        return None
+
+    child_dir = jobs_mod.create_job_dir(state_root, child_id)
+    _write_child_prompt(child_dir, prompt)
+    _write_child_command(
+        child_dir,
+        advisor=advisor,
+        model=model,
+        cwd=cwd,
+        registry_path=registry_path,
+        check=check,
+        check_timeout=check_timeout,
+        scope_paths=scope_paths,
+        pass_env=pass_env,
+        verify_with=verify_with,
+        verify_model=verify_model,
+        escalate_to=remaining,
+    )
+    child = Job(
+        job_id=child_id,
+        status=jobs_mod.JobState.PENDING,
+        advisor=advisor.name,
+        name="",
+        cwd=cwd,
+        redacted_command="",
+        started_at=_now(),
+        updated_at=_now(),
+        last_activity_at=_now(),
+        last_event="escalation.created",
+        max_runtime_seconds=failed_job.max_runtime_seconds,
+        termination_grace_seconds=failed_job.termination_grace_seconds,
+        parent_job_id=parent_id,
+        trace_id=trace_id,
+        orchestrator_label=label,
+        nesting_depth=depth,
+    )
+    jobs_mod.save_state(child_dir, child)
+
+    jobs_mod.append_event(
+        job_dir,
+        "escalation",
+        actor="system:escalate",
+        child_job_id=child_id,
+        advisor=advisor.name,
+        model=model,
+        trace_id=trace_id,
+        depth=depth,
+    )
+
+    launch = launcher if launcher is not None else _default_launcher
+    try:
+        launch(child_id, state_root)
+    except OSError as exc:
+        _audit_skip(job_dir, reason=f"escalation worker failed to launch: {exc}")
+        return None
+    return child_id
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _drop_first_nonblank(rungs: Optional[list[str]]) -> list[str]:
+    """Return the non-blank rungs with the first one (the spawned rung) removed."""
+    cleaned = [raw for raw in (rungs or []) if raw.strip()]
+    return cleaned[1:]
+
+
+def _write_child_prompt(child_dir: Path, prompt: str) -> None:
+    prompt_path = child_dir / "prompt"
+    prompt_path.write_text(prompt, encoding="utf-8")
+    try:
+        prompt_path.chmod(0o600)
+    except OSError:
+        pass
+
+
+def _write_child_command(
+    child_dir: Path,
+    *,
+    advisor: Advisor,
+    model: Optional[str],
+    cwd: str,
+    registry_path: str,
+    check: Optional[str],
+    check_timeout: float,
+    scope_paths: Optional[list[str]],
+    pass_env: list[str],
+    verify_with: Optional[str],
+    verify_model: Optional[str],
+    escalate_to: list[str],
+) -> None:
+    info = {
+        "command": _build_child_argv(advisor, model),
+        "prompt_delivery": advisor.prompt_delivery,
+        "cwd": cwd,
+        "result_parser": advisor.result_parser,
+        "registry_path": registry_path,
+        "key": "",
+        "name": None,
+        "model": model or "",
+        "advisor": advisor.name,
+        "check": check,
+        "check_timeout": check_timeout,
+        # Security posture propagated to the larger peer (same write boundary,
+        # same credential withholding) — see the module docstring.
+        "scope_paths": scope_paths,
+        "pass_env": pass_env,
+        # The verification gate is re-run on the escalated output, and the
+        # remaining ladder lets a further failure climb the next rung.
+        "verify_with": verify_with,
+        "verify_model": verify_model,
+        "escalate_to": escalate_to,
+    }
+    jobs_mod.atomic_json_write(info, child_dir / "command.json")
+
+
+def _audit_skip(job_dir: Path, *, reason: str) -> None:
+    jobs_mod.append_event(
+        job_dir, "escalation_skipped", actor="system:escalate", reason=reason
+    )
+
+
+def _default_launcher(child_id: str, state_root: Path) -> Any:
+    # Lazy import breaks the worker <-> escalate import cycle.
+    from .worker import start_worker
+
+    return start_worker(child_id, state_root)
+
+
+def _now() -> str:
+    from datetime import datetime, timezone
+
+    return datetime.now(timezone.utc).isoformat()