From 24c6707780fa59157c9b03bd71518517b2925745 Mon Sep 17 00:00:00 2001 From: Toby Martin Date: Fri, 21 Aug 2026 10:34:54 +1000 Subject: [PATCH] feat(ai): settle proven Codex interrupts --- Cargo.lock | 2 +- crates/graphql-orm-ai/CHANGELOG.md | 33 +- crates/graphql-orm-ai/Cargo.toml | 2 +- crates/graphql-orm-ai/MIGRATION.md | 40 +- crates/graphql-orm-ai/README.md | 7 + .../docs/session-reliability-adoption.md | 52 ++- crates/graphql-orm-ai/src/orm_coordinator.rs | 380 +++++++++++++++++- .../src/orm_provider_session.rs | 297 +++++++++++++- crates/graphql-orm-ai/src/persistence.rs | 2 +- crates/graphql-orm-ai/src/provider.rs | 13 + crates/graphql-orm-ai/src/provider_calls.rs | 24 +- crates/graphql-orm-ai/src/provider_run.rs | 137 ++++++- crates/graphql-orm-ai/src/provider_session.rs | 91 +++++ .../src/providers/codex_app_server.rs | 206 +++++++++- crates/graphql-orm-ai/src/runtime.rs | 48 ++- .../graphql-orm-ai/tests/provider_sessions.rs | 96 ++++- crates/graphql-orm-ai/tests/schema_module.rs | 2 +- docs/reference/workspace-packages.md | 2 +- 18 files changed, 1364 insertions(+), 70 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e945c4b4..5771df23 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3104,7 +3104,7 @@ dependencies = [ [[package]] name = "graphql-orm-ai" -version = "0.82.0" +version = "0.83.0" dependencies = [ "agql-auth", "async-graphql", diff --git a/crates/graphql-orm-ai/CHANGELOG.md b/crates/graphql-orm-ai/CHANGELOG.md index 4e432401..972b9276 100644 --- a/crates/graphql-orm-ai/CHANGELOG.md +++ b/crates/graphql-orm-ai/CHANGELOG.md @@ -3,7 +3,7 @@ title: "Changelog" kind: reference status: active owner: graphql-orm-ai-maintainers -last_reviewed: 2026-08-16 +last_reviewed: 2026-08-21 review_by: 2027-02-01 supersedes: [] --- @@ -18,6 +18,37 @@ checkpoint facts. For the current workspace baseline and active gates, use the [implementation status](docs/implementation-status.md) and the central [AI production-readiness plan](../../docs/plans/active/ai-production-readiness/README.md). +## [0.83.0] - 2026-08-21 + +Persistent schema module: **0.62.0**. + +### Added + +- Codex app-server interrupts can now report a proven settled turn for the + reviewed `codex-cli 0.148.0` / `gpt-5.4` deployment when the exact interrupt + is acknowledged and no dynamic tool call remains unresolved. The guarantee + is explicitly version-observed and must be reverified before a Codex upgrade. +- `AiProviderRunInterruptOutcome::RequestedSettled` and + `AiRuntime::interrupt_all_provider_runs_with_settlement` carry the + provider-side proof without changing the existing interruption-count API. +- `AiProviderSessionService::settle_interrupted_turn` and + `require_cleanup_for_run` let a durable store either advance the transcript + watermark across the retained unanswered user prompt or invalidate the + binding under the exact run fence. Alternate stores deny both by default. + +### Changed + +- A cancelled retained Codex turn keeps its provider thread only after three + independent proofs: exact provider acknowledgement, no unresolved dynamic + call plus version-observed partial-output discard, and a transactional ORM + check that the run persisted no assistant message, tool call, or checkpoint. + Every uncertain or refused case continues through the disclosed + cleanup-required path. +- The persistent schema module advances semantically to 0.62.0 because an + existing provider-session binding may now advance its watermark and + transcript fingerprint after a settled interrupt. No table, column, index, + row rewrite, or backfill is required. + ## [0.82.0] - 2026-08-20 Persistent schema module: **0.61.0**. diff --git a/crates/graphql-orm-ai/Cargo.toml b/crates/graphql-orm-ai/Cargo.toml index 228694f2..8df49548 100644 --- a/crates/graphql-orm-ai/Cargo.toml +++ b/crates/graphql-orm-ai/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "graphql-orm-ai" -version = "0.82.0" +version = "0.83.0" edition = "2024" authors = ["Toby Martin "] description = "Project-agnostic AI agent runtime for graphql-orm applications" diff --git a/crates/graphql-orm-ai/MIGRATION.md b/crates/graphql-orm-ai/MIGRATION.md index e72e6142..812d2c91 100644 --- a/crates/graphql-orm-ai/MIGRATION.md +++ b/crates/graphql-orm-ai/MIGRATION.md @@ -3,7 +3,7 @@ title: "Migration Guide" kind: reference status: active owner: graphql-orm-ai-maintainers -last_reviewed: 2026-08-16 +last_reviewed: 2026-08-21 review_by: 2027-02-01 supersedes: [] --- @@ -19,6 +19,44 @@ they describe. For the current workspace baseline and active delivery gates, use [implementation status](docs/implementation-status.md) and the central [AI production-readiness plan](../../docs/plans/active/ai-production-readiness/README.md). +## 0.82.0 to 0.83.0: settled retained Codex interruption + +Adopt `graphql-orm-ai` 0.83.0 at one reviewed full monorepo revision. + +### Schema module + +The AI schema module advances **0.61.0 to 0.62.0** as a persistent-semantic +version. There is no DDL, table, column, index, constraint, protected-payload, +row-rewrite, or backfill change. Apply and verify the module while AI workers +are stopped, then restart all workers on the same revision. The semantic bump +records that an existing provider-session binding can now advance its durable +watermark and transcript fingerprint after a settled interrupt. + +### Provider adoption + +`AiProviderRunInterruptOutcome` gains the non-exhaustive +`RequestedSettled` variant. Ordinary adapters should continue returning +`Requested`; only an adapter with exact acknowledged-interrupt, unresolved-tool +absence, and provider-thread discard proof may report the new variant. + +`AiRuntime::interrupt_all_provider_runs` keeps its existing count result. The +new `interrupt_all_provider_runs_with_settlement` returns the aggregate proof +for coordinator implementations. A caller must still apply durable evidence; +provider acknowledgement alone never permits retaining a thread. + +The Codex app-server proof is version-observed for `codex-cli 0.148.0` with +`gpt-5.4`, not guaranteed by the empty `turn/interrupt` response. Re-run the +documented interrupt probe before changing the Codex version or admitted model. +The adapter fails closed when a dynamic call remains in flight or starts after +interruption begins. The ORM implementation then transactionally rechecks the +cancelled run, exact claim, message watermark, and absence of an assistant +message, tool call, or checkpoint before retaining the binding. A failure uses +the existing disclosed cleanup-required path. + +The interrupted user message remains in Codex and in the durable transcript, +with no assistant reply. That unanswered prompt is expected and is incorporated +into the provider-session transcript fingerprint before a later run resumes. + ## 0.81.0 to 0.82.0: session reliability and failure disposition Adopt `graphql-orm-ai` 0.82.0 at one reviewed full monorepo revision. diff --git a/crates/graphql-orm-ai/README.md b/crates/graphql-orm-ai/README.md index 0d3b0c52..b58a0d3b 100644 --- a/crates/graphql-orm-ai/README.md +++ b/crates/graphql-orm-ai/README.md @@ -141,6 +141,13 @@ Invalidating a retained provider thread emits `provider_session_reset` or `provider_session_rebound` so a host can tell the user the model's context was reset even though the durable transcript reads as continuous. +The reviewed Codex app-server adapter can retain a thread after Stop only when +the exact interrupt was acknowledged, no dynamic tool call remains unresolved, +and the ORM transaction proves that the cancelled turn persisted no assistant +message, tool call, or checkpoint. This discard guarantee is version-observed +for `codex-cli 0.148.0` with `gpt-5.4`; reverify it before upgrading Codex. Any +missing proof continues through the disclosed cleanup-and-rebind path. + See the [session reliability adoption contract](docs/session-reliability-adoption.md). ## Features and capability boundary diff --git a/crates/graphql-orm-ai/docs/session-reliability-adoption.md b/crates/graphql-orm-ai/docs/session-reliability-adoption.md index 86b460b7..17ee8997 100644 --- a/crates/graphql-orm-ai/docs/session-reliability-adoption.md +++ b/crates/graphql-orm-ai/docs/session-reliability-adoption.md @@ -3,14 +3,14 @@ title: "Session reliability adoption contract" kind: reference status: active owner: graphql-orm-ai-maintainers -last_reviewed: 2026-08-20 +last_reviewed: 2026-08-21 review_by: 2027-02-01 supersedes: [] --- # Session reliability adoption contract -This is the exact contract for the 0.82.0 session-reliability work: which +This is the exact contract for the 0.82.0–0.83.0 session-reliability work: which public APIs are new or changed, what a client must do differently, and which behaviours changed with no API change at all. It complements [MIGRATION.md](../MIGRATION.md), which records the schema and source-breaking @@ -176,22 +176,35 @@ a changed rule fingerprint, an incomplete dynamic turn, and an exceeded budget are all ordinary user behaviour rather than faults. **Interruption reports what it proved.** `AiRunInterruptSettlement` replaces -`()` from `interrupt_run`. `retains_thread()` is true only for `Settled`, which -no adapter currently reports, so it fails closed to invalidation. - -Acknowledgement is not settlement. The Codex app-server `turn/interrupt` -response is an empty object, `TurnStatus` has a first-class `interrupted` -value, and a resumed thread pages prior turns back through -`thread/turns/list` — so an acknowledgement cannot distinguish a discarded -partial turn from a retained one. Treating it as settlement would let the model -carry content the durable transcript never recorded, which is the same -divergence the disclosure events above exist to expose. The variant exists so -an adapter that can prove settlement may report it without a further breaking -change. - -Interrupting an in-flight turn already invalidates the retained binding through -the executor's own ambiguous-turn cleanup. That path is now *disclosed* rather -than silent, so a mid-generation stop is visible to the user. +`()` from `interrupt_run`. `retains_thread()` is true only for `Settled`, and +settlement requires three independent legs: + +1. the exact provider interrupt was acknowledged; +2. the adapter proves no unresolved dynamic tool call and version-observed + discard of the partial provider output; and +3. the durable provider-session store transactionally proves that the cancelled + run persisted no assistant message, tool call, or checkpoint. + +Acknowledgement alone is not settlement. The Codex app-server adapter reports +its provider-side proof only for the reviewed `codex-cli 0.148.0` / `gpt-5.4` +deployment. Direct measurement showed that an interrupted assistant stream has +`status=interrupted`, no assistant item in `thread/turns/list`, no partial text +in the rollout file, and no partial text in the resumed model context. The +empty `turn/interrupt` response does not promise that behaviour, so repeat the +probe before upgrading Codex or the admitted model. + +The adapter fences interruption against dynamic-call dispatch. A call still in +flight, or one first dispatched after interruption begins, keeps the result +unsettled. A call completed before interruption is no longer unresolved, but +the independent durable leg still refuses retention once tool traffic or a +checkpoint was persisted. + +When all three proofs hold, the ORM store advances the binding watermark and +transcript fingerprint across the interrupted user message and releases the +claim. Codex retains that user message with no assistant reply, which matches +the durable transcript. Any missing proof, ORM conflict, or alternate store's +default-deny implementation invalidates the binding through the same disclosed +cleanup path, so a mid-generation context reset remains visible to the user. ## Messages accepted during cleanup @@ -217,6 +230,9 @@ reconciliation keeps owning that case. | `AiSessionEventEnvelope` | Added `closed`; construct via `delivered`/`ended` | | `AiSessionStreamClose` | New enum | | `AiRunInterruptSettlement` | New enum; `interrupt_run` returns it instead of `()` | +| `AiProviderRunInterruptOutcome::RequestedSettled` | Provider-side settled-interrupt proof | +| `AiRuntime::interrupt_all_provider_runs_with_settlement` | Aggregate provider proof; existing count API unchanged | +| `AiProviderSessionService::settle_interrupted_turn`, `require_cleanup_for_run` | Default-deny durable retain/invalidate boundary | | `AiRunFailure`, `AiRunRetryAdmission`, `AiRunRetryEvidence`, `classify_run_retry` | New | | `AiRunDisposition`, `AiRunDispositionView`, `AiRunRetryRefusal` | New | | `RetryAiRunInput`, `AcknowledgeAiRunFailureInput` | New GraphQL inputs | diff --git a/crates/graphql-orm-ai/src/orm_coordinator.rs b/crates/graphql-orm-ai/src/orm_coordinator.rs index 3e62b412..6a1ae2e5 100644 --- a/crates/graphql-orm-ai/src/orm_coordinator.rs +++ b/crates/graphql-orm-ai/src/orm_coordinator.rs @@ -1119,6 +1119,48 @@ impl AiReadOnlyAgentCoordinator { } } + /// Applies an interrupt settlement to the retained provider session of a + /// cancelled run. + /// + /// A settled interrupt keeps the binding: the provider proved it discarded + /// the interrupted partial turn, and the durable evidence below proves the + /// turn persisted nothing, so the retained thread holds exactly the + /// interrupted prompt with no reply — which is what the durable transcript + /// records too. Everything else invalidates the binding through the same + /// disclosed cleanup funnel as any other reset, so the user learns the + /// model's context was lost. + /// + /// The durable leg is deliberately conservative. Settlement needs a run + /// that never observed a completed provider turn, never executed a tool, + /// and holds no checkpoint; a later turn of the same run has already put + /// tool traffic into the thread that the message transcript does not + /// reproduce on its own. + async fn settle_interrupted_provider_session( + &self, + lease: &AiRunLease, + guard: &AiAgentLoopGuard, + settlement: crate::AiRunInterruptSettlement, + ) { + let Some(service) = &self.provider_session_service else { + return; + }; + let no_uncertain_persisted_output = guard.provider_turns() == 0 + && guard.total_tool_calls() == 0 + && lease.latest_checkpoint_id().is_none(); + let settlement = settlement.with_durable_turn_evidence(no_uncertain_persisted_output); + if settlement.retains_thread() + && service + .settle_interrupted_turn(lease, settlement) + .await + .is_ok() + { + return; + } + let _ = service + .require_cleanup_for_run(lease, "provider_session_interrupted_unsettled") + .await; + } + /// Executes one freshly claimed lease through a bounded terminal outcome. /// /// A successful return means the corresponding terminal or @@ -1465,7 +1507,9 @@ impl AiReadOnlyAgentCoordinator { .await; } Err(ProviderTurnFailure::LeaseLost(error)) => return Err(error), - Err(ProviderTurnFailure::Cancelled) => { + Err(ProviderTurnFailure::Cancelled(settlement)) => { + self.settle_interrupted_provider_session(&lease, &guard, settlement) + .await; return Ok(Cancelled { provider_turns: guard.provider_turns(), total_tool_calls: guard.total_tool_calls(), @@ -1892,8 +1936,12 @@ impl AiReadOnlyAgentCoordinator { result = &mut cancellation => { match result.map_err(ProviderTurnFailure::LeaseLost)? { Some(_) => { - let _ = self.provider_executor.interrupt_run(lease).await; - return Err(ProviderTurnFailure::Cancelled); + let settlement = self + .provider_executor + .interrupt_run(lease) + .await + .unwrap_or(crate::AiRunInterruptSettlement::RequestedUnsettled); + return Err(ProviderTurnFailure::Cancelled(settlement)); } None => { match self.run_control.heartbeat(lease).await { @@ -1937,9 +1985,15 @@ impl AiReadOnlyAgentCoordinator { match result.map_err(ProviderTurnFailure::LeaseLost)? { Some(_) => { let current = lease_state.lock().await.clone(); - let _ = self.provider_executor.interrupt_run(¤t).await; + // A failed or unrecognized interrupt proves nothing, + // so it fails closed into invalidation. + let settlement = self + .provider_executor + .interrupt_run(¤t) + .await + .unwrap_or(crate::AiRunInterruptSettlement::RequestedUnsettled); *lease = current; - return Err(ProviderTurnFailure::Cancelled); + return Err(ProviderTurnFailure::Cancelled(settlement)); } None => { let mut current = lease_state.lock().await; @@ -2002,9 +2056,15 @@ impl AiReadOnlyAgentCoordinator { match result.map_err(ProviderTurnFailure::LeaseLost)? { Some(_) => { let current = lease_state.lock().await.clone(); - let _ = self.provider_executor.interrupt_run(¤t).await; + // A failed or unrecognized interrupt proves nothing, + // so it fails closed into invalidation. + let settlement = self + .provider_executor + .interrupt_run(¤t) + .await + .unwrap_or(crate::AiRunInterruptSettlement::RequestedUnsettled); *lease = current; - return Err(ProviderTurnFailure::Cancelled); + return Err(ProviderTurnFailure::Cancelled(settlement)); } None => { let mut current = lease_state.lock().await; @@ -2193,7 +2253,9 @@ enum ProviderTurnFailure { Provider, Deferred, LeaseLost(AiError), - Cancelled, + /// Owner cancellation won the fence; the value reports what the resulting + /// interrupt proved about the provider's retained thread. + Cancelled(crate::AiRunInterruptSettlement), } #[cfg(all(test, feature = "sqlite"))] @@ -3657,6 +3719,308 @@ mod tests { ); } + struct PendingRetainedProviderExecutor { + settlement: crate::AiRunInterruptSettlement, + interrupts: AtomicUsize, + /// Cancellation must land *while* the provider turn is in flight; a run + /// cancelled before the turn starts never reaches an interrupt. + run: Arc, + } + + #[async_trait] + impl AiAgentProviderTurnExecutor for PendingRetainedProviderExecutor { + async fn execute_turn( + &self, + _lease: &AiRunLease, + _plan: AiProviderCallPlan, + ) -> Result { + Err(AiError::Conflict) + } + + async fn execute_retained_turn( + &self, + _lease: Arc>, + _plan: AiProviderCallPlan, + _session_plan: crate::AiProviderSessionTurnPlan, + _session_service: Arc, + _execution: Option>, + ) -> Result { + self.run.cancelled.store(true, Ordering::SeqCst); + std::future::pending().await + } + + async fn interrupt_run( + &self, + _lease: &AiRunLease, + ) -> Result { + self.interrupts.fetch_add(1, Ordering::SeqCst); + Ok(self.settlement) + } + } + + #[derive(Default)] + struct InterruptSessionService { + settlements: AtomicUsize, + run_cleanups: AtomicUsize, + settlement_fails: bool, + } + + #[async_trait] + impl crate::AiProviderSessionService for InterruptSessionService { + async fn inspect_for_run( + &self, + _lease: &AiRunLease, + ) -> Result, AiError> { + Ok(None) + } + + async fn bind_for_run( + &self, + _lease: &AiRunLease, + _request: crate::AiProviderSessionBindRequest, + ) -> Result { + Err(AiError::Conflict) + } + + async fn claim_for_run( + &self, + _lease: &AiRunLease, + _expected: &crate::AiProviderSessionDescriptor, + _expected_transcript_fingerprint: &str, + ) -> Result { + Err(AiError::Conflict) + } + + async fn open_for_run( + &self, + _lease: &AiRunLease, + _claim: &crate::AiProviderSessionClaim, + ) -> Result { + Err(AiError::Conflict) + } + + async fn heartbeat( + &self, + _lease: &AiRunLease, + _claim: &crate::AiProviderSessionClaim, + ) -> Result { + Err(AiError::Conflict) + } + + async fn commit_turn( + &self, + _lease: &AiRunLease, + _claim: &crate::AiProviderSessionClaim, + _commit: crate::AiProviderSessionCommit, + ) -> Result { + Err(AiError::Conflict) + } + + async fn settle_interrupted_turn( + &self, + lease: &AiRunLease, + settlement: crate::AiRunInterruptSettlement, + ) -> Result { + assert!( + settlement.retains_thread(), + "the durable boundary must never be asked to retain an unsettled thread" + ); + self.settlements.fetch_add(1, Ordering::SeqCst); + if self.settlement_fails { + return Err(AiError::Conflict); + } + Ok(crate::AiProviderSessionBindingView { + binding_id: Uuid::from_u128(7), + session_id: lease.session_id(), + scope: test_scope(), + descriptor: retained_descriptor(), + state: crate::AiProviderSessionState::Active, + through_message_sequence: 1, + transcript_fingerprint: "d".repeat(64), + provider_expires_at: None, + idle_expires_at: time::OffsetDateTime::now_utc() + Duration::minutes(5), + absolute_expires_at: time::OffsetDateTime::now_utc() + Duration::hours(1), + row_version: 2, + }) + } + + async fn require_cleanup( + &self, + _claim: &crate::AiProviderSessionClaim, + _reason_code: &str, + ) -> Result<(), AiError> { + Err(AiError::Conflict) + } + + async fn require_cleanup_for_run( + &self, + _lease: &AiRunLease, + reason_code: &str, + ) -> Result<(), AiError> { + assert_eq!(reason_code, "provider_session_interrupted_unsettled"); + self.run_cleanups.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + + async fn claim_cleanup( + &self, + _worker_id: &str, + ) -> Result, AiError> { + Ok(None) + } + + async fn open_for_cleanup( + &self, + _claim: &crate::AiProviderSessionCleanupClaim, + _policy: &crate::AiContentProtectionPolicy, + ) -> Result { + Err(AiError::Conflict) + } + + async fn complete_cleanup( + &self, + _claim: &crate::AiProviderSessionCleanupClaim, + _proof: crate::AiProviderSessionAbsenceProof, + ) -> Result<(), AiError> { + Err(AiError::Conflict) + } + + async fn schedule_cleanup_retry( + &self, + _claim: &crate::AiProviderSessionCleanupClaim, + _delay: Duration, + _reason_code: &str, + ) -> Result<(), AiError> { + Err(AiError::Conflict) + } + } + + fn interrupted_retained_coordinator( + settlement: crate::AiRunInterruptSettlement, + settlement_fails: bool, + ) -> ( + AiReadOnlyAgentCoordinator, + Arc, + Arc, + ) { + let run = Arc::new(TestRunControl::new()); + let descriptor = retained_descriptor(); + let planner = Arc::new(TestRetainedChatPlanner { + scope: test_scope(), + provider_session: crate::AiProviderSessionTurnPlan::new(descriptor, "c".repeat(64)) + .expect("retained turn plan should validate"), + }); + let provider = Arc::new(PendingRetainedProviderExecutor { + settlement, + interrupts: AtomicUsize::new(0), + run: run.clone(), + }); + let session_service = Arc::new(InterruptSessionService { + settlement_fails, + ..InterruptSessionService::default() + }); + let forbidden = Arc::new(ChatForbiddenBoundaries::default()); + let coordinator = AiReadOnlyAgentCoordinator::new( + run.clone(), + provider.clone(), + forbidden.clone(), + Arc::new(TestOutputWriter), + forbidden.clone(), + Arc::new(TestCheckpointWriter), + Arc::new(TestRuleResolver), + planner, + limits(50), + ) + .with_provider_session_service(session_service.clone()); + + (coordinator, session_service, provider) + } + + async fn interrupted_retained_run( + settlement: crate::AiRunInterruptSettlement, + lease: &AiRunLease, + settlement_fails: bool, + ) -> ( + Arc, + Arc, + ) { + let (coordinator, session_service, provider) = + interrupted_retained_coordinator(settlement, settlement_fails); + + let outcome = coordinator + .execute_claimed(lease) + .await + .expect("an interrupted run should close as cancelled"); + assert!(matches!(outcome, Cancelled { .. })); + assert_eq!(provider.interrupts.load(Ordering::SeqCst), 1); + (session_service, provider) + } + + /// A settled interrupt asks the durable boundary to keep the binding and + /// never routes through the invalidation funnel. + #[tokio::test] + async fn settled_interrupt_retains_the_provider_session_binding() { + let lease = AiRunLease::test_running(principal_reference()); + let (service, _) = + interrupted_retained_run(crate::AiRunInterruptSettlement::Settled, &lease, false).await; + + assert_eq!(service.settlements.load(Ordering::SeqCst), 1); + assert_eq!( + service.run_cleanups.load(Ordering::SeqCst), + 0, + "a settled interrupt must not invalidate the retained thread" + ); + } + + /// An acknowledged-but-unsettled interrupt still invalidates through the + /// disclosed cleanup funnel. + #[tokio::test] + async fn unsettled_interrupt_invalidates_the_provider_session_binding() { + let lease = AiRunLease::test_running(principal_reference()); + let (service, _) = interrupted_retained_run( + crate::AiRunInterruptSettlement::RequestedUnsettled, + &lease, + false, + ) + .await; + + assert_eq!(service.settlements.load(Ordering::SeqCst), 0); + assert_eq!(service.run_cleanups.load(Ordering::SeqCst), 1); + } + + /// Uncertain persisted output demotes an adapter-proven settlement before + /// the durable boundary is ever asked to retain the binding. + #[tokio::test] + async fn uncertain_persisted_output_demotes_a_settled_interrupt() { + let lease = + AiRunLease::test_running(principal_reference()).test_with_checkpoint(Uuid::new_v4()); + let (coordinator, service, _) = + interrupted_retained_coordinator(crate::AiRunInterruptSettlement::Settled, false); + let guard = AiAgentLoopGuard::new(&lease, limits(50).loop_limits); + coordinator + .settle_interrupted_provider_session( + &lease, + &guard, + crate::AiRunInterruptSettlement::Settled, + ) + .await; + + assert_eq!(service.settlements.load(Ordering::SeqCst), 0); + assert_eq!(service.run_cleanups.load(Ordering::SeqCst), 1); + } + + /// A durable boundary that refuses the settlement falls back to the same + /// disclosed invalidation. + #[tokio::test] + async fn refused_durable_settlement_falls_back_to_invalidation() { + let lease = AiRunLease::test_running(principal_reference()); + let (service, _) = + interrupted_retained_run(crate::AiRunInterruptSettlement::Settled, &lease, true).await; + + assert_eq!(service.settlements.load(Ordering::SeqCst), 1); + assert_eq!(service.run_cleanups.load(Ordering::SeqCst), 1); + } + /// Work item 4.3: once the bounded retry allowance is exhausted while /// cleanup is still pending, the run must close as a visible failure rather /// than being left to expire into `RecoveryRequired`, which is both diff --git a/crates/graphql-orm-ai/src/orm_provider_session.rs b/crates/graphql-orm-ai/src/orm_provider_session.rs index 5bf50bc3..44c4a7ed 100644 --- a/crates/graphql-orm-ai/src/orm_provider_session.rs +++ b/crates/graphql-orm-ai/src/orm_provider_session.rs @@ -147,11 +147,12 @@ mod service { content_context, map_orm, map_protection, map_transaction, principal_identity, record_scope, }; use crate::persistence::{ - AiApprovalRecord, AiAuditEventRecord, AiMessageRecord, AiRunCheckpointRecord, - AiRunCheckpointRecordWhereInput, AiRunRecord, AiSessionEventRecord, AiSessionRecord, - AiSessionRecordWhereInput, AiSubscriptionWaitAdoptionRecord, + AiApprovalRecord, AiAuditEventRecord, AiMessageRecord, AiMessageRecordWhereInput, + AiRunCheckpointRecord, AiRunCheckpointRecordWhereInput, AiRunRecord, AiSessionEventRecord, + AiSessionRecord, AiSessionRecordWhereInput, AiSubscriptionWaitAdoptionRecord, AiSubscriptionWaitAdoptionRecordWhereInput, AiSubscriptionWaiterRecord, AiToolCallRecord, - CreateAiAuditEventRecordInput, CreateAiSessionEventRecordInput, UpdateAiSessionRecordInput, + AiToolCallRecordWhereInput, CreateAiAuditEventRecordInput, CreateAiSessionEventRecordInput, + UpdateAiSessionRecordInput, }; use crate::{ AiAccessPolicy, AiContentProtectionPolicy, AiContentProtectionPolicyResolver, @@ -1896,6 +1897,127 @@ mod service { binding_view(&record) } + async fn settle_interrupted_turn( + &self, + lease: &AiRunLease, + settlement: crate::AiRunInterruptSettlement, + ) -> Result { + // The caller's settlement is necessary, never sufficient: every + // condition below is re-proven from committed rows inside the + // transaction that keeps the binding. + if !settlement.retains_thread() { + return Err(AiError::Conflict); + } + let (current, observed_session, scope) = self.load_owned_active_context(lease).await?; + self.protection_policy(current.principal(), &scope).await?; + let now = canonical_second(self.clock.now()); + let idle_expires_at = now + .checked_add(self.limits.idle_ttl()) + .ok_or(AiError::PersistenceFailed)?; + let lease = lease.clone(); + let record = self + .database + .transaction(TransactionMode::StateMachine, move |tx| { + Box::pin(async move { + let _run = load_and_validate_interrupted_run(tx, &lease).await?; + let session = tx + .find_by_id::(&lease.session_id().0) + .await + .map_err(OrmPublicError::from)? + .ok_or_else(OrmPublicError::not_found)?; + if session != observed_session { + return Err(OrmPublicError::new(OrmErrorCode::Conflict)); + } + let input = tx + .find_by_id::(&lease.input_message_id()) + .await + .map_err(OrmPublicError::from)? + .ok_or_else(|| OrmPublicError::new(OrmErrorCode::Conflict))?; + if input.session_id != session.id + || input.message_role != "user" + || input.completion_state != "complete" + || input.sequence <= 0 + { + return Err(OrmPublicError::new(OrmErrorCode::Conflict)); + } + if !run_persisted_no_output(tx, lease.run_id()).await? { + return Err(OrmPublicError::new(OrmErrorCode::Conflict)); + } + let binding = load_single_session_binding(tx, session.id).await?; + validate_binding_record(&binding)?; + if binding.owner_principal_kind != session.owner_principal_kind + || binding.owner_subject != session.owner_subject + || record_scope_from_binding(&binding) != scope + || binding.state != AiProviderSessionState::Claimed.as_str() + || binding.claimed_run_id != Some(lease.run_id().0) + || binding.claimed_attempt_id != Some(lease.attempt_id()) + || binding.claimed_run_lease_generation + != Some(lease.lease_generation()) + || binding.claim_owner.as_deref() != Some(lease.worker_id()) + || binding + .claim_expires_at + .is_none_or(|expiry| expiry <= now.unix_timestamp()) + || binding.through_message_sequence.checked_add(1) + != Some(input.sequence) + { + return Err(OrmPublicError::new(OrmErrorCode::Conflict)); + } + let transcript_fingerprint = + crate::provider_session::interrupted_turn_transcript_fingerprint( + &binding.transcript_fingerprint, + AiSessionId(binding.session_id), + lease.run_id(), + input.id, + input.sequence, + ); + let bounded_idle_expiry = idle_expires_at + .unix_timestamp() + .min(binding.absolute_expires_at) + .min(binding.provider_expires_at.unwrap_or(i64::MAX)); + if bounded_idle_expiry <= now.unix_timestamp() { + return Err(OrmPublicError::new(OrmErrorCode::Conflict)); + } + let outcome = tx + .compare_and_swap::( + &binding.id, + binding.row_version, + AiProviderSessionBindingRecordWhereInput::default(), + UpdateAiProviderSessionBindingRecordInput { + through_message_sequence: Some(input.sequence), + transcript_fingerprint: Some(transcript_fingerprint), + last_run_id: Some(Some(lease.run_id().0)), + state: Some(AiProviderSessionState::Active.as_str().to_owned()), + claimed_run_id: Some(None), + claimed_attempt_id: Some(None), + claimed_run_lease_generation: Some(None), + claim_owner: Some(None), + claim_expires_at: Some(None), + idle_expires_at: Some(bounded_idle_expiry), + ..Default::default() + }, + ) + .await + .map_err(OrmPublicError::from)?; + let ConditionalUpdateOutcome::Updated(updated) = outcome else { + return Err(OrmPublicError::new(OrmErrorCode::Conflict)); + }; + append_audit( + tx, + "ai.provider_session.interrupt_settled", + binding.id, + "provider_session_interrupt_settled", + lease.run_id().0, + now, + ) + .await?; + Ok(updated) + }) + }) + .await + .map_err(map_transaction)?; + binding_view(&record) + } + async fn require_cleanup( &self, claim: &AiProviderSessionClaim, @@ -1951,6 +2073,66 @@ mod service { .map_err(map_transaction) } + async fn require_cleanup_for_run( + &self, + lease: &AiRunLease, + reason_code: &str, + ) -> Result<(), AiError> { + validate_reason_code(reason_code)?; + let now = canonical_second(self.clock.now()); + let lease = lease.clone(); + let reason_code = reason_code.to_owned(); + self.database + .transaction(TransactionMode::StateMachine, move |tx| { + Box::pin(async move { + let binding = load_single_session_binding(tx, lease.session_id().0).await?; + validate_binding_record(&binding)?; + if binding.state != AiProviderSessionState::Claimed.as_str() + || binding.claimed_run_id != Some(lease.run_id().0) + || binding.claimed_attempt_id != Some(lease.attempt_id()) + || binding.claimed_run_lease_generation + != Some(lease.lease_generation()) + || binding.claim_owner.as_deref() != Some(lease.worker_id()) + { + return Err(OrmPublicError::new(OrmErrorCode::Conflict)); + } + let outcome = tx + .compare_and_swap::( + &binding.id, + binding.row_version, + AiProviderSessionBindingRecordWhereInput::default(), + cleanup_required_update(reason_code.clone(), lease.run_id().0), + ) + .await + .map_err(OrmPublicError::from)?; + if !matches!(outcome, ConditionalUpdateOutcome::Updated(_)) { + return Err(OrmPublicError::new(OrmErrorCode::Conflict)); + } + append_audit( + tx, + "ai.provider_session.cleanup_required", + binding.id, + &reason_code, + lease.run_id().0, + now, + ) + .await?; + append_provider_session_disclosure( + tx, + crate::orm_runs::PROVIDER_SESSION_RESET_EVENT, + binding.session_id, + binding.id, + &reason_code, + Some(lease.run_id().0), + now, + ) + .await + }) + }) + .await + .map_err(map_transaction) + } + async fn claim_cleanup( &self, worker_id: &str, @@ -2401,6 +2583,113 @@ mod service { Ok(run) } + /// Loads the run behind an interrupted turn and proves it recorded nothing + /// for that turn. + /// + /// Owner cancellation is what makes an interrupt observable, so the run row + /// is already `Cancelled` with its lease released while the caller's lease + /// snapshot still reads `Running`. The absent checkpoint is the durable + /// proof that no provider turn, adopted wait, or assistant output of this + /// run was persisted. + async fn load_and_validate_interrupted_run( + tx: &mut MutationContext<'_, DefaultWriteBackend>, + lease: &AiRunLease, + ) -> Result { + let run = tx + .find_by_id::(&lease.run_id().0) + .await + .map_err(OrmPublicError::from)? + .ok_or_else(OrmPublicError::not_found)?; + let stored_reference: PrincipalReference = + serde_json::from_value(run.principal_reference.clone()) + .map_err(|_| OrmPublicError::new(OrmErrorCode::InternalError))?; + if lease.state() != AiRunState::Running + || run.session_id != lease.session_id().0 + || run.input_message_id != lease.input_message_id() + || stored_reference != *lease.principal_reference() + || run.attempt_id != Some(lease.attempt_id()) + || run.lease_generation != lease.lease_generation() + || run.retry_count != i64::from(lease.retry_count()) + || run.state != AiRunState::Cancelled.as_str() + || run.latest_checkpoint_id.is_some() + || lease.latest_checkpoint_id().is_some() + { + return Err(OrmPublicError::new(OrmErrorCode::Conflict)); + } + Ok(run) + } + + /// Returns whether one run committed no assistant message and no tool call. + /// + /// Both are read inside the settling transaction, so a concurrent writer + /// cannot slip persisted output past the check. + async fn run_persisted_no_output( + tx: &mut MutationContext<'_, DefaultWriteBackend>, + run_id: AiRunId, + ) -> Result { + let messages = tx + .query::() + .filter(AiMessageRecordWhereInput { + run_id: Some(UuidFilter { + eq: Some(run_id.0), + ..Default::default() + }), + message_role: Some(StringFilter { + eq: Some("assistant".to_owned()), + ..Default::default() + }), + ..Default::default() + }) + .limit(1) + .fetch_all() + .await + .map_err(OrmPublicError::from)?; + if !messages.is_empty() { + return Ok(false); + } + let tool_calls = tx + .query::() + .filter(AiToolCallRecordWhereInput { + run_id: Some(UuidFilter { + eq: Some(run_id.0), + ..Default::default() + }), + ..Default::default() + }) + .limit(1) + .fetch_all() + .await + .map_err(OrmPublicError::from)?; + Ok(tool_calls.is_empty()) + } + + /// Loads the single provider-session binding of one session. + async fn load_single_session_binding( + tx: &mut MutationContext<'_, DefaultWriteBackend>, + session_id: Uuid, + ) -> Result { + let records = tx + .query::() + .filter(AiProviderSessionBindingRecordWhereInput { + session_id: Some(UuidFilter { + eq: Some(session_id), + ..Default::default() + }), + ..Default::default() + }) + .limit(2) + .fetch_all() + .await + .map_err(OrmPublicError::from)?; + if records.len() > 1 { + return Err(OrmPublicError::new(OrmErrorCode::InternalError)); + } + records + .into_iter() + .next() + .ok_or_else(|| OrmPublicError::new(OrmErrorCode::Conflict)) + } + fn validate_active_claim( record: &AiProviderSessionBindingRecord, claim: &AiProviderSessionClaim, diff --git a/crates/graphql-orm-ai/src/persistence.rs b/crates/graphql-orm-ai/src/persistence.rs index 2f78079d..ca2b97a6 100644 --- a/crates/graphql-orm-ai/src/persistence.rs +++ b/crates/graphql-orm-ai/src/persistence.rs @@ -2424,7 +2424,7 @@ pub(crate) struct AiRuntimeRecoveryRecord { /// Stable schema module ID. pub const AI_SCHEMA_MODULE_ID: &str = "com.dastari.graphql-orm-ai"; /// Current AI schema module version. -pub const AI_SCHEMA_MODULE_VERSION: &str = "0.61.0"; +pub const AI_SCHEMA_MODULE_VERSION: &str = "0.62.0"; /// Reserved table namespace. pub const AI_TABLE_NAMESPACE: &str = "graphql_orm_ai_"; diff --git a/crates/graphql-orm-ai/src/provider.rs b/crates/graphql-orm-ai/src/provider.rs index 1c227b21..535ad870 100644 --- a/crates/graphql-orm-ai/src/provider.rs +++ b/crates/graphql-orm-ai/src/provider.rs @@ -2976,6 +2976,19 @@ pub trait AiProvider: Send + Sync { /// The default is inert for ordinary request-scoped adapters. This signal /// contains no cancellation authority; the caller must first observe the /// authoritative durable cancellation or lease-loss fence. + /// + /// An adapter reports + /// [`AiProviderRunInterruptOutcome::RequestedSettled`] only when it can + /// prove, for the exact interrupted turn, that the provider acknowledged + /// the interrupt, that no dynamic tool call is unresolved, and that the + /// interrupted partial turn is discarded from the provider payload, the + /// provider's durable thread artifact, and the model's later context. An + /// adapter that cannot prove all three keeps + /// [`AiProviderRunInterruptOutcome::Requested`], which fails closed to + /// invalidating the retained thread. + /// + /// [`AiProviderRunInterruptOutcome::RequestedSettled`]: crate::AiProviderRunInterruptOutcome::RequestedSettled + /// [`AiProviderRunInterruptOutcome::Requested`]: crate::AiProviderRunInterruptOutcome::Requested async fn interrupt_run( &self, _binding: &crate::AiProviderRunBinding, diff --git a/crates/graphql-orm-ai/src/provider_calls.rs b/crates/graphql-orm-ai/src/provider_calls.rs index 6f0053e0..18872c6a 100644 --- a/crates/graphql-orm-ai/src/provider_calls.rs +++ b/crates/graphql-orm-ai/src/provider_calls.rs @@ -2594,19 +2594,17 @@ impl AiProviderCallExecutor { lease: &AiRunLease, ) -> Result { let binding = crate::AiProviderRunBinding::from_lease(lease)?; - let requested = self - .runtime - .interrupt_all_provider_runs(&binding) - .await - .map_err(|_| AiError::ProviderFailed)?; - // Acknowledgement is not settlement: no adapter can currently prove the - // interrupted turn left its retained thread consistent with the durable - // transcript, so this never reports `Settled`. - Ok(if requested == 0 { - crate::AiRunInterruptSettlement::NotActive - } else { - crate::AiRunInterruptSettlement::RequestedUnsettled - }) + // Acknowledgement is not settlement. The aggregate carries only the two + // provider-side legs — the interrupt was acknowledged, and the adapter + // proved no unresolved dynamic tool call plus version-observed discard + // of the interrupted partial turn. The caller owns the third leg and + // must apply + // `AiRunInterruptSettlement::with_durable_turn_evidence` before it + // treats `retains_thread` as permission to keep a binding. + self.runtime + .interrupt_all_provider_runs_with_settlement(&binding) + .await + .map_err(|_| AiError::ProviderFailed) } pub(crate) async fn close_run( diff --git a/crates/graphql-orm-ai/src/provider_run.rs b/crates/graphql-orm-ai/src/provider_run.rs index 0321718b..f7bec0aa 100644 --- a/crates/graphql-orm-ai/src/provider_run.rs +++ b/crates/graphql-orm-ai/src/provider_run.rs @@ -155,29 +155,59 @@ pub enum AiProviderRunCloseReason { } /// Result of a bounded provider-run interruption request. +/// +/// [`Self::RequestedSettled`] is an adapter-level proof, not an +/// acknowledgement: only an adapter that can show its interrupt leaves the +/// retained provider thread consistent with the durable transcript may report +/// it. Every other adapter keeps the fail-closed +/// [`Self::Requested`]/[`Self::NotActive`] pair, and an unrecognized value is +/// treated as unsettled. #[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[non_exhaustive] pub enum AiProviderRunInterruptOutcome { /// No live resource existed for the exact binding. NotActive, /// An active provider turn accepted the interruption request. Requested, + /// An active provider turn accepted the interruption request, and the + /// adapter proved its own settlement conditions for that exact turn. + /// + /// An adapter may return this only when all of the following hold: + /// + /// 1. the provider acknowledged the interrupt for the exact fenced turn; + /// 2. no dynamic tool call for that turn is unresolved; and + /// 3. the adapter has version-observed evidence that the interrupted + /// partial turn is discarded from the provider payload, the provider's + /// durable thread artifact, and the model's later context. + /// + /// This still proves nothing about the *caller's* persistence. The durable + /// leg — that no output of the interrupted turn was persisted, or may have + /// been persisted uncertainly — belongs to the caller and is applied + /// through [`AiRunInterruptSettlement::with_durable_turn_evidence`]. + RequestedSettled, } /// What an interrupt request proved about the provider turn it stopped. /// /// This is deliberately separate from [`AiProviderRunInterruptOutcome`], which -/// reports only whether a live resource accepted the request. Settlement is a -/// stronger claim: that the provider's retained thread, after interruption, is -/// consistent with the durable transcript. +/// reports one adapter's view of one resource. Settlement is a stronger claim: +/// that the provider's retained thread, after interruption, is consistent with +/// the durable transcript. +/// +/// [`Self::Settled`] requires three independent legs, and anything short of +/// all three stays [`Self::RequestedUnsettled`]: +/// +/// 1. the interrupt was acknowledged by the provider; +/// 2. the adapter proved no unresolved dynamic tool call for that turn and +/// version-observed discard of the interrupted partial turn +/// ([`AiProviderRunInterruptOutcome::RequestedSettled`]); and +/// 3. the caller proved from committed durable rows that the interrupted turn +/// left no uncertain persisted output +/// ([`Self::with_durable_turn_evidence`]). /// -/// No adapter currently reports [`Self::Settled`]. The Codex app-server -/// `turn/interrupt` response is an empty object, `TurnStatus` has a -/// first-class `interrupted` value, and resumed threads page prior turns back -/// through `thread/turns/list`, so an acknowledgement does not distinguish a -/// discarded partial turn from a retained one. Treating acknowledgement as -/// settlement would let the model carry content the durable transcript never -/// recorded. The variant exists so an adapter that can prove settlement may -/// report it without another breaking change. +/// Acknowledgement alone is never settlement. Treating it as settlement would +/// let the model carry content the durable transcript never recorded, which is +/// the divergence the retained-session disclosure events exist to expose. #[derive(Clone, Copy, Debug, PartialEq, Eq)] #[non_exhaustive] pub enum AiRunInterruptSettlement { @@ -197,9 +227,50 @@ impl AiRunInterruptSettlement { /// /// Anything other than proven settlement is false, so an unrecognized or /// merely acknowledged interruption fails closed into invalidation. + /// + /// A true value permits retention; it does not perform it. The durable + /// provider-session boundary re-proves the same conditions from committed + /// rows before it keeps a binding, so a caller that skips + /// [`Self::with_durable_turn_evidence`] still cannot retain a thread whose + /// turn persisted output. pub const fn retains_thread(self) -> bool { matches!(self, Self::Settled) } + + /// Applies the caller-owned durable leg of the settlement guard. + /// + /// `no_uncertain_persisted_output` must be derived from committed durable + /// state for the exact interrupted turn: no assistant output, tool result, + /// or turn checkpoint was persisted, and none may have been persisted + /// uncertainly. Anything else demotes provider-proven settlement to + /// [`Self::RequestedUnsettled`], so an adapter can never widen retention + /// past what the caller's transcript can reproduce. + #[must_use] + pub const fn with_durable_turn_evidence(self, no_uncertain_persisted_output: bool) -> Self { + match self { + Self::Settled if !no_uncertain_persisted_output => Self::RequestedUnsettled, + other => other, + } + } + + /// Folds one adapter outcome into an aggregate settlement. + /// + /// Aggregation is fail-closed: a single acknowledged-but-unsettled adapter + /// keeps the whole interruption unsettled even when another adapter proved + /// its own resource settled. + pub(crate) const fn fold_provider_outcome( + self, + outcome: AiProviderRunInterruptOutcome, + ) -> Self { + match (self, outcome) { + (_, AiProviderRunInterruptOutcome::Requested) | (Self::RequestedUnsettled, _) => { + Self::RequestedUnsettled + } + (Self::NotActive, AiProviderRunInterruptOutcome::RequestedSettled) + | (Self::Settled, _) => Self::Settled, + (Self::NotActive, AiProviderRunInterruptOutcome::NotActive) => Self::NotActive, + } + } } /// Result of closing one exact provider-run resource. @@ -255,4 +326,48 @@ mod tests { assert_ne!(binding, later_generation); assert_ne!(binding, another_owner); } + + #[test] + fn settlement_requires_acknowledgement_adapter_proof_and_durable_evidence() { + let settled = AiRunInterruptSettlement::NotActive + .fold_provider_outcome(AiProviderRunInterruptOutcome::RequestedSettled) + .with_durable_turn_evidence(true); + assert_eq!(settled, AiRunInterruptSettlement::Settled); + assert!(settled.retains_thread()); + + // Acknowledgement without adapter proof is never settlement. + let acknowledged = AiRunInterruptSettlement::NotActive + .fold_provider_outcome(AiProviderRunInterruptOutcome::Requested) + .with_durable_turn_evidence(true); + assert_eq!(acknowledged, AiRunInterruptSettlement::RequestedUnsettled); + assert!(!acknowledged.retains_thread()); + + // Uncertain persisted output demotes an adapter-proven settlement. + let uncertain = AiRunInterruptSettlement::NotActive + .fold_provider_outcome(AiProviderRunInterruptOutcome::RequestedSettled) + .with_durable_turn_evidence(false); + assert_eq!(uncertain, AiRunInterruptSettlement::RequestedUnsettled); + assert!(!uncertain.retains_thread()); + + // An inert adapter still reports no live resource. + assert_eq!( + AiRunInterruptSettlement::NotActive + .fold_provider_outcome(AiProviderRunInterruptOutcome::NotActive) + .with_durable_turn_evidence(true), + AiRunInterruptSettlement::NotActive + ); + } + + #[test] + fn one_unsettled_adapter_keeps_the_whole_interruption_unsettled() { + let mixed = AiRunInterruptSettlement::NotActive + .fold_provider_outcome(AiProviderRunInterruptOutcome::RequestedSettled) + .fold_provider_outcome(AiProviderRunInterruptOutcome::Requested); + assert_eq!(mixed, AiRunInterruptSettlement::RequestedUnsettled); + + let reversed = AiRunInterruptSettlement::NotActive + .fold_provider_outcome(AiProviderRunInterruptOutcome::Requested) + .fold_provider_outcome(AiProviderRunInterruptOutcome::RequestedSettled); + assert_eq!(reversed, AiRunInterruptSettlement::RequestedUnsettled); + } } diff --git a/crates/graphql-orm-ai/src/provider_session.rs b/crates/graphql-orm-ai/src/provider_session.rs index c2b7cd3b..38d83ebd 100644 --- a/crates/graphql-orm-ai/src/provider_session.rs +++ b/crates/graphql-orm-ai/src/provider_session.rs @@ -1163,6 +1163,32 @@ impl AiProviderSessionCommit { } } +/// Derives the transcript fingerprint of a thread that retained one exact +/// interrupted user prompt and no reply. +/// +/// The value chains the binding's previous fingerprint, so it is an opaque +/// crate-owned continuation of the same commit chain rather than anything a +/// host can recompute from message content alone. Hosts echo the stored +/// fingerprint from the binding view into the next turn plan exactly as they +/// do after an ordinary commit. +#[cfg(any(feature = "sqlite", feature = "postgres"))] +pub(crate) fn interrupted_turn_transcript_fingerprint( + previous_transcript_fingerprint: &str, + session_id: AiSessionId, + run_id: AiRunId, + input_message_id: Uuid, + through_message_sequence: i64, +) -> String { + let mut digest = Sha256::new(); + digest.update(b"graphql-orm-ai/provider-session-interrupted-turn/v1\0"); + digest.update(previous_transcript_fingerprint.as_bytes()); + digest.update(session_id.0.as_bytes()); + digest.update(run_id.0.as_bytes()); + digest.update(input_message_id.as_bytes()); + digest.update(through_message_sequence.to_be_bytes()); + hex::encode(digest.finalize()) +} + /// Fenced cleanup claim containing no opened provider cursor. #[derive(Clone, Debug, PartialEq, Eq)] pub struct AiProviderSessionCleanupClaim { @@ -1488,6 +1514,47 @@ pub trait AiProviderSessionService: Send + Sync { commit: AiProviderSessionCommit, ) -> Result; + /// Retains one exact binding across an interrupt the provider settled. + /// + /// This is the only path that releases a claim without an assistant + /// message. It is admissible because a settled interrupt leaves the + /// retained thread holding exactly the interrupted user prompt and no + /// reply, which is what the durable transcript also records for a + /// cancelled run. The implementation must therefore re-prove, from + /// committed rows and inside the same transaction, that: + /// + /// - the caller's `settlement` reports [`AiRunInterruptSettlement::Settled`]; + /// - the binding is claimed by exactly this run, attempt, lease + /// generation, and claim owner, and the claim has not expired; + /// - the run persisted no assistant message and no run checkpoint, so no + /// output of the interrupted turn is persisted or uncertainly persisted; + /// and + /// - the durable input message is the exact next message after the + /// binding's watermark. + /// + /// It then advances the watermark to that input message, derives a new + /// transcript fingerprint, and returns the binding to a resumable state + /// with no claim. Anything short of the full proof must fail, and the + /// caller must fall back to [`Self::require_cleanup_for_run`]. + /// + /// The default denies, so an alternate store never acquires interrupt + /// retention implicitly. + /// + /// [`AiRunInterruptSettlement::Settled`]: crate::AiRunInterruptSettlement::Settled + /// + /// # Errors + /// + /// Returns an error when settlement, the run fence, current + /// principal/session access, claim ownership, durable output absence, or + /// the transcript watermark cannot be validated exactly. + async fn settle_interrupted_turn( + &self, + _lease: &AiRunLease, + _settlement: crate::AiRunInterruptSettlement, + ) -> Result { + Err(AiError::RuntimeNotReady) + } + /// Irreversibly removes a claim from reuse after cancellation, transport /// ambiguity, cursor rejection, policy drift, or another safe reason. async fn require_cleanup( @@ -1496,6 +1563,30 @@ pub trait AiProviderSessionService: Send + Sync { reason_code: &str, ) -> Result<(), AiError>; + /// Invalidates the binding claimed by one exact run fence. + /// + /// This is [`Self::require_cleanup`] for a caller that observed durable + /// cancellation or an unsettled interrupt and therefore never received the + /// claim value — the claim is identified by the run fence itself. It must + /// apply the same state transition and the same retained-thread + /// invalidation disclosure event, so a user whose model context was reset + /// still learns about it. + /// + /// The default denies rather than silently leaving a claimed binding to + /// expire. + /// + /// # Errors + /// + /// Returns an error when no binding is claimed by the exact run fence, the + /// reason code is unsafe, or the transition cannot be committed. + async fn require_cleanup_for_run( + &self, + _lease: &AiRunLease, + _reason_code: &str, + ) -> Result<(), AiError> { + Err(AiError::RuntimeNotReady) + } + /// Claims one expired, invalidated, or deletion-required binding. async fn claim_cleanup( &self, diff --git a/crates/graphql-orm-ai/src/providers/codex_app_server.rs b/crates/graphql-orm-ai/src/providers/codex_app_server.rs index 265692d3..dbba26c8 100644 --- a/crates/graphql-orm-ai/src/providers/codex_app_server.rs +++ b/crates/graphql-orm-ai/src/providers/codex_app_server.rs @@ -12,7 +12,7 @@ use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; use std::time::Duration; use agql_auth::Clock; @@ -1603,9 +1603,78 @@ struct RunEntry { turn_count: AtomicU32, turn_active: AtomicBool, poisoned: AtomicBool, + /// Dynamic tool calls currently executing through the coordinator-owned + /// responder for this exact run process. + dynamic_tool_calls_in_flight: AtomicU64, + /// Closes the race between beginning interruption and checking the + /// in-flight counter after the provider acknowledgement. + interrupt_started: AtomicBool, + /// Remembers a tool call dispatched after interruption began. Once set it + /// is never cleared: that ordering is outside the measured settlement + /// evidence even if the responder finishes before the acknowledgement. + dynamic_tool_call_after_interrupt: AtomicBool, empty_thread: Mutex, } +/// Counts coordinator-owned dynamic tool calls for one exact run process. +/// +/// The wrapper adds no authority: it forwards the exact call to the +/// coordinator responder and returns its exact result. It exists only so the +/// adapter can prove, at interrupt time, that the interrupted turn has no +/// unresolved dynamic tool call and did not dispatch one after interruption +/// began. +struct DynamicToolCallCounter { + entry: Arc, + inner: Arc, +} + +/// Decrements the in-flight counter even when the responder future is dropped +/// by interruption or task cancellation. +struct DynamicToolCallInFlightGuard { + entry: Arc, +} + +impl Drop for DynamicToolCallInFlightGuard { + fn drop(&mut self) { + self.entry + .dynamic_tool_calls_in_flight + .fetch_sub(1, Ordering::SeqCst); + } +} + +impl DynamicToolCallCounter { + fn wrap( + entry: &Arc, + inner: Arc, + ) -> Arc { + Arc::new(Self { + entry: entry.clone(), + inner, + }) + } +} + +#[async_trait] +impl ProviderDynamicToolResponder for DynamicToolCallCounter { + async fn respond( + &self, + call: ProviderDynamicToolCall, + ) -> Result { + self.entry + .dynamic_tool_calls_in_flight + .fetch_add(1, Ordering::SeqCst); + let _in_flight = DynamicToolCallInFlightGuard { + entry: self.entry.clone(), + }; + if self.entry.interrupt_started.load(Ordering::SeqCst) { + self.entry + .dynamic_tool_call_after_interrupt + .store(true, Ordering::SeqCst); + } + self.inner.respond(call).await + } +} + enum EmptyThreadActivation { Vacant, Creating, @@ -2033,6 +2102,7 @@ impl AiCodexAppServerRunPool { let entry = self .begin_bound_turn(binding, ®istration, &session, &input) .await?; + let responder = DynamicToolCallCounter::wrap(&entry, responder); let turn_deadline = tokio::time::Instant::now() + self.inner.limits.turn_timeout; let stream = match tokio::time::timeout_at( turn_deadline, @@ -2346,6 +2416,7 @@ impl AiCodexAppServerRunPool { entry.turn_active.store(false, Ordering::Release); return Err(ProviderError::RateLimited); } + let responder = DynamicToolCallCounter::wrap(&entry, responder); let turn_deadline = tokio::time::Instant::now() + self.inner.limits.turn_timeout; let stream = match tokio::time::timeout_at( turn_deadline, @@ -2448,6 +2519,7 @@ impl AiCodexAppServerRunPool { entry.turn_active.store(false, Ordering::Release); return Err(ProviderError::RateLimited); } + let responder = DynamicToolCallCounter::wrap(&entry, responder); let turn_deadline = tokio::time::Instant::now() + self.inner.limits.turn_timeout; let stream = match tokio::time::timeout_at( turn_deadline, @@ -2573,6 +2645,9 @@ impl AiCodexAppServerRunPool { turn_count: AtomicU32::new(0), turn_active: AtomicBool::new(false), poisoned: AtomicBool::new(false), + dynamic_tool_calls_in_flight: AtomicU64::new(0), + interrupt_started: AtomicBool::new(false), + dynamic_tool_call_after_interrupt: AtomicBool::new(false), empty_thread: Mutex::new(EmptyThreadActivation::Vacant), }); if identity_is_new { @@ -2617,6 +2692,31 @@ impl AiCodexAppServerRunPool { /// This method contains no cancellation authority. Callers must first /// observe authoritative durable cancellation or lease loss. /// + /// # Settlement provenance + /// + /// A confirmed interrupt of a turn with no unresolved dynamic tool call + /// reports [`AiProviderRunInterruptOutcome::RequestedSettled`]. That is + /// **version-observed behaviour, not a protocol promise**: + /// `TurnInterruptResponse` is an empty object and nothing in the app-server + /// contract guarantees discard. It rests on a measurement against + /// `codex-cli 0.148.0` with `gpt-5.4`, interrupting mid-assistant-stream, + /// which found that the interrupted partial turn is absent from + /// `turn/completed` (`status=interrupted`, `items=0`), from + /// `thread/turns/list` (the interrupted turn holds only its `userMessage`), + /// from the on-disk rollout JSONL a later resume reloads, and from the + /// model's context when the resumed thread is asked to quote its last line. + /// The interrupted user message *is* retained, so a settled interrupt + /// leaves a prompt with no reply — consistent with a durable transcript + /// that recorded the same prompt and no answer. + /// + /// **Re-verify this on a Codex upgrade.** Behaviour with a dynamic tool + /// call in flight was never measured, so an unresolved call or one first + /// dispatched after interruption begins keeps the fail-closed + /// [`AiProviderRunInterruptOutcome::Requested`]. A call that completed + /// before interruption is not unresolved; the coordinator's separate + /// durable-output guard still decides whether that turn may retain its + /// binding. + /// /// # Errors /// /// Returns a non-sensitive provider error when bounded interruption cannot @@ -2632,12 +2732,20 @@ impl AiCodexAppServerRunPool { if !entry.turn_active.load(Ordering::Acquire) { return Ok(AiProviderRunInterruptOutcome::NotActive); } + entry.interrupt_started.store(true, Ordering::SeqCst); tokio::time::timeout( self.inner.limits.interrupt_timeout, entry.process.interrupt(), ) .await .map_err(|_| provider_timeout_error())??; + if entry.dynamic_tool_calls_in_flight.load(Ordering::SeqCst) == 0 + && !entry + .dynamic_tool_call_after_interrupt + .load(Ordering::SeqCst) + { + return Ok(AiProviderRunInterruptOutcome::RequestedSettled); + } Ok(AiProviderRunInterruptOutcome::Requested) } @@ -8582,6 +8690,94 @@ pub(crate) mod tests { assert_eq!(counters.turns.load(Ordering::SeqCst), 1); } + /// A dynamic tool call that has not been answered is outside the measured + /// interrupt evidence, so the adapter refuses to report settlement. + struct BlockingDynamicResponder { + entered: Arc, + } + + #[async_trait] + impl ProviderDynamicToolResponder for BlockingDynamicResponder { + async fn respond( + &self, + _call: ProviderDynamicToolCall, + ) -> Result { + self.entered.notify_one(); + std::future::pending::<()>().await; + Err(ProviderError::Rejected) + } + } + + #[tokio::test] + async fn unresolved_dynamic_tool_call_keeps_the_interrupt_unsettled() { + let counters = Arc::new(Counters::new()); + let provider: Arc = Arc::new(AiCodexAppServerProvider::new( + dynamic_registration("1.0.0"), + pool(counters.clone(), 1, 2), + )); + let request = dynamic_model_request(); + let context = provider_context("profile-1", &request); + let binding = context + .run_binding() + .expect("test context should carry the exact run binding"); + let entered = Arc::new(tokio::sync::Notify::new()); + let responder = Arc::new(BlockingDynamicResponder { + entered: entered.clone(), + }); + let started = tokio::spawn({ + let provider = provider.clone(); + async move { + let _ = provider + .stream_with_dynamic_tools(request, context, responder) + .await; + } + }); + entered.notified().await; + + assert_eq!( + provider + .interrupt_run(&binding) + .await + .expect("interrupt should dispatch"), + AiProviderRunInterruptOutcome::Requested + ); + + started.abort(); + let _ = started.await; + assert_eq!(counters.interrupts.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn resolved_dynamic_tool_call_has_no_unresolved_adapter_work() { + let counters = Arc::new(Counters::new()); + let provider: Arc = Arc::new(AiCodexAppServerProvider::new( + dynamic_registration("1.0.0"), + pool(counters.clone(), 1, 2), + )); + let request = dynamic_model_request(); + let context = provider_context("profile-1", &request); + let binding = context + .run_binding() + .expect("test context should carry the exact run binding"); + let active = provider + .stream_with_dynamic_tools(request, context, Arc::new(FakeDynamicResponder)) + .await + .expect("dynamic provider turn should start"); + + // The turn answered its dynamic tool call before interruption, so the + // adapter has no unresolved tool work. The coordinator's independent + // durable-output leg will still demote this settlement before retaining + // a binding that persisted tool traffic. + assert_eq!( + provider + .interrupt_run(&binding) + .await + .expect("interrupt should dispatch"), + AiProviderRunInterruptOutcome::RequestedSettled + ); + drop(active); + } + #[tokio::test] async fn active_dynamic_turn_uses_exact_interrupt_and_close_lifecycle() { let counters = Arc::new(Counters::new()); @@ -8599,12 +8795,14 @@ pub(crate) mod tests { .stream_with_dynamic_tools(request, context, Arc::new(FakeDynamicResponder)) .await .expect("dynamic provider turn should start"); + // No dynamic tool call was dispatched, so this interrupt is inside the + // measured evidence and reports settlement. assert_eq!( provider .interrupt_run(&binding) .await .expect("dynamic interrupt should dispatch"), - AiProviderRunInterruptOutcome::Requested + AiProviderRunInterruptOutcome::RequestedSettled ); assert_eq!( provider @@ -8641,7 +8839,7 @@ pub(crate) mod tests { .interrupt_run(&binding) .await .expect("interrupt should dispatch"), - AiProviderRunInterruptOutcome::Requested + AiProviderRunInterruptOutcome::RequestedSettled ); assert_eq!( provider @@ -8806,7 +9004,7 @@ pub(crate) mod tests { pool.interrupt_run(&binding) .await .expect("interrupt should succeed"), - AiProviderRunInterruptOutcome::Requested + AiProviderRunInterruptOutcome::RequestedSettled ); assert_eq!(counters.interrupts.load(Ordering::SeqCst), 1); assert_eq!( diff --git a/crates/graphql-orm-ai/src/runtime.rs b/crates/graphql-orm-ai/src/runtime.rs index 7cec5943..f7d980c5 100644 --- a/crates/graphql-orm-ai/src/runtime.rs +++ b/crates/graphql-orm-ai/src/runtime.rs @@ -1073,7 +1073,41 @@ impl AiRuntime { &self, binding: &crate::AiProviderRunBinding, ) -> Result { + self.interrupt_provider_runs(binding) + .await + .map(|(requested, _)| requested) + } + + /// Requests interruption and reports the aggregate settlement proof. + /// + /// This is the proof-carrying counterpart to + /// [`Self::interrupt_all_provider_runs`]. It visits the same complete + /// adapter set, but returns what those adapters proved rather than the + /// count that accepted the request. Aggregation is fail-closed: one adapter + /// that merely acknowledged keeps the whole result + /// [`crate::AiRunInterruptSettlement::RequestedUnsettled`], and settlement + /// still needs the caller's durable evidence through + /// [`crate::AiRunInterruptSettlement::with_durable_turn_evidence`]. + /// + /// # Errors + /// + /// Returns the first non-sensitive provider error after attempting every + /// registered adapter. + pub async fn interrupt_all_provider_runs_with_settlement( + &self, + binding: &crate::AiProviderRunBinding, + ) -> Result { + self.interrupt_provider_runs(binding) + .await + .map(|(_, settlement)| settlement) + } + + async fn interrupt_provider_runs( + &self, + binding: &crate::AiProviderRunBinding, + ) -> Result<(u64, crate::AiRunInterruptSettlement), ProviderError> { let mut requested = 0_u64; + let mut settlement = crate::AiRunInterruptSettlement::NotActive; let mut first_error = None; for (kind, provider) in &self.providers { if provider.provider_kind() != *kind { @@ -1085,16 +1119,22 @@ impl AiRuntime { continue; } match provider.interrupt_run(binding).await { - Ok(crate::AiProviderRunInterruptOutcome::Requested) => { - requested = requested.saturating_add(1); + Ok(outcome) => { + if matches!( + outcome, + crate::AiProviderRunInterruptOutcome::Requested + | crate::AiProviderRunInterruptOutcome::RequestedSettled + ) { + requested = requested.saturating_add(1); + } + settlement = settlement.fold_provider_outcome(outcome); } - Ok(crate::AiProviderRunInterruptOutcome::NotActive) => {} Err(error) => { first_error.get_or_insert(error); } } } - first_error.map_or(Ok(requested), Err) + first_error.map_or(Ok((requested, settlement)), Err) } /// Closes every registered adapter resource for one exact run binding. diff --git a/crates/graphql-orm-ai/tests/provider_sessions.rs b/crates/graphql-orm-ai/tests/provider_sessions.rs index 5eecd51d..738e63c5 100644 --- a/crates/graphql-orm-ai/tests/provider_sessions.rs +++ b/crates/graphql-orm-ai/tests/provider_sessions.rs @@ -113,6 +113,7 @@ fn principal(subject: &str) -> AuthPrincipal { struct ProviderSessionFixture { sessions: OrmAiSessionService, runs: OrmAiRunService, + cancellation: Arc, provider_sessions: OrmAiProviderSessionService, owner: AuthPrincipal, other: AuthPrincipal, @@ -166,12 +167,25 @@ async fn provider_session_fixture() -> ProviderSessionFixture { protection_resolver.clone(), content_protector.clone(), ); + let cancellation_hub = + Arc::new(AiRunCancellationHub::new(32).expect("cancellation hub should validate")); let runs = OrmAiRunService::new( database.clone(), clock.clone(), AiRunServiceLimits::new(Duration::minutes(1), Duration::minutes(1), 16, 3, 3) .expect("run limits should validate"), - ); + ) + .with_cancellation_hub(cancellation_hub.clone()); + let cancellation = Arc::new(OrmAiRunCancellationService::new( + database.clone(), + access_policy.clone(), + protection_resolver.clone(), + content_protector.clone(), + principal_resolver.clone(), + clock.clone(), + AiRunCancellationLimits::default(), + cancellation_hub, + )); let provider_sessions = OrmAiProviderSessionService::new( database, access_policy, @@ -186,6 +200,7 @@ async fn provider_session_fixture() -> ProviderSessionFixture { ProviderSessionFixture { sessions, runs, + cancellation, provider_sessions, owner, other, @@ -523,6 +538,85 @@ async fn provider_session_cursor_is_owner_run_fenced_revocable_and_exactly_clean )); } +#[tokio::test] +async fn settled_interrupt_advances_the_durable_watermark_and_reuses_the_thread() { + let fixture = provider_session_fixture().await; + let owner = fixture.owner.clone(); + let interrupted = active_run(&fixture, &owner, "settled-interrupt-workspace").await; + let descriptor = AiProviderSessionDescriptor::new( + ProviderKind::LocalHarness, + "reviewed-local-profile", + "reviewed-model", + "a".repeat(64), + "codex-app-server/v2", + "b".repeat(64), + ) + .expect("provider descriptor should validate"); + let initial_fingerprint = "c".repeat(64); + let claim = fixture + .provider_sessions + .bind_for_run( + &interrupted, + AiProviderSessionBindRequest::new( + descriptor.clone(), + AiProviderSessionCursor::new("codex.thread", "settled-interrupt-thread") + .expect("cursor should validate"), + initial_fingerprint.clone(), + Some(fixture.clock.now() + Duration::hours(1)), + ) + .expect("bind request should validate"), + ) + .await + .expect("empty provider thread should bind to the interrupted run"); + assert_eq!(claim.through_message_sequence(), 0); + + fixture + .cancellation + .request_cancellation( + &owner, + CancelAiRunInput { + session_id: interrupted.session_id().0, + run_id: interrupted.run_id().0, + client_request_id: Uuid::new_v4(), + }, + ) + .await + .expect("owner cancellation should become durable"); + + assert!(matches!( + fixture + .provider_sessions + .settle_interrupted_turn(&interrupted, AiRunInterruptSettlement::RequestedUnsettled,) + .await, + Err(AiError::Conflict) + )); + let settled = fixture + .provider_sessions + .settle_interrupted_turn(&interrupted, AiRunInterruptSettlement::Settled) + .await + .expect("exact cancelled run with no output should settle"); + assert_eq!(settled.state(), AiProviderSessionState::Active); + assert_eq!(settled.through_message_sequence(), 1); + assert_ne!(settled.transcript_fingerprint(), initial_fingerprint); + + let next = next_active_run(&fixture, &owner, interrupted.session_id()).await; + let resumed = fixture + .provider_sessions + .claim_for_run(&next, &descriptor, settled.transcript_fingerprint()) + .await + .expect("the next run should claim the same settled provider thread"); + assert_eq!(resumed.through_message_sequence(), 1); + let opened = fixture + .provider_sessions + .open_for_run(&next, &resumed) + .await + .expect("the retained cursor should reopen under the next exact fence"); + assert_eq!( + opened.cursor().expose_to_provider_adapter(), + "settled-interrupt-thread" + ); +} + #[tokio::test] async fn exact_absence_authorizes_one_fenced_rebind_with_a_fresh_cursor() { let fixture = provider_session_fixture().await; diff --git a/crates/graphql-orm-ai/tests/schema_module.rs b/crates/graphql-orm-ai/tests/schema_module.rs index 6fd17444..5ae69679 100644 --- a/crates/graphql-orm-ai/tests/schema_module.rs +++ b/crates/graphql-orm-ai/tests/schema_module.rs @@ -8,7 +8,7 @@ fn ai_schema_module_owns_only_reserved_namespace_tables() { assert_eq!(catalog.modules().len(), 1); assert_eq!(catalog.modules()[0].version, AI_SCHEMA_MODULE_VERSION); - assert_eq!(AI_SCHEMA_MODULE_VERSION, "0.61.0"); + assert_eq!(AI_SCHEMA_MODULE_VERSION, "0.62.0"); assert_eq!(catalog.entities().len(), 47); assert!( catalog diff --git a/docs/reference/workspace-packages.md b/docs/reference/workspace-packages.md index 2c7626ce..bb93e2e9 100644 --- a/docs/reference/workspace-packages.md +++ b/docs/reference/workspace-packages.md @@ -19,7 +19,7 @@ changes. | Package | Version | Path | Default features | Direct internal dependencies | | --- | --- | --- | --- | --- | | `graphql-orm` | `0.23.0` | `crates/graphql-orm` | `sqlite` | `graphql-orm-macros`, `graphql-orm-operation-catalog`, `graphql-orm-router-protocol` (dev-only) | -| `graphql-orm-ai` | `0.82.0` | `crates/graphql-orm-ai` | `sqlite` | `graphql-orm`, `graphql-orm-ai-tool-profiles`, `graphql-orm-storage` | +| `graphql-orm-ai` | `0.83.0` | `crates/graphql-orm-ai` | `sqlite` | `graphql-orm`, `graphql-orm-ai-tool-profiles`, `graphql-orm-storage` | | `graphql-orm-ai-tool-profiles` | `0.6.0` | `crates/graphql-orm-ai-tool-profiles` | none | `graphql-orm-operation-catalog`, `graphql-orm-router-protocol` (dev-only) | | `graphql-orm-backup` | `0.7.1` | `crates/graphql-orm-backup` | `local` | `graphql-orm` (optional), `graphql-orm-storage` | | `graphql-orm-macros` | `0.23.0` | `crates/graphql-orm-macros` | `sqlite` | none |