diff --git a/bin/fm-azure-runner-guest.sh b/bin/fm-azure-runner-guest.sh index 010000e82f4..fa98a1e5313 100755 --- a/bin/fm-azure-runner-guest.sh +++ b/bin/fm-azure-runner-guest.sh @@ -56,13 +56,102 @@ run_bootstrap_network() { return "$status" } +# BEGIN FM_AZURE_RUNNER_APT_LOCK_HELPERS +wait_for_apt_locks() { + local timeout=$1 poll=$2 + shift 2 + python3 - "$timeout" "$poll" "$@" <<'PY' +import errno +import fcntl +import os +import pathlib +import sys +import time + +timeout = float(sys.argv[1]) +poll = float(sys.argv[2]) +paths = [pathlib.Path(value) for value in sys.argv[3:]] +if not 0 <= timeout <= 600 or not 0.01 <= poll <= 5 or not paths: + raise SystemExit("guest bootstrap: unsafe apt/dpkg lock wait configuration") +deadline = time.monotonic() + timeout +while True: + descriptors = [] + held = [] + try: + for path in paths: + try: + descriptor = os.open(path, os.O_RDWR | os.O_CREAT, 0o644) + except FileNotFoundError: + continue + descriptors.append(descriptor) + try: + fcntl.lockf(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as exc: + if exc.errno not in (errno.EACCES, errno.EAGAIN): + raise + held.append(str(path)) + if not held: + raise SystemExit(0) + finally: + for descriptor in descriptors: + os.close(descriptor) + remaining = deadline - time.monotonic() + if remaining <= 0: + print( + "guest bootstrap: timed out waiting for apt/dpkg lock(s): " + + ", ".join(held), + file=sys.stderr, + ) + raise SystemExit(1) + time.sleep(min(poll, remaining)) +PY +} + +APT_LOCK_WAIT_SECONDS=180 +APT_LOCK_PATHS=( + /var/lib/dpkg/lock-frontend + /var/lib/dpkg/lock + /var/cache/apt/archives/lock + /var/lib/apt/lists/lock +) +run_bootstrap_apt() { + local deadline=$((SECONDS + APT_LOCK_WAIT_SECONDS)) remaining stderr_file status + stderr_file=$(mktemp) + while true; do + remaining=$((deadline - SECONDS)) + if [ "$remaining" -le 0 ]; then + echo "guest bootstrap: timed out retrying apt/dpkg lock contention" >&2 + rm -f "$stderr_file" + return 1 + fi + wait_for_apt_locks "$remaining" 0.2 "${APT_LOCK_PATHS[@]}" || { + rm -f "$stderr_file" + return 1 + } + : >"$stderr_file" + if run_bootstrap_network apt-get -o "DPkg::Lock::Timeout=$remaining" "$@" 2>"$stderr_file"; then + cat "$stderr_file" >&2 + rm -f "$stderr_file" + return 0 + else + status=$? + fi + cat "$stderr_file" >&2 + if ! grep -Eq 'Could not (get|open) lock |Unable to acquire (the )?(dpkg frontend|download directory|lists directory|archive) lock|is another process using it' "$stderr_file"; then + rm -f "$stderr_file" + return "$status" + fi + done +} +# END FM_AZURE_RUNNER_APT_LOCK_HELPERS + missing=() for tool in mkfs.ext4 mount runuser groupadd useradd getent tmux node npm xz; do command -v "$tool" >/dev/null 2>&1 || missing+=("$tool"); done if [ "${#missing[@]}" -gt 0 ]; then export DEBIAN_FRONTEND=noninteractive bootstrap_packages() { - run_bootstrap_network apt-get update -qq && - run_bootstrap_network apt-get install -y --no-install-recommends ca-certificates curl git python3 python3-venv e2fsprogs util-linux passwd systemd tmux jq nodejs npm xz-utils ripgrep + run_bootstrap_apt update -qq && + run_bootstrap_apt install -y --no-install-recommends ca-certificates curl git python3 python3-venv e2fsprogs util-linux passwd systemd tmux jq nodejs npm xz-utils ripgrep } if ! bootstrap_packages; then # The regional azure.archive mirror rides plain port 80, whose egress can diff --git a/bin/fm-crosscheck-pi-reviewer.py b/bin/fm-crosscheck-pi-reviewer.py index f3355a84d71..5b13acf95c5 100755 --- a/bin/fm-crosscheck-pi-reviewer.py +++ b/bin/fm-crosscheck-pi-reviewer.py @@ -16,6 +16,14 @@ class ReviewError(RuntimeError): """A fail-closed Pi launch or verdict-protocol failure.""" +class VerdictProtocolError(ReviewError): + """A repairable final-verdict shape failure with its incurred telemetry.""" + + def __init__(self, message: str, telemetry: dict[str, Any]) -> None: + super().__init__(message) + self.telemetry = telemetry + + def session_id(model: str, extension: Path) -> str: extension_digest = hashlib.sha256(extension.read_bytes()).hexdigest() seed = f"{model}\n{extension_digest}\n".encode() @@ -54,6 +62,89 @@ def recover_single_object(value: str) -> dict[str, Any]: return recovered +def usage_telemetry( + tokens: dict[str, int], + *, + tokens_complete: bool, + pi_cost: float, + cost_complete: bool, + turns: int, +) -> dict[str, Any]: + rates = {"input": 1.40, "cache_read": 0.14, "cache_write": 1.40, "output": 4.40} + declared = ( + sum(tokens[name] * rates[name] / 1_000_000 for name in rates) + if tokens_complete + else None + ) + return { + "tokens": { + **(tokens if tokens_complete else dict.fromkeys(tokens)), + "source": "pi-turn-end-message-usage" if tokens_complete else "unavailable", + }, + "costs_usd": { + "provider_reported": None, + "provider_reported_source": "unavailable-in-pi-events", + "pi_calculated": round(pi_cost, 12) if cost_complete else None, + "pi_calculated_source": ( + "pi-turn-end-message-usage-cost-total" if cost_complete else "unavailable" + ), + "declared": round(declared, 12) if declared is not None else None, + "declared_source": ( + "pinned-fireworks-regular-rates" if declared is not None else "unavailable" + ), + }, + "turns": turns, + } + + +def merge_telemetry(attempts: list[dict[str, Any]]) -> dict[str, Any]: + """Add the rejected initial attempt to the admitted repair attempt's spend.""" + + token_names = ("input", "output", "cache_read", "cache_write") + token_rows = [attempt["tokens"] for attempt in attempts] + tokens_complete = all( + row.get("source") == "pi-turn-end-message-usage" + and all( + isinstance(row.get(name), int) + and not isinstance(row.get(name), bool) + and row[name] >= 0 + for name in token_names + ) + for row in token_rows + ) + tokens = { + name: sum(row[name] for row in token_rows) if tokens_complete else None + for name in token_names + } + + costs: dict[str, Any] = { + "provider_reported": None, + "provider_reported_source": "unavailable-in-pi-events", + } + for name, source, source_name in ( + ("pi_calculated", "pi-turn-end-message-usage-cost-total", "pi_calculated_source"), + ("declared", "pinned-fireworks-regular-rates", "declared_source"), + ): + values = [attempt["costs_usd"].get(name) for attempt in attempts] + complete = all( + isinstance(value, (int, float)) + and not isinstance(value, bool) + and value >= 0 + for value in values + ) + costs[name] = round(sum(values), 12) if complete else None + costs[source_name] = source if complete else "unavailable" + + return { + "tokens": { + **tokens, + "source": "pi-turn-end-message-usage" if tokens_complete else "unavailable", + }, + "costs_usd": costs, + "turns": sum(attempt["turns"] for attempt in attempts), + } + + def parse_events(source: Path, expected_provider: str, expected_model: str) -> dict[str, Any]: calls: dict[str, Any] = {} turns = 0 @@ -66,6 +157,7 @@ def parse_events(source: Path, expected_provider: str, expected_model: str) -> d pi_cost = 0.0 tokens_complete = True cost_complete = True + verdict_protocol_error: str | None = None for line_number, line in enumerate(source.read_text(encoding="utf-8").splitlines(), start=1): if not line.strip(): @@ -123,7 +215,10 @@ def parse_events(source: Path, expected_provider: str, expected_model: str) -> d continue call_id = part.get("id") if not isinstance(call_id, str) or not call_id or call_id in calls: - raise ReviewError("model guest: Pi verdict tool call id is invalid or duplicated") + verdict_protocol_error = ( + "model guest: Pi verdict tool call id is invalid or duplicated" + ) + continue calls[call_id] = part.get("arguments") elif event.get("type") == "agent_end": if agent_ended: @@ -140,47 +235,43 @@ def parse_events(source: Path, expected_provider: str, expected_model: str) -> d final_provider = None final_model = None calls.clear() + verdict_protocol_error = None if not agent_ended or turns < 1 or attempt_turns < 1: raise ReviewError("model guest: Pi did not complete a reviewer turn") - if final_stop != "toolUse": - raise ReviewError(f"model guest: Pi final stopReason was {final_stop!r}, not 'toolUse'") if final_provider != expected_provider or final_model != expected_model: raise ReviewError("model guest: Pi final provider/model identity mismatch") + telemetry = usage_telemetry( + tokens, + tokens_complete=tokens_complete, + pi_cost=pi_cost, + cost_complete=cost_complete, + turns=turns, + ) + if final_stop != "toolUse": + message = f"model guest: Pi final stopReason was {final_stop!r}, not 'toolUse'" + if not calls: + raise VerdictProtocolError(message, telemetry) + raise ReviewError(message) + if verdict_protocol_error is not None: + raise VerdictProtocolError(verdict_protocol_error, telemetry) if len(calls) != 1: - raise ReviewError("model guest: Pi must submit exactly one verdict tool call") + raise VerdictProtocolError( + "model guest: Pi must submit exactly one verdict tool call", telemetry + ) value = next(iter(calls.values())) if isinstance(value, str): - value = recover_single_object(value) + try: + value = recover_single_object(value) + except ReviewError as exc: + raise VerdictProtocolError(str(exc), telemetry) from exc if not isinstance(value, dict) or not isinstance(value.get("verdict"), dict): - raise ReviewError("model guest: reviewer omitted its verdict") + raise VerdictProtocolError("model guest: reviewer omitted its verdict", telemetry) if not isinstance(value.get("evidence_files"), list): - raise ReviewError("model guest: reviewer omitted its evidence manifest") - rates = {"input": 1.40, "cache_read": 0.14, "cache_write": 1.40, "output": 4.40} - declared = ( - sum(tokens[name] * rates[name] / 1_000_000 for name in rates) - if tokens_complete - else None - ) - value["telemetry"] = { - "tokens": { - **(tokens if tokens_complete else dict.fromkeys(tokens)), - "source": "pi-turn-end-message-usage" if tokens_complete else "unavailable", - }, - "costs_usd": { - "provider_reported": None, - "provider_reported_source": "unavailable-in-pi-events", - "pi_calculated": round(pi_cost, 12) if cost_complete else None, - "pi_calculated_source": ( - "pi-turn-end-message-usage-cost-total" if cost_complete else "unavailable" - ), - "declared": round(declared, 12) if declared is not None else None, - "declared_source": ( - "pinned-fireworks-regular-rates" if declared is not None else "unavailable" - ), - }, - "turns": turns, - } + raise VerdictProtocolError( + "model guest: reviewer omitted its evidence manifest", telemetry + ) + value["telemetry"] = telemetry return value @@ -192,11 +283,7 @@ def run(argv: list[str]) -> int: prompt = Path(prompt_raw) schema = Path(schema_raw) result = Path(result_raw) - events = result.with_name("pi-events.jsonl") - stderr_path = result.with_name("pi.stderr") result.unlink(missing_ok=True) - events.unlink(missing_ok=True) - stderr_path.unlink(missing_ok=True) environment = dict(os.environ) environment["PI_CODING_AGENT_DIR"] = account environment["FM_CROSSCHECK_REVIEW_SCHEMA"] = str(schema) @@ -206,53 +293,82 @@ def run(argv: list[str]) -> int: "the enabled tools and submit the complete final verdict exactly once " "with submit_crosscheck_verdict." ) - command = [ - "pi", - "--mode", - "json", - "--offline", - "--provider", - provider, - "--model", - model, - "--thinking", - effort, - "--tools", - "submit_crosscheck_verdict", - "--extension", - str(extension), - "--system-prompt", - system_prompt, - "--session-id", - session_id(model, extension), - "--no-session", - "--no-extensions", - "--no-skills", - "--no-prompt-templates", - "--no-themes", - "--no-context-files", - "--no-approve", - f"@{prompt}", - ] - with events.open("wb") as stdout_file, stderr_path.open("wb") as stderr_file: - completed = subprocess.run( - command, - check=False, - env=environment, - stdin=subprocess.DEVNULL, - stdout=stdout_file, - stderr=stderr_file, + repair_prompt = result.with_name("repair-prompt.txt") + attempt_telemetry: list[dict[str, Any]] = [] + for attempt in range(2): + active_prompt = prompt if attempt == 0 else repair_prompt + events = result.with_name(f"pi-events-{attempt + 1}.jsonl") + stderr_path = result.with_name(f"pi-{attempt + 1}.stderr") + events.unlink(missing_ok=True) + stderr_path.unlink(missing_ok=True) + command = [ + "pi", + "--mode", + "json", + "--offline", + "--provider", + provider, + "--model", + model, + "--thinking", + effort, + "--tools", + "submit_crosscheck_verdict", + "--extension", + str(extension), + "--system-prompt", + system_prompt, + "--session-id", + session_id(model, extension), + "--no-session", + "--no-extensions", + "--no-skills", + "--no-prompt-templates", + "--no-themes", + "--no-context-files", + "--no-approve", + f"@{active_prompt}", + ] + with events.open("wb") as stdout_file, stderr_path.open("wb") as stderr_file: + completed = subprocess.run( + command, + check=False, + env=environment, + stdin=subprocess.DEVNULL, + stdout=stdout_file, + stderr=stderr_file, + ) + if completed.returncode != 0: + sys.stderr.buffer.write(stderr_path.read_bytes()[:1024]) + return 125 + try: + value = parse_events(events, provider, model) + except VerdictProtocolError as exc: + attempt_telemetry.append(exc.telemetry) + if attempt == 1: + raise ReviewError( + f"{exc}; one bounded verdict repair was exhausted" + ) from exc + repair_prompt.write_text( + prompt.read_text(encoding="utf-8") + + "\n\nVERDICT PROTOCOL REPAIR (trusted controller instruction):\n" + + "The preceding isolated attempt did not produce exactly one valid " + + "submit_crosscheck_verdict call. Re-evaluate this same exact-head " + + "packet. Do not end with prose and do not call the tool more than " + + "once. Submit the complete schema-valid verdict through " + + "submit_crosscheck_verdict exactly once.\n", + encoding="utf-8", + ) + continue + attempt_telemetry.append(value["telemetry"]) + value["telemetry"] = merge_telemetry(attempt_telemetry) + result.write_text( + json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n", + encoding="utf-8", ) - if completed.returncode != 0: - sys.stderr.buffer.write(stderr_path.read_bytes()[:1024]) - return 125 - value = parse_events(events, provider, model) - result.write_text( - json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n", - encoding="utf-8", - ) - stderr_path.unlink(missing_ok=True) - return 0 + stderr_path.unlink(missing_ok=True) + return 0 + raise ReviewError("model guest: Pi verdict repair loop ended without a result") def main() -> int: diff --git a/bin/fm-crosscheck.py b/bin/fm-crosscheck.py index 04459a58f22..129a0426185 100755 --- a/bin/fm-crosscheck.py +++ b/bin/fm-crosscheck.py @@ -3165,20 +3165,24 @@ def run_is_compartment_lane(run: dict[str, Any]) -> bool: def stamp_durations(run: dict[str, Any], measured: dict[str, int]) -> None: - """Record this run's measurement, or drop it rather than brick the ledger. + """Record this run's compatible measurement, or drop an invalid one. Everything that later reads this ledger validates it, so an unvalidated write is a durable outage waiting to happen: one writer bug and `run`, `verify` and `timings` all refuse the task until a human edits the JSON by - hand. The measurement is the disposable half of that trade. A timing bug - should cost the operator a breakdown, never the durable findings, so a - measurement that fails its own contract is dropped loudly and the record - is written without it. + hand. A failed compartment attempt has no completed Azure identity to bind + its lane-only phases, so those incompatible fields are omitted while its + total and ordinary phases remain useful. Any other contract failure still + drops the measurement loudly and never affects the durable findings. """ + candidate = dict(measured) + if not run_is_compartment_lane(run): + for phase in CROSSCHECK_COMPARTMENT_PHASES: + candidate.pop(phase, None) try: validate_durations( - measured, + candidate, "recorded durations_ms", compartment=run_is_compartment_lane(run), ) @@ -3191,7 +3195,7 @@ def stamp_durations(run: dict[str, Any], measured: dict[str, int]) -> None: file=sys.stderr, ) return - run["durations_ms"] = measured + run["durations_ms"] = candidate def validate_ledger(value: Any, task_id: str, url: str) -> dict[str, Any]: diff --git a/docs/azure-crosscheck.md b/docs/azure-crosscheck.md index 1f1b0065573..6ff3047ead9 100644 --- a/docs/azure-crosscheck.md +++ b/docs/azure-crosscheck.md @@ -102,7 +102,7 @@ The final terminal event must report the exact `fireworks-glm` provider and `acc Reporting the historical Fast selector or another route fails before a verdict can publish. Pi's explicit `auto_retry_start` may open a continuation only after a completed attempt executed a turn and did not stop successfully. The continuation resets attempt-local terminal and verdict state, preserves aggregate usage for economics, and must execute its own turn before completing. -Zero, multiple, malformed, or truncated verdict submissions fail closed without a second model invocation. +The bounded verdict-repair contract owned by [`docs/crosscheck.md`](crosscheck.md) applies unchanged inside the isolated model compartment. The prompt is passed by `@file`, Pi starts with `--offline`, and a deterministic session identifier enables Fireworks session affinity without persisting a Pi conversation. The stable system prompt and byte-stable verdict tool schema precede all untrusted pull-request material. The guest returns input, output, cache-read, cache-write, turn, and Pi-calculated cost data from the complete event stream when available. diff --git a/docs/azure-runner.md b/docs/azure-runner.md index ed187018855..b6d180c1bc4 100644 --- a/docs/azure-runner.md +++ b/docs/azure-runner.md @@ -69,6 +69,7 @@ Declared dependency paths are rehashed after the VM clones the bundle. Package installation performed by a repository command must remain rootless and derive from committed lockfiles or the selected reviewed image. Missing toolchain capability fails the command rather than triggering a local retry or privileged repository-controlled bootstrap. The fixed root bootstrap installs a hard-coded Ubuntu transport and Linux test-tool package closure before repository code starts when the pinned Canonical image lacks it. +Before each package operation it waits up to three minutes for the standard apt/dpkg locks, and apt carries the same bounded dpkg timeout, so normal image maintenance can finish without turning into a repository-command failure while a stuck lock still fails closed. Repository code cannot alter that privileged package list, all package and staging traffic is shaped to one megabit per second and ends before deny-all command networking starts, and invocation evidence must record the resolved package/image versions during real acceptance. The request records the exact-size, checksum-pinned ShellCheck 0.11.0 and uv 0.9.10 releases plus, when the snapshot contains the Agent Fleet `uv.lock`, the complete Linux x86_64 pytest/ruff wheel closure selected from that lock. When that lock is present, trusted root verifies the lock, archive, file set, sizes, and hashes, creates the Agent Fleet environment with an empty cache and networking disabled, and forces repository `uv run --locked` commands to use that already-synchronized offline environment. diff --git a/docs/crosscheck.md b/docs/crosscheck.md index f8c93f4c840..9de6d453ae4 100644 --- a/docs/crosscheck.md +++ b/docs/crosscheck.md @@ -79,8 +79,10 @@ Every reviewer disables reviewed-repository instruction discovery at launch: Cod Pi is launched through the resolved installed executable at `xhigh` with JSON event output, offline startup, an ephemeral session, a deterministic session-affinity identifier, and only the read and Bash-capable review tools plus the explicit verdict tool. The prompt is passed by `@file` so repository and claim size cannot exceed the process argument limit. Extension discovery remains disabled while the tracked verdict extension is loaded explicitly. -That extension registers a strict JSON-schema-constrained `submit_crosscheck_verdict` tool whose successful execution terminates the run without a second paid model turn. -Crosscheck accepts exactly one verdict tool call from the successful final attempt, preserves usage across Pi auto-retries, and refuses zero or multiple calls. +That extension registers a strict JSON-schema-constrained `submit_crosscheck_verdict` tool whose successful execution terminates that attempt without another model turn. +Crosscheck accepts exactly one verdict tool call from the successful final attempt and preserves usage across Pi auto-retries. +If an otherwise completed attempt makes zero, multiple, or malformed verdict calls, the same isolated reviewer identity receives one fresh, fixed repair prompt containing the same exact-head packet. +The repair is attempted once, its usage is included in the run economics, and a second protocol miss fails closed instead of selecting a convenient call or rotating to another reviewer. The model decides the provider slot through an explicit mapping derived from the lane registry that maps each registered model to its own slot, maps `gpt-5.6-sol` to `openai-codex`, and refuses an unmapped model rather than guessing. For the installed npm entrypoint, Crosscheck also resolves Pi's sibling Node runtime before launch instead of allowing the reviewer environment's `PATH` to substitute another interpreter. That pin recognizes every `env`-based Node shebang, including `#!/usr/bin/env -S node --flag`, and preserves the flags; an `env` shebang naming no interpreter fails closed rather than silently falling back to `PATH`. @@ -93,7 +95,7 @@ On that lane it authenticates against the same upstream OpenAI accounts the Code What Pi adds is an independent client path and a reviewer that is separate from a Claude author by construction. A usage-limited reviewer account records a `tool-failure`, never a verdict about code, and Crosscheck then advances to the next independent entry rather than refusing the merge. Failover is limited to faults that prevented a verdict: a launch failure, an unusable credential, a provider that was never reached, or an exhausted account. -A reviewer that reached the model and then declined clearance, returned no valid artifact, or returned a malformed one ends the run on the spot, because that is the reviewer's own conclusion and a second account must not be used to shop for a friendlier one. +A reviewer that reached the model and then declined clearance, or still returned no valid artifact after the one bounded protocol repair, ends the run on the spot because a second account must not be used to shop for a friendlier conclusion. Each abandoned attempt is recorded as its own `tool-failure` run, so the ledger names every account that was tried and why it was left, and each attempt gets its own pristine exact-head checkout so no reviewer inherits an earlier reviewer's helpers or scratch state. Selection therefore makes the gate as available as the roster rather than as available as its first entry. Every candidate passed the configured reviewer-profile and model policy. @@ -157,10 +159,12 @@ An empty `FM_STATE_OVERRIDE` falls back to the home state directory, so task met Every run record carries `durations_ms`, an integer millisecond breakdown of where that invocation's wall clock went (C1, `docs/azure-requirements.md`). The local lane records `snapshot` (task metadata, the GitHub head/claims lookup, reviewer selection, and the exact-head review checkout), `reviewer` (the bounded reviewer subprocess), `proofs` (the reproduction and mutation verification the gate re-executes for itself), `ledger` (reading and validating the durable ledger plus this invocation's earlier writes), and `total`. -The Azure compartment lane additionally records `create`, `stage`, `boot`, and `collect`, which only that lane performs. +The Azure compartment lane additionally records `create`, `stage`, `boot`, and `collect` after its completed Azure identity binds those phases to that lane. -A phase appears only if the run actually entered it. -An absent phase means the lane never did that work; it never means the work was free, so a run that failed before reviewer launch records no `reviewer` key rather than `reviewer: 0`. +Every recorded phase represents work the run actually entered, but a failed compartment attempt may omit lane-only detail it cannot bind. +An absent ordinary phase means the run never entered it, so a run that failed before reviewer launch records no `reviewer` key rather than `reviewer: 0`. +A failed compartment attempt has no complete Azure identity to bind lane-only detail, so the writer omits `create`, `stage`, `boot`, and `collect` from that run while retaining `total` and compatible ordinary phases. +The difference remains real unattributed time rather than a fabricated zero. Durations are measured on `time.monotonic()`, so a clock change cannot move them, while the record's `at` stamp remains the wall clock it has always been. Phases never nest and named phases round down while `total` rounds up, so `total >= sum(named phases)` holds exactly; the difference is real unattributed time between phases, not rounding. Two reviewer attempts inside one invocation accumulate into one `reviewer` phase, because the invocation really did spend both. @@ -186,28 +190,12 @@ Summing the `total` column therefore double counts; read the last row of an invo `durations_ms` is additive. A run recorded before this field existed still validates and still renders, and shows `-` in every phase column rather than a fabricated zero. A record that does carry one is held to the full contract: integers only, never negative, only phase names the gate defines, a `snapshot` phase (every run that reaches a record has performed it), and a `total` that covers the phases it names. -The compartment phases are lane-bound rather than writer-asserted: `create`, `stage`, `boot`, and `collect` are admitted only on a record whose own reviewer entry carries `execution_mode: azure-compartment-v1`, so "absent means this lane did not do it" is enforced by the gate and not merely by the writer's good behavior. +The compartment phases are lane-bound rather than writer-asserted: `create`, `stage`, `boot`, and `collect` are admitted only on a record whose own reviewer entry carries `execution_mode: azure-compartment-v1`; a failed attempt without that completed identity retains its total but cannot claim the lane-only breakdown. A run's `at` stamp is pinned to `YYYY-MM-DDTHH:MM:SSZ` for the same reason: it is the one free-form string the table renders, and an embedded newline would let one record forge extra rows. -The writer validates the measurement against that same contract before writing it, and drops it if it fails. +The writer removes compartment-only phases from a run that lacks a completed compartment identity, then validates the compatible measurement against that same contract before writing it. Everything that later reads this ledger validates it, so an unvalidated write would be a durable outage: one writer bug and `run`, `verify` and `timings` all refuse the task until a human edits the JSON by hand. -The measurement is the disposable half of that trade, so a timing bug loudly costs one run its breakdown and never the durable findings. -#### Known gap, bound to the `fm-ccm` image rebake: stamp the compartment lane at its START - -The compartment lane's discriminator only exists on a run that COMPLETED that lane, so a compartment review which fails part-way loses its measurement entirely. -`bin/fm-crosscheck-azure.py` `_run_azure_review_in_lane` stamps `execution_mode: azure-compartment-v1` onto the reviewer record only at the end, after a digest-bound result exists. -A review that fails during `create`, `stage`, `boot`, or `collect` therefore reaches `append_failed_run` with those phases measured but no `execution_mode`, fails `validate_durations`' lane check, and has its whole `durations_ms` dropped by `stamp_durations` rather than written. -That is precisely the run whose numbers would be most diagnostic: a compartment review that died in create or boot is the one an operator most needs a breakdown of. - -The obvious fix does not work as written, which is why this is recorded rather than done. -Stamping `execution_mode` earlier makes `validate_ledger` route the record into `bin/fm-crosscheck-azure.py` `validate_azure_reviewer_record`, which demands a complete `azure_identity` (review generation, request and credential digests, model/tool/verifier identities). -A failed part-way review has no such identity, so an early `execution_mode` would refuse the record outright: it trades a lost measurement for a bricked ledger, which is the worse end of the same trade. - -**Follow-up, due when the compartment lane becomes executable again:** before any compartment timing is relied on, stamp the lane at its start, either by making `validate_azure_reviewer_record` apply only to records that claim a completed compartment review, or by adding a separate early lane marker that `run_is_compartment_lane` also accepts. -Do this at the rebake, not before: while the `fm-ccm` image carries no `pi` binary the compartment lane cannot run at all, so a failed compartment review is impossible and the lost-measurement cost is exactly zero. -Adding schema surface for a lane that cannot execute would freeze a contract nothing has exercised. - -Until then the current behavior is deliberate and is preferred over both alternatives, which are bricking the ledger and letting any record claim compartment phases it never performed. +Any other timing-contract bug still loudly costs one run its breakdown and never the durable findings. `bin/fm-pr-merge.sh` calls the verification form automatically after approval. Do not call the verification form as a substitute for running a reviewer. diff --git a/tests/fm-azure-runner.test.sh b/tests/fm-azure-runner.test.sh index 7d5ab99a199..e184431f6ae 100755 --- a/tests/fm-azure-runner.test.sh +++ b/tests/fm-azure-runner.test.sh @@ -1926,6 +1926,110 @@ PY pass "per-run routing selects a class from a file the run can carry, an absent file stays local and says so, and every present-but-broken file refuses by name and runs the command nowhere" } +apt_lock_wait_contract() { + local tmp helpers lock ready release holder_pid release_pid out rc call_log attempts + tmp=$(mktemp -d) + helpers="$tmp/apt-lock-helpers.sh" + awk ' + /BEGIN FM_AZURE_RUNNER_APT_LOCK_HELPERS/ {emit=1; next} + /END FM_AZURE_RUNNER_APT_LOCK_HELPERS/ {emit=0} + emit + ' "$GUEST" >"$helpers" + bash -n "$helpers" || fail "the extracted Azure runner apt-lock helpers are invalid" + # shellcheck source=/dev/null + . "$helpers" + + lock="$tmp/lock-frontend" + ready="$tmp/ready" + release="$tmp/release" + python3 - "$lock" "$ready" "$release" <<'PY' & +import fcntl +import os +from pathlib import Path +import sys +import time + +descriptor = os.open(sys.argv[1], os.O_RDWR | os.O_CREAT, 0o644) +fcntl.lockf(descriptor, fcntl.LOCK_EX) +Path(sys.argv[2]).touch() +while not Path(sys.argv[3]).exists(): + time.sleep(0.01) +os.close(descriptor) +PY + holder_pid=$! + fm_test_wait_for_file "$ready" "$holder_pid" 0.02 \ + || fail "the apt-lock release regression did not acquire its fixture lock" + (sleep 0.15; touch "$release") & + release_pid=$! + wait_for_apt_locks 2 0.02 "$lock" \ + || fail "the guest bootstrap did not continue after a normal apt lock released" + wait "$release_pid" || fail "the apt-lock release fixture failed" + wait "$holder_pid" || fail "the apt-lock holder failed" + call_log="$tmp/apt-call" + APT_LOCK_WAIT_SECONDS=1 + APT_LOCK_PATHS=("$lock") + wait_for_apt_locks "$APT_LOCK_WAIT_SECONDS" 0.02 "${APT_LOCK_PATHS[@]}" \ + || fail "the test apt lock set was not available for wrapper execution" + # shellcheck disable=SC2329 # Invoked indirectly by the extracted run_bootstrap_apt helper. + run_bootstrap_network() { printf '%s\n' "$*" >"$call_log"; } + run_bootstrap_apt install -y fixture-package \ + || fail "the bounded apt wrapper did not reach apt-get after lock admission" + [ "$(<"$call_log")" = \ + "apt-get -o DPkg::Lock::Timeout=1 install -y fixture-package" ] \ + || fail "the executable apt wrapper lost its dpkg timeout or package arguments" + + attempts="$tmp/apt-attempts" + printf '0\n' >"$attempts" + run_bootstrap_network() { + local count + count=$(<"$attempts") + count=$((count + 1)) + printf '%s\n' "$count" >"$attempts" + if [ "$count" -eq 1 ]; then + echo "E: Could not get lock $lock. It is held by process 42" >&2 + return 100 + fi + return 0 + } + APT_LOCK_WAIT_SECONDS=2 + run_bootstrap_apt update -qq \ + || fail "the apt wrapper did not retry an invocation-boundary lock race" + [ "$(<"$attempts")" -eq 2 ] \ + || fail "the apt wrapper did not perform exactly one lock-race retry" + + ready="$tmp/ready-timeout" + release="$tmp/release-timeout" + python3 - "$lock" "$ready" "$release" <<'PY' & +import fcntl +import os +from pathlib import Path +import sys +import time + +descriptor = os.open(sys.argv[1], os.O_RDWR | os.O_CREAT, 0o644) +fcntl.lockf(descriptor, fcntl.LOCK_EX) +Path(sys.argv[2]).touch() +while not Path(sys.argv[3]).exists(): + time.sleep(0.01) +os.close(descriptor) +PY + holder_pid=$! + fm_test_wait_for_file "$ready" "$holder_pid" 0.02 \ + || fail "the apt-lock timeout regression did not acquire its fixture lock" + set +e + out=$(wait_for_apt_locks 0.1 0.02 "$lock" 2>&1) + rc=$? + set -e + touch "$release" + wait "$holder_pid" || fail "the timed-out apt-lock holder failed" + [ "$rc" -ne 0 ] || fail "the guest bootstrap waited forever on an apt lock" + assert_contains "$out" "timed out waiting for apt/dpkg lock(s): $lock" \ + "the bounded apt-lock refusal did not name the held lock" + + pass "Azure runner bootstrap waits for apt/dpkg maintenance races and times out deterministically" +} + +apt_lock_wait_contract dispatch_ambient_binding_contract no_mistakes_test_step_offload_contract routing_file_contract diff --git a/tests/fm-crosscheck-azure.test.sh b/tests/fm-crosscheck-azure.test.sh index 765a0d88690..271b2bf7029 100755 --- a/tests/fm-crosscheck-azure.test.sh +++ b/tests/fm-crosscheck-azure.test.sh @@ -2926,13 +2926,20 @@ import os from pathlib import Path import sys -Path(os.environ["CAPTURE"]).write_text(json.dumps({ +capture_path = Path(os.environ["CAPTURE"]) +captures = json.loads(capture_path.read_text()) if capture_path.exists() else [] +captures.append({ "argv": sys.argv[1:], "account": os.environ.get("PI_CODING_AGENT_DIR"), "schema": os.environ.get("FM_CROSSCHECK_REVIEW_SCHEMA"), -})) + "prompt_text": Path(sys.argv[-1][1:]).read_text(), +}) +capture_path.write_text(json.dumps(captures)) scenario = os.environ["SCENARIO"] -if scenario == "nonzero": +effective = scenario +if scenario.endswith("-then-valid"): + effective = scenario.removesuffix("-then-valid") if len(captures) == 1 else "valid" +if effective == "nonzero": print("bounded fake provider failure", file=sys.stderr) raise SystemExit(17) outer = { @@ -2942,22 +2949,22 @@ outer = { ], } arguments = outer -if scenario == "string": +if effective == "string": arguments = "Reviewed. " + json.dumps(outer) + " Done." -elif scenario == "multiple-json": +elif effective == "multiple-json": arguments = json.dumps(outer) + json.dumps({"second": True}) call = { "type": "toolCall", "id": "verdict-1", "name": "submit_crosscheck_verdict", "arguments": arguments, } content = [call] -if scenario == "missing": +if effective == "missing": content = [] -elif scenario == "multiple": +elif effective == "multiple": content = [call, {**call, "id": "verdict-2"}] reported_model = ( "accounts/fireworks/routers/glm-5p2-fast" - if scenario == "wrong-model" else "accounts/fireworks/models/glm-5p2" + if effective == "wrong-model" else "accounts/fireworks/models/glm-5p2" ) message = { "role": "assistant", @@ -2974,8 +2981,17 @@ message = { }, }, } +if effective == "internal-retry": + message["content"] = [{**call, "id": ""}] + message["stopReason"] = "error" print(json.dumps({"type": "turn_end", "message": message})) print(json.dumps({"type": "agent_end"})) +if effective == "internal-retry": + print(json.dumps({"type": "auto_retry_start"})) + message["content"] = [call] + message["stopReason"] = "toolUse" + print(json.dumps({"type": "turn_end", "message": message})) + print(json.dumps({"type": "agent_end"})) ''' @@ -3016,7 +3032,7 @@ def run(scenario): return completed, captured, value, account, prompt, schema -def assert_launch(captured, account, prompt, schema): +def assert_launch(captured, account, prompt, schema, attempts=1): extension_digest = hashlib.sha256(extension.read_bytes()).hexdigest() seed = f"{model}\n{extension_digest}\n".encode() session = "fm-crosscheck-" + hashlib.sha256(seed).hexdigest()[:32] @@ -3026,18 +3042,25 @@ def assert_launch(captured, account, prompt, schema): "the enabled tools and submit the complete final verdict exactly once " "with submit_crosscheck_verdict." ) - assert captured["argv"] == [ - "--mode", "json", "--offline", "--provider", provider, - "--model", model, "--thinking", "xhigh", - "--tools", "submit_crosscheck_verdict", - "--extension", str(extension), - "--system-prompt", system_prompt, - "--session-id", session, "--no-session", "--no-extensions", - "--no-skills", "--no-prompt-templates", "--no-themes", - "--no-context-files", "--no-approve", f"@{prompt}", - ], captured["argv"] - assert captured["account"] == str(account), captured - assert captured["schema"] == str(schema), captured + assert len(captured) == attempts, captured + for index, launch in enumerate(captured): + active_prompt = prompt if index == 0 else prompt.parent / "repair-prompt.txt" + assert launch["argv"] == [ + "--mode", "json", "--offline", "--provider", provider, + "--model", model, "--thinking", "xhigh", + "--tools", "submit_crosscheck_verdict", + "--extension", str(extension), + "--system-prompt", system_prompt, + "--session-id", session, "--no-session", "--no-extensions", + "--no-skills", "--no-prompt-templates", "--no-themes", + "--no-context-files", "--no-approve", f"@{active_prompt}", + ], launch["argv"] + assert launch["account"] == str(account), launch + assert launch["schema"] == str(schema), launch + if attempts == 2: + repair = captured[1]["prompt_text"] + assert repair.startswith("PROMPT BY FILE\n\nVERDICT PROTOCOL REPAIR"), repair + assert "submit_crosscheck_verdict exactly once" in repair, repair completed, captured, value, account, prompt, schema = run("valid") @@ -3056,15 +3079,42 @@ completed, captured, value, account, prompt, schema = run("string") assert completed.returncode == 0 and value["verdict"] == outer["verdict"] assert_launch(captured, account, prompt, schema) -for scenario in ("missing", "multiple", "multiple-json", "wrong-model", "nonzero"): +completed, captured, value, account, prompt, schema = run("internal-retry") +assert completed.returncode == 0 and value["verdict"] == outer["verdict"], ( + completed.returncode, completed.stderr, value, +) +assert_launch(captured, account, prompt, schema) + +for scenario in ("missing-then-valid", "multiple-then-valid", "multiple-json-then-valid"): + completed, captured, value, account, prompt, schema = run(scenario) + assert completed.returncode == 0 and value["verdict"] == outer["verdict"], ( + scenario, completed.returncode, completed.stderr, value, + ) + assert_launch(captured, account, prompt, schema, attempts=2) + assert value["telemetry"]["tokens"] == { + "input": 20, "output": 4, "cache_read": 8, "cache_write": 0, + "source": "pi-turn-end-message-usage", + }, value["telemetry"] + assert value["telemetry"]["costs_usd"]["declared"] == 0.00004672 + assert value["telemetry"]["turns"] == 2 + +for scenario in ("missing", "multiple", "multiple-json"): + completed, captured, value, account, prompt, schema = run(scenario) + assert completed.returncode == 125 and value is None, ( + scenario, completed.returncode, completed.stderr, value, + ) + assert "one bounded verdict repair was exhausted" in completed.stderr + assert_launch(captured, account, prompt, schema, attempts=2) + +for scenario in ("wrong-model", "nonzero"): completed, captured, value, account, prompt, schema = run(scenario) assert completed.returncode == 125 and value is None, ( scenario, completed.returncode, completed.stderr, value, ) assert_launch(captured, account, prompt, schema) -print("PI RUNTIME executes the shipped strict-tool launch and fails closed") +print("PI RUNTIME repairs one verdict-protocol miss, then fails closed") PY - pass "the digest-bound Pi runtime executes one isolated strict-tool review" + pass "the digest-bound Pi runtime bounds verdict repair and remains fail closed" } azure_pi_review_contract_unit() { diff --git a/tests/fm-crosscheck.test.sh b/tests/fm-crosscheck.test.sh index 23b965f8924..c8af5b31bd4 100755 --- a/tests/fm-crosscheck.test.sh +++ b/tests/fm-crosscheck.test.sh @@ -5632,12 +5632,18 @@ assert run["summary"] == "kept", run assert "dropping this run's phase measurement" in noise.getvalue(), noise.getvalue() assert "does not cover its named phases" in noise.getvalue(), noise.getvalue() -# A stale measurement from an earlier persist in the same invocation is -# removed too, rather than left behind as the record's answer. +# A failed Azure attempt has no completed compartment identity to bind its +# lane-only phases. Those incompatible fields are omitted, while the honest +# total and ordinary phases remain available and the writer stays quiet. run["durations_ms"] = {"snapshot": 1, "total": 2} -with contextlib.redirect_stderr(io.StringIO()): - module.stamp_durations(run, {"create": 1, "snapshot": 1, "total": 30}) -assert "durations_ms" not in run, run +partial = io.StringIO() +with contextlib.redirect_stderr(partial): + module.stamp_durations( + run, + {"create": 1, "stage": 2, "boot": 3, "snapshot": 4, "total": 30}, + ) +assert run["durations_ms"] == {"snapshot": 4, "total": 30}, run +assert partial.getvalue() == "", partial.getvalue() # An honest measurement is still written, and the drop path is not the norm. good = {"snapshot": 10, "reviewer": 20, "total": 40} @@ -5670,7 +5676,7 @@ source = inspect.getsource(module.run_crosscheck) assert "stamp_durations(run, timer.durations_ms())" in source, source assert 'run["durations_ms"] =' not in source, source PY - pass "a measurement failing its own contract is dropped loudly, never written into the ledger" + pass "failed compartment timings omit incompatible phases and invalid measurements never land" } test_explicit_pi_tool_loads_with_discovery_disabled() {