diff --git a/CHANGELOG.md b/CHANGELOG.md index 9cb630e..82a2de1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,32 @@ A running server reports its version at `/health`. ## [Unreleased] +### Added + +- Durable trip progress (INT-416): each conversation's turn (identity, action, status, + timestamps, deadline, active seconds, cancel request/outcome) and approval history are + persisted in the existing store, and the new `trip_progress` MCP tool reports every leg + revision's stage, the pending or cancelled approval, completed legs and uncertain + generations without invoking the model or a provider. After a restart, unfinished work is + reported as `interrupted` instead of `running`. +- Cooperative cancellation and turn deadlines (INT-417): the new `cancel_trip` MCP tool stops + queued or running work at the next safe point (model/tool boundary, browser queue wait or + immediately before Generate), withdraws a pending approval and approved revisions that were + never clicked, and keeps completed plans and briefings. A Generate click already made is + never interrupted or repeated; it is reported as generated or uncertain. + `IFD_TURN_DEADLINE_SECONDS` (default 600) bounds each turn's active execution; waiting for + a human approval happens between turns and is not counted. +- Separate `ifd-workflow-v050-trip-control` evaluation suite (WF-029–033) over the production + MCP tools with controlled providers. + +### Changed + +- `turn_status` falls back to the persisted turn record when no in-process turn exists, and + `conversation_status` reports `approval_cancelled` for a withdrawn approval; + `approve_dispatch` refuses to approve a cancelled approval. Existing statuses are unchanged. +- Workflow verifier `ifd-workflow-oracles-v4` adds MCP-output assertions and the cancelled + outcome; v1–v3 manifests remain accepted. + ## [0.4.7] - 2026-11-12 ### Fixed diff --git a/docs/implementation/agent-usage.md b/docs/implementation/agent-usage.md index 31820c7..05aa8be 100644 --- a/docs/implementation/agent-usage.md +++ b/docs/implementation/agent-usage.md @@ -116,6 +116,24 @@ The review queue scores whole conversations. Automatic workflow feedback is sepa owner acceptance. Provider usage reports FlightAware list-price estimates and Browserbase elapsed time; unknown invoice costs stay unknown. +## v0.5.0 trip control (unreleased) + +Remote MCP tools added for INT-416/417. Neither invokes the model or a provider. + +| Tool | Use | +| --- | --- | +| `trip_progress(thread_id)` | Durable progress: the current/last turn (`queued`, `running`, `approval_required`, `reply`, `cancelled`, `deadline_exceeded`, `failed`, or `interrupted` after a restart), pending/cancelled approval and history, and each leg revision's stage (`awaiting_generation`, `generating`, `generated_awaiting_finalize`, `generation_uncertain`, `generation_failed`, `cancelled`, `plan_retained`, `briefing_ready`, `infeasible`), with `completed_legs`, `uncertain_generations` and a `next` instruction | +| `cancel_trip(thread_id, reason?)` | Only when the user asks to stop. Records the request, signals a running turn and waits up to 25 seconds for it to reach a safe point. Returns `cancelled` (with `outcome` `stopped` or `completed` when the turn finished first) or `cancelling` while a Generate click is finishing | + +Cancellation withdraws the pending approval (`approve_dispatch` then returns `cancelled`) and +every approved revision whose Generate was never clicked; continuing needs a new approved +revision. Completed plans and briefings are kept. A clicked or uncertain generation is never +withdrawn, cleared or repeated: check SimBrief, then finalize or request a new revision. +`IFD_TURN_DEADLINE_SECONDS` (default 600) stops a turn at its next safe point and, like a +cancel, withdraws unclicked approved revisions; worker threads +are not killed, so a stalled model or provider call ends when it returns. Browser limits are +unchanged: one active workflow, two waiters, 30-second queue wait, 180-second session. + ## Bounded browser management Deployed in [v0.4.2](deployment.md#current-release-v042) after exact-image staging acceptance diff --git a/docs/technical-design.md b/docs/technical-design.md index a8f1366..4bcf550 100644 --- a/docs/technical-design.md +++ b/docs/technical-design.md @@ -132,6 +132,37 @@ Use separate component statuses for research/parking, plan generation, thrust, a Before retrying generation after a timeout, reconcile using any supported external identity/receipt; never blindly generate again. If outcome cannot be reconciled, report it as unknown and request the necessary action. Retry limits, timeouts, and research bounds should be explicit operational settings selected from measured behavior. +### Implemented trip progress and cancellation (v0.5.0, INT-416/417) + +State lives in the existing store without a schema change. Each conversation thread has one +`turn_progress:` JSON record in `settings` (SQLite and Postgres): `turn_id`, +`action`, `status`, `boot_id`, `started_at`/`finished_at`/`deadline_at`, `active_seconds` and +`active_seconds_total`, `cancel` (`requested_at`, `reason`, `while`, `outcome`, +`confirmed_at`), `cancelled_interrupt` and the approval decision history. Leg stages are +derived from the saved request, `leg_meta` and snapshot of each revision; a cancelled +revision carries `cancelled` in its metadata, which also fences `claim_generation`. The pending +approval comes from the LangGraph checkpoint. + +Turn states: `queued → running → approval_required | reply | cancelled | deadline_exceeded | +failed`. A record left `queued`/`running` by another server process (`boot_id` differs) or +with no in-process turn is reported as `interrupted`. Cancel requests move +`requested → stopped` (the turn reached a safe point) or `requested → completed` (it +finished first); while a Generate click is in flight `cancel_trip` reports `cancelling`. + +Leg stages: `awaiting_generation → generating → generated_awaiting_finalize → plan_retained → +briefing_ready | infeasible`, with `generation_failed`, `cancelled` (approval withdrawn before +any click) and `generation_uncertain` (clicked without an observed result, or a `clicking` +claim older than one bounded browser workflow or left by an inactive turn). + +Cancellation is cooperative: checks run before each model call and tool call, while waiting for +browser admission and immediately before the Generate claim. Python threads are never killed +and a click is never interrupted, cleared or retried. A cancel that lands while the model is +proposing a dispatch withdraws that new approval instead of offering it. The turn deadline +(`IFD_TURN_DEADLINE_SECONDS`, default 600 seconds) covers active execution only; the human +approval wait falls between turns. The default exceeds one 180-second browser session plus the +30-second queue and model steps; recorded live sessions took 50–80 seconds. Per-turn +`active_seconds` is now recorded so the limit can be recalibrated from observed turns. + ## MCP and presentation contract Proposed task-level interface: diff --git a/evals/README.md b/evals/README.md index 74b8271..cfa2db9 100644 --- a/evals/README.md +++ b/evals/README.md @@ -245,6 +245,28 @@ A product exception is a valid observed failure; an invalid run means the harnes cannot support a comparison. Manifest/fixture drift aborts before execution. Reports should remain under ignored `artifacts/evals/` because they retain complete observations. +### v0.5.0 trip-control suite (INT-416/417) + +`workflow_cases_v050.json` is the separate `ifd-workflow-v050-trip-control` manifest +(verifier `ifd-workflow-oracles-v4`, base commit v0.4.6). Its cases require `--mcp`; without +it a run is marked invalid. Cancellation comes from a concurrent MCP `cancel_trip` call and +the deadline from real elapsed time against `IFD_TURN_DEADLINE_SECONDS`; no clock patching or +live session is involved. `mcp_assertions` check `cancel_trip`, `approve_dispatch` and +`trip_progress` outputs. + +| Case | Scenario | +| --- | --- | +| WF-029 | Cancel at the first approval; the approval cannot then be approved | +| WF-030 | Cancel at leg 2's approval after leg 1's briefing; leg 1 kept across a restart | +| WF-031 | Cancel before the Generate click; no click, revision withdrawn | +| WF-032 | Cancel after a click whose outcome times out; reported uncertain, one attempt only | +| WF-033 | Turn deadline passes before the click; no click, revision withdrawn | + +```sh +uv run python evals/run_workflow_replay.py run \ + --manifest evals/workflow_cases_v050.json --mcp --output artifacts/evals/v050-trip-control.json +``` + ### Optional LangSmith experiment upload The installed `langsmith==0.14.16` supports [`Client.evaluate`](https://docs.langchain.com/langsmith/evaluate-llm-application). diff --git a/evals/mcp_replay.py b/evals/mcp_replay.py index 8447bde..e8710c9 100644 --- a/evals/mcp_replay.py +++ b/evals/mcp_replay.py @@ -107,6 +107,11 @@ def invoke(self, payload, config, **kwargs): } ) ] + elif result["status"] == "cancelled": + # A cancelled or expired turn ends without a model reply; the harness inspects + # durable progress next instead of treating it as a transport failure. + self.run["cancelled"] = True + self.run.setdefault("mcp_outputs", []).append({"tool": name, "output": result}) elif result["status"] != "reply": raise RuntimeError(f"unexpected MCP outcome: {result['status']}") return SimpleNamespace(value=state, interrupts=interrupts) diff --git a/evals/run_workflow_replay.py b/evals/run_workflow_replay.py index 7d9f99c..f72da0a 100644 --- a/evals/run_workflow_replay.py +++ b/evals/run_workflow_replay.py @@ -8,6 +8,7 @@ import json import os import tempfile +import threading import time import sqlite3 from collections import Counter @@ -32,6 +33,7 @@ flights, observability, operations, + progress, remote, workflow, ) @@ -244,6 +246,30 @@ def now(_cls, tz=None): continuation_turns = 0 turns = [] source = (ROOT / manifest["fixture"]["path"]).read_bytes() + control = case.get("control") or {} + concurrent = [] # cancel_trip calls made by a concurrent client while a turn runs + driver = None + + def mcp_call(name, args): + output = driver.call(name, args) + run.setdefault("mcp_outputs", []).append({"tool": name, "output": output}) + return output + + def cancel_during_turn(): + """Cancel from a second client mid-turn, then continue once the turn sees it.""" + if control.get("cancel_at") not in {"pre_click", "post_click"}: + return + caller = threading.Thread( + target=mcp_call, + args=("cancel_trip", {"thread_id": case["id"], "reason": "owner cancelled"}), + ) + caller.start() + concurrent.append(caller) + deadline = time.monotonic() + 10 + while not (case["id"] in remote._controls and remote._controls[case["id"]].event.is_set()): + if time.monotonic() > deadline: + raise RuntimeError("controlled cancellation was not received") + time.sleep(0.01) def blocked(*args, **kwargs): run["invalid"] = True @@ -290,6 +316,11 @@ def generate(fields, click, shots_dir, before_click=None): ) if failure == "timeout_before_click": raise TimeoutError("controlled pre-click timeout") + if control.get("cancel_at") == "pre_click": + cancel_during_turn() + if control.get("cancel_at") == "deadline_pre_click": + # Real elapsed time past the configured turn deadline; no clock patching. + time.sleep(float(control["deadline_seconds"]) + 0.5) if case["scenario"] == "retry_duplicate" and len(run["provider_attempts"]) == 1: return { "status": "fill_mismatch", @@ -313,6 +344,8 @@ def generate(fields, click, shots_dir, before_click=None): raw["navlog"] = {"fix": [{"ident": s} for s in case["route_evidence"]]} body = json.dumps(raw).encode() run["provider_plans"].append({"key": ident, "sha256": sha256(body), "fields": fields}) + if control.get("cancel_at") == "post_click": + cancel_during_turn() if failure == "timeout_after_click": raise TimeoutError("controlled post-click timeout") return { @@ -383,6 +416,11 @@ def harvest(current): "IFD_AIRPORT_CACHE_DIR": "disabled", "IFD_INTERNAL_TOKEN": "controlled-workflow-replay-token", "LANGSMITH_TRACING": "false", + **( + {"IFD_TURN_DEADLINE_SECONDS": str(control["deadline_seconds"])} + if "deadline_seconds" in control + else {} + ), }, ) ) @@ -410,12 +448,18 @@ def harvest(current): (flights, "datetime", Clock), (agent, "datetime", Clock), (operations, "datetime", Clock), + (progress, "datetime", Clock), (remote, "paid_operations_enabled", lambda: True), ): stack.enter_context(patch.object(module, name, replacement)) # Do not export .env tracing settings or send ambient traces; explicit upload is separate. stack.enter_context(tracing_context(enabled=False)) try: + if control and not mcp_url: + run["invalid"] = True + run["invalid_reasons"].append("trip-control case requires --mcp transport") + run.update(verify(case, run)) + return run if case.get("airline_flight"): flight = case["airline_flight"] required = { @@ -528,6 +572,20 @@ def flight_get(path, params=None): messages = result.value["messages"] harvest(messages) observe() + if ( + result.interrupts + and control.get("cancel_at") == "approval" + and len(run["approvals"]) == control.get("approval_index", 0) + ): + # The owner stops at this approval instead of answering it. + pending = result.interrupts[0].value["action_requests"][0] + run["cancelled_approvals"] = [pending] + cancelled = mcp_call( + "cancel_trip", {"thread_id": case["id"], "reason": "owner cancelled"} + ) + run["cancelled"] = cancelled["status"] == "cancelled" + mcp_call("approve_dispatch", {"thread_id": case["id"], "approve": True}) + break if result.interrupts: decisions = [] for interrupt in result.interrupts: @@ -600,6 +658,11 @@ def flight_get(path, params=None): graph = agent.build_agent(model=selected_model, checkpointer=replacement) driver.graph = graph run["recovered"] = driver.call("conversation_status", {"thread_id": case["id"]}) + if control: + for caller in concurrent: + caller.join(timeout=remote.WAIT_SECONDS + 15) + mcp_call("turn_status", {"thread_id": case["id"]}) + mcp_call("trip_progress", {"thread_id": case["id"]}) observe() except Exception as exc: diff --git a/evals/workflow_cases_v050.json b/evals/workflow_cases_v050.json new file mode 100644 index 0000000..659ac61 --- /dev/null +++ b/evals/workflow_cases_v050.json @@ -0,0 +1,614 @@ +{ + "suite_id": "ifd-workflow-v050-trip-control", + "schema_version": 1, + "verifier_version": "ifd-workflow-oracles-v4", + "clock_utc": "2026-11-08T11:00:00Z", + "base_commit": "0cbbe49d53944947824e18941782dcd5edb930f1", + "fixture": { + "path": "docs/research/evidence/simbrief-b38m-parser-fixture.json", + "sha256": "c4c22b55a9db64ea533e57a229946fb88243f5480253ed5d96cb324a8dd0b3af", + "provenance_path": "docs/research/evidence/simbrief-b38m-parser-fixture-provenance.json", + "kind": "controlled-derived-partial-ofp", + "transformation": "run_workflow_replay.controlled_ofp v1: explicit case identity, clock, profile, fuel and feasibility fields; no live/simulator claim", + "provenance_sha256": "0e5b65dfa34596d605ccacf7eea2bd423ce7032337dc6bb5cac559d33c094c6d" + }, + "airframe_fixture_kind": "synthetic-standard-profile-options; no custom account-profile claim", + "airframe_options": { + "A20N": [ + { + "id": "A20N", + "name": "Controlled standard A20N" + } + ], + "B738": [ + { + "id": "B738", + "name": "Controlled standard B738" + } + ], + "CL35": [ + { + "id": "CL35", + "name": "Controlled standard CL35" + } + ] + }, + "purpose": "INT-416/417 trip progress, cancellation and deadline scenarios over the production MCP tools with controlled providers. Separate suite; the v0.4.4 manifest and v0.4.5 supplement are unchanged.", + "cases": [ + { + "id": "WF-029", + "scenario": "trip_control", + "legs": [ + { + "trip": "WF-029", + "leg": "1", + "revision": 1, + "origin": "KIAD", + "destination": "KEWR", + "sim_aircraft": "A20N", + "departure_local_date": "2026-11-08", + "departure_utc": "2026-11-08T12:00:00Z", + "private_callsign": "N320HD", + "units": "KGS", + "passengers": 4, + "airframe_id": "A20N" + } + ], + "no_dispatch": true, + "expected_outcome": "cancelled", + "max_clarifications": 0, + "prompt": "Prepare this exact private itinerary. Use the approval prompt for my decision, then generate and retain each briefing. [{\"trip\": \"WF-029\", \"leg\": \"1\", \"revision\": 1, \"origin\": \"KIAD\", \"destination\": \"KEWR\", \"sim_aircraft\": \"A20N\", \"departure_local_date\": \"2026-11-08\", \"departure_utc\": \"2026-11-08T12:00:00Z\", \"private_callsign\": \"N320HD\", \"units\": \"KGS\", \"passengers\": 4, \"airframe_id\": \"A20N\"}]", + "control": { + "cancel_at": "approval", + "approval_index": 0 + }, + "script": [ + { + "tool": "dispatch_flight", + "args": { + "trip": "WF-029", + "leg": "1", + "revision": 1, + "origin": "KIAD", + "destination": "KEWR", + "sim_aircraft": "A20N", + "departure_local_date": "2026-11-08", + "departure_utc": "2026-11-08T12:00:00Z", + "private_callsign": "N320HD", + "units": "KGS", + "passengers": 4, + "airframe_id": "A20N" + } + }, + { + "text": "Cancelled before dispatch." + } + ], + "mcp_assertions": [ + { + "id": "cancel_withdrew_pending_approval", + "tool": "cancel_trip", + "path": [ + "withdrawn_approval", + "inputs", + "leg" + ], + "equals": "1" + }, + { + "id": "cancelled_approval_cannot_be_approved", + "tool": "approve_dispatch", + "path": [ + "status" + ], + "equals": "cancelled" + }, + { + "id": "progress_reports_cancelled_approval", + "tool": "trip_progress", + "path": [ + "approval", + "pending", + "status" + ], + "equals": "cancelled" + }, + { + "id": "progress_reports_cancelled_turn", + "tool": "trip_progress", + "path": [ + "turn", + "status" + ], + "equals": "cancelled" + }, + { + "id": "no_completed_legs", + "tool": "trip_progress", + "path": [ + "completed_legs" + ], + "equals": [] + } + ] + }, + { + "id": "WF-030", + "scenario": "trip_control", + "legs": [ + { + "trip": "WF-030", + "leg": "1", + "revision": 1, + "origin": "KIAD", + "destination": "KEWR", + "sim_aircraft": "A20N", + "departure_local_date": "2026-11-08", + "departure_utc": "2026-11-08T12:00:00Z", + "private_callsign": "N320HD", + "units": "KGS", + "passengers": 4, + "airframe_id": "A20N" + } + ], + "expected_outcome": "cancelled", + "max_clarifications": 0, + "prompt": "Prepare this exact private itinerary. Use the approval prompt for my decision, then generate and retain each briefing. After the first briefing I may stop the trip. [{\"trip\": \"WF-030\", \"leg\": \"1\", \"revision\": 1, \"origin\": \"KIAD\", \"destination\": \"KEWR\", \"sim_aircraft\": \"A20N\", \"departure_local_date\": \"2026-11-08\", \"departure_utc\": \"2026-11-08T12:00:00Z\", \"private_callsign\": \"N320HD\", \"units\": \"KGS\", \"passengers\": 4, \"airframe_id\": \"A20N\"}, {\"trip\": \"WF-030\", \"leg\": \"2\", \"revision\": 1, \"origin\": \"KEWR\", \"destination\": \"KBOS\", \"sim_aircraft\": \"A20N\", \"departure_local_date\": \"2026-11-08\", \"departure_utc\": \"2026-11-08T16:00:00Z\", \"private_callsign\": \"N320HD\", \"units\": \"KGS\", \"passengers\": 4, \"airframe_id\": \"A20N\"}]", + "restart": true, + "recovered_status": "approval_cancelled", + "control": { + "cancel_at": "approval", + "approval_index": 1 + }, + "expected_clicks": 1, + "expected_snapshots": 1, + "script": [ + { + "tool": "dispatch_flight", + "args": { + "trip": "WF-030", + "leg": "1", + "revision": 1, + "origin": "KIAD", + "destination": "KEWR", + "sim_aircraft": "A20N", + "departure_local_date": "2026-11-08", + "departure_utc": "2026-11-08T12:00:00Z", + "private_callsign": "N320HD", + "units": "KGS", + "passengers": 4, + "airframe_id": "A20N" + } + }, + { + "tool": "generate_in_simbrief", + "args": { + "trip": "WF-030", + "leg": "1", + "revision": 1 + } + }, + { + "tool": "finalize_dispatch", + "args": { + "trip": "WF-030", + "leg": "1", + "revision": 1 + } + }, + { + "tool": "dispatch_flight", + "args": { + "trip": "WF-030", + "leg": "2", + "revision": 1, + "origin": "KEWR", + "destination": "KBOS", + "sim_aircraft": "A20N", + "departure_local_date": "2026-11-08", + "departure_utc": "2026-11-08T16:00:00Z", + "private_callsign": "N320HD", + "units": "KGS", + "passengers": 4, + "airframe_id": "A20N" + } + }, + { + "text": "Leg 2 was cancelled; leg 1 briefing is kept." + } + ], + "mcp_assertions": [ + { + "id": "cancel_withdrew_leg2_approval", + "tool": "cancel_trip", + "path": [ + "withdrawn_approval", + "inputs", + "leg" + ], + "equals": "2" + }, + { + "id": "cancel_keeps_completed_leg", + "tool": "cancel_trip", + "path": [ + "progress", + "completed_legs", + 0, + "stage" + ], + "equals": "briefing_ready" + }, + { + "id": "cancelled_approval_cannot_be_approved", + "tool": "approve_dispatch", + "path": [ + "status" + ], + "equals": "cancelled" + }, + { + "id": "progress_after_restart_keeps_leg1", + "tool": "trip_progress", + "path": [ + "completed_legs", + 0 + ], + "equals": { + "trip": "WF-030", + "leg": "1", + "revision": 1, + "stage": "briefing_ready" + } + }, + { + "id": "progress_after_restart_turn_cancelled", + "tool": "trip_progress", + "path": [ + "turn", + "status" + ], + "equals": "cancelled" + }, + { + "id": "no_withdrawn_completed_revision", + "tool": "cancel_trip", + "path": [ + "withdrawn_revisions" + ], + "equals": [] + } + ] + }, + { + "id": "WF-031", + "scenario": "trip_control", + "legs": [ + { + "trip": "WF-031", + "leg": "1", + "revision": 1, + "origin": "KIAD", + "destination": "KEWR", + "sim_aircraft": "A20N", + "departure_local_date": "2026-11-08", + "departure_utc": "2026-11-08T12:00:00Z", + "private_callsign": "N320HD", + "units": "KGS", + "passengers": 4, + "airframe_id": "A20N" + } + ], + "expected_outcome": "cancelled", + "max_clarifications": 1, + "prompt": "Prepare this exact private itinerary. Use the approval prompt for my decision, then generate and retain each briefing. [{\"trip\": \"WF-031\", \"leg\": \"1\", \"revision\": 1, \"origin\": \"KIAD\", \"destination\": \"KEWR\", \"sim_aircraft\": \"A20N\", \"departure_local_date\": \"2026-11-08\", \"departure_utc\": \"2026-11-08T12:00:00Z\", \"private_callsign\": \"N320HD\", \"units\": \"KGS\", \"passengers\": 4, \"airframe_id\": \"A20N\"}]", + "control": { + "cancel_at": "pre_click" + }, + "followups": [ + "Continue with the same revision now." + ], + "expected_clicks": 0, + "expected_snapshots": 0, + "script": [ + { + "tool": "dispatch_flight", + "args": { + "trip": "WF-031", + "leg": "1", + "revision": 1, + "origin": "KIAD", + "destination": "KEWR", + "sim_aircraft": "A20N", + "departure_local_date": "2026-11-08", + "departure_utc": "2026-11-08T12:00:00Z", + "private_callsign": "N320HD", + "units": "KGS", + "passengers": 4, + "airframe_id": "A20N" + } + }, + { + "tool": "generate_in_simbrief", + "args": { + "trip": "WF-031", + "leg": "1", + "revision": 1 + } + }, + { + "tool": "generate_in_simbrief", + "args": { + "trip": "WF-031", + "leg": "1", + "revision": 1 + } + }, + { + "text": "That revision was cancelled before Generate. A new approved revision is needed to continue." + } + ], + "tool_assertions": [ + { + "id": "cancelled_revision_not_regenerated", + "tool": "generate_in_simbrief", + "path": [ + "status" + ], + "equals": "cancelled" + } + ], + "mcp_assertions": [ + { + "id": "turn_stopped_by_cancel", + "tool": "approve_dispatch", + "path": [ + "status" + ], + "equals": "cancelled" + }, + { + "id": "cancel_reported_no_running_turn_left", + "tool": "cancel_trip", + "path": [ + "status" + ], + "equals": "cancelled" + }, + { + "id": "progress_leg_cancelled", + "tool": "trip_progress", + "path": [ + "trips", + 0, + "legs", + 0, + "stage" + ], + "equals": "cancelled" + }, + { + "id": "progress_no_click", + "tool": "trip_progress", + "path": [ + "trips", + 0, + "legs", + 0, + "generation", + "clicked" + ], + "equals": false + } + ] + }, + { + "id": "WF-032", + "scenario": "trip_control", + "legs": [ + { + "trip": "WF-032", + "leg": "1", + "revision": 1, + "origin": "KIAD", + "destination": "KEWR", + "sim_aircraft": "A20N", + "departure_local_date": "2026-11-08", + "departure_utc": "2026-11-08T12:00:00Z", + "private_callsign": "N320HD", + "units": "KGS", + "passengers": 4, + "airframe_id": "A20N" + } + ], + "expected_outcome": "cancelled", + "max_clarifications": 0, + "prompt": "Prepare this exact private itinerary. Use the approval prompt for my decision, then generate and retain each briefing. [{\"trip\": \"WF-032\", \"leg\": \"1\", \"revision\": 1, \"origin\": \"KIAD\", \"destination\": \"KEWR\", \"sim_aircraft\": \"A20N\", \"departure_local_date\": \"2026-11-08\", \"departure_utc\": \"2026-11-08T12:00:00Z\", \"private_callsign\": \"N320HD\", \"units\": \"KGS\", \"passengers\": 4, \"airframe_id\": \"A20N\"}]", + "control": { + "cancel_at": "post_click" + }, + "provider_failure": "timeout_after_click", + "expected_clicks": 1, + "expected_snapshots": 0, + "script": [ + { + "tool": "dispatch_flight", + "args": { + "trip": "WF-032", + "leg": "1", + "revision": 1, + "origin": "KIAD", + "destination": "KEWR", + "sim_aircraft": "A20N", + "departure_local_date": "2026-11-08", + "departure_utc": "2026-11-08T12:00:00Z", + "private_callsign": "N320HD", + "units": "KGS", + "passengers": 4, + "airframe_id": "A20N" + } + }, + { + "tool": "generate_in_simbrief", + "args": { + "trip": "WF-032", + "leg": "1", + "revision": 1 + } + }, + { + "tool": "generate_in_simbrief", + "args": { + "trip": "WF-032", + "leg": "1", + "revision": 1 + } + }, + { + "text": "unreachable after cancellation" + } + ], + "tool_assertions": [ + { + "id": "uncertain_generation_reported", + "tool": "generate_in_simbrief", + "path": [ + "status" + ], + "equals": "uncertain" + } + ], + "mcp_assertions": [ + { + "id": "turn_stopped_by_cancel", + "tool": "approve_dispatch", + "path": [ + "status" + ], + "equals": "cancelled" + }, + { + "id": "progress_reports_uncertain", + "tool": "trip_progress", + "path": [ + "uncertain_generations", + 0, + "stage" + ], + "equals": "generation_uncertain" + }, + { + "id": "uncertain_revision_not_withdrawn", + "tool": "cancel_trip", + "path": [ + "withdrawn_revisions" + ], + "equals": [] + }, + { + "id": "progress_click_recorded", + "tool": "trip_progress", + "path": [ + "trips", + 0, + "legs", + 0, + "generation", + "clicked" + ], + "equals": true + } + ] + }, + { + "id": "WF-033", + "scenario": "trip_control", + "legs": [ + { + "trip": "WF-033", + "leg": "1", + "revision": 1, + "origin": "KIAD", + "destination": "KEWR", + "sim_aircraft": "A20N", + "departure_local_date": "2026-11-08", + "departure_utc": "2026-11-08T12:00:00Z", + "private_callsign": "N320HD", + "units": "KGS", + "passengers": 4, + "airframe_id": "A20N" + } + ], + "expected_outcome": "cancelled", + "max_clarifications": 0, + "prompt": "Prepare this exact private itinerary. Use the approval prompt for my decision, then generate and retain each briefing. [{\"trip\": \"WF-033\", \"leg\": \"1\", \"revision\": 1, \"origin\": \"KIAD\", \"destination\": \"KEWR\", \"sim_aircraft\": \"A20N\", \"departure_local_date\": \"2026-11-08\", \"departure_utc\": \"2026-11-08T12:00:00Z\", \"private_callsign\": \"N320HD\", \"units\": \"KGS\", \"passengers\": 4, \"airframe_id\": \"A20N\"}]", + "control": { + "cancel_at": "deadline_pre_click", + "deadline_seconds": 2 + }, + "expected_clicks": 0, + "expected_snapshots": 0, + "script": [ + { + "tool": "dispatch_flight", + "args": { + "trip": "WF-033", + "leg": "1", + "revision": 1, + "origin": "KIAD", + "destination": "KEWR", + "sim_aircraft": "A20N", + "departure_local_date": "2026-11-08", + "departure_utc": "2026-11-08T12:00:00Z", + "private_callsign": "N320HD", + "units": "KGS", + "passengers": 4, + "airframe_id": "A20N" + } + }, + { + "tool": "generate_in_simbrief", + "args": { + "trip": "WF-033", + "leg": "1", + "revision": 1 + } + }, + { + "text": "unreachable after the deadline" + } + ], + "mcp_assertions": [ + { + "id": "deadline_stops_turn", + "tool": "approve_dispatch", + "path": [ + "reason" + ], + "equals": "deadline_exceeded" + }, + { + "id": "progress_turn_deadline", + "tool": "trip_progress", + "path": [ + "turn", + "status" + ], + "equals": "deadline_exceeded" + }, + { + "id": "progress_leg_cancelled", + "tool": "trip_progress", + "path": [ + "trips", + 0, + "legs", + 0, + "stage" + ], + "equals": "cancelled" + } + ] + } + ] +} diff --git a/evals/workflow_evaluations.py b/evals/workflow_evaluations.py index d4dfe42..e386e49 100644 --- a/evals/workflow_evaluations.py +++ b/evals/workflow_evaluations.py @@ -11,7 +11,7 @@ ROOT = Path(__file__).resolve().parents[1] MANIFEST = ROOT / "evals/workflow_cases.json" -VERIFIER_VERSION = "ifd-workflow-oracles-v3" +VERIFIER_VERSION = "ifd-workflow-oracles-v4" EVIDENCE_MODES = { "scripted-controlled", "real-model-replay", @@ -34,6 +34,7 @@ def load_manifest(path=MANIFEST) -> dict: if manifest.get("schema_version") != 1 or manifest.get("verifier_version") not in { "ifd-workflow-oracles-v1", "ifd-workflow-oracles-v2", + "ifd-workflow-oracles-v3", VERIFIER_VERSION, }: raise ValueError("unsupported workflow manifest/verifier version") @@ -126,6 +127,21 @@ def same_input(field, left, right, aircraft): return same_value(field, left, right) +_MISSING = object() + + +def at_path(value, path): + """Follow dict keys and list indices; a missing step yields a sentinel, never a match.""" + for part in path: + if isinstance(value, dict) and part in value: + value = value[part] + elif isinstance(value, list) and isinstance(part, int) and -len(value) <= part < len(value): + value = value[part] + else: + return _MISSING + return value + + def verify(case: dict, run: dict) -> dict: """Check observed end state, not a prescribed model tool sequence or a blended score.""" state = run.get("state", []) @@ -239,7 +255,10 @@ def verify(case: dict, run: dict) -> dict: if case["scenario"] == "revision_recovery": # A recovered refusal is a blocked latest tool step while the saved plan stays usable. checks["completion"] = bool(snapshots) and bool(run.get("final_reply")) - if declined: + if case["expected_outcome"] == "cancelled": + # Cancellation ends a turn at a safe point without a model reply. + checks["completion"] = bool(run.get("cancelled")) and not run.get("error") + elif declined: checks["completion"] = not actual and not clicks and bool(run.get("final_reply")) elif blocked or case.get("no_dispatch"): checks["completion"] = bool(run.get("final_reply")) and not snapshots @@ -333,6 +352,14 @@ def matches(output): return output == assertion["equals"] checks[assertion["id"]] = any(matches(o) for o in outputs) + for assertion in case.get("mcp_assertions", []): + # Production MCP control/progress outputs (cancel_trip, approve_dispatch, trip_progress). + outputs = [ + o["output"] for o in run.get("mcp_outputs", []) if o["tool"] == assertion["tool"] + ] + checks[assertion["id"]] = any( + at_path(o, assertion["path"]) == assertion["equals"] for o in outputs + ) if case.get("unknown_connection"): unknown_results = [ t["output"] @@ -426,7 +453,12 @@ def matches(output): ) and len(run.get("provider_attempts", [])) == 1 ) - if case.get("restart"): + if case.get("restart") and case.get("recovered_status"): + # A cancelled approval must not come back as approvable after reconstruction. + checks["restart_recovery"] = ( + run.get("recovered", {}).get("status") == case["recovered_status"] + ) + elif case.get("restart"): checks["restart_recovery"] = run.get("recovered", {}).get("status") == "saved" and run.get( "recovered", {} ).get("reply") == run.get("final_reply") diff --git a/src/ifdispatcher/agent.py b/src/ifdispatcher/agent.py index dc16404..019e980 100644 --- a/src/ifdispatcher/agent.py +++ b/src/ifdispatcher/agent.py @@ -30,6 +30,7 @@ from ifdispatcher.config import data_dir, export_tracing_settings, setting from ifdispatcher.siwc import chatgpt_model from ifdispatcher.browser_middleware import BrowserBudgetMiddleware +from ifdispatcher.progress import TurnControlMiddleware from ifdispatcher.browser_budget import BrowserLimit INSTRUCTIONS = (Path(__file__).resolve().parents[2] / "instructions.md").read_text() @@ -578,7 +579,11 @@ def build_agent(model=None, checkpointer=None): tools=TOOLS, system_prompt=INSTRUCTIONS, interrupt_on=INTERRUPT_ON, - middleware=[DispatchApproval(interrupt_on=INTERRUPT_ON), BrowserBudgetMiddleware()], + middleware=[ + TurnControlMiddleware(), + DispatchApproval(interrupt_on=INTERRUPT_ON), + BrowserBudgetMiddleware(), + ], checkpointer=checkpointer, name="infinite-flight-dispatcher", ) diff --git a/src/ifdispatcher/browser_budget.py b/src/ifdispatcher/browser_budget.py index 99d574d..6bd608c 100644 --- a/src/ifdispatcher/browser_budget.py +++ b/src/ifdispatcher/browser_budget.py @@ -9,7 +9,7 @@ import time import uuid -from ifdispatcher import observability +from ifdispatcher import observability, progress from ifdispatcher.core import open_store SESSION_SECONDS = 180 @@ -82,6 +82,7 @@ def claim(state, now): "browserbase_admission", queue_wait_s=round(time.monotonic() - started, 3) ) return + progress.check() # a cancelled or expired turn leaves the queue unopened if time.monotonic() - started >= QUEUE_SECONDS: raise BrowserLimit("browser_queue_timeout: no browser opened; try later") time.sleep(0.1) diff --git a/src/ifdispatcher/core.py b/src/ifdispatcher/core.py index 72f0975..62d6121 100644 --- a/src/ifdispatcher/core.py +++ b/src/ifdispatcher/core.py @@ -253,10 +253,15 @@ def claim_generation(self, trip: str, leg: str, revision: int) -> dict | None: ).fetchone() meta = json.loads(row[0]) if row else {} previous = meta.get("generation") or {} - if previous.get("at") or previous.get("status") in ( - "generated", - "uncertain", - "clicking", + if ( + meta.get("cancelled") + or previous.get("at") + or previous.get("status") + in ( + "generated", + "uncertain", + "clicking", + ) ): return None meta["generation"] = { diff --git a/src/ifdispatcher/operations.py b/src/ifdispatcher/operations.py index 23931c3..08b91f3 100644 --- a/src/ifdispatcher/operations.py +++ b/src/ifdispatcher/operations.py @@ -4,7 +4,7 @@ import json from datetime import datetime, timedelta, timezone -from ifdispatcher import planning, thrust +from ifdispatcher import planning, progress, thrust from ifdispatcher.core import open_store @@ -96,6 +96,8 @@ def saved_trips(trip: str | None = None) -> dict: else "plan_retained" if snap else (meta.get("generation") or {}).get("status", "prepared"), + "stage": progress.leg_stage(meta, bool(snap)), + "cancelled": meta.get("cancelled"), "briefing_markdown": meta.get("briefing"), "task_outcome": meta.get("task_outcome"), "snapshot_sha256": snap["raw_sha256"] if snap else None, diff --git a/src/ifdispatcher/progress.py b/src/ifdispatcher/progress.py new file mode 100644 index 0000000..b50b6fb --- /dev/null +++ b/src/ifdispatcher/progress.py @@ -0,0 +1,389 @@ +"""Durable trip progress and cooperative turn cancellation/deadlines (INT-416/417). + +Each conversation thread has one turn record in the existing `settings` table, so it works on +SQLite and Postgres without a schema change. Leg progress is derived from the saved requests, +leg metadata and snapshots that the workflow already writes; pending approval comes from the +agent checkpoint. Nothing here calls a model or provider. + +Cancellation is cooperative: a running turn stops at its next model or tool boundary, while +waiting for browser admission, or immediately before the Generate click. A click already in +flight is never interrupted or repeated; its outcome is saved as generated or uncertain. +""" + +import json +import threading +import time +import uuid +from contextvars import ContextVar +from datetime import datetime, timedelta, timezone + +from langchain.agents.middleware import AgentMiddleware + +from ifdispatcher import core +from ifdispatcher.config import setting + +BOOT_ID = uuid.uuid4().hex # identifies this server process; other values mean a restart +PREFIX = "turn_progress:" +LIVE = {"queued", "running"} +# A Generate claim older than one bounded browser workflow cannot still be in progress. +GENERATION_SETTLE_SECONDS = 180 + 8 * 3 + 12 + + +def turn_deadline_seconds() -> float: + """Wall-clock budget for one plan_flight/approve_dispatch turn (operational setting).""" + return float(setting("IFD_TURN_DEADLINE_SECONDS") or 600) + + +class TurnCancelled(Exception): + """Stops the current turn at a safe boundary. Not a ValueError, so tools do not absorb it.""" + + def __init__(self, reason: str): + super().__init__(reason) + self.reason = reason + + +class TurnControl: + def __init__(self, thread_id: str, deadline_seconds: float): + self.thread_id = thread_id + self.deadline = time.monotonic() + deadline_seconds + self.event = threading.Event() + self.reason = None + + def cancel(self, reason: str): + self.reason = self.reason or reason + self.event.set() + + def check(self): + if self.event.is_set(): + raise TurnCancelled(self.reason or "cancelled") + if time.monotonic() >= self.deadline: + self.cancel("deadline_exceeded") + raise TurnCancelled("deadline_exceeded") + + +_control: ContextVar[TurnControl | None] = ContextVar("ifd_turn_control", default=None) + + +def check() -> None: + """Raise TurnCancelled if this turn was cancelled or passed its deadline; else no-op.""" + control = _control.get() + if control is not None: + control.check() + + +def bind(control: TurnControl): + return _control.set(control) + + +def unbind(token) -> None: + _control.reset(token) + + +class TurnControlMiddleware(AgentMiddleware): + """Check cancellation/deadline before every model call and tool call of a turn.""" + + def before_model(self, state, runtime): + check() + + async def abefore_model(self, state, runtime): + check() + + def wrap_tool_call(self, request, handler): + check() + return handler(request) + + async def awrap_tool_call(self, request, handler): + check() + return await handler(request) + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def update(thread_id: str, change) -> dict: + """Atomically read-modify-write a thread's turn record; returns the new record.""" + with core.open_store().connect() as db: + db.execute("BEGIN IMMEDIATE") + row = db.execute( + "SELECT value FROM settings WHERE name=?", (PREFIX + thread_id,) + ).fetchone() + record = json.loads(row[0]) if row else {"thread_id": thread_id, "approvals": []} + change(record) + record["updated_at"] = _now() + db.execute( + "INSERT INTO settings (name,value) VALUES (?,?) " + "ON CONFLICT (name) DO UPDATE SET value=excluded.value", + (PREFIX + thread_id, json.dumps(record)), + ) + return record + + +def read(thread_id: str) -> dict | None: + raw = core.open_store().read_setting(PREFIX + thread_id) + return json.loads(raw) if raw else None + + +def start_turn(thread_id: str, action: str) -> dict: + deadline = turn_deadline_seconds() + + def change(record): + record.update( + turn_id=uuid.uuid4().hex, + action=action, + status="queued", + boot_id=BOOT_ID, + started_at=_now(), + deadline_seconds=deadline, + deadline_at=(datetime.now(timezone.utc) + timedelta(seconds=deadline)).isoformat(), + cancel=None, + error=None, + ) + + return update(thread_id, change) + + +def set_status(thread_id: str, status: str, **fields) -> dict: + return update(thread_id, lambda record: record.update(status=status, **fields)) + + +def finish_turn( + thread_id: str, + status: str, + active_seconds: float, + cancel_outcome: str | None = None, + cancelled_interrupt: str | None = None, + **fields, +) -> dict: + """Record a turn's terminal status and its active execution time. + + Only time inside a turn counts; waiting for a human approval happens between turns, so it + is excluded from `active_seconds_total` by construction. + """ + + def change(record): + record.update(status=status, finished_at=_now(), **fields) + record["active_seconds"] = round(active_seconds, 3) + record["active_seconds_total"] = round( + (record.get("active_seconds_total") or 0) + active_seconds, 3 + ) + if cancel_outcome and record.get("cancel"): + # requested -> stopped (at a safe point) or completed (finished before a boundary) + record["cancel"]["outcome"] = cancel_outcome + record["cancel"]["confirmed_at"] = _now() + if cancelled_interrupt: + record["cancelled_interrupt"] = cancelled_interrupt + + return update(thread_id, change) + + +def turn_view(record: dict | None, live: bool) -> dict | None: + """The persisted turn, with work left behind by a restart reported as interrupted.""" + if record is None: + return None + view = { + k: record.get(k) + for k in ( + "turn_id", + "action", + "status", + "started_at", + "deadline_at", + "finished_at", + "updated_at", + "active_seconds", + "active_seconds_total", + "cancel", + "error", + ) + } + if record.get("status") in LIVE and not live: + view["status"] = "interrupted" + view["interrupted_reason"] = ( + "server_restarted" if record.get("boot_id") != BOOT_ID else "turn_not_running" + ) + return view + + +def leg_stage(meta: dict, snapshot: bool, turn_live: bool = True) -> str: + """One leg revision's durable stage, derived only from saved records.""" + generation = meta.get("generation") or {} + status = generation.get("status") + if meta.get("briefing"): + return "infeasible" if meta.get("task_outcome") == "infeasible" else "briefing_ready" + if snapshot: + return "plan_retained" + if status == "generated": + return "generated_awaiting_finalize" + if status == "uncertain": + return "generation_uncertain" + if status == "clicking": + try: + age = ( + datetime.now(timezone.utc) - datetime.fromisoformat(generation["at"]) + ).total_seconds() + except (KeyError, TypeError, ValueError): + age = GENERATION_SETTLE_SECONDS + return ( + "generating" + if turn_live and age < GENERATION_SETTLE_SECONDS + else ("generation_uncertain") + ) + if meta.get("cancelled") or status == "cancelled": + return "cancelled" + if status: + return "generation_failed" + return "awaiting_generation" + + +COMPLETED = {"briefing_ready", "infeasible", "plan_retained"} + + +def trips_in(state) -> list[str]: + """Trip ids this conversation dispatched or is asking to dispatch, in first-use order.""" + found = [] + values = getattr(state, "values", None) or {} + calls = [c for m in values.get("messages", []) for c in (getattr(m, "tool_calls", None) or [])] + for interrupt in getattr(state, "interrupts", None) or (): + calls += interrupt.value.get("action_requests", []) + for call in calls: + trip = (call.get("args") or {}).get("trip") + if isinstance(trip, str) and trip not in found: + found.append(trip) + return found + + +def legs(trip: str, turn_live: bool = True) -> list[dict]: + store = core.open_store() + result = [] + for leg, request, meta in store.trip_legs(trip): + try: + snap = store.read(trip, leg, request["revision"]) + except ValueError: + snap = None + generation = meta.get("generation") or {} + stage = leg_stage(meta, bool(snap), turn_live) + result.append( + { + "trip": trip, + "leg": leg, + "revision": request["revision"], + "stage": stage, + "approval": "approved", # a saved request exists only after exact approval + "completed": stage in COMPLETED, + "uncertain": stage == "generation_uncertain", + "generation": {k: generation.get(k) for k in ("status", "clicked", "at")}, + "cancelled": meta.get("cancelled"), + "snapshot_sha256": snap["raw_sha256"] if snap else None, + } + ) + return result + + +def pending_approval(state, record: dict | None) -> dict | None: + interrupts = getattr(state, "interrupts", None) or () + if not interrupts: + return None + request = interrupts[0].value["action_requests"][0] + interrupt_id = getattr(interrupts[0], "id", None) + cancelled = bool(record and interrupt_id and record.get("cancelled_interrupt") == interrupt_id) + return { + "status": "cancelled" if cancelled else "pending", + "interrupt_id": interrupt_id, + "action": request["name"], + "inputs": request["args"], + } + + +def snapshot(thread_id: str, state, live: bool) -> dict: + """Whole-conversation progress for MCP clients; reads storage and checkpoint only.""" + record = read(thread_id) + turn = turn_view(record, live) + if not getattr(state, "values", None) and record is None: + return {"status": "not_found", "thread_id": thread_id} + turn_live = bool(turn and turn["status"] in LIVE) + trips = [{"trip": trip, "legs": legs(trip, turn_live)} for trip in trips_in(state)] + all_legs = [leg for trip in trips for leg in trip["legs"]] + approval = pending_approval(state, record) + return { + "status": "ok", + "thread_id": thread_id, + "turn": turn, + "approval": {"pending": approval, "history": (record or {}).get("approvals", [])}, + "trips": trips, + "completed_legs": [_ident(leg) for leg in all_legs if leg["completed"]], + "uncertain_generations": [_ident(leg) for leg in all_legs if leg["uncertain"]], + "interrupted": bool(turn and turn["status"] == "interrupted"), + "next": _next(turn, approval, all_legs), + } + + +def _ident(leg: dict) -> dict: + return {k: leg[k] for k in ("trip", "leg", "revision", "stage")} + + +UNCERTAIN_NEXT = ( + "Generate was clicked but completion was not observed. Do not generate this revision again. " + "Check SimBrief; if the plan exists, ask the dispatcher to finalize it, otherwise ask for a " + "new approved revision." +) + + +def _next(turn, approval, all_legs) -> str: + if turn and turn["status"] in LIVE: + return "Work is still running. Call turn_status, or cancel_trip to stop it." + if any(leg["uncertain"] for leg in all_legs): + return UNCERTAIN_NEXT + if approval and approval["status"] == "pending": + return "Show the pending inputs to the user, then call approve_dispatch with their answer." + if any(leg["stage"] == "cancelled" for leg in all_legs): + return ( + "Cancelled revisions were never generated and their approval was withdrawn. To " + "continue one, ask the user and request a new approved revision with plan_flight." + ) + if turn and turn["status"] == "interrupted": + return ( + "The previous turn stopped when the server restarted. Completed legs are kept. Send " + "plan_flight on this thread to continue; do not repeat a generation shown above." + ) + return "Continue with plan_flight on this thread, or read briefings with list_saved_trips." + + +def mark_cancelled_legs(trips: list[str], thread_id: str, reason: str) -> list[dict]: + """Withdraw approval for approved revisions whose Generate was never clicked. + + Completed plans, generated or uncertain revisions are left untouched. A cancelled revision + can only proceed through a new approved revision. + """ + store = core.open_store() + marked = [] + for trip in trips: + for leg, request, _ in store.trip_legs(trip): + revision = request["revision"] + with store.connect() as db: + db.execute("BEGIN IMMEDIATE") + row = db.execute( + "SELECT payload FROM leg_meta WHERE trip=? AND leg=? AND revision=?", + (trip, leg, revision), + ).fetchone() + meta = json.loads(row[0]) if row else {} + snap = db.execute( + "SELECT 1 FROM snapshots WHERE trip=? AND leg=? AND revision=?", + (trip, leg, revision), + ).fetchone() + generation = meta.get("generation") or {} + if ( + snap + or meta.get("briefing") + or meta.get("cancelled") + or generation.get("at") + or generation.get("status") in ("clicking", "generated", "uncertain") + ): + continue + meta["cancelled"] = {"at": _now(), "thread_id": thread_id, "reason": reason} + db.execute( + "INSERT OR REPLACE INTO leg_meta VALUES (?,?,?,?)", + (trip, leg, revision, core.encode(meta)), + ) + marked.append({"trip": trip, "leg": leg, "revision": revision}) + return marked diff --git a/src/ifdispatcher/remote.py b/src/ifdispatcher/remote.py index 19789af..3902c78 100644 --- a/src/ifdispatcher/remote.py +++ b/src/ifdispatcher/remote.py @@ -12,6 +12,7 @@ import asyncio import hmac import os +import time import uuid from concurrent.futures import Future, ThreadPoolExecutor from typing import Any @@ -24,13 +25,14 @@ from ifdispatcher import __version__ from ifdispatcher.config import export_tracing_settings, paid_operations_enabled, setting -from ifdispatcher import observability, operations, simbrief_browser +from ifdispatcher import observability, operations, progress, simbrief_browser TOKEN_HEADER = "x-ifd-internal-token" WAIT_SECONDS = 25 MAX_OUTSTANDING_TURNS = 6 # four workers and at most two waiting turns _agent = None _turns: dict[str, Future] = {} # latest turn per thread_id +_controls: dict[str, progress.TurnControl] = {} # cancellation/deadline of that turn _pool = ThreadPoolExecutor(max_workers=4) @@ -75,7 +77,63 @@ def _turn(result, thread_id: str, seen: set) -> dict[str, Any]: return {"status": "reply", "thread_id": thread_id, "reply": result.value["messages"][-1].text} -def _run(payload, thread_id: str) -> dict[str, Any]: +def _live(thread_id: str) -> bool: + turn = _turns.get(thread_id) + return bool(turn and not turn.done()) + + +def _config(thread_id: str) -> dict: + return {"configurable": {"thread_id": thread_id}} + + +def _run(payload, thread_id: str, control: progress.TurnControl | None = None) -> dict[str, Any]: + control = control or progress.TurnControl(thread_id, progress.turn_deadline_seconds()) + bound = progress.bind(control) + started = time.monotonic() + try: + control.check() # cancelled while queued: start no work + progress.set_status(thread_id, "running") + response = _invoke(payload, thread_id) + if response["status"] == "approval_required" and control.event.is_set(): + # Cancel arrived while the model proposed this dispatch: withdraw it, never offer it. + raise progress.TurnCancelled(control.reason or "cancelled") + progress.finish_turn( + thread_id, + response["status"], + time.monotonic() - started, + cancel_outcome="completed" if control.event.is_set() else None, + ) + return response + except progress.TurnCancelled as exc: + state = _get_agent().get_state(_config(thread_id)) + pending = progress.pending_approval(state, None) + progress.finish_turn( + thread_id, + "deadline_exceeded" if exc.reason == "deadline_exceeded" else "cancelled", + time.monotonic() - started, + cancel_outcome="stopped", + cancelled_interrupt=pending["interrupt_id"] if pending else None, + ) + # A dispatch saved after cancel_trip looked (or before a deadline) needs a new approval. + progress.mark_cancelled_legs(progress.trips_in(state), thread_id, exc.reason) + report = progress.snapshot(thread_id, state, False) + return { + "status": "cancelled", + "thread_id": thread_id, + "reason": exc.reason, + "progress": report, + "next": "Work stopped at a safe point; completed legs are kept. " + report["next"], + } + except Exception as exc: + progress.finish_turn( + thread_id, "failed", time.monotonic() - started, error=type(exc).__name__ + ) + raise + finally: + progress.unbind(bound) + + +def _invoke(payload, thread_id: str) -> dict[str, Any]: if not paid_operations_enabled(): return {"status": "paid_operations_disabled", "thread_id": thread_id} from ifdispatcher.agent import message_ids @@ -118,7 +176,7 @@ async def _wait(thread_id: str) -> dict[str, Any]: } -async def _start(payload, thread_id: str) -> dict[str, Any]: +async def _start(payload, thread_id: str, action: str) -> dict[str, Any]: running = _turns.get(thread_id) if running and not running.done(): return { @@ -132,7 +190,10 @@ async def _start(payload, thread_id: str) -> dict[str, Any]: "thread_id": thread_id, "next": "Dispatcher busy; try later. No work was started.", } - _turns[thread_id] = _pool.submit(_run, payload, thread_id) + record = progress.start_turn(thread_id, action) + control = progress.TurnControl(thread_id, record["deadline_seconds"]) + _controls[thread_id] = control + _turns[thread_id] = _pool.submit(_run, payload, thread_id, control) return await _wait(thread_id) @@ -148,7 +209,7 @@ async def plan_flight(message: str, thread_id: str | None = None) -> dict[str, A """ thread_id = thread_id or f"trip-{uuid.uuid4().hex[:10]}" payload = {"messages": [{"role": "user", "content": message}]} - return await _start(payload, thread_id) + return await _start(payload, thread_id, "plan_flight") @mcp.tool() @@ -160,9 +221,32 @@ async def approve_dispatch( then returns the briefing; approve=false declines with the user's note. Only call after the user explicitly answered. """ + if not _live(thread_id): + state = _get_agent().get_state(_config(thread_id)) + pending = progress.pending_approval(state, progress.read(thread_id)) + if pending and pending["status"] == "cancelled" and approve: + return { + "status": "cancelled", + "thread_id": thread_id, + "pending": pending, + "next": "This approval was cancelled and cannot be approved. Nothing was " + "dispatched or generated. Send plan_flight to start again with a new approval.", + } + if pending: + entry = { + "interrupt_id": pending["interrupt_id"], + "action": pending["action"], + **{ + k: pending["inputs"].get(k) + for k in ("trip", "leg", "revision", "base_revision") + }, + "decision": "approve" if approve else "reject", + "at": progress._now(), + } + progress.update(thread_id, lambda r: r.setdefault("approvals", []).append(entry)) decision = {"type": "approve"} if approve else {"type": "reject", "message": note or "declined"} payload = Command(resume={"decisions": [decision]}) - return await _start(payload, thread_id) + return await _start(payload, thread_id, "approve_dispatch") @mcp.tool() @@ -171,6 +255,15 @@ async def turn_status(thread_id: str) -> dict[str, Any]: status running. Returns running again (call later), or the reply / approval request. """ if thread_id not in _turns: + turn = progress.turn_view(progress.read(thread_id), live=False) + if turn: + return { + "status": turn["status"], + "thread_id": thread_id, + "turn": turn, + "next": "Call trip_progress for completed legs, pending approval and any " + "uncertain generation before continuing with plan_flight on this thread.", + } return { "status": "not_found", "thread_id": thread_id, @@ -180,6 +273,93 @@ async def turn_status(thread_id: str) -> dict[str, Any]: return await _wait(thread_id) +@mcp.tool() +def trip_progress(thread_id: str) -> dict: + """Durable progress of a conversation without invoking the model or any provider. + + Returns the current turn (running, approval_required, reply, cancelled, deadline_exceeded, + failed, or interrupted by a restart), the pending or cancelled approval, and each dispatched + leg's stage: awaiting_generation, generating, generated_awaiting_finalize, + generation_uncertain, generation_failed, cancelled, plan_retained, briefing_ready or + infeasible. Never retry a generation_uncertain leg. + """ + state = _get_agent().get_state(_config(thread_id)) + return progress.snapshot(thread_id, state, _live(thread_id)) + + +@mcp.tool() +async def cancel_trip(thread_id: str, reason: str | None = None) -> dict: + """Stop this conversation's queued or running work and withdraw a pending approval. + + Only call when the user asks to stop. Completed plans and briefings are kept. Approved + revisions not yet generated need a new approval to continue. A Generate click already in + progress cannot be undone or repeated; it is reported as generated or uncertain. + """ + state = _get_agent().get_state(_config(thread_id)) + record = progress.read(thread_id) + if not state.values and record is None: + return {"status": "not_found", "thread_id": thread_id} + reason = reason or "cancelled_by_user" + running = _live(thread_id) + pending = progress.pending_approval(state, record) + + def change(r): + r["cancel"] = { + "requested_at": progress._now(), + "reason": reason, + "while": "running" if running else "approval_pending" if pending else "idle", + } + if pending: + r["cancelled_interrupt"] = pending["interrupt_id"] + if not running: + # Nothing is executing, so the request is confirmed immediately. + r["cancel"].update(outcome="stopped", confirmed_at=r["cancel"]["requested_at"]) + r["status"] = "cancelled" + + # Persist the request before signalling, so the turn's terminal record can confirm it. + progress.update(thread_id, change) + if running: + _controls[thread_id].cancel(reason) + withdrawn = progress.mark_cancelled_legs(progress.trips_in(state), thread_id, reason) + if running: + try: + await asyncio.wait_for( + asyncio.shield(asyncio.wrap_future(_turns[thread_id])), WAIT_SECONDS + ) + except Exception: + pass # timeout: still finishing; a turn's own failure is recorded in its progress + live = _live(thread_id) + state = _get_agent().get_state(_config(thread_id)) + if running and not live: + # The turn may have finished before seeing the signal: confirm the request as + # completed, and still withdraw any approval or unclicked revision it left behind. + late = progress.pending_approval(state, None) + + def settle(r): + if r.get("cancel"): + r["cancel"].setdefault("outcome", "completed") + r["cancel"].setdefault("confirmed_at", progress._now()) + if late: + r["cancelled_interrupt"] = late["interrupt_id"] + + progress.update(thread_id, settle) + pending = pending or late + withdrawn += progress.mark_cancelled_legs(progress.trips_in(state), thread_id, reason) + report = progress.snapshot(thread_id, state, live) + return { + "status": "cancelling" if live else "cancelled", + "thread_id": thread_id, + "stopped_running_turn": running, + # stopped (at a safe point) or completed (finished first); None while still cancelling + "outcome": ((report.get("turn") or {}).get("cancel") or {}).get("outcome"), + "withdrawn_approval": pending if pending else None, + "withdrawn_revisions": withdrawn, + "progress": report, + "next": ("A Generate click may be finishing; call trip_progress shortly. " if live else "") + + report["next"], + } + + @mcp.tool() def browser_status() -> dict: """Inspect browser admission and active provider sessions without opening a browser.""" @@ -201,6 +381,15 @@ def conversation_status(thread_id: str) -> dict: state = _get_agent().get_state({"configurable": {"thread_id": thread_id}}) if not state.values: return {"status": "not_found", "thread_id": thread_id} + pending = progress.pending_approval(state, progress.read(thread_id)) + if pending and pending["status"] == "cancelled": + return { + "status": "approval_cancelled", + "thread_id": thread_id, + "inputs": pending["inputs"], + "next": "The owner cancelled this approval; it cannot be approved. Call trip_progress " + "for kept legs, or send plan_flight on this thread to start again.", + } if state.interrupts: request = state.interrupts[0].value["action_requests"][0] return { diff --git a/src/ifdispatcher/workflow.py b/src/ifdispatcher/workflow.py index 1b694f5..376f97c 100644 --- a/src/ifdispatcher/workflow.py +++ b/src/ifdispatcher/workflow.py @@ -10,7 +10,7 @@ from datetime import datetime, timedelta, timezone from typing import Any -from ifdispatcher import artifacts, flights, planning, simbrief_browser, thrust +from ifdispatcher import artifacts, flights, planning, progress, simbrief_browser, thrust from ifdispatcher.config import data_dir, setting from ifdispatcher.core import Dispatch, Store, fetch_sync, normalize, open_store @@ -319,6 +319,19 @@ class GenerationClaimed(ValueError): """Another execution already owns this leg revision's Generate click.""" +def _cancelled_result(meta: dict) -> dict[str, Any]: + return { + "status": "cancelled", + "reason": "This approved revision was cancelled before its Generate click; its approval " + "no longer authorizes generation.", + "next": "Ask the user, then request a new approved revision with revise_dispatch to continue.", + "cancelled": meta.get("cancelled"), + "clicked": False, + "retryable": False, + "automatic_retry_allowed": False, + } + + def generate_plan(trip: str, leg: str, revision: int = 1, click: bool = True) -> dict[str, Any]: """Fill SimBrief's dispatch form in the logged-in cloud browser and (if click) Generate once.""" store = _store() @@ -327,6 +340,8 @@ def generate_plan(trip: str, leg: str, revision: int = 1, click: bool = True) -> return {"status": "missing_dispatch", "reason": "call dispatch_flight for this leg first"} meta = store.read_meta(trip, leg, revision) previous = (meta.get("generation") or {}).get("status") + if click and meta.get("cancelled"): + return _cancelled_result(meta) if ( click and previous in ("clicking", "generated", "uncertain", "error", "fill_mismatch") @@ -362,8 +377,11 @@ def generate_plan(trip: str, leg: str, revision: int = 1, click: bool = True) -> def mark_clicking() -> None: # just before the click: earlier failures leave the leg retryable nonlocal meta + progress.check() # the last safe point to honor cancellation or the turn deadline claimed = store.claim_generation(trip, leg, revision) if claimed is None: + if store.read_meta(trip, leg, revision).get("cancelled"): + raise progress.TurnCancelled("cancelled") raise GenerationClaimed("Generation already claimed for this revision") meta = claimed @@ -378,6 +396,27 @@ def mark_clicking() -> None: # just before the click: earlier failures leave th "retryable": False, "reason": "Another execution already claimed Generate for this revision", } + except progress.TurnCancelled as exc: + # Raised only before the click (browser admission or the pre-click check). + meta = store.read_meta(trip, leg, revision) + meta["generation"] = { + **meta.get("generation", {}), + "status": "cancelled", + "clicked": False, + "retryable": False, + "automatic_retry_allowed": False, + "reason_code": exc.reason, + } + meta.setdefault( + "cancelled", + { + "at": datetime.now(timezone.utc).isoformat(), + "reason": exc.reason, + "before_click": True, + }, + ) + store.save_meta(trip, leg, revision, meta) + raise except Exception as exc: clicked = (meta.get("generation") or {}).get("status") == "clicking" result = { diff --git a/tests/conftest.py b/tests/conftest.py index fc405ca..ee40bf2 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -53,3 +53,5 @@ async def handle_async(self, request): monkeypatch.setattr(httpx.HTTPTransport, "handle_request", handle) monkeypatch.setattr(httpx.AsyncHTTPTransport, "handle_async_request", handle_async) monkeypatch.setenv("IFD_AIRPORT_CACHE_DIR", str(tmp_path_factory.mktemp("airports"))) + # Turn progress and other Store writes never land in the checkout's data directory. + monkeypatch.setenv("IFD_DATA_DIR", str(tmp_path_factory.mktemp("data"))) diff --git a/tests/test_release_workflow_evaluations.py b/tests/test_release_workflow_evaluations.py index 10c1d50..b116266 100644 --- a/tests/test_release_workflow_evaluations.py +++ b/tests/test_release_workflow_evaluations.py @@ -23,7 +23,19 @@ def release_manifest(): @pytest.fixture(scope="module") def mcp_batch(tmp_path_factory): - directory = tmp_path_factory.mktemp("release-mcp") + return run_mcp_batch( + tmp_path_factory.mktemp("release-mcp"), ROOT / "evals/workflow_cases_v044.json" + ) + + +@pytest.fixture(scope="module") +def trip_control_batch(tmp_path_factory): + return run_mcp_batch( + tmp_path_factory.mktemp("trip-control-mcp"), ROOT / "evals/workflow_cases_v050.json" + ) + + +def run_mcp_batch(directory, manifest): production = directory / "production" production.mkdir() marker = production / "do-not-change" @@ -68,7 +80,7 @@ def mcp_batch(tmp_path_factory): str(ROOT / "evals/run_workflow_replay.py"), "run", "--manifest", - str(ROOT / "evals/workflow_cases_v044.json"), + str(manifest), "--mcp", "--output", str(output), @@ -114,6 +126,29 @@ def test_release_matrix_preserves_original_cases_and_completes_over_mcp( assert all(row["snapshot"] and row["meta"]["briefing"] for row in revised) +def test_trip_control_suite_over_mcp(trip_control_batch): + """INT-416/417 scenarios: cancel at approval/before/after the click, deadline, restart.""" + manifest = load_manifest(ROOT / "evals/workflow_cases_v050.json") + assert trip_control_batch["case_ids"] == [c["id"] for c in manifest["cases"]] + runs = {row["case_id"]: row for row in trip_control_batch["results"]} + for row in runs.values(): + assert row["passed"] and not row["invalid"], (row["case_id"], row["constraint_failures"]) + assert row["metrics"]["transport"] == "mcp-http" + assert row["metrics"]["flightaware_requests"] == 0 + assert row["metrics"]["browserbase_sessions"] == 0 + assert row["cancelled"] is True + # Completed work survives cancellation; uncertain generation is reported, never repeated. + assert runs["WF-030"]["state"][0]["meta"]["briefing"].startswith("**SPAWN:**") + assert len(runs["WF-032"]["provider_attempts"]) == 1 + assert runs["WF-031"]["clicks"] == [] and runs["WF-033"]["clicks"] == [] + + +def test_trip_control_cases_are_invalid_without_mcp(): + manifest = load_manifest(ROOT / "evals/workflow_cases_v050.json") + run = run_case(manifest["cases"][0], manifest) + assert run["invalid"] and "requires --mcp" in run["invalid_reasons"][0] + + @pytest.mark.parametrize( "corruption,failed", [ diff --git a/tests/test_remote.py b/tests/test_remote.py index b34ac27..6dfcd6b 100644 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -56,9 +56,11 @@ async def list_tools(): assert asyncio.run(list_tools()) == [ "approve_dispatch", "browser_status", + "cancel_trip", "conversation_status", "list_saved_trips", "plan_flight", + "trip_progress", "turn_status", ] @@ -66,7 +68,7 @@ async def list_tools(): def test_long_turn_returns_running_then_result(monkeypatch): release = threading.Event() - def slow_run(payload, thread_id): + def slow_run(payload, thread_id, control=None): release.wait(5) return {"status": "reply", "thread_id": thread_id, "reply": payload} diff --git a/tests/test_trip_progress.py b/tests/test_trip_progress.py new file mode 100644 index 0000000..2c78165 --- /dev/null +++ b/tests/test_trip_progress.py @@ -0,0 +1,382 @@ +"""INT-416/417: durable trip progress, cancellation and deadlines (no model/provider calls).""" + +import asyncio +import json +import time +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + +import pytest + +from ifdispatcher import browser_budget, core, progress, remote, workflow +from ifdispatcher.core import Dispatch + +DEPARTURE = "2026-11-08T12:00:00Z" + + +@pytest.fixture +def store(): + return core.open_store() + + +def prepare(store, trip="T1", leg="1", revision=1): + store.prepare( + Dispatch( + trip=trip, + leg=leg, + revision=revision, + origin="KIAD", + destination="KEWR", + aircraft="A20N", + departure=DEPARTURE, + units="KGS", + passengers=4, + ) + ) + store.save_meta(trip, leg, revision, {"prepared_at": DEPARTURE}) + + +@pytest.fixture +def bound(): + """A turn control bound to the current context, as remote._run does.""" + control = progress.TurnControl("thread-1", 60) + token = progress.bind(control) + yield control + progress.unbind(token) + + +def test_browser_limits_unchanged(): + assert ( + browser_budget.SESSION_SECONDS, + browser_budget.IDLE_SECONDS, + browser_budget.QUEUE_SECONDS, + browser_budget.MAX_WAITERS, + ) == (180, 30, 30, 2) + + +@pytest.mark.parametrize( + "meta,snapshot,live,stage", + [ + ({}, False, True, "awaiting_generation"), + ({"generation": {"status": "fill_mismatch"}}, False, True, "generation_failed"), + ({"cancelled": {"reason": "x"}}, False, True, "cancelled"), + ({"generation": {"status": "cancelled"}, "cancelled": {}}, False, True, "cancelled"), + ({"generation": {"status": "generated"}}, False, True, "generated_awaiting_finalize"), + ({"generation": {"status": "uncertain"}}, False, True, "generation_uncertain"), + ({"generation": {"status": "generated"}}, True, True, "plan_retained"), + ({"briefing": "**SPAWN:**"}, True, True, "briefing_ready"), + ({"briefing": "x", "task_outcome": "infeasible"}, True, True, "infeasible"), + ], +) +def test_leg_stage_from_saved_records(meta, snapshot, live, stage): + assert progress.leg_stage(meta, snapshot, live) == stage + + +def test_click_in_progress_becomes_uncertain_when_not_live_or_too_old(): + now = datetime.now(timezone.utc) + recent = {"generation": {"status": "clicking", "at": now.isoformat()}} + old = {"generation": {"status": "clicking", "at": (now - timedelta(minutes=10)).isoformat()}} + assert progress.leg_stage(recent, False, turn_live=True) == "generating" + assert progress.leg_stage(recent, False, turn_live=False) == "generation_uncertain" + assert progress.leg_stage(old, False, turn_live=True) == "generation_uncertain" + + +def test_check_is_noop_without_a_turn_and_honors_cancel_and_deadline(): + progress.check() + control = progress.TurnControl("t", 60) + token = progress.bind(control) + try: + progress.check() + control.cancel("owner") + with pytest.raises(progress.TurnCancelled, match="owner"): + progress.check() + finally: + progress.unbind(token) + expired = progress.TurnControl("t", 0) + with pytest.raises(progress.TurnCancelled, match="deadline_exceeded"): + expired.check() + + +def test_middleware_stops_before_model_and_tool(bound): + middleware = progress.TurnControlMiddleware() + called = [] + assert middleware.wrap_tool_call("request", lambda r: called.append(r) or "ok") == "ok" + bound.cancel("owner") + with pytest.raises(progress.TurnCancelled): + middleware.before_model({}, None) + with pytest.raises(progress.TurnCancelled): + middleware.wrap_tool_call("request", lambda r: called.append(r)) + assert called == ["request"] + + +def generation_stub(monkeypatch, clicks, after_click=None): + def generate(fields, click, shots_dir, before_click=None): + before_click() + clicks.append(fields["tag"]) + if after_click: + after_click() + return {"status": "generated", "session": "controlled"} + + monkeypatch.setattr(workflow.simbrief_browser, "generate", generate) + monkeypatch.setattr(workflow.artifacts, "upload", lambda *a, **k: []) + + +def test_cancel_before_click_never_clicks_and_withdraws_revision(store, bound, monkeypatch): + prepare(store) + clicks = [] + generation_stub(monkeypatch, clicks) + bound.cancel("owner cancelled") + with pytest.raises(progress.TurnCancelled): + workflow.generate_plan("T1", "1", 1) + assert clicks == [] + meta = store.read_meta("T1", "1", 1) + assert meta["generation"]["status"] == "cancelled" + assert meta["generation"]["clicked"] is False + assert meta["cancelled"]["before_click"] is True + # A later turn under the old approval is refused, and the store fence agrees. + token = progress.bind(progress.TurnControl("later", 60)) + try: + refused = workflow.generate_plan("T1", "1", 1) + finally: + progress.unbind(token) + assert refused["status"] == "cancelled" and refused["automatic_retry_allowed"] is False + assert store.claim_generation("T1", "1", 1) is None + assert clicks == [] + + +def test_cancel_after_click_keeps_the_provider_result(store, bound, monkeypatch): + prepare(store) + clicks = [] + generation_stub(monkeypatch, clicks, after_click=lambda: bound.cancel("owner")) + result = workflow.generate_plan("T1", "1", 1) + assert result["status"] == "generated" and result["clicked"] is True + assert clicks == ["IFD_T1_1_R1"] + assert store.read_meta("T1", "1", 1)["generation"]["status"] == "generated" + + +def test_browser_queue_wait_honors_cancellation(bound): + other = browser_budget.Lease() + other.acquire() + try: + bound.cancel("owner") + started = time.monotonic() + with pytest.raises(progress.TurnCancelled): + browser_budget.Lease().acquire() + assert time.monotonic() - started < 2 + assert browser_budget.admission_status()["waiting"] == 0 + finally: + other.finish(True) + + +def test_mark_cancelled_legs_keeps_completed_clicked_and_uncertain_work(store): + for leg in "12345": + prepare(store, leg=leg) + store.save_meta("T1", "2", 1, {"briefing": "**SPAWN:**"}) + store.save_meta("T1", "3", 1, {"generation": {"status": "uncertain", "at": DEPARTURE}}) + store.save_meta("T1", "4", 1, {"generation": {"status": "generated", "at": DEPARTURE}}) + store.save_meta("T1", "5", 1, {"generation": {"status": "fill_mismatch", "retryable": True}}) + marked = progress.mark_cancelled_legs(["T1"], "thread-1", "owner") + assert [m["leg"] for m in marked] == ["1", "5"] + stages = {leg["leg"]: leg["stage"] for leg in progress.legs("T1")} + assert stages == { + "1": "cancelled", + "2": "briefing_ready", + "3": "generation_uncertain", + "4": "generated_awaiting_finalize", + "5": "cancelled", + } + assert progress.mark_cancelled_legs(["T1"], "thread-1", "owner") == [] + + +def test_turn_record_persists_and_restart_is_reported_as_interrupted(monkeypatch): + record = progress.start_turn("thread-r", "approve_dispatch") + progress.set_status("thread-r", "running") + assert progress.turn_view(progress.read("thread-r"), live=True)["status"] == "running" + # A different server process (restart) cannot be running this turn. + monkeypatch.setattr(progress, "BOOT_ID", "replacement-process") + view = progress.turn_view(progress.read("thread-r"), live=False) + assert view["status"] == "interrupted" + assert view["interrupted_reason"] == "server_restarted" + assert view["turn_id"] == record["turn_id"] + monkeypatch.setattr(remote, "_turns", {}) + status = asyncio.run(remote.turn_status("thread-r")) + assert status["status"] == "interrupted" + assert "trip_progress" in status["next"] + + +class NoModel: + """Checkpoint reader that fails the test if a turn would run the model.""" + + def __init__(self, state): + self.state = state + + def get_state(self, config): + return self.state + + def invoke(self, *args, **kwargs): + raise AssertionError("cancelled work must not invoke the model") + + +def empty_state(): + return SimpleNamespace(values={}, interrupts=()) + + +def test_cancelled_queued_turn_starts_no_work(monkeypatch): + monkeypatch.setattr(remote, "_get_agent", lambda: NoModel(empty_state())) + progress.start_turn("thread-q", "plan_flight") + control = progress.TurnControl("thread-q", 60) + control.cancel("owner") + result = remote._run({"messages": []}, "thread-q", control) + assert result["status"] == "cancelled" + assert progress.read("thread-q")["status"] == "cancelled" + + +def test_turn_deadline_is_recorded(monkeypatch): + monkeypatch.setattr(remote, "_get_agent", lambda: NoModel(empty_state())) + monkeypatch.setenv("IFD_TURN_DEADLINE_SECONDS", "0") + record = progress.start_turn("thread-d", "plan_flight") + assert record["deadline_seconds"] == 0 + result = remote._run({"messages": []}, "thread-d", progress.TurnControl("thread-d", 0)) + assert result["reason"] == "deadline_exceeded" + assert progress.read("thread-d")["status"] == "deadline_exceeded" + + +def pending_state(store): + interrupt = SimpleNamespace( + id="interrupt-1", + value={ + "action_requests": [ + {"name": "dispatch_flight", "args": {"trip": "T1", "leg": "2", "revision": 1}} + ] + }, + ) + message = SimpleNamespace( + tool_calls=[{"name": "dispatch_flight", "args": {"trip": "T1", "leg": "1"}}] + ) + return SimpleNamespace(values={"messages": [message]}, interrupts=(interrupt,)) + + +def test_cancel_pending_approval_keeps_completed_legs_and_refuses_approval(store, monkeypatch): + prepare(store, leg="1") + store.save_meta("T1", "1", 1, {"briefing": "**SPAWN:** gate"}) + state = pending_state(store) + monkeypatch.setattr(remote, "_get_agent", lambda: NoModel(state)) + monkeypatch.setattr(remote, "_turns", {}) + + cancelled = asyncio.run(remote.cancel_trip("thread-a")) + assert cancelled["status"] == "cancelled" + assert cancelled["withdrawn_approval"]["inputs"]["leg"] == "2" + assert cancelled["withdrawn_revisions"] == [] + assert cancelled["progress"]["completed_legs"] == [ + {"trip": "T1", "leg": "1", "revision": 1, "stage": "briefing_ready"} + ] + + refused = asyncio.run(remote.approve_dispatch("thread-a", approve=True)) + assert refused["status"] == "cancelled" + report = remote.trip_progress("thread-a") + assert report["approval"]["pending"]["status"] == "cancelled" + assert report["turn"]["status"] == "cancelled" + assert remote.conversation_status("thread-a")["status"] == "approval_cancelled" + assert store.read_meta("T1", "1", 1)["briefing"] == "**SPAWN:** gate" + + +def test_trip_progress_not_found_and_records_approval_history(store, monkeypatch): + monkeypatch.setattr(remote, "_get_agent", lambda: NoModel(empty_state())) + assert remote.trip_progress("missing")["status"] == "not_found" + state = pending_state(store) + monkeypatch.setattr(remote, "_get_agent", lambda: NoModel(state)) + monkeypatch.setattr(remote, "_turns", {}) + + async def answer(): + async def started(payload, thread_id, action): + return {"status": "reply", "thread_id": thread_id} + + monkeypatch.setattr(remote, "_start", started) + return await remote.approve_dispatch("thread-h", approve=False, note="no") + + assert asyncio.run(answer())["status"] == "reply" + history = progress.read("thread-h")["approvals"] + assert [(h["leg"], h["decision"]) for h in history] == [("2", "reject")] + assert json.dumps(history) # durable JSON record + + +def test_cancel_during_proposal_withdraws_the_new_approval_and_unclicked_dispatch( + store, monkeypatch +): + """Cancel lands while the model is proposing: no approval is offered, nothing generates.""" + state = pending_state(store) + monkeypatch.setattr(remote, "_get_agent", lambda: NoModel(state)) + progress.start_turn("thread-p", "plan_flight") + progress.update("thread-p", lambda r: r.update(cancel={"reason": "owner"})) + control = progress.TurnControl("thread-p", 60) + + def invoke(payload, thread_id): + prepare(store, leg="1") # an approved dispatch saved just before the signal + control.cancel("owner") + return {"status": "approval_required", "thread_id": thread_id} + + monkeypatch.setattr(remote, "_invoke", invoke) + result = remote._run({"messages": []}, "thread-p", control) + assert result["status"] == "cancelled" + record = progress.read("thread-p") + assert record["status"] == "cancelled" + assert record["cancel"]["outcome"] == "stopped" + assert record["cancelled_interrupt"] == "interrupt-1" + assert progress.pending_approval(state, record)["status"] == "cancelled" + assert store.read_meta("T1", "1", 1)["cancelled"]["reason"] == "owner" + assert store.claim_generation("T1", "1", 1) is None + + +def test_turn_that_finishes_before_seeing_cancel_is_confirmed_completed(monkeypatch): + monkeypatch.setattr(remote, "_get_agent", lambda: NoModel(empty_state())) + progress.start_turn("thread-c", "plan_flight") + progress.update("thread-c", lambda r: r.update(cancel={"reason": "owner"})) + control = progress.TurnControl("thread-c", 60) + + def invoke(payload, thread_id): + control.cancel("owner") # after the last boundary: the reply already exists + return {"status": "reply", "thread_id": thread_id, "reply": "done"} + + monkeypatch.setattr(remote, "_invoke", invoke) + assert remote._run({"messages": []}, "thread-c", control)["status"] == "reply" + record = progress.read("thread-c") + assert record["status"] == "reply" and record["cancel"]["outcome"] == "completed" + + +def test_stalled_call_is_stopped_at_the_next_boundary_by_the_deadline(monkeypatch): + """Threads are not killed: a call stalled past the deadline stops at the next boundary.""" + monkeypatch.setattr(remote, "_get_agent", lambda: NoModel(empty_state())) + progress.start_turn("thread-s", "plan_flight") + middleware = progress.TurnControlMiddleware() + after_stall = [] + + def invoke(payload, thread_id): + time.sleep(0.3) # a model or provider call that outlives the turn deadline + middleware.before_model({}, None) + after_stall.append("next model call") + return {"status": "reply", "thread_id": thread_id, "reply": "late"} + + monkeypatch.setattr(remote, "_invoke", invoke) + result = remote._run({"messages": []}, "thread-s", progress.TurnControl("thread-s", 0.2)) + assert result["reason"] == "deadline_exceeded" and after_stall == [] + record = progress.read("thread-s") + assert record["status"] == "deadline_exceeded" + assert record["active_seconds"] >= 0.3 + + +def test_active_time_counts_turns_only_not_the_approval_wait(monkeypatch): + monkeypatch.setattr(remote, "_get_agent", lambda: NoModel(empty_state())) + + def invoke(payload, thread_id): + time.sleep(0.1) + return {"status": "reply", "thread_id": thread_id, "reply": "ok"} + + monkeypatch.setattr(remote, "_invoke", invoke) + for _ in range(2): + progress.start_turn("thread-t", "plan_flight") + remote._run({"messages": []}, "thread-t", progress.TurnControl("thread-t", 60)) + time.sleep(0.4) # the owner deciding between turns + record = progress.read("thread-t") + assert 0.2 <= record["active_seconds_total"] < 0.6 + view = progress.turn_view(record, live=False) + assert view["finished_at"] and view["active_seconds"] >= 0.1