diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a9f4e787..c8111ca8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -7,6 +7,16 @@ on: - main workflow_dispatch: +# The companions job relinks every graphql-orm-ai test binary once per provider +# feature combination. With full debug info that exhausted the runner disk and +# the linker died with SIGBUS mid-link. `line-tables-only` keeps file and line +# numbers in failure backtraces, which is what CI diagnosis actually needs, +# at a fraction of the size. +env: + CARGO_PROFILE_DEV_DEBUG: line-tables-only + CARGO_PROFILE_TEST_DEBUG: line-tables-only + CARGO_INCREMENTAL: "0" + jobs: workspace-integrity: runs-on: ubuntu-latest @@ -158,11 +168,12 @@ jobs: manifest-path: crates/graphql-orm/Cargo.toml baseline-root: ../graphql-orm-baseline/crates/graphql-orm feature-group: default-features - - uses: obi1kenobi/cargo-semver-checks-action@v2 - with: - manifest-path: crates/graphql-orm-macros/Cargo.toml - baseline-root: ../graphql-orm-baseline/crates/graphql-orm-macros - feature-group: default-features + # graphql-orm-macros is deliberately absent. It is a proc-macro crate, so + # it exposes no library API surface for cargo-semver-checks to analyze; + # newer versions of the tool now fail the job outright rather than + # silently checking nothing. scripts/check-semver.sh has excluded it for + # the same reason. Macro compatibility is covered by the aligned + # package-version gate and the full compile/trybuild fixture matrix. - if: steps.companion-baseline.outputs.exists == 'true' uses: obi1kenobi/cargo-semver-checks-action@v2 with: diff --git a/Cargo.lock b/Cargo.lock index 6d78a05a..e945c4b4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3104,7 +3104,7 @@ dependencies = [ [[package]] name = "graphql-orm-ai" -version = "0.81.0" +version = "0.82.0" dependencies = [ "agql-auth", "async-graphql", @@ -3126,6 +3126,7 @@ dependencies = [ "serde_json", "sha2 0.10.9", "subtle", + "tempfile", "thiserror 2.0.18", "time", "tokio", diff --git a/crates/graphql-orm-ai/CHANGELOG.md b/crates/graphql-orm-ai/CHANGELOG.md index c087a495..4e432401 100644 --- a/crates/graphql-orm-ai/CHANGELOG.md +++ b/crates/graphql-orm-ai/CHANGELOG.md @@ -18,6 +18,69 @@ 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.82.0] - 2026-08-20 + +Persistent schema module: **0.61.0**. + +### Added + +- `retryAiRun` and `acknowledgeAiRunFailure` dispose of a failed or + recovery-required run. Retry authors a *new* run over the same durable user + message under current policy and is idempotent under a client request id; + acknowledge durably dismisses the failure. Neither mutates or deletes the + source run, so its state, immutable attempt outcomes, and session and inbox + events survive. At most one disposition wins per run. +- `run_failed` and `run_recovery_required` events carry a bounded failure + record with a stable code, a retryable flag, and the admission reason. + Retry admission is computed from committed rows only: `RecoveryRequired` is + never retryable, an absent or unrecognized failure code is never retryable, + and a run that already produced a durable assistant message is refused. +- `provider_session_reset` and `provider_session_rebound` events disclose that + a retained provider thread stopped being usable, carrying only the + server-owned reason class. Previously the durable transcript read as + continuous while the model had silently lost all prior context. +- `AiSessionEventEnvelope.closed` carries a typed + [`AiSessionStreamClose`] on the final envelope of any server-ended session + stream, so a client can distinguish "stream over, resubscribe" from network + silence. +- `AiRunInterruptSettlement` reports what an interrupt proved about the turn it + stopped. It fails closed: only proven settlement retains a thread, and no + adapter currently reports it. +- `OrmAiSessionService::session_stream_head` performs one bounded authorized + head-sequence read, used as a durable fallback delivery path. + +### Changed + +- **Conversation bootstrap no longer fails while an assistant is streaming.** + The snapshot retry predicate compared `row_version` and `stream_head`, which + every coalesced live delta advances at roughly the coalescer rate, so the + bounded snapshot returned `Conflict` for exactly the sessions a user is most + likely to open. The predicate now covers only fields the bootstrap returns. + The watermark is a resume floor: nothing at or below it is missing, run and + tool-call rows may already reflect a later event, and the message window + never leads it. +- Session-event subscriptions survive a brief authorization-service restart + within a bounded, per-session jittered grace window instead of failing on the + first `resolve` error. An authoritative denial, or any class this crate does + not recognize, still denies immediately. +- Session-event subscriptions run a periodic bounded durable head check, so + single-replica delivery no longer depends solely on the process-local wakeup + channel and a missed hint is no longer unrecoverable. +- A message accepted while provider-session cleanup is pending now converges + without an operator. Exhausting the bounded retry allowance closes the run as + a clean visible failure instead of leaving it to expire into + `RecoveryRequired`, which was misclassified because nothing had executed. +- `AiRunCompletion::outcome_code` is now readable. + +### Breaking + +- `AiAgentProviderTurnExecutor::interrupt_run` returns + `AiRunInterruptSettlement` instead of `()`. +- `AiSessionEventEnvelope` gained a `closed` field. Construct it through + `AiSessionEventEnvelope::delivered` or `AiSessionEventEnvelope::ended`. +- The persistent schema module advances to 0.61.0 for the new run-failure + disposition entity. + ## [0.81.0] - 2026-08-16 ### Added diff --git a/crates/graphql-orm-ai/Cargo.toml b/crates/graphql-orm-ai/Cargo.toml index 71197eab..228694f2 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.81.0" +version = "0.82.0" edition = "2024" authors = ["Toby Martin "] description = "Project-agnostic AI agent runtime for graphql-orm applications" @@ -62,4 +62,5 @@ url = "2" uuid = { version = "1", features = ["serde", "v4"] } [dev-dependencies] +tempfile = "3" tokio = { version = "1", features = ["fs", "io-util", "macros", "net", "rt-multi-thread", "sync", "time"] } diff --git a/crates/graphql-orm-ai/MIGRATION.md b/crates/graphql-orm-ai/MIGRATION.md index 362c5ee9..e72e6142 100644 --- a/crates/graphql-orm-ai/MIGRATION.md +++ b/crates/graphql-orm-ai/MIGRATION.md @@ -19,6 +19,71 @@ 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.81.0 to 0.82.0: session reliability and failure disposition + +Adopt `graphql-orm-ai` 0.82.0 at one reviewed full monorepo revision. + +### Schema module + +The AI schema module advances **0.60.0 to 0.61.0** and adds one entity, +`graphql_orm_ai_run_failure_dispositions`, with a unique index on +`source_run_id` and a session/decision index. Apply and verify the module +before serving traffic. There is no backfill, no column change to an existing +table, and no protected-payload migration. Existing rows and events remain +readable. + +### Source-breaking changes + +`AiAgentProviderTurnExecutor::interrupt_run` now returns +`AiRunInterruptSettlement` instead of `()`. Existing implementations that +interrupt without proving settlement should return +`AiRunInterruptSettlement::RequestedUnsettled`, and one that finds no live +resource should return `NotActive`. Do not return `Settled` unless the adapter +can prove the interrupted turn left the provider's retained thread consistent +with the durable transcript; `retains_thread()` is the only thing that keeps a +binding, and it fails closed. + +`AiSessionEventEnvelope` gained a nullable `closed` field. Build envelopes with +`AiSessionEventEnvelope::delivered` or `AiSessionEventEnvelope::ended` instead +of struct literals. + +### Behavioural changes with no API change + +`conversation_bootstrap` no longer returns `Conflict` while an assistant is +streaming. Its `watermark` is now documented as a **resume floor** rather than +an equality point. Subscribe with `after_sequence = watermark`; no event at or +below it is missing, and the message window never leads it, but run and +tool-call rows may already reflect an event after it. Apply replayed events by +identifier so re-applying one the snapshot already reflects is idempotent. A +client that assumed every replayed event was unseen must be updated. + +Session-event streams now tolerate a briefly unavailable authorization +dependency within a bounded per-session jittered grace window, and emit a typed +close envelope before ending. Authoritative denials are unchanged: the stream +still fails immediately, and the existing `AiError` still follows the close +envelope, so a client reading only errors keeps working. + +A run whose provider-session cleanup stays pending past its retry allowance now +closes as `Failed` with `provider_session_cleanup_unavailable` instead of +expiring into `RecoveryRequired`. That code is retryable, because nothing +executed. + +### New GraphQL surface + +`retryAiRun` and `acknowledgeAiRunFailure` are additive; regenerate typed and +PascalCase clients. Install `Arc` in schema data +or both mutations return a configuration error. `OrmAiRunDispositionService` +is the generated-ORM implementation. + +Four event types are additive on the existing session stream: +`run_retry_queued`, `run_failure_acknowledged`, `provider_session_reset`, and +`provider_session_rebound`. Clients that reject unknown event types must be +updated to ignore them. + +The `run_failed` and `run_recovery_required` payloads advance from the tagged +`...-v1` shape to `...-v2` and carry a `failure` record. Readers accept both +shapes; a v1 payload written before this release stays readable. + ## 0.80.0 to 0.81.0: capability discovery and durable provider loops Adopt `graphql-orm-ai` 0.81.0 and diff --git a/crates/graphql-orm-ai/README.md b/crates/graphql-orm-ai/README.md index b3d2724e..0d3b0c52 100644 --- a/crates/graphql-orm-ai/README.md +++ b/crates/graphql-orm-ai/README.md @@ -28,7 +28,7 @@ for AI, ORM, storage, backup, and tool-profile packages: ```toml [dependencies] -graphql-orm-ai = { git = "https://github.com/Dastari/graphql-orm.git", rev = "", version = "0.81.0", default-features = false, features = ["sqlite"] } +graphql-orm-ai = { git = "https://github.com/Dastari/graphql-orm.git", rev = "", version = "0.82.0", default-features = false, features = ["sqlite"] } ``` Exactly one persistence backend is required: `sqlite` (default), `postgres`, @@ -118,6 +118,31 @@ authenticated registered replay source and the existing run queue; it rehydrates current authority at open, event and adoption boundaries. See [durable bounded subscription waits](docs/durable-subscription-waits.md). +## Session reliability + +One bounded `aiConversationBootstrap` snapshot plus durable event replay is the +supported way to open a conversation. Its watermark is a **resume floor**: +nothing at or below it is missing from the snapshot, the message window never +leads it, and run and tool-call rows may already reflect a later event, so +replayed events are applied by identifier. + +Session-event streams end with a typed close envelope rather than silence, +tolerate a briefly unavailable authorization dependency inside a bounded +jittered grace window while denying authoritative revocation immediately, and +run a periodic bounded durable head check so single-replica delivery does not +depend solely on the process-local wakeup channel. + +Terminal `run_failed` and `run_recovery_required` events carry a bounded, +content-free failure record with a stable code and a retryable flag computed +from committed rows. `retryAiRun` authors a new run over the same durable user +message under current policy where re-execution is provably safe; +`acknowledgeAiRunFailure` dismisses a failure without deleting audit history. +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. + +See the [session reliability adoption contract](docs/session-reliability-adoption.md). + ## Features and capability boundary | Feature | Default | Meaning | diff --git a/crates/graphql-orm-ai/docs/README.md b/crates/graphql-orm-ai/docs/README.md index 6a8e372a..02ed0ceb 100644 --- a/crates/graphql-orm-ai/docs/README.md +++ b/crates/graphql-orm-ai/docs/README.md @@ -36,6 +36,7 @@ tool-free, and network-free; production integration comes later. - [Attachments](attachments.md), [provider files](provider-files.md), [live streaming](live-streaming.md), [context compaction](context-compaction.md), [skills and UI intents](skills-and-ui-intents.md), and [remote GraphQL execution](remote-graphql-execution.md). - [Capability index, compact planning, durable broker and conversation bootstrap](capability-discovery-and-execution.md). +- [Session reliability adoption contract](session-reliability-adoption.md) — bootstrap watermark semantics, stream close envelopes, run failure records, retry/acknowledge, and retained-thread disclosure. - [Migration guide](../MIGRATION.md) and [changelog](../CHANGELOG.md). ## Concepts and operations diff --git a/crates/graphql-orm-ai/docs/capability-discovery-and-execution.md b/crates/graphql-orm-ai/docs/capability-discovery-and-execution.md index 2d822e03..4dddaa09 100644 --- a/crates/graphql-orm-ai/docs/capability-discovery-and-execution.md +++ b/crates/graphql-orm-ai/docs/capability-discovery-and-execution.md @@ -211,8 +211,31 @@ returns the session shell, newest messages plus backward cursor, durable watermark, active runs, recent terminal codes, related tool calls, safe provider activity and reset-required state. It never returns prompts, tool results, provider payloads, credentials or authorization details. The ORM -implementation uses a bounded optimistic snapshot and retries if the session -watermark changes during assembly. +implementation uses a bounded optimistic snapshot and retries only when a +field it actually returns changed during assembly. + +### The watermark is a resume floor + +The returned watermark is captured before the snapshot is assembled and is a +lower bound, not an equality point: + +- every durable effect at or before the watermark is reflected in the + snapshot, so subscribing with `after_sequence = watermark` cannot miss an + event; +- the message window never leads the watermark, because a new message changes + the message head and forces the snapshot to be reassembled; +- run and tool-call rows may already reflect an effect after the watermark. + Both are identified state keyed by row ID, so re-applying the replayed event + that produced them is idempotent. A client must apply replayed events by ID + rather than assuming every replayed event is unseen. + +Deliberately excluded from the retry predicate are the session stream head, +last-activity timestamp and CAS row version. A coalesced live delta advances +all three at roughly the streaming coalescer rate while an assistant is +answering, but appends only a session event and cannot change anything the +bootstrap returns. Including that churn made the bounded snapshot fail with +`Conflict` for exactly the sessions a user is most likely to open, which a +client cannot distinguish from a disconnection. A client renders the snapshot, begins durable event replay strictly after the returned watermark, drains to the captured/current head, then attaches live diff --git a/crates/graphql-orm-ai/docs/session-reliability-adoption.md b/crates/graphql-orm-ai/docs/session-reliability-adoption.md new file mode 100644 index 00000000..86b460b7 --- /dev/null +++ b/crates/graphql-orm-ai/docs/session-reliability-adoption.md @@ -0,0 +1,226 @@ +--- +title: "Session reliability adoption contract" +kind: reference +status: active +owner: graphql-orm-ai-maintainers +last_reviewed: 2026-08-20 +review_by: 2027-02-01 +supersedes: [] +--- + +# Session reliability adoption contract + +This is the exact contract for the 0.82.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 +facts. + +The guardrail behind every decision here is unchanged: **not every error is +recoverable.** Where it is uncertain whether protected output was persisted, +whether a provider saw a response, or whether a consequential operation +occurred, the run keeps `RecoveryRequired` and nothing below fabricates an +absence proof, a cursor, or settled state. + +## Conversation bootstrap: the watermark is a resume floor + +`conversation_bootstrap` previously reassembled its snapshot whenever the +session's `row_version` or `stream_head` changed between two reads. Every +coalesced live delta advances both at roughly the streaming coalescer rate, so +the bounded snapshot failed structurally for any session with an assistant +currently answering, and the caller saw `Conflict`. + +The retry predicate now covers only fields the bootstrap actually returns. The +stream head, last-activity timestamp, and CAS version are excluded: a live +delta appends a session event and changes nothing in the returned payload. + +The resulting watermark contract is: + +| Guarantee | Holds | +| --- | --- | +| Every durable effect at or before `watermark` is in the snapshot | Yes | +| The message window may lead `watermark` | No | +| Run and tool-call rows may lead `watermark` | Yes | +| `after_sequence = watermark` can miss an event | No | + +**What a client must do.** Subscribe with `after_sequence = watermark`, exactly +as before. Apply replayed events *by identifier*. A run or tool-call event +replayed just after the snapshot may describe a row the snapshot already +reflects; both are identified state, so re-applying is idempotent. A client +that assumed every replayed event was previously unseen must be updated. +Messages are unaffected, because a new message changes the message head and +forces the snapshot to be reassembled. + +## Session-event streams: typed close, reauthorization grace, durable fallback + +`AiSessionEventEnvelope` gained `closed: Option`. It is +`None` on every ordinary delivery and set on the final envelope of any stream +the server ended: + +| Value | Meaning | Client action | +| --- | --- | --- | +| `ResetRequired` | Retention removed needed history | Discard derived state, reload from bootstrap | +| `WakeupChannelClosed` | Host wakeup channel closed, normally shutdown | Resubscribe from the last delivered watermark | +| `AuthorizationRevoked` | Authoritative denial, or session no longer visible | Do not resubscribe with the same credentials | +| `ReauthorizationUnavailable` | Grace window expired with the dependency down | Resubscribe after backing off | + +A stream that ends with no close envelope and no error is a client unsubscribe +or a transport failure. `reset_required` stays coupled to `ResetRequired`, and +the existing `AiError` still follows the close envelope, so a client reading +only the boolean or only the error keeps working. + +Reauthorization now distinguishes an unavailable dependency from a denial. +Only `AuthServiceUnavailable`, `Store`, `AuthThrottled`, and `AuthLocked` are +retried, inside a bounded grace window with per-session jittered backoff so one +authorization restart cannot produce a synchronized bootstrap storm. Every +other class, including one this crate does not recognize, denies immediately. +Both bounds are crate-owned and configurable through +`with_reauthorization_grace` and `with_replay_check_interval`. + +A periodic bounded head-sequence read (`session_stream_head`) now runs +independently of the wakeup channel. Single-replica delivery therefore no +longer depends solely on an in-process `tokio::broadcast`, and a dropped or +missed hint is recoverable rather than terminal. + +**What multi-replica delivery would additionally require**, and what this +release deliberately does not supply: the head check makes *one* replica +self-healing, but the wakeup hint is still process-local, so a subscriber on +replica A learns about a commit from replica B only at the next poll. Bounded +staleness, not loss. Real multi-replica delivery needs a cross-process commit +notification — a database `LISTEN`/`NOTIFY` channel or an external bus — fanned +out to subscribers, plus a shared retention floor so a replica cannot serve a +watermark another replica has already pruned past. Neither is in this release. + +## Failed and recovery-required runs + +`run_failed` and `run_recovery_required` events already existed and were +already emitted in the same transaction as the terminal run write, through the +same sequenced replayable channel as every other session event. What they +lacked was any classification. + +The payload advances from the tagged `...-v1` shape to `...-v2` and adds a +`failure` record, mirroring the existing safe failure envelope: + +```json +{ + "version": 1, + "ok": false, + "code": "provider_turn_uncertain", + "retryable": false, + "admission": "refused_uncertain" +} +``` + +`failure` is `null` for `run_completed`. Readers accept v1 and v2 and fail +closed on anything else, so events written before this release stay readable. +The record carries only server-owned classification, never provider content. + +**Retry admission** is computed from committed rows and means "a new run may be +authored for the same durable user message", not a state-machine transition: + +| Terminal state | Admission | +| --- | --- | +| `RecoveryRequired` | Never. Re-execution is what the guardrail forbids | +| `Completed` | Never. The message already has its answer | +| `Cancelled` | Only when the run produced no durable assistant message | +| `Failed` | Only for an explicitly allowlisted, proven-clean code | + +`Cancelled` is deliberately not a class. Cancellation is observed at two +points with opposite correct answers: after the assistant output was durably +persisted the message is fully answered and a second run would produce a second +answer, while before persistence the result was discarded and retry is +meaningful. Read the flag rather than the event type. + +An absent or unrecognized failure code is refused. The allowlist is opt-in, so +adding a new failure classification cannot silently make it retryable. + +## Retry and acknowledge + +`retryAiRun` and `acknowledgeAiRunFailure` require +`Arc` in schema data; +`OrmAiRunDispositionService` is the generated-ORM implementation. + +Retry authors a new `queued` run over the **same** `input_message_id`. It never +duplicates the prompt, never resumes the source run, and carries a fresh +principal reference so it executes under current policy rather than the +authority the source run captured. Admission is re-decided from committed rows +inside the same transaction that authors the new run. + +Acknowledge is always available for a terminal failed or recovery-required run, +*including one whose retry is refused*: dismissing a failure asserts nothing +about whether re-execution would be safe. + +Both are idempotent under `clientRequestId` and at most one disposition wins +per run; a second key for an already-disposed run conflicts. Neither deletes a +row or an event, so the source run, its immutable attempt outcomes, and its +durable session and inbox events all survive. `run_retry_queued` and +`run_failure_acknowledged` are appended to the ordinary session stream. + +## Retained provider sessions + +**Every invalidation is now disclosed.** Each funnel that marks a retained +binding cleanup-required, and each explicit rebind, appends a durable session +event carrying only the server-owned reason class: + +- `provider_session_reset` — the retained thread stopped being usable; +- `provider_session_rebound` — a new binding replaced a prior one. + +This closes the case where the durable transcript rendered as continuous while +the model had silently lost all prior context. The payload uses the existing +content-free tagged envelope, so it discloses no cursor, prompt, provider +payload, tool argument, or authorization detail. A host should tell the user +the model's context was reset when it sees either event. + +Reason classes worth distinguishing when rendering: cancellation after a turn, +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. + +## Messages accepted during cleanup + +A message accepted while provider-session cleanup is pending converges without +operator intervention, and did so across a host restart before this release: +the deferred turn is scheduled as a durable retry, and `claim_next` reclaims +queued and retry-scheduled runs. The `Deferred` outcome is a report, not the +delivery mechanism, so delivery never depended on it reaching the executor. + +What changed is the end of that allowance. Exhausting the bounded retry +allowance while cleanup stayed pending previously propagated a conflict and +left the run running until its lease expired into `RecoveryRequired` — both +misclassified, because nothing had executed, and stuck until an operator looked +at it. The run now closes as `Failed` with +`provider_session_cleanup_unavailable`, which the classifier admits for retry. +A stale fence still fails the terminal write, so ordinary expired-lease +reconciliation keeps owning that case. + +## Summary of public API changes + +| Item | Change | +| --- | --- | +| `AiSessionEventEnvelope` | Added `closed`; construct via `delivered`/`ended` | +| `AiSessionStreamClose` | New enum | +| `AiRunInterruptSettlement` | New enum; `interrupt_run` returns it instead of `()` | +| `AiRunFailure`, `AiRunRetryAdmission`, `AiRunRetryEvidence`, `classify_run_retry` | New | +| `AiRunDisposition`, `AiRunDispositionView`, `AiRunRetryRefusal` | New | +| `RetryAiRunInput`, `AcknowledgeAiRunFailureInput` | New GraphQL inputs | +| `AiRunDispositionService`, `OrmAiRunDispositionService`, `AiRunDispositionLimits` | New | +| `OrmAiSessionService::session_stream_head` | New | +| `OrmAiSubscriptionService::with_reauthorization_grace`, `with_replay_check_interval` | New | +| `AiRunCompletion::outcome_code` | Now readable | diff --git a/crates/graphql-orm-ai/src/lib.rs b/crates/graphql-orm-ai/src/lib.rs index a8b7c23c..ecaa3890 100644 --- a/crates/graphql-orm-ai/src/lib.rs +++ b/crates/graphql-orm-ai/src/lib.rs @@ -77,6 +77,8 @@ mod orm_rules; #[cfg(any(feature = "sqlite", feature = "postgres"))] mod orm_run_cancellation; #[cfg(any(feature = "sqlite", feature = "postgres"))] +mod orm_run_disposition; +#[cfg(any(feature = "sqlite", feature = "postgres"))] mod orm_runs; #[cfg(any(feature = "sqlite", feature = "postgres"))] mod orm_session_retention; @@ -121,6 +123,7 @@ mod remote_execution; mod restore; mod rules; mod run_cancellation; +mod run_disposition; mod run_state; mod runtime; mod secrets; @@ -193,6 +196,8 @@ pub use orm_rules::*; #[cfg(any(feature = "sqlite", feature = "postgres"))] pub use orm_run_cancellation::*; #[cfg(any(feature = "sqlite", feature = "postgres"))] +pub use orm_run_disposition::*; +#[cfg(any(feature = "sqlite", feature = "postgres"))] pub use orm_runs::*; #[cfg(any(feature = "sqlite", feature = "postgres"))] pub use orm_session_retention::*; @@ -240,6 +245,7 @@ pub use remote_execution::*; pub use restore::*; pub use rules::*; pub use run_cancellation::*; +pub use run_disposition::*; pub use run_state::*; pub use runtime::*; pub use secrets::*; diff --git a/crates/graphql-orm-ai/src/orm_background.rs b/crates/graphql-orm-ai/src/orm_background.rs index 02934221..5f54329b 100644 --- a/crates/graphql-orm-ai/src/orm_background.rs +++ b/crates/graphql-orm-ai/src/orm_background.rs @@ -1539,8 +1539,14 @@ impl OrmAiOpenAiBackgroundSubmissionService { now, ) .await?; - append_terminal_run_event(tx, ¤t, AiRunState::RecoveryRequired, now) - .await?; + append_terminal_run_event( + tx, + ¤t, + AiRunState::RecoveryRequired, + Some(safe_error_code), + now, + ) + .await?; tx.insert::(CreateAiAuditEventRecordInput { actor_principal_kind: "system".to_owned(), actor_subject: "provider-background".to_owned(), @@ -3308,7 +3314,7 @@ async fn commit_background_terminal_graph( }) .await .map_err(OrmPublicError::from)?; - append_terminal_run_event(tx, &run, run_state, now).await?; + append_terminal_run_event(tx, &run, run_state, Some(outcome_code), now).await?; tx.insert::(CreateAiAuditEventRecordInput { actor_principal_kind: "system".to_owned(), actor_subject: submission @@ -4711,7 +4717,14 @@ async fn close_background_recovery( }) .await .map_err(OrmPublicError::from)?; - append_terminal_run_event(tx, &run, AiRunState::RecoveryRequired, now).await?; + append_terminal_run_event( + tx, + &run, + AiRunState::RecoveryRequired, + Some(safe_error_code), + now, + ) + .await?; close_matched_receipt_recovery(tx, submission, safe_error_code, now).await?; tx.insert::(CreateAiAuditEventRecordInput { actor_principal_kind: "system".to_owned(), diff --git a/crates/graphql-orm-ai/src/orm_coordinator.rs b/crates/graphql-orm-ai/src/orm_coordinator.rs index f459ea60..3e62b412 100644 --- a/crates/graphql-orm-ai/src/orm_coordinator.rs +++ b/crates/graphql-orm-ai/src/orm_coordinator.rs @@ -446,8 +446,15 @@ pub trait AiAgentProviderTurnExecutor: Send + Sync { /// Interrupts an active run-scoped provider resource after durable /// cancellation or lease loss has already been observed. - async fn interrupt_run(&self, _lease: &AiRunLease) -> Result<(), AiError> { - Ok(()) + /// + /// The returned settlement reports what the interruption proved. A default + /// executor proves nothing, so it reports no live resource rather than + /// implying the retained thread may be kept. + async fn interrupt_run( + &self, + _lease: &AiRunLease, + ) -> Result { + Ok(crate::AiRunInterruptSettlement::NotActive) } /// Closes all provider resources belonging to one exact run fence. @@ -492,7 +499,10 @@ impl AiAgentProviderTurnExecutor for AiProviderCallExecutor { .await } - async fn interrupt_run(&self, lease: &AiRunLease) -> Result<(), AiError> { + async fn interrupt_run( + &self, + lease: &AiRunLease, + ) -> Result { AiProviderCallExecutor::interrupt_run(self, lease).await } @@ -1394,19 +1404,48 @@ impl AiReadOnlyAgentCoordinator { Ok(result) => result, Err(ProviderTurnFailure::Deferred) => { const RETRY_AFTER: std::time::Duration = std::time::Duration::from_secs(5); - self.run_control + // The durable retry is what makes a message accepted during + // provider-session cleanup converge without an operator: the + // run becomes retry-scheduled and is reclaimed after the + // deadline, across a host restart. The `Deferred` outcome is + // a report, not the mechanism. + match self + .run_control .schedule_retry( &lease, time::Duration::seconds(5), "provider_session_cleanup_pending", ) - .await?; - return Ok(Deferred { - reason: AiProviderSessionDeferralReason::CleanupPending, - retry_after: RETRY_AFTER, - provider_turns: guard.provider_turns(), - total_tool_calls: guard.total_tool_calls(), - }); + .await + { + Ok(()) => { + return Ok(Deferred { + reason: AiProviderSessionDeferralReason::CleanupPending, + retry_after: RETRY_AFTER, + provider_turns: guard.provider_turns(), + total_tool_calls: guard.total_tool_calls(), + }); + } + Err(AiError::Conflict) => { + // The retry ceiling ran out while cleanup stayed + // pending. Nothing was executed on this attempt: no + // provider call, no tool, no persisted output. Close + // the run as a clean visible failure rather than + // letting the lease expire into `RecoveryRequired`, + // which would be both misclassified and stuck until + // an operator looked at it. A stale fence makes the + // terminal write fail too, so ordinary expired-lease + // reconciliation still owns that case. + return self + .finish_failed( + &lease, + &guard, + "provider_session_cleanup_unavailable", + ) + .await; + } + Err(error) => return Err(error), + } } Err(ProviderTurnFailure::Provider) => { if self.run_control.cancellation(&lease).await?.is_some() { @@ -2176,21 +2215,41 @@ mod tests { struct TestRunControl { finishes: Mutex>, + finish_codes: Mutex>, heartbeat_count: AtomicUsize, fail_heartbeat: AtomicBool, cancelled: AtomicBool, + scheduled_retries: Mutex>, + retry_ceiling_reached: AtomicBool, } impl TestRunControl { fn new() -> Self { Self { finishes: Mutex::new(Vec::new()), + finish_codes: Mutex::new(Vec::new()), heartbeat_count: AtomicUsize::new(0), fail_heartbeat: AtomicBool::new(false), cancelled: AtomicBool::new(false), + scheduled_retries: Mutex::new(Vec::new()), + retry_ceiling_reached: AtomicBool::new(false), } } + fn scheduled_retry_codes(&self) -> Vec { + self.scheduled_retries + .lock() + .expect("test retry lock should not be poisoned") + .clone() + } + + fn final_codes(&self) -> Vec { + self.finish_codes + .lock() + .expect("test finish code lock should not be poisoned") + .clone() + } + fn final_states(&self) -> Vec { self.finishes .lock() @@ -2242,12 +2301,34 @@ mod tests { _lease: &AiRunLease, completion: AiRunCompletion, ) -> Result<(), AiError> { + self.finish_codes + .lock() + .expect("test finish code lock should not be poisoned") + .push(completion.outcome_code().to_owned()); self.finishes .lock() .expect("test finish lock should not be poisoned") .push(completion.final_state()); Ok(()) } + + async fn schedule_retry( + &self, + _lease: &AiRunLease, + _delay: time::Duration, + error_code: &str, + ) -> Result<(), AiError> { + if self.retry_ceiling_reached.load(Ordering::SeqCst) { + // Exactly what the durable service returns once the run has + // exhausted its bounded retry allowance. + return Err(AiError::Conflict); + } + self.scheduled_retries + .lock() + .expect("test retry lock should not be poisoned") + .push(error_code.to_owned()); + Ok(()) + } } struct TestProviderExecutor { @@ -2382,6 +2463,30 @@ mod tests { } } + struct DeferringRetainedProviderExecutor; + + #[async_trait] + impl AiAgentProviderTurnExecutor for DeferringRetainedProviderExecutor { + 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 { + Err(AiError::ProviderSessionDeferred) + } + } + struct TestProviderSessionService { run: Arc, commits: AtomicUsize, @@ -3493,6 +3598,128 @@ mod tests { } } + /// Work item 4.3: a message accepted while provider-session cleanup is + /// pending must converge without an operator. + /// + /// While the run still has retry allowance it becomes durably + /// retry-scheduled, which is what survives a host restart: `claim_next` + /// reclaims queued and retry-scheduled runs, so the `Deferred` outcome is a + /// report rather than the delivery mechanism. + #[tokio::test] + async fn cleanup_pending_defers_through_a_durable_retry() { + let lease = AiRunLease::test_running(principal_reference()); + 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 session_service = Arc::new(TestProviderSessionService { + run: run.clone(), + commits: AtomicUsize::new(0), + cleanups: AtomicUsize::new(0), + fail_commit: false, + }); + let forbidden = Arc::new(ChatForbiddenBoundaries::default()); + let coordinator = AiReadOnlyAgentCoordinator::new( + run.clone(), + Arc::new(DeferringRetainedProviderExecutor), + forbidden.clone(), + Arc::new(TestOutputWriter), + forbidden.clone(), + Arc::new(TestCheckpointWriter), + Arc::new(TestRuleResolver), + planner, + limits(50), + ) + .with_provider_session_service(session_service); + + let outcome = coordinator + .execute_claimed(&lease) + .await + .expect("a pending cleanup should defer rather than fail"); + assert!(matches!( + outcome, + Deferred { + reason: AiProviderSessionDeferralReason::CleanupPending, + .. + } + )); + assert_eq!( + run.scheduled_retry_codes(), + vec!["provider_session_cleanup_pending".to_owned()], + "the durable retry, not the reported outcome, is what converges" + ); + assert!( + run.final_states().is_empty(), + "a deferral must not close the run" + ); + } + + /// 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 + /// misclassified and stuck until an operator looks at it. + #[tokio::test] + async fn exhausted_cleanup_retries_converge_to_a_clean_visible_failure() { + let lease = AiRunLease::test_running(principal_reference()); + let run = Arc::new(TestRunControl::new()); + run.retry_ceiling_reached.store(true, Ordering::SeqCst); + 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 session_service = Arc::new(TestProviderSessionService { + run: run.clone(), + commits: AtomicUsize::new(0), + cleanups: AtomicUsize::new(0), + fail_commit: false, + }); + let forbidden = Arc::new(ChatForbiddenBoundaries::default()); + let coordinator = AiReadOnlyAgentCoordinator::new( + run.clone(), + Arc::new(DeferringRetainedProviderExecutor), + forbidden.clone(), + Arc::new(TestOutputWriter), + forbidden.clone(), + Arc::new(TestCheckpointWriter), + Arc::new(TestRuleResolver), + planner, + limits(50), + ) + .with_provider_session_service(session_service); + + let outcome = coordinator + .execute_claimed(&lease) + .await + .expect("an exhausted retry allowance should still close the run"); + assert!(matches!(outcome, Failed { .. })); + assert_eq!(run.final_states(), vec![AiRunState::Failed]); + assert_eq!( + run.final_codes(), + vec!["provider_session_cleanup_unavailable".to_owned()] + ); + assert!( + run.scheduled_retry_codes().is_empty(), + "no retry may be recorded once the allowance is exhausted" + ); + // Nothing executed on this attempt, so the failure is provably clean + // and the owner may author a new run for the same message. + assert_eq!( + crate::classify_run_retry( + crate::AiRunRetryEvidence { + terminal: crate::AiRunTerminalEvent::Failed, + produced_assistant_output: false, + }, + Some("provider_session_cleanup_unavailable"), + ), + crate::AiRunRetryAdmission::Allowed + ); + } + #[tokio::test] async fn retained_turn_cancelled_after_output_never_advances_and_requires_cleanup() { let lease = AiRunLease::test_running(principal_reference()); diff --git a/crates/graphql-orm-ai/src/orm_provider_session.rs b/crates/graphql-orm-ai/src/orm_provider_session.rs index 48728ffb..5bf50bc3 100644 --- a/crates/graphql-orm-ai/src/orm_provider_session.rs +++ b/crates/graphql-orm-ai/src/orm_provider_session.rs @@ -148,9 +148,10 @@ mod service { }; use crate::persistence::{ AiApprovalRecord, AiAuditEventRecord, AiMessageRecord, AiRunCheckpointRecord, - AiRunCheckpointRecordWhereInput, AiRunRecord, AiSessionRecord, - AiSubscriptionWaitAdoptionRecord, AiSubscriptionWaitAdoptionRecordWhereInput, - AiSubscriptionWaiterRecord, AiToolCallRecord, CreateAiAuditEventRecordInput, + AiRunCheckpointRecordWhereInput, AiRunRecord, AiSessionEventRecord, AiSessionRecord, + AiSessionRecordWhereInput, AiSubscriptionWaitAdoptionRecord, + AiSubscriptionWaitAdoptionRecordWhereInput, AiSubscriptionWaiterRecord, AiToolCallRecord, + CreateAiAuditEventRecordInput, CreateAiSessionEventRecordInput, UpdateAiSessionRecordInput, }; use crate::{ AiAccessPolicy, AiContentProtectionPolicy, AiContentProtectionPolicyResolver, @@ -1006,6 +1007,20 @@ mod service { now, ) .await?; + // A rebind replaces the provider's thread even though + // the durable transcript is continuous. Disclose it so + // the host can tell the user the model's context was + // reset rather than silently continuing. + append_provider_session_disclosure( + tx, + crate::orm_runs::PROVIDER_SESSION_REBOUND_EVENT, + updated.session_id, + updated.id, + "provider_session_absence_rebound", + Some(lease.run_id().0), + now, + ) + .await?; Ok(updated) }) }) @@ -1574,6 +1589,16 @@ mod service { parked.source_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(parked.source_run_id.0), + now, + ) .await }) }) @@ -1625,6 +1650,16 @@ mod service { request.claim.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(request.claim.run_id.0), + now, + ) .await }) }) @@ -1899,6 +1934,16 @@ mod service { claim.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(claim.run_id.0), + now, + ) .await }) }) @@ -3171,6 +3216,77 @@ mod service { Ok(()) } + /// Discloses that a retained provider thread stopped being usable, so the + /// model's context was reset even though the durable transcript stays + /// continuous. + /// + /// The event carries only the server-owned reason class: never a cursor, + /// prompt, provider payload, tool argument, or authorization detail. It is + /// appended to the ordinary sequenced session stream so it replays, + /// retains, and authorizes exactly like every other session event. + async fn append_provider_session_disclosure( + tx: &mut graphql_orm::graphql::orm::MutationContext<'_, DefaultWriteBackend>, + event_type: &str, + session_id: Uuid, + binding_id: Uuid, + reason_code: &str, + run_id: Option, + now: OffsetDateTime, + ) -> Result<(), OrmPublicError> { + let session = tx + .find_by_id::(&session_id) + .await + .map_err(OrmPublicError::from)? + .ok_or_else(OrmPublicError::not_found)?; + // A deleting or removed session has no client to disclose to, and its + // stream is being torn down; skip rather than fail the lifecycle write. + if !matches!(session.state.as_str(), "active" | "archived") + || session.deleted_at.is_some() + || session.stream_head < 0 + { + return Ok(()); + } + let sequence = session + .stream_head + .checked_add(1) + .filter(|sequence| *sequence <= i64::from(i32::MAX)) + .ok_or_else(|| OrmPublicError::new(OrmErrorCode::Conflict))?; + if !matches!( + tx.compare_and_swap::( + &session.id, + session.row_version, + AiSessionRecordWhereInput::default(), + UpdateAiSessionRecordInput { + stream_head: Some(sequence), + last_activity_at: Some(now.unix_timestamp()), + ..Default::default() + }, + ) + .await + .map_err(OrmPublicError::from)?, + ConditionalUpdateOutcome::Updated(_) + ) { + return Err(OrmPublicError::new(OrmErrorCode::Conflict)); + } + tx.insert::(CreateAiSessionEventRecordInput { + id: Uuid::new_v4(), + session_id: session.id, + sequence, + event_type: event_type.to_owned(), + run_id, + causation_id: Some(binding_id.to_string()), + correlation_id: binding_id.to_string(), + protected_payload: crate::orm_runs::provider_session_event_metadata(reason_code)?, + }) + .await + .map_err(OrmPublicError::from)?; + tx.queue_event(crate::AiSessionWakeup { + session_id: session.id, + sequence, + }); + Ok(()) + } + async fn append_cleanup_completion_audit( tx: &mut graphql_orm::graphql::orm::MutationContext<'_, DefaultWriteBackend>, record: &AiProviderSessionBindingRecord, diff --git a/crates/graphql-orm-ai/src/orm_run_disposition.rs b/crates/graphql-orm-ai/src/orm_run_disposition.rs new file mode 100644 index 00000000..29210fad --- /dev/null +++ b/crates/graphql-orm-ai/src/orm_run_disposition.rs @@ -0,0 +1,514 @@ +//! ORM-backed owner-authorized disposition of failed runs. + +#![cfg(any(feature = "sqlite", feature = "postgres"))] + +use std::sync::Arc; + +use agql_auth::{ + AuthPrincipal, Clock, CurrentPrincipalResolver, PrincipalReference, ResolvedPrincipal, +}; +use async_trait::async_trait; +use graphql_orm::db::Database; +use graphql_orm::graphql::errors::{OrmErrorCode, OrmPublicError}; +use graphql_orm::graphql::filters::UuidFilter; +use graphql_orm::graphql::orm::{ConditionalUpdateOutcome, DefaultWriteBackend, TransactionMode}; +use serde_json::json; +use time::Duration; +use uuid::Uuid; + +use crate::orm_inbox::{PreparedAiInboxEvent, append_inbox_event}; +use crate::orm_runs::run_produced_assistant_output; +use crate::orm_sessions::{ + content_context, map_orm, map_protection, map_transaction, principal_identity, record_scope, +}; +use crate::persistence::*; +use crate::{ + AcknowledgeAiRunFailureInput, AiAccessPolicy, AiContentProtectionPolicy, + AiContentProtectionPolicyResolver, AiContentProtector, AiError, AiRunDisposition, + AiRunDispositionService, AiRunDispositionView, AiRunRetryAdmission, AiRunRetryEvidence, + AiRunState, AiRunTerminalEvent, AiScope, AiSessionAction, AiSessionId, AiSessionWakeup, + RetryAiRunInput, classify_run_retry, +}; + +/// Deployment bounds for owner disposition and current-principal freshness. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct AiRunDispositionLimits { + maximum_principal_age: Duration, +} + +impl AiRunDispositionLimits { + /// Creates validated disposition bounds. + /// + /// # Errors + /// + /// Returns [`AiError::InvalidConfiguration`] for a non-positive or + /// longer-than-one-hour principal age. + pub fn new(maximum_principal_age: Duration) -> Result { + if !maximum_principal_age.is_positive() || maximum_principal_age > Duration::hours(1) { + return Err(AiError::InvalidConfiguration( + "invalid run disposition limits".to_owned(), + )); + } + Ok(Self { + maximum_principal_age, + }) + } +} + +impl Default for AiRunDispositionLimits { + fn default() -> Self { + Self { + maximum_principal_age: Duration::minutes(5), + } + } +} + +/// Generated-ORM failure-disposition service for application hosts. +pub struct OrmAiRunDispositionService { + database: Database, + access_policy: Arc, + protection_policy: Arc, + content_protector: Arc, + principal_resolver: Arc, + clock: Arc, + limits: AiRunDispositionLimits, +} + +impl OrmAiRunDispositionService { + /// Creates an owner-authorized failure-disposition service. + pub fn new( + database: Database, + access_policy: Arc, + protection_policy: Arc, + content_protector: Arc, + principal_resolver: Arc, + clock: Arc, + limits: AiRunDispositionLimits, + ) -> Self { + Self { + database, + access_policy, + protection_policy, + content_protector, + principal_resolver, + clock, + limits, + } + } + + async fn resolve_current( + &self, + reference: &PrincipalReference, + ) -> Result { + let resolved = self + .principal_resolver + .resolve(reference) + .await + .map_err(|_| AiError::ReauthorizationFailed)?; + let now = self.clock.now(); + if resolved.reference() != reference + || resolved.resolved_at() > now + || now - resolved.resolved_at() >= self.limits.maximum_principal_age + || reference + .expires_at + .is_some_and(|expires_at| expires_at <= now) + { + return Err(AiError::ReauthorizationFailed); + } + Ok(resolved) + } + + async fn authorize( + &self, + resolved: &ResolvedPrincipal, + session: &AiSessionRecord, + ) -> Result<(), AiError> { + let principal = resolved.principal(); + let (kind, subject) = principal_identity(principal); + if session.owner_principal_kind != kind + || session.owner_subject != subject + || session.deleted_at.is_some() + { + return Err(AiError::NotFound); + } + let scope = record_scope(session); + if !self + .access_policy + .can_access_session(principal, AiSessionId(session.id), AiSessionAction::Write) + .await + .is_allowed() + || !self + .access_policy + .can_access_scope(principal, &scope, AiSessionAction::Write) + .await + .is_allowed() + { + return Err(AiError::Forbidden); + } + Ok(()) + } + + async fn protect( + &self, + policy: &AiContentProtectionPolicy, + entity: &str, + row_id: Uuid, + scope: &AiScope, + value: serde_json::Value, + ) -> Result { + let envelope = self + .content_protector + .protect( + policy, + &content_context(entity, row_id, "protected_payload", scope), + value, + ) + .await + .map_err(map_protection)?; + serde_json::to_value(envelope).map_err(|_| AiError::PersistenceFailed) + } + + /// Runs the shared admission, idempotency, and durable write path. + async fn dispose( + &self, + principal: &AuthPrincipal, + session_id: Uuid, + run_id: Uuid, + client_request_id: Uuid, + disposition: AiRunDisposition, + ) -> Result { + if session_id.is_nil() || run_id.is_nil() || client_request_id.is_nil() { + return Err(AiError::InvalidInput( + "invalid run disposition identity".to_owned(), + )); + } + let requested_reference = principal.reference(); + let current = self.resolve_current(&requested_reference).await?; + let session = AiSessionRecord::find_by_id(&self.database, &session_id) + .await + .map_err(|error| map_orm(OrmPublicError::from(error)))? + .ok_or(AiError::NotFound)?; + self.authorize(¤t, &session).await?; + + // Idempotent replay: the same key returns the original decision without + // authoring a second run. + if let Some(existing) = + AiRunFailureDispositionRecord::find_by_id(&self.database, &client_request_id) + .await + .map_err(|error| map_orm(OrmPublicError::from(error)))? + { + if existing.session_id != session.id || existing.source_run_id != run_id { + return Err(AiError::Conflict); + } + return disposition_view(&existing); + } + + let run = AiRunRecord::find_by_id(&self.database, &run_id) + .await + .map_err(|error| map_orm(OrmPublicError::from(error)))? + .filter(|run| run.session_id == session.id) + .ok_or(AiError::NotFound)?; + let state = AiRunState::from_persisted(&run.state).ok_or(AiError::PersistenceFailed)?; + let terminal = match state { + AiRunState::Failed => AiRunTerminalEvent::Failed, + AiRunState::RecoveryRequired => AiRunTerminalEvent::RecoveryRequired, + AiRunState::Cancelled => AiRunTerminalEvent::Cancelled, + _ => return Err(AiError::Conflict), + }; + + // Reauthorize immediately before the durable write so a revocation + // between the read and the commit cannot be used. + let current = self.resolve_current(&requested_reference).await?; + self.authorize(¤t, &session).await?; + let scope = record_scope(&session); + let policy = self + .protection_policy + .resolve(current.principal(), &scope) + .await?; + if !policy.ready || policy.scope != scope { + return Err(AiError::RuntimeNotReady); + } + // The retry runs under the current principal, not the one the source + // run captured. A stale reference must never be resurrected. + let principal_reference = + serde_json::to_value(current.reference()).map_err(|_| AiError::PersistenceFailed)?; + let (principal_kind, principal_subject) = principal_identity(current.principal()); + let principal_subject = principal_subject.to_owned(); + + let event_id = Uuid::new_v4(); + let inbox_event_id = Uuid::new_v4(); + let retry_run_id = matches!(disposition, AiRunDisposition::Retried).then(Uuid::new_v4); + let event_type = match disposition { + AiRunDisposition::Retried => "run_retry_queued", + AiRunDisposition::Acknowledged => "run_failure_acknowledged", + }; + let identifiers = json!({ + "sessionId": session_id, + "sourceRunId": run_id, + "retryRunId": retry_run_id, + "clientRequestId": client_request_id, + "disposition": disposition.as_str(), + }); + let protected_event = self + .protect( + &policy, + "graphql_orm_ai_session_events", + event_id, + &scope, + identifiers.clone(), + ) + .await?; + let protected_inbox_event = self + .protect( + &policy, + "graphql_orm_ai_inbox_events", + inbox_event_id, + &scope, + identifiers, + ) + .await?; + + let now = self + .clock + .now() + .replace_nanosecond(0) + .unwrap_or_else(|_| self.clock.now()); + let now_unix = now.unix_timestamp(); + let input_message_id = run.input_message_id; + let source_state = run.state.clone(); + let source_outcome_code = run.error_code.clone(); + let expected_run = run.clone(); + let owner_kind = principal_kind.clone(); + let owner_subject = principal_subject.clone(); + + let record = self + .database + .transaction(TransactionMode::StateMachine, move |tx| { + let principal_reference = principal_reference.clone(); + let protected_event = protected_event.clone(); + let protected_inbox_event = protected_inbox_event.clone(); + let owner_kind = owner_kind.clone(); + let owner_subject = owner_subject.clone(); + let source_state = source_state.clone(); + let source_outcome_code = source_outcome_code.clone(); + let expected_run = expected_run.clone(); + Box::pin(async move { + let session = tx + .find_by_id::(&session_id) + .await + .map_err(OrmPublicError::from)? + .ok_or_else(OrmPublicError::not_found)?; + if !matches!(session.state.as_str(), "active") || session.deleted_at.is_some() { + return Err(OrmPublicError::new(OrmErrorCode::Conflict)); + } + let run = tx + .find_by_id::(&run_id) + .await + .map_err(OrmPublicError::from)? + .ok_or_else(OrmPublicError::not_found)?; + // The run must not have changed between admission and the + // commit; a terminal run should be immutable, and anything + // else means the decision was made against stale evidence. + if run != expected_run { + return Err(OrmPublicError::new(OrmErrorCode::Conflict)); + } + // At most one disposition wins per source run. + let existing = tx + .query::() + .filter(AiRunFailureDispositionRecordWhereInput { + source_run_id: Some(UuidFilter { + eq: Some(run_id), + ..Default::default() + }), + ..Default::default() + }) + .limit(1) + .fetch_all() + .await + .map_err(OrmPublicError::from)?; + if !existing.is_empty() { + return Err(OrmPublicError::new(OrmErrorCode::Conflict)); + } + + if let Some(retry_run_id) = retry_run_id { + // Re-decide admission from committed rows inside the + // same transaction that authors the new run. + let evidence = AiRunRetryEvidence { + terminal, + produced_assistant_output: run_produced_assistant_output( + tx, session_id, run_id, + ) + .await?, + }; + if classify_run_retry(evidence, run.error_code.as_deref()) + != AiRunRetryAdmission::Allowed + { + return Err(OrmPublicError::new(OrmErrorCode::Conflict)); + } + let message = tx + .find_by_id::(&input_message_id) + .await + .map_err(OrmPublicError::from)? + .ok_or_else(OrmPublicError::not_found)?; + // Retry reuses the durable user message; it never + // rewrites one, and it refuses a purged one because the + // prompt no longer exists. + if message.session_id != session_id + || message.message_role != "user" + || message.content_purged_at.is_some() + { + return Err(OrmPublicError::new(OrmErrorCode::Conflict)); + } + tx.insert::(CreateAiRunRecordInput { + id: retry_run_id, + session_id, + input_message_id, + principal_reference, + state: AiRunState::Queued.as_str().to_owned(), + attempt_id: None, + lease_owner: None, + lease_generation: 0, + lease_expires_at: None, + lease_heartbeat_at: None, + retry_count: 0, + next_attempt_at: Some(now_unix), + error_code: None, + latest_checkpoint_id: None, + cancellation_request_id: None, + cancellation_requested_at: None, + }) + .await + .map_err(OrmPublicError::from)?; + } + + let sequence = session + .stream_head + .checked_add(1) + .filter(|sequence| *sequence <= i64::from(i32::MAX)) + .ok_or_else(|| OrmPublicError::new(OrmErrorCode::Conflict))?; + if !matches!( + tx.compare_and_swap::( + &session.id, + session.row_version, + AiSessionRecordWhereInput::default(), + UpdateAiSessionRecordInput { + stream_head: Some(sequence), + last_activity_at: Some(now_unix), + ..Default::default() + }, + ) + .await + .map_err(OrmPublicError::from)?, + ConditionalUpdateOutcome::Updated(_) + ) { + return Err(OrmPublicError::new(OrmErrorCode::Conflict)); + } + + let record = tx + .insert::( + CreateAiRunFailureDispositionRecordInput { + id: client_request_id, + session_id, + source_run_id: run_id, + input_message_id, + disposition: disposition.as_str().to_owned(), + retry_run_id, + source_state, + source_outcome_code, + principal_kind: owner_kind.clone(), + principal_subject: owner_subject.clone(), + decided_at: now_unix, + }, + ) + .await + .map_err(OrmPublicError::from)?; + tx.insert::(CreateAiSessionEventRecordInput { + id: event_id, + session_id, + sequence, + event_type: event_type.to_owned(), + run_id: Some(run_id), + causation_id: Some(client_request_id.to_string()), + correlation_id: client_request_id.to_string(), + protected_payload: protected_event, + }) + .await + .map_err(OrmPublicError::from)?; + tx.queue_event(AiSessionWakeup { + session_id, + sequence, + }); + append_inbox_event( + tx, + PreparedAiInboxEvent { + id: inbox_event_id, + principal_kind: owner_kind, + principal_subject: owner_subject, + scope: AiScope { + kind: session.scope_kind.clone(), + id: session.scope_id.clone(), + tenant_id: session.tenant_id.clone(), + }, + session_id, + event_type: event_type.to_owned(), + protected_payload: protected_inbox_event, + created_at: now_unix, + }, + ) + .await?; + Ok(record) + }) + }) + .await + .map_err(map_transaction)?; + disposition_view(&record) + } +} + +fn disposition_view( + record: &AiRunFailureDispositionRecord, +) -> Result { + Ok(AiRunDispositionView { + session_id: record.session_id, + run_id: record.source_run_id, + client_request_id: record.id, + disposition: AiRunDisposition::from_persisted(&record.disposition) + .ok_or(AiError::PersistenceFailed)?, + retry_run_id: record.retry_run_id, + input_message_id: record.input_message_id, + decided_at: record.decided_at, + }) +} + +#[async_trait] +impl AiRunDispositionService for OrmAiRunDispositionService { + async fn retry_run( + &self, + principal: &AuthPrincipal, + input: RetryAiRunInput, + ) -> Result { + self.dispose( + principal, + input.session_id, + input.run_id, + input.client_request_id, + AiRunDisposition::Retried, + ) + .await + } + + async fn acknowledge_run_failure( + &self, + principal: &AuthPrincipal, + input: AcknowledgeAiRunFailureInput, + ) -> Result { + self.dispose( + principal, + input.session_id, + input.run_id, + input.client_request_id, + AiRunDisposition::Acknowledged, + ) + .await + } +} diff --git a/crates/graphql-orm-ai/src/orm_runs.rs b/crates/graphql-orm-ai/src/orm_runs.rs index 17837cda..76113156 100644 --- a/crates/graphql-orm-ai/src/orm_runs.rs +++ b/crates/graphql-orm-ai/src/orm_runs.rs @@ -21,9 +21,9 @@ use crate::orm_inbox::{PreparedAiInboxEvent, append_inbox_event}; use crate::orm_provider_session::AiProviderSessionBindingRecord; use crate::persistence::*; use crate::{ - AiApprovalId, AiBudgetAmounts, AiError, AiRunCancellation, AiRunCancellationHub, AiRunId, - AiRunState, AiRunTerminalEvent, AiScope, AiSessionId, AiSessionWakeup, AiToolCallId, - ProtectedContentEnvelope, + AiApprovalId, AiBudgetAmounts, AiError, AiRunCancellation, AiRunCancellationHub, AiRunFailure, + AiRunId, AiRunRetryEvidence, AiRunState, AiRunTerminalEvent, AiScope, AiSessionId, + AiSessionWakeup, AiToolCallId, ProtectedContentEnvelope, classify_run_retry, }; const MAXIMUM_WORKER_ID_BYTES: usize = 256; @@ -268,6 +268,14 @@ impl AiRunCompletion { pub const fn final_state(&self) -> AiRunState { self.final_state } + + /// Stable redacted outcome code recorded on the immutable attempt outcome. + /// + /// This is server-owned classification, never provider diagnostics, a + /// prompt, a tool argument, or response content. + pub fn outcome_code(&self) -> &str { + &self.outcome_code + } } /// Bounded result of expired-lease reconciliation. @@ -914,6 +922,12 @@ impl OrmAiRunService { if !matches!(outcome, ConditionalUpdateOutcome::Updated(_)) { return Err(OrmPublicError::new(OrmErrorCode::Conflict)); } + // The event must classify from the value that lands on the + // run row, because that is what the retry mutation will + // later re-evaluate. Classifying from `outcome_code` would + // let a completion with an outcome code but no error code + // advertise a retry the mutation then refuses. + let terminal_error_code = completion.error_code.clone(); append_attempt_outcome( tx, &lease, @@ -923,7 +937,14 @@ impl OrmAiRunService { now, ) .await?; - append_terminal_run_event(tx, ¤t, completion.final_state, now).await + append_terminal_run_event( + tx, + ¤t, + completion.final_state, + terminal_error_code.as_deref(), + now, + ) + .await }) }) .await @@ -1453,7 +1474,14 @@ impl OrmAiRunService { ) .await?; if AiRunTerminalEvent::from_run_state(next_state).is_some() { - append_terminal_run_event(tx, ¤t, next_state, now).await?; + append_terminal_run_event( + tx, + ¤t, + next_state, + Some(outcome_code), + now, + ) + .await?; } } Ok(report) @@ -4015,6 +4043,7 @@ impl OrmAiRunService { database .transaction(TransactionMode::StateMachine, move |tx| { Box::pin(async move { + let terminal_outcome_code = reconciliation.outcome_code.clone(); let current = tx .find_by_id::(&reconciliation.expected_run.id) .await @@ -4469,7 +4498,7 @@ impl OrmAiRunService { resource_kind: "ai_run".to_owned(), resource_reference: current.id.to_string(), outcome: audit_outcome.to_owned(), - reason_code: reconciliation.outcome_code.clone(), + reason_code: terminal_outcome_code.clone(), correlation_id: approval_id.unwrap_or(current.id).to_string(), causation_id: tool_call_id.map(|id| id.to_string()), policy_version: reconciliation.policy_version, @@ -4481,13 +4510,20 @@ impl OrmAiRunService { tx, lease, final_state, - reconciliation.outcome_code, + terminal_outcome_code.clone(), provider_response_id, now, ) .await?; } - append_terminal_run_event(tx, ¤t, final_state, now).await?; + append_terminal_run_event( + tx, + ¤t, + final_state, + Some(terminal_outcome_code.as_str()), + now, + ) + .await?; tx.queue_event(AiSessionWakeup { session_id: session.id, sequence: event_sequence, @@ -4708,6 +4744,89 @@ fn reservation_usage_matches( } const TERMINAL_EVENT_METADATA_FORMAT: &str = "graphql-orm-ai-run-terminal-event-v1"; +const TERMINAL_EVENT_METADATA_FORMAT_V2: &str = "graphql-orm-ai-run-terminal-event-v2"; +pub(crate) const PROVIDER_SESSION_EVENT_METADATA_FORMAT: &str = + "graphql-orm-ai-provider-session-event-v1"; + +/// Stable event type disclosing that a retained provider thread stopped being +/// usable, so the model's context was reset even though the durable transcript +/// is continuous. +pub(crate) const PROVIDER_SESSION_RESET_EVENT: &str = "provider_session_reset"; +/// Stable event type disclosing that a new provider session binding replaced a +/// prior one for the same session. +pub(crate) const PROVIDER_SESSION_REBOUND_EVENT: &str = "provider_session_rebound"; + +/// Opens any content-free tagged event metadata without consulting a scope +/// content key. +/// +/// Returns `Ok(None)` for an event whose payload is ordinary protected content, +/// which the caller must open through the configured protector. +pub(crate) fn open_metadata_only_event( + event_type: &str, + protected_payload: &serde_json::Value, +) -> Result, AiError> { + if matches!( + event_type, + PROVIDER_SESSION_RESET_EVENT | PROVIDER_SESSION_REBOUND_EVENT + ) { + return open_provider_session_event_metadata(protected_payload); + } + open_terminal_event_metadata(event_type, protected_payload) +} + +/// Opens the content-free provider-session lifecycle envelope. +/// +/// The reason class is a server-owned safe code describing *why* a retained +/// thread was dropped. It never carries provider, prompt, tool, cursor, or +/// authorization content, so it needs no scope content key. +fn open_provider_session_event_metadata( + protected_payload: &serde_json::Value, +) -> Result, AiError> { + let Some(value) = protected_payload + .as_object() + .and_then(|envelope| envelope.get("value")) + .and_then(serde_json::Value::as_object) + else { + return Ok(None); + }; + if value.get("format").and_then(serde_json::Value::as_str) + != Some(PROVIDER_SESSION_EVENT_METADATA_FORMAT) + { + return Ok(None); + } + if protected_payload + .get("protection") + .and_then(serde_json::Value::as_str) + != Some("database_managed") + || protected_payload + .as_object() + .is_none_or(|object| object.len() != 2) + || value.len() != 2 + || !value + .get("reason") + .and_then(serde_json::Value::as_str) + .is_some_and(valid_safe_code) + { + return Err(AiError::PersistenceFailed); + } + Ok(Some(serde_json::Value::Object(value.clone()))) +} + +/// Builds the content-free provider-session lifecycle envelope. +pub(crate) fn provider_session_event_metadata( + reason_code: &str, +) -> Result { + if !valid_safe_code(reason_code) { + return Err(OrmPublicError::new(OrmErrorCode::InvalidInput)); + } + serde_json::to_value(ProtectedContentEnvelope::DatabaseManaged { + value: serde_json::json!({ + "format": PROVIDER_SESSION_EVENT_METADATA_FORMAT, + "reason": reason_code, + }), + }) + .map_err(|_| OrmPublicError::new(OrmErrorCode::InternalError)) +} /// Opens the deliberately content-free metadata envelope used by canonical /// run terminal events without consulting a scope content key. @@ -4727,11 +4846,15 @@ pub(crate) fn open_terminal_event_metadata( else { return Ok(None); }; - if value.get("format").and_then(serde_json::Value::as_str) - != Some(TERMINAL_EVENT_METADATA_FORMAT) - { - return Ok(None); - } + // Rows written before the failure record used the v1 shape and must stay + // readable. Anything that is neither tagged shape falls through to the + // configured content protector; anything that claims a tagged shape but + // does not match it exactly fails closed. + let expected_len = match value.get("format").and_then(serde_json::Value::as_str) { + Some(TERMINAL_EVENT_METADATA_FORMAT) => 2, + Some(TERMINAL_EVENT_METADATA_FORMAT_V2) => 3, + _ => return Ok(None), + }; let terminal = AiRunTerminalEvent::from_event_type(event_type).ok_or(AiError::PersistenceFailed)?; if protected_payload @@ -4741,22 +4864,62 @@ pub(crate) fn open_terminal_event_metadata( || protected_payload .as_object() .is_none_or(|object| object.len() != 2) - || value.len() != 2 + || value.len() != expected_len || value.get("state").and_then(serde_json::Value::as_str) != Some(terminal.run_state().as_str()) { return Err(AiError::PersistenceFailed); } + if expected_len == 3 && !valid_terminal_failure_record(value.get("failure")) { + return Err(AiError::PersistenceFailed); + } Ok(Some(serde_json::Value::Object(value.clone()))) } +/// Validates the bounded failure record without opening any protected content. +/// +/// `null` is the successful-terminal case. Every other shape must be the exact +/// versioned record this crate writes. +fn valid_terminal_failure_record(failure: Option<&serde_json::Value>) -> bool { + let Some(failure) = failure else { + return false; + }; + if failure.is_null() { + return true; + } + let Some(record) = failure.as_object() else { + return false; + }; + record.len() == 5 + && record.get("version").and_then(serde_json::Value::as_u64) + == Some(u64::from(crate::AI_RUN_FAILURE_VERSION)) + && record.get("ok") == Some(&serde_json::Value::Bool(false)) + && record + .get("retryable") + .is_some_and(serde_json::Value::is_boolean) + && record + .get("admission") + .and_then(serde_json::Value::as_str) + .is_some_and(|admission| { + matches!( + admission, + "allowed" | "refused_uncertain" | "refused_already_answered" + ) + }) + && record + .get("code") + .is_some_and(|code| code.is_null() || code.as_str().is_some_and(valid_safe_code)) +} + fn terminal_event_metadata( terminal: AiRunTerminalEvent, + failure: Option<&AiRunFailure>, ) -> Result { serde_json::to_value(ProtectedContentEnvelope::DatabaseManaged { value: serde_json::json!({ - "format": TERMINAL_EVENT_METADATA_FORMAT, + "format": TERMINAL_EVENT_METADATA_FORMAT_V2, "state": terminal.run_state().as_str(), + "failure": failure.map(AiRunFailure::to_json), }), }) .map_err(|_| OrmPublicError::new(OrmErrorCode::InternalError)) @@ -4768,6 +4931,7 @@ pub(crate) async fn append_terminal_run_event( tx: &mut MutationContext<'_, DefaultWriteBackend>, run: &AiRunRecord, final_state: AiRunState, + outcome_code: Option<&str>, now: OffsetDateTime, ) -> Result<(), OrmPublicError> { let terminal = AiRunTerminalEvent::from_run_state(final_state) @@ -4835,8 +4999,32 @@ pub(crate) async fn append_terminal_run_event( }; let event_id = Uuid::new_v4(); let inbox_event_id = Uuid::new_v4(); - let protected_event = terminal_event_metadata(terminal)?; - let protected_inbox_event = terminal_event_metadata(terminal)?; + // Retry admission is decided from committed rows only. The caller supplies + // the outcome code from the completion it is committing, because `run` is + // the pre-transition record and its `error_code` is still the previous + // attempt's value. + let failure = match terminal { + AiRunTerminalEvent::Completed => None, + AiRunTerminalEvent::Failed + | AiRunTerminalEvent::Cancelled + | AiRunTerminalEvent::RecoveryRequired => { + let evidence = AiRunRetryEvidence { + terminal, + produced_assistant_output: run_produced_assistant_output( + tx, + run.session_id, + run.id, + ) + .await?, + }; + Some(AiRunFailure::new( + classify_run_retry(evidence, outcome_code), + outcome_code.map(str::to_owned), + )) + } + }; + let protected_event = terminal_event_metadata(terminal, failure.as_ref())?; + let protected_inbox_event = terminal_event_metadata(terminal, failure.as_ref())?; tx.insert::(CreateAiSessionEventRecordInput { id: event_id, session_id: session.id, @@ -5662,8 +5850,9 @@ mod tests { .expect("terminal metadata should validate") .expect("terminal event should use metadata-only envelope"), serde_json::json!({ - "format": TERMINAL_EVENT_METADATA_FORMAT, + "format": TERMINAL_EVENT_METADATA_FORMAT_V2, "state": "completed", + "failure": serde_json::Value::Null, }) ); let inbox = inbox_events(&fixture).await; @@ -5845,3 +6034,39 @@ mod tests { assert_eq!(retry.lease_generation(), 2); } } + +/// Returns whether one run already produced a durable assistant message. +/// +/// This is the only evidence allowed to decide "already answered". It reads +/// committed rows in the caller's transaction and never consults worker state, +/// elapsed time, or a provider report. +pub(crate) async fn run_produced_assistant_output( + tx: &mut MutationContext<'_, DefaultWriteBackend>, + session_id: Uuid, + run_id: Uuid, +) -> Result { + let rows = tx + .query::() + .filter(AiMessageRecordWhereInput { + // Scoped by session first so the read stays inside one + // conversation's indexed keyset rather than scanning globally. + session_id: Some(UuidFilter { + eq: Some(session_id), + ..Default::default() + }), + run_id: Some(UuidFilter { + eq: Some(run_id), + ..Default::default() + }), + message_role: Some(StringFilter { + eq: Some("assistant".to_owned()), + ..Default::default() + }), + ..Default::default() + }) + .limit(1) + .fetch_all() + .await + .map_err(OrmPublicError::from)?; + Ok(!rows.is_empty()) +} diff --git a/crates/graphql-orm-ai/src/orm_sessions.rs b/crates/graphql-orm-ai/src/orm_sessions.rs index eebdfa9e..43aec4cf 100644 --- a/crates/graphql-orm-ai/src/orm_sessions.rs +++ b/crates/graphql-orm-ai/src/orm_sessions.rs @@ -157,6 +157,30 @@ impl OrmAiSessionService { Ok(Some(record)) } + /// Returns the session's current durable stream head for an authorized + /// owner, or `None` when the session is not visible. + /// + /// This is the bounded head-sequence read used as the durable fallback for + /// live delivery. It exists so a subscriber does not depend solely on the + /// process-local wakeup channel: a dropped or missed in-process hint is + /// otherwise unrecoverable. It reads one row and opens no protected + /// content. + /// + /// # Errors + /// + /// Returns a safe library error for a policy denial or persistence + /// failure. + pub async fn session_stream_head( + &self, + principal: &AuthPrincipal, + session_id: AiSessionId, + ) -> Result, AiError> { + Ok(self + .visible_session(principal, session_id, AiSessionAction::Read) + .await? + .map(|record| record.stream_head)) + } + async fn protection_policy( &self, principal: &AuthPrincipal, @@ -343,10 +367,7 @@ impl AiSessionService for OrmAiSessionService { .visible_session(principal, session_id, AiSessionAction::Read) .await? .ok_or(AiError::NotFound)?; - if first.row_version != second.row_version - || first.stream_head != second.stream_head - || first.message_head != second.message_head - { + if !bootstrap_snapshot_is_stable(&first, &second) { continue; } let has_older_messages = messages.page_info.has_previous_page; @@ -681,7 +702,7 @@ impl AiSessionService for OrmAiSessionService { let has_more = rows.last().is_some_and(|row| row.sequence < watermark); let mut events = Vec::with_capacity(rows.len()); for row in rows { - let payload = match crate::orm_runs::open_terminal_event_metadata( + let payload = match crate::orm_runs::open_metadata_only_event( &row.event_type, &row.protected_payload, )? { @@ -1571,6 +1592,34 @@ fn validate_scope(scope: &AiScope) -> Result<(), AiError> { Ok(()) } +/// Returns whether two reads of the same session admit one bootstrap snapshot. +/// +/// The predicate deliberately ignores `stream_head`, `last_activity_at`, and +/// `row_version`. A coalesced live delta advances all three at roughly the +/// streaming coalescer rate while an assistant is answering, but it appends +/// only a session event: it cannot change the session shell, the message +/// window, a run row, or a tool-call row that the bootstrap returns. Including +/// that churn in the predicate made the bounded snapshot fail structurally for +/// exactly the sessions a user is most likely to open. +/// +/// Every remaining field either appears in the returned session view or bounds +/// the returned message window, so an unstable value means the assembled +/// snapshot could not have existed at one instant. +fn bootstrap_snapshot_is_stable(first: &AiSessionRecord, second: &AiSessionRecord) -> bool { + first.id == second.id + && first.message_head == second.message_head + && first.state == second.state + && first.archived_at == second.archived_at + && first.deleted_at == second.deleted_at + && first.title == second.title + && first.title_revision == second.title_revision + && first.owner_principal_kind == second.owner_principal_kind + && first.owner_subject == second.owner_subject + && first.tenant_id == second.tenant_id + && first.scope_kind == second.scope_kind + && first.scope_id == second.scope_id +} + pub(crate) fn session_view(record: &AiSessionRecord) -> AiSessionView { AiSessionView { id: record.id, @@ -1727,3 +1776,141 @@ pub(crate) fn map_orm(error: OrmPublicError) -> AiError { | OrmErrorCode::AuthorizationMisconfigured => AiError::PersistenceFailed, } } + +#[cfg(test)] +mod bootstrap_snapshot_tests { + use super::{AiSessionRecord, bootstrap_snapshot_is_stable}; + + fn record() -> AiSessionRecord { + AiSessionRecord { + id: uuid::Uuid::from_u128(1), + owner_principal_kind: "user".to_owned(), + owner_subject: "bootstrap-owner".to_owned(), + tenant_id: Some("tenant".to_owned()), + scope_kind: "workspace".to_owned(), + scope_id: "scope".to_owned(), + title: "Research".to_owned(), + title_revision: 4, + title_source: "user".to_owned(), + state: "active".to_owned(), + stream_head: 12, + message_head: 3, + last_activity_at: 1_700_000_000, + archived_at: None, + deleted_at: None, + row_version: 9, + } + } + + #[test] + fn live_delta_churn_admits_the_snapshot() { + let first = record(); + // Exactly what a coalesced live delta advances: the stream head, the + // activity timestamp, and the CAS version. Nothing the bootstrap + // returns depends on any of them. + let second = AiSessionRecord { + stream_head: first.stream_head + 41, + last_activity_at: first.last_activity_at + 7, + row_version: first.row_version + 41, + ..record() + }; + assert!(bootstrap_snapshot_is_stable(&first, &second)); + } + + #[test] + fn every_returned_field_forces_a_retry() { + let first = record(); + let mutations: Vec<(&str, AiSessionRecord)> = vec![ + ( + "id", + AiSessionRecord { + id: uuid::Uuid::from_u128(2), + ..record() + }, + ), + ( + "message_head", + AiSessionRecord { + message_head: 4, + ..record() + }, + ), + ( + "state", + AiSessionRecord { + state: "archived".to_owned(), + ..record() + }, + ), + ( + "archived_at", + AiSessionRecord { + archived_at: Some(1_700_000_100), + ..record() + }, + ), + ( + "deleted_at", + AiSessionRecord { + deleted_at: Some(1_700_000_100), + ..record() + }, + ), + ( + "title", + AiSessionRecord { + title: "Renamed".to_owned(), + ..record() + }, + ), + ( + "title_revision", + AiSessionRecord { + title_revision: 5, + ..record() + }, + ), + ( + "owner_principal_kind", + AiSessionRecord { + owner_principal_kind: "api_token:service".to_owned(), + ..record() + }, + ), + ( + "owner_subject", + AiSessionRecord { + owner_subject: "someone-else".to_owned(), + ..record() + }, + ), + ( + "tenant_id", + AiSessionRecord { + tenant_id: Some("other-tenant".to_owned()), + ..record() + }, + ), + ( + "scope_kind", + AiSessionRecord { + scope_kind: "project".to_owned(), + ..record() + }, + ), + ( + "scope_id", + AiSessionRecord { + scope_id: "other-scope".to_owned(), + ..record() + }, + ), + ]; + for (field, second) in mutations { + assert!( + !bootstrap_snapshot_is_stable(&first, &second), + "an unstable {field} must reject the assembled snapshot" + ); + } + } +} diff --git a/crates/graphql-orm-ai/src/orm_subscription_waits.rs b/crates/graphql-orm-ai/src/orm_subscription_waits.rs index a48793dd..2ebe486f 100644 --- a/crates/graphql-orm-ai/src/orm_subscription_waits.rs +++ b/crates/graphql-orm-ai/src/orm_subscription_waits.rs @@ -2051,7 +2051,7 @@ impl OrmAiSubscriptionWaitService { UpdateAiRunRecordInput { state: Some(final_state.as_str().to_owned()), next_attempt_at: Some(None), - error_code: Some(Some(reason_code)), + error_code: Some(Some(reason_code.clone())), ..Default::default() }, ) @@ -2061,7 +2061,7 @@ impl OrmAiSubscriptionWaitService { ) { return Err(OrmPublicError::new(OrmErrorCode::Conflict)); } - append_terminal_run_event(tx, &run, final_state, now).await + append_terminal_run_event(tx, &run, final_state, Some(&reason_code), now).await }) }) .await diff --git a/crates/graphql-orm-ai/src/orm_subscriptions.rs b/crates/graphql-orm-ai/src/orm_subscriptions.rs index 41b9c8cb..62cf7c9d 100644 --- a/crates/graphql-orm-ai/src/orm_subscriptions.rs +++ b/crates/graphql-orm-ai/src/orm_subscriptions.rs @@ -5,28 +5,69 @@ use std::sync::Arc; use std::time::Duration; -use agql_auth::CurrentPrincipalResolver; +use agql_auth::{AuthError, CurrentPrincipalResolver}; use async_trait::async_trait; +use sha2::{Digest, Sha256}; use tokio::sync::broadcast::error::RecvError; use tokio::time::{Instant, MissedTickBehavior}; +use uuid::Uuid; use crate::{ AiError, AiSessionEventEnvelope, AiSessionEventStream, AiSessionId, AiSessionService, - AiSessionWakeup, AiSubscriptionService, OrmAiSessionService, + AiSessionStreamClose, AiSessionWakeup, AiSubscriptionService, OrmAiSessionService, }; +/// Classification of one reauthorization failure. +/// +/// The default is deny: only a dependency that is explicitly unavailable is +/// worth waiting for. Every other class, including one this crate does not +/// recognize, ends the stream immediately. +fn is_transient_authorization_failure(error: &AuthError) -> bool { + matches!( + error, + AuthError::AuthServiceUnavailable + | AuthError::Store(_) + | AuthError::AuthThrottled { .. } + | AuthError::AuthLocked { .. } + ) +} + +/// Returns a bounded per-session jittered backoff. +/// +/// The jitter is derived from the session ID rather than a random source so +/// that concurrent subscribers to *different* sessions desynchronize (which is +/// the point: a single authorization restart must not produce one synchronized +/// bootstrap storm) while one session's behavior stays reproducible in tests. +fn jittered_backoff(base: Duration, attempt: u32, session_id: Uuid) -> Duration { + let scaled = base.saturating_mul(1_u32 << attempt.min(4)); + let mut hasher = Sha256::new(); + hasher.update(b"graphql-orm-ai-reauthorization-backoff-v1"); + hasher.update(session_id.as_bytes()); + hasher.update(attempt.to_be_bytes()); + let digest = hasher.finalize(); + // Map the digest into [0.5, 1.5) of the scaled delay. + let fraction = u32::from(digest[0]) * 2 + 256; + scaled + .saturating_mul(fraction) + .checked_div(512) + .unwrap_or(scaled) +} + /// Durable subscription service. Broadcast events are commit-only wakeup hints; /// every client item is re-read from protected durable storage. pub struct OrmAiSubscriptionService { sessions: Arc, principal_resolver: Arc, reauthorization_interval: Duration, + reauthorization_grace: Duration, + replay_check_interval: Duration, replay_page_size: i64, } impl OrmAiSubscriptionService { - /// Creates a service with a 30-second reauthorization interval and bounded - /// 100-event replay pages. + /// Creates a service with a 30-second reauthorization interval, a + /// 2-minute reauthorization grace window, a 10-second durable replay + /// fallback check, and bounded 100-event replay pages. pub fn new( sessions: Arc, principal_resolver: Arc, @@ -35,6 +76,8 @@ impl OrmAiSubscriptionService { sessions, principal_resolver, reauthorization_interval: Duration::from_secs(30), + reauthorization_grace: Duration::from_secs(120), + replay_check_interval: Duration::from_secs(10), replay_page_size: 100, } } @@ -47,6 +90,28 @@ impl OrmAiSubscriptionService { self } + /// Overrides how long an unavailable authorization dependency may be + /// tolerated before the stream closes. + /// + /// An authoritative denial is never subject to this window. Zero disables + /// the grace period and restores fail-on-first-failure behavior. + #[must_use] + pub fn with_reauthorization_grace(mut self, grace: Duration) -> Self { + self.reauthorization_grace = grace; + self + } + + /// Overrides the durable replay fallback interval. Zero is rejected when a + /// stream opens. + /// + /// This bounded head-sequence read is the delivery path that does not + /// depend on the process-local wakeup channel. + #[must_use] + pub fn with_replay_check_interval(mut self, interval: Duration) -> Self { + self.replay_check_interval = interval; + self + } + /// Overrides the durable replay page size, bounded to 1..=500 when a stream /// opens. #[must_use] @@ -66,6 +131,7 @@ impl AiSubscriptionService for OrmAiSubscriptionService { ) -> Result { if after_sequence < 0 || self.reauthorization_interval.is_zero() + || self.replay_check_interval.is_zero() || !(1..=500).contains(&self.replay_page_size) { return Err(AiError::InvalidConfiguration( @@ -81,6 +147,8 @@ impl AiSubscriptionService for OrmAiSubscriptionService { let sessions = self.sessions.clone(); let resolver = self.principal_resolver.clone(); let reauthorization_interval = self.reauthorization_interval; + let reauthorization_grace = self.reauthorization_grace; + let replay_check_interval = self.replay_check_interval; let replay_page_size = self.replay_page_size; Ok(Box::pin(async_stream::try_stream! { @@ -92,6 +160,14 @@ impl AiSubscriptionService for OrmAiSubscriptionService { reauthorization_interval, ); reauthorize.set_missed_tick_behavior(MissedTickBehavior::Skip); + let mut replay_check = tokio::time::interval_at( + Instant::now() + replay_check_interval, + replay_check_interval, + ); + replay_check.set_missed_tick_behavior(MissedTickBehavior::Skip); + // Grace state for an unavailable authorization dependency. + let mut grace_deadline: Option = None; + let mut grace_attempt: u32 = 0; loop { if replay_required { @@ -105,11 +181,10 @@ impl AiSubscriptionService for OrmAiSubscriptionService { .await?; let target_watermark = page.watermark; if target_watermark < delivered_sequence || page.reset_required { - yield AiSessionEventEnvelope { - event: None, - watermark: target_watermark, - reset_required: true, - }; + yield AiSessionEventEnvelope::ended( + AiSessionStreamClose::ResetRequired, + target_watermark, + ); return; } @@ -124,21 +199,16 @@ impl AiSubscriptionService for OrmAiSubscriptionService { continue; } delivered_sequence = event.sequence; - yield AiSessionEventEnvelope { - event: Some(event), - watermark: target_watermark, - reset_required: false, - }; + yield AiSessionEventEnvelope::delivered(event, target_watermark); } if delivered_sequence >= target_watermark || crossed_watermark { break; } if !page.has_more { - yield AiSessionEventEnvelope { - event: None, - watermark: target_watermark, - reset_required: true, - }; + yield AiSessionEventEnvelope::ended( + AiSessionStreamClose::ResetRequired, + target_watermark, + ); return; } page = sessions @@ -150,19 +220,26 @@ impl AiSubscriptionService for OrmAiSubscriptionService { ) .await?; if page.reset_required { - yield AiSessionEventEnvelope { - event: None, - watermark: target_watermark, - reset_required: true, - }; + yield AiSessionEventEnvelope::ended( + AiSessionStreamClose::ResetRequired, + target_watermark, + ); return; } } replay_required = false; } - let should_reauthorize = tokio::select! { - _ = reauthorize.tick() => Some(true), + enum Tick { + Reauthorize, + ReplayCheck, + Wakeup, + WakeupChannelClosed, + } + + let tick = tokio::select! { + _ = reauthorize.tick() => Tick::Reauthorize, + _ = replay_check.tick() => Tick::ReplayCheck, wakeup = wakeups.recv() => { match wakeup { Ok(wakeup) @@ -170,26 +247,104 @@ impl AiSubscriptionService for OrmAiSubscriptionService { && wakeup.sequence > delivered_sequence => { replay_required = true; + Tick::Wakeup + } + Ok(_) => Tick::Wakeup, + Err(RecvError::Lagged(_)) => { + replay_required = true; + Tick::Wakeup } - Ok(_) => {} - Err(RecvError::Lagged(_)) => replay_required = true, - Err(RecvError::Closed) => return, + Err(RecvError::Closed) => Tick::WakeupChannelClosed, } - Some(false) } }; - if should_reauthorize == Some(true) { - let resolved = resolver - .resolve(&principal_reference) - .await - .map_err(|_| AiError::ReauthorizationFailed)?; - current_principal = resolved.into_principal(); - if sessions - .session(¤t_principal, session_id) - .await? - .is_none() - { - Err(AiError::Forbidden)?; + + match tick { + Tick::Wakeup => {} + Tick::WakeupChannelClosed => { + // Durable history is intact; only the process-local + // hint path is gone. Tell the client so it can + // resubscribe instead of reading silence. + yield AiSessionEventEnvelope::ended( + AiSessionStreamClose::WakeupChannelClosed, + delivered_sequence, + ); + return; + } + Tick::ReplayCheck => { + // Fallback delivery path: a missed or dropped in-process + // wakeup is otherwise unrecoverable. One bounded + // authorized head read decides whether to replay. + match sessions.session_stream_head(¤t_principal, session_id).await? { + Some(head) if head > delivered_sequence => replay_required = true, + Some(_) => {} + None => { + yield AiSessionEventEnvelope::ended( + AiSessionStreamClose::AuthorizationRevoked, + delivered_sequence, + ); + Err(AiError::Forbidden)?; + } + } + } + Tick::Reauthorize => { + match resolver.resolve(&principal_reference).await { + Ok(resolved) => { + grace_deadline = None; + grace_attempt = 0; + current_principal = resolved.into_principal(); + if sessions + .session(¤t_principal, session_id) + .await? + .is_none() + { + yield AiSessionEventEnvelope::ended( + AiSessionStreamClose::AuthorizationRevoked, + delivered_sequence, + ); + Err(AiError::Forbidden)?; + } + } + Err(error) if is_transient_authorization_failure(&error) => { + // A brief authorization restart must not drop + // every open stream at once. Keep the existing + // principal only until the bounded grace window + // expires, and retry on a per-session jittered + // schedule so recovery is not synchronized. + let now = Instant::now(); + let deadline = *grace_deadline + .get_or_insert_with(|| now + reauthorization_grace); + if reauthorization_grace.is_zero() || now >= deadline { + yield AiSessionEventEnvelope::ended( + AiSessionStreamClose::ReauthorizationUnavailable, + delivered_sequence, + ); + Err(AiError::ReauthorizationFailed)?; + } + let backoff = jittered_backoff( + replay_check_interval, + grace_attempt, + session_id.0, + ); + grace_attempt = grace_attempt.saturating_add(1); + let next = (now + backoff).min(deadline); + reauthorize = tokio::time::interval_at( + next, + reauthorization_interval, + ); + reauthorize + .set_missed_tick_behavior(MissedTickBehavior::Skip); + } + Err(_) => { + // Authoritative denial, or a class this crate + // does not recognize. Deny fast. + yield AiSessionEventEnvelope::ended( + AiSessionStreamClose::AuthorizationRevoked, + delivered_sequence, + ); + Err(AiError::ReauthorizationFailed)?; + } + } } } } diff --git a/crates/graphql-orm-ai/src/persistence.rs b/crates/graphql-orm-ai/src/persistence.rs index fdac2fa0..2f78079d 100644 --- a/crates/graphql-orm-ai/src/persistence.rs +++ b/crates/graphql-orm-ai/src/persistence.rs @@ -888,6 +888,7 @@ pub(crate) struct AiMessageRecord { /// Hash binding the idempotency reference to text and attachment IDs. pub content_hash: Option, /// Producing run. + #[filterable(type = "uuid")] pub run_id: Option, /// Provider kind/model metadata. pub provider_kind: Option, @@ -1168,6 +1169,65 @@ pub(crate) struct AiRunCancellationRequestRecord { pub requested_at: i64, } +/// Owner-authored terminal disposition of one failed or recovery-required run. +/// +/// A failed run is either superseded by a newly authored run or explicitly +/// dismissed. Both are recorded here rather than by mutating or deleting the +/// original run: the source run row, its immutable attempt outcomes, and its +/// durable session/inbox events all remain intact for audit. +#[backend_selected_graphql_entity( + table = "graphql_orm_ai_run_failure_dispositions", + plural = "GraphqlOrmAiRunFailureDispositions", + default_sort = "decided_at ASC, id ASC", + unique_index = "source_run_id", + index( + name = "idx_graphql_orm_ai_run_failure_dispositions_session", + columns = ["session_id", "decided_at", "id"], + directions = ["asc", "asc", "asc"] + ) +)] +#[cfg_attr(feature = "mssql", derive(GraphQLSchemaEntity))] +#[cfg_attr( + any(feature = "sqlite", feature = "postgres"), + derive(GraphQLEntity, GraphQLOperations) +)] +#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq)] +pub(crate) struct AiRunFailureDispositionRecord { + /// Client-generated idempotency key. + #[primary_key] + #[graphql_orm(auto_generated = false)] + pub id: graphql_orm::uuid::Uuid, + /// Exact owning session. + #[filterable(type = "uuid")] + pub session_id: graphql_orm::uuid::Uuid, + /// Failed or recovery-required run being disposed of. At most one + /// disposition may win per run. + #[unique] + #[filterable(type = "uuid")] + pub source_run_id: graphql_orm::uuid::Uuid, + /// Durable user message the source run consumed. A retry authors a new run + /// over this same message rather than duplicating it. + #[filterable(type = "uuid")] + pub input_message_id: graphql_orm::uuid::Uuid, + /// Closed disposition: `retried` or `acknowledged`. + #[filterable(type = "string")] + pub disposition: String, + /// Newly authored run, present only for `retried`. + #[filterable(type = "uuid")] + pub retry_run_id: Option, + /// Terminal state the source run had reached when it was disposed of. + pub source_state: String, + /// Bounded safe outcome code observed on the source run. + pub source_outcome_code: Option, + /// Safe owner principal kind. + pub principal_kind: String, + /// Safe owner subject. + pub principal_subject: String, + /// Server timestamp at which the disposition won. + #[sortable] + pub decided_at: i64, +} + /// Private, protected one-shot replay-then-live subscription waiter. /// /// The row contains only safe drift/fencing metadata in ordinary columns. @@ -2364,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.60.0"; +pub const AI_SCHEMA_MODULE_VERSION: &str = "0.61.0"; /// Reserved table namespace. pub const AI_TABLE_NAMESPACE: &str = "graphql_orm_ai_"; @@ -2431,6 +2491,7 @@ impl OrmSchemaModule for AiSchemaModule { AiAttachmentArtifactRecord::metadata(), AiRunRecord::metadata(), AiRunCancellationRequestRecord::metadata(), + AiRunFailureDispositionRecord::metadata(), AiSubscriptionWaiterRecord::metadata(), AiSubscriptionWaitAdoptionRecord::metadata(), AiRunAttemptRecord::metadata(), diff --git a/crates/graphql-orm-ai/src/provider_calls.rs b/crates/graphql-orm-ai/src/provider_calls.rs index fa0213ab..6f0053e0 100644 --- a/crates/graphql-orm-ai/src/provider_calls.rs +++ b/crates/graphql-orm-ai/src/provider_calls.rs @@ -2589,13 +2589,24 @@ impl AiProviderCallExecutor { self } - pub(crate) async fn interrupt_run(&self, lease: &AiRunLease) -> Result<(), AiError> { + pub(crate) async fn interrupt_run( + &self, + lease: &AiRunLease, + ) -> Result { let binding = crate::AiProviderRunBinding::from_lease(lease)?; - self.runtime + let requested = self + .runtime .interrupt_all_provider_runs(&binding) .await - .map(|_| ()) - .map_err(|_| AiError::ProviderFailed) + .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 + }) } 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 cbee8cfa..0321718b 100644 --- a/crates/graphql-orm-ai/src/provider_run.rs +++ b/crates/graphql-orm-ai/src/provider_run.rs @@ -163,6 +163,45 @@ pub enum AiProviderRunInterruptOutcome { Requested, } +/// 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. +/// +/// 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. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub enum AiRunInterruptSettlement { + /// No live provider resource existed for the exact fence. + NotActive, + /// Interruption was requested and acknowledged, but the retained thread's + /// post-interrupt content is not proven to match the durable transcript. + RequestedUnsettled, + /// The provider proved the interrupted turn left its retained thread + /// consistent with the durable transcript, with no unresolved dynamic tool + /// call and no uncertain persisted output. + Settled, +} + +impl AiRunInterruptSettlement { + /// Returns whether the retained thread may be kept bound. + /// + /// Anything other than proven settlement is false, so an unrecognized or + /// merely acknowledged interruption fails closed into invalidation. + pub const fn retains_thread(self) -> bool { + matches!(self, Self::Settled) + } +} + /// Result of closing one exact provider-run resource. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum AiProviderRunCloseOutcome { diff --git a/crates/graphql-orm-ai/src/run_disposition.rs b/crates/graphql-orm-ai/src/run_disposition.rs new file mode 100644 index 00000000..2e41e54a --- /dev/null +++ b/crates/graphql-orm-ai/src/run_disposition.rs @@ -0,0 +1,174 @@ +//! Owner-authorized disposition of a failed or recovery-required run. +//! +//! A terminal run never resumes. Its durable state, immutable attempt +//! outcomes, and session/inbox events are permanent. This module adds the two +//! things an owner may still do about a failure: +//! +//! - **Retry** authors a *new* run over the same already-persisted user +//! message, under current policy, and only when the server can prove +//! re-execution is safe. +//! - **Acknowledge** durably dismisses the failure so a client can stop +//! surfacing it, without removing any audit history. +//! +//! Neither operation mutates the source run, deletes a row, or grants +//! provider, application-tool, approval, or run-state authority. + +use agql_auth::AuthPrincipal; +use async_graphql::{Enum, InputObject, SimpleObject}; +use async_trait::async_trait; +use uuid::Uuid; + +use crate::{AiError, AiRunRetryAdmission}; + +/// Closed owner-authored disposition of one failed run. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Enum)] +#[cfg_attr(feature = "graphql-case-pascal", graphql(rename_items = "PascalCase"))] +pub enum AiRunDisposition { + /// A new run was authored over the same durable user message. + Retried, + /// The failure was dismissed without authoring a new run. + Acknowledged, +} + +impl AiRunDisposition { + /// Stable durable storage value. + pub const fn as_str(self) -> &'static str { + match self { + Self::Retried => "retried", + Self::Acknowledged => "acknowledged", + } + } + + /// Parses one stable durable storage value. + pub const fn from_persisted(value: &str) -> Option { + match value.as_bytes() { + b"retried" => Some(Self::Retried), + b"acknowledged" => Some(Self::Acknowledged), + _ => None, + } + } +} + +/// Exact owner request to author a new run for a failed run's user message. +#[derive(Clone, Debug, InputObject)] +#[cfg_attr(feature = "graphql-case-pascal", graphql(rename_fields = "PascalCase"))] +pub struct RetryAiRunInput { + /// Owning session. + pub session_id: Uuid, + /// Failed or recovery-required run to supersede. + pub run_id: Uuid, + /// Client-generated idempotency key. + pub client_request_id: Uuid, +} + +/// Exact owner request to dismiss a failed run without retrying it. +#[derive(Clone, Debug, InputObject)] +#[cfg_attr(feature = "graphql-case-pascal", graphql(rename_fields = "PascalCase"))] +pub struct AcknowledgeAiRunFailureInput { + /// Owning session. + pub session_id: Uuid, + /// Failed or recovery-required run to dismiss. + pub run_id: Uuid, + /// Client-generated idempotency key. + pub client_request_id: Uuid, +} + +/// Authoritative result of an accepted disposition request. +#[derive(Clone, Debug, SimpleObject)] +#[cfg_attr(feature = "graphql-case-pascal", graphql(rename_fields = "PascalCase"))] +pub struct AiRunDispositionView { + /// Owning session. + pub session_id: Uuid, + /// Disposed source run. + pub run_id: Uuid, + /// Idempotency key that won the disposition fence. + pub client_request_id: Uuid, + /// Closed disposition that was recorded. + pub disposition: AiRunDisposition, + /// Newly authored run, present only for a retry. + pub retry_run_id: Option, + /// Durable user message the source run consumed. + pub input_message_id: Uuid, + /// Server timestamp at which the disposition won. + pub decided_at: i64, +} + +/// Why a retry request was refused, without disclosing provider detail. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Enum)] +#[cfg_attr(feature = "graphql-case-pascal", graphql(rename_items = "PascalCase"))] +pub enum AiRunRetryRefusal { + /// The source run is not in a terminal failed state. + NotFailed, + /// Re-execution could not be proven safe. + Uncertain, + /// The user message already has a durable assistant answer. + AlreadyAnswered, + /// A different disposition already won for this run. + AlreadyDisposed, +} + +impl AiRunRetryRefusal { + /// Stable public value. + pub const fn as_str(self) -> &'static str { + match self { + Self::NotFailed => "not_failed", + Self::Uncertain => "uncertain", + Self::AlreadyAnswered => "already_answered", + Self::AlreadyDisposed => "already_disposed", + } + } + + /// Maps a refusing admission to its public reason. + pub const fn from_admission(admission: AiRunRetryAdmission) -> Option { + match admission { + AiRunRetryAdmission::Allowed => None, + AiRunRetryAdmission::RefusedUncertain => Some(Self::Uncertain), + AiRunRetryAdmission::RefusedAlreadyAnswered => Some(Self::AlreadyAnswered), + } + } +} + +/// Current-owner failure-disposition boundary used by the GraphQL mutations. +#[async_trait] +pub trait AiRunDispositionService: Send + Sync { + /// Authors a new run for the same durable user message as one failed run. + /// + /// Implementations must rehydrate current authority, apply session/scope + /// access, and re-decide retry admission from committed rows inside the + /// same transaction that records the disposition. Replaying the same + /// `client_request_id` returns the original result. The new run carries a + /// fresh principal reference so it executes under current policy; it never + /// inherits the source run's lease, attempt, checkpoint, approval, or + /// provider session. + /// + /// # Errors + /// + /// Returns [`AiError::Forbidden`] for an unauthorized principal, + /// [`AiError::NotFound`] for an invisible session or run, and + /// [`AiError::Conflict`] when retry is refused or another disposition + /// already won. + async fn retry_run( + &self, + principal: &AuthPrincipal, + input: RetryAiRunInput, + ) -> Result; + + /// Durably dismisses one failed run's failure. + /// + /// Acknowledgement is always available for a terminal failed or + /// recovery-required run, including one whose retry is refused: dismissing + /// a failure asserts nothing about whether re-execution would be safe. It + /// removes no row and no event. + /// + /// # Errors + /// + /// Returns [`AiError::Forbidden`] for an unauthorized principal, + /// [`AiError::NotFound`] for an invisible session or run, and + /// [`AiError::Conflict`] when the run is not terminally failed or another + /// disposition already won. + async fn acknowledge_run_failure( + &self, + principal: &AuthPrincipal, + input: AcknowledgeAiRunFailureInput, + ) -> Result; +} diff --git a/crates/graphql-orm-ai/src/run_state.rs b/crates/graphql-orm-ai/src/run_state.rs index dc7ed79c..5232efaf 100644 --- a/crates/graphql-orm-ai/src/run_state.rs +++ b/crates/graphql-orm-ai/src/run_state.rs @@ -300,3 +300,149 @@ pub enum AiRunTransitionError { #[error(transparent)] Lease(#[from] LeaseError), } + +/// Wire version of the bounded run-failure record carried by terminal run +/// events. +pub const AI_RUN_FAILURE_VERSION: u16 = 1; + +/// Whether a client may author a new run for the same durable user message. +/// +/// This is not a state-machine transition. `RecoveryRequired`, `Failed`, and +/// `Cancelled` are all terminal and never resume; retry means *authoring a new +/// run* over the same already-persisted user message under current policy. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum AiRunRetryAdmission { + /// A new run may be authored for the same durable user message. + Allowed, + /// Re-execution is refused because the original stop left an effect that + /// could not be proven safe. This is the fail-closed default. + RefusedUncertain, + /// Re-execution is refused because the run already produced a durable + /// assistant answer for that user message. + RefusedAlreadyAnswered, +} + +impl AiRunRetryAdmission { + /// Stable public value. + pub const fn as_str(self) -> &'static str { + match self { + Self::Allowed => "allowed", + Self::RefusedUncertain => "refused_uncertain", + Self::RefusedAlreadyAnswered => "refused_already_answered", + } + } + + /// Returns whether a retry request will be admitted. + pub const fn is_allowed(self) -> bool { + matches!(self, Self::Allowed) + } +} + +/// Durable evidence a retry decision is allowed to consider. +/// +/// Every field must come from committed rows. Nothing here may be inferred +/// from elapsed time, in-memory worker state, or provider reports. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct AiRunRetryEvidence { + /// Terminal state the run actually reached. + pub terminal: AiRunTerminalEvent, + /// Whether a durable assistant message exists for this run. + pub produced_assistant_output: bool, +} + +/// Classifies whether a terminal run may be retried as a new run. +/// +/// The rules are deliberately conservative: +/// +/// - `RecoveryRequired` is never retryable. It exists precisely because an +/// external effect could not be proven safe, and re-execution is what the +/// safe-failure guardrail forbids. +/// - `Completed` is never retryable; the message already has its answer. +/// - `Cancelled` is retryable only when the run produced no durable assistant +/// output. Cancellation observed *after* a provider turn was persisted +/// leaves a fully answered message, and authoring a second run over it would +/// produce a second answer. +/// - `Failed` is retryable only for an explicitly proven-clean code. An absent +/// or unrecognized code is refused, because an unclassified failure is +/// exactly the case where safety cannot be proven. +pub fn classify_run_retry( + evidence: AiRunRetryEvidence, + outcome_code: Option<&str>, +) -> AiRunRetryAdmission { + match evidence.terminal { + AiRunTerminalEvent::RecoveryRequired => AiRunRetryAdmission::RefusedUncertain, + AiRunTerminalEvent::Completed => AiRunRetryAdmission::RefusedAlreadyAnswered, + AiRunTerminalEvent::Cancelled => { + if evidence.produced_assistant_output { + AiRunRetryAdmission::RefusedAlreadyAnswered + } else { + AiRunRetryAdmission::Allowed + } + } + AiRunTerminalEvent::Failed => { + if evidence.produced_assistant_output { + return AiRunRetryAdmission::RefusedAlreadyAnswered; + } + match outcome_code { + Some(code) if is_retryable_failure_code(code) => AiRunRetryAdmission::Allowed, + _ => AiRunRetryAdmission::RefusedUncertain, + } + } + } +} + +/// Closed allowlist of failure codes that leave no unproven external effect. +/// +/// Membership is opt-in. A code absent from this list is refused, so adding a +/// new failure classification cannot silently make it retryable. +const fn is_retryable_failure_code(code: &str) -> bool { + matches!( + code.as_bytes(), + b"provider_session_cleanup_unavailable" + | b"agent_rule_budget_exceeded" + | b"agent_rule_changed_after_provider" + | b"agent_turn_limit_reached" + | b"provider_unavailable" + | b"provider_rate_limited" + | b"provider_request_rejected" + | b"runtime_not_ready" + ) +} + +/// Bounded, content-free failure record a client may render. +/// +/// It mirrors the model-visible safe failure envelope shape and carries only +/// server-owned classification: never a prompt, provider payload, tool +/// argument, stack, or authorization detail. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct AiRunFailure { + /// Stable redacted outcome code, absent when the writer supplied none. + pub code: Option, + /// Whether a new run may be authored for the same durable user message. + pub retryable: bool, + /// Why retry is or is not admitted. + pub admission: AiRunRetryAdmission, +} + +impl AiRunFailure { + /// Builds the record for one classified terminal outcome. + pub fn new(admission: AiRunRetryAdmission, code: Option) -> Self { + Self { + code, + retryable: admission.is_allowed(), + admission, + } + } + + /// Serializes the versioned, content-free record. + pub fn to_json(&self) -> serde_json::Value { + serde_json::json!({ + "version": AI_RUN_FAILURE_VERSION, + "ok": false, + "code": self.code, + "retryable": self.retryable, + "admission": self.admission.as_str(), + }) + } +} diff --git a/crates/graphql-orm-ai/src/sessions.rs b/crates/graphql-orm-ai/src/sessions.rs index 953acd16..5acddbf7 100644 --- a/crates/graphql-orm-ai/src/sessions.rs +++ b/crates/graphql-orm-ai/src/sessions.rs @@ -11,8 +11,9 @@ use graphql_orm::graphql::pagination::{ use uuid::Uuid; use crate::{ - AiError, AiInboxEventPage, AiInboxService, AiRunCancellationService, AiRunCancellationView, - AiScope, AiSessionId, AiUsageConnection, AiUsageFilterInput, CancelAiRunInput, + AcknowledgeAiRunFailureInput, AiError, AiInboxEventPage, AiInboxService, + AiRunCancellationService, AiRunCancellationView, AiRunDispositionService, AiRunDispositionView, + AiScope, AiSessionId, AiUsageConnection, AiUsageFilterInput, CancelAiRunInput, RetryAiRunInput, }; /// Scope input for session creation/configuration. @@ -255,7 +256,11 @@ pub struct AiConversationBootstrap { pub backward_cursor: Option, /// Whether older messages remain. pub has_older_messages: bool, - /// Durable event watermark captured with this state. + /// Durable event resume floor captured before this state was assembled. + /// + /// Subscribe with `after_sequence = watermark`. Run and tool-call rows may + /// already reflect events after it; replaying those is idempotent because + /// both are identified state. Messages never lead it. pub watermark: i64, /// All active runs admitted by the bootstrap bound. pub active_runs: Vec, @@ -344,6 +349,13 @@ pub struct SendAiMessagePayload { pub trait AiSessionService: Send + Sync { /// Returns one bounded authoritative conversation bootstrap suitable for /// replay from its watermark followed by live subscription. + /// + /// The returned `watermark` is a resume floor, not an equality point. Every + /// durable effect at or before it is reflected in the snapshot, so a client + /// that subscribes with `after_sequence = watermark` cannot miss an event. + /// Run and tool-call rows may already reflect a later effect; those rows are + /// identified state, so re-applying the replayed event that produced them is + /// idempotent. The message window never leads the watermark. async fn conversation_bootstrap( &self, _principal: &AuthPrincipal, @@ -639,6 +651,36 @@ impl AiMutationRoot { .map_err(extend) } + /// Authors a new run for the same durable user message as one failed run. + /// + /// This is not a resume: the failed run stays terminal and every audit row + /// it produced is preserved. The request is refused when the server cannot + /// prove re-execution is safe. + async fn retry_ai_run( + &self, + context: &Context<'_>, + input: RetryAiRunInput, + ) -> async_graphql::Result { + let principal = agql_auth::principal_from_ctx(context)?; + disposition_service(context)? + .retry_run(&principal, input) + .await + .map_err(extend) + } + + /// Durably dismisses one failed run's failure without deleting history. + async fn acknowledge_ai_run_failure( + &self, + context: &Context<'_>, + input: AcknowledgeAiRunFailureInput, + ) -> async_graphql::Result { + let principal = agql_auth::principal_from_ctx(context)?; + disposition_service(context)? + .acknowledge_run_failure(&principal, input) + .await + .map_err(extend) + } + /// Creates a private owner-only session. async fn create_ai_session( &self, @@ -754,6 +796,18 @@ fn cancellation_service( }) } +fn disposition_service( + context: &Context<'_>, +) -> async_graphql::Result> { + context + .data_opt::>() + .cloned() + .ok_or_else(|| { + AiError::InvalidConfiguration("AI run disposition service is not installed".to_owned()) + .extend() + }) +} + fn tool_result_preview_service( context: &Context<'_>, ) -> async_graphql::Result> { diff --git a/crates/graphql-orm-ai/src/subscriptions.rs b/crates/graphql-orm-ai/src/subscriptions.rs index 1df62a0e..8556874d 100644 --- a/crates/graphql-orm-ai/src/subscriptions.rs +++ b/crates/graphql-orm-ai/src/subscriptions.rs @@ -4,7 +4,7 @@ use std::pin::Pin; use std::sync::Arc; use agql_auth::AuthPrincipal; -use async_graphql::{Context, ErrorExtensions, SimpleObject, Subscription}; +use async_graphql::{Context, Enum, ErrorExtensions, SimpleObject, Subscription}; use async_trait::async_trait; use futures::{Stream, StreamExt}; use uuid::Uuid; @@ -21,16 +21,75 @@ pub struct AiSessionWakeup { pub sequence: i64, } -/// Subscription item supporting explicit retention-gap reset signaling. +/// Closed reason for a session-event stream that ended without the client +/// unsubscribing. +/// +/// A stream that simply stops producing items is indistinguishable from +/// network silence, so every server-side end carries exactly one of these on a +/// final envelope. None of these values disclose provider, prompt, tool, or +/// authorization detail. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Enum)] +#[cfg_attr(feature = "graphql-case-pascal", graphql(rename_items = "PascalCase"))] +pub enum AiSessionStreamClose { + /// Retention removed history the client still needs. Discard derived state + /// and reload from `aiConversationBootstrap`. + ResetRequired, + /// The host's in-process wakeup channel closed, normally because the + /// process is shutting down. Durable history is intact; resubscribe from + /// the last delivered watermark. + WakeupChannelClosed, + /// Reauthorization returned an authoritative denial, or the session is no + /// longer visible to the principal. Do not resubscribe with the same + /// credentials. + AuthorizationRevoked, + /// Reauthorization could not be completed within the bounded grace window + /// because the authorization dependency was unavailable. Durable history is + /// intact; resubscribe after backing off. + ReauthorizationUnavailable, +} + +/// Subscription item supporting explicit retention-gap reset signaling and +/// typed terminal close signaling. #[derive(Clone, Debug, SimpleObject)] #[cfg_attr(feature = "graphql-case-pascal", graphql(rename_fields = "PascalCase"))] pub struct AiSessionEventEnvelope { - /// Durable event, absent only for a reset signal. + /// Durable event, absent only for a reset or close signal. pub event: Option, /// Replay watermark associated with this delivery. pub watermark: i64, /// Whether retention removed required history and the client must reload. pub reset_required: bool, + /// Set only on the final envelope of a server-ended stream, and `None` on + /// every ordinary delivery. An end with no preceding close envelope and no + /// error is a client unsubscribe or a transport failure. + pub closed: Option, +} + +impl AiSessionEventEnvelope { + /// Creates an ordinary delivery for one durable event. + #[must_use] + pub const fn delivered(event: AiSessionEventView, watermark: i64) -> Self { + Self { + event: Some(event), + watermark, + reset_required: false, + closed: None, + } + } + + /// Creates the final envelope for a server-ended stream. + /// + /// `reset_required` stays coupled to [`AiSessionStreamClose::ResetRequired`] + /// so existing clients that only read the boolean keep working. + #[must_use] + pub const fn ended(reason: AiSessionStreamClose, watermark: i64) -> Self { + Self { + event: None, + watermark, + reset_required: matches!(reason, AiSessionStreamClose::ResetRequired), + closed: Some(reason), + } + } } /// Type-erased bounded event stream. diff --git a/crates/graphql-orm-ai/tests/orm_run_cancellation.rs b/crates/graphql-orm-ai/tests/orm_run_cancellation.rs index 156f72c4..75ca5e0b 100644 --- a/crates/graphql-orm-ai/tests/orm_run_cancellation.rs +++ b/crates/graphql-orm-ai/tests/orm_run_cancellation.rs @@ -421,3 +421,111 @@ async fn cancellation_wakes_the_fenced_worker_and_wrong_pairs_fail_closed() { assert!(observed.expect("wait should succeed").is_some()); assert_eq!(view.expect("request should succeed").state, "cancelled"); } + +/// Work item 1: a bootstrap opened *while* durable session events are being +/// appended must return a snapshot rather than `Conflict`. +/// +/// Cancellation is used as the event source because it appends two session +/// events per run and touches no field the bootstrap returns, which is exactly +/// the shape of coalesced live-delta churn during streaming. The writer runs +/// concurrently with the readers so the append lands between the bootstrap's +/// two session reads. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn bootstrap_survives_concurrent_session_event_churn() { + let fixture = Arc::new(fixture().await); + let (session, _first_run) = active_run(&fixture).await; + + // Enqueue every run up front so no message lands during the bootstraps: a + // new message legitimately invalidates the snapshot, and this test is + // about event churn alone. + let mut runs = Vec::new(); + for _ in 0..64 { + let sent = fixture + .sessions + .send_message( + &fixture.owner, + SendAiMessageInput { + session_id: session.id, + text: "Keep counting".to_owned(), + attachment_ids: vec![], + client_message_id: Uuid::new_v4(), + }, + ) + .await + .expect("message should enqueue a run"); + runs.push(sent.run_id); + } + let expected_messages = runs.len() + 1; + + let before = fixture + .sessions + .conversation_bootstrap(&fixture.owner, AiSessionId(session.id), 200, 20, 500) + .await + .expect("quiescent bootstrap should succeed"); + + let writer_fixture = fixture.clone(); + let writer_session = session.id; + let writer = tokio::spawn(async move { + for run_id in runs { + writer_fixture + .cancellation + .request_cancellation( + &writer_fixture.owner, + CancelAiRunInput { + session_id: writer_session, + run_id, + client_request_id: Uuid::new_v4(), + }, + ) + .await + .expect("owner should cancel the queued run"); + tokio::task::yield_now().await; + } + }); + + let mut bootstraps = 0_u32; + while !writer.is_finished() { + let bootstrap = fixture + .sessions + .conversation_bootstrap(&fixture.owner, AiSessionId(session.id), 200, 20, 500) + .await + .expect("bootstrap must not fail while session events are appended"); + assert_eq!(bootstrap.messages.len(), expected_messages); + assert!(!bootstrap.reset_required); + + // The watermark is a resume floor: replay strictly after it must never + // skip an event, and the snapshot must already cover everything at or + // below it. + let page = fixture + .sessions + .session_event_page( + &fixture.owner, + AiSessionId(session.id), + bootstrap.watermark, + 500, + ) + .await + .expect("replay should start from the returned watermark"); + assert!(!page.reset_required); + assert!(page.watermark >= bootstrap.watermark); + for event in &page.events { + assert!(event.sequence > bootstrap.watermark); + } + bootstraps += 1; + } + writer.await.expect("writer task should not panic"); + + let after = fixture + .sessions + .conversation_bootstrap(&fixture.owner, AiSessionId(session.id), 200, 20, 500) + .await + .expect("final bootstrap should succeed"); + assert!( + after.watermark > before.watermark, + "the churn source must actually advance the durable stream head" + ); + assert!( + bootstraps > 0, + "the reader must have raced the writer at least once" + ); +} diff --git a/crates/graphql-orm-ai/tests/orm_run_disposition.rs b/crates/graphql-orm-ai/tests/orm_run_disposition.rs new file mode 100644 index 00000000..b651a841 --- /dev/null +++ b/crates/graphql-orm-ai/tests/orm_run_disposition.rs @@ -0,0 +1,526 @@ +#![cfg(feature = "sqlite")] +//! Owner-authorized retry and acknowledgement of failed runs. + +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; + +use agql_auth::{ + AccessTokenMetadata, AuthPrincipal, AuthUser, Clock, CurrentPrincipalResolver, FixedClock, + PrincipalReference, ResolvedPrincipal, SessionContext, +}; +use async_trait::async_trait; +use graphql_orm::graphql::orm::{ApplyOptions, OrmSchemaModule}; +use graphql_orm::prelude::{Database, SqliteBackend}; +use graphql_orm_ai::*; +use time::{Duration, OffsetDateTime}; +use uuid::Uuid; + +struct AllowAll; + +#[async_trait] +impl AiAccessPolicy for AllowAll { + async fn can_access_scope( + &self, + _principal: &AuthPrincipal, + _scope: &AiScope, + _action: AiSessionAction, + ) -> AiAccessDecision { + AiAccessDecision::allow("run-disposition-test", "v1") + } + + async fn can_access_session( + &self, + _principal: &AuthPrincipal, + _session_id: AiSessionId, + _action: AiSessionAction, + ) -> AiAccessDecision { + AiAccessDecision::allow("run-disposition-test", "v1") + } +} + +struct ProtectionPolicy; + +#[async_trait] +impl AiContentProtectionPolicyResolver for ProtectionPolicy { + async fn resolve( + &self, + _principal: &AuthPrincipal, + scope: &AiScope, + ) -> Result { + Ok(AiContentProtectionPolicy { + scope: scope.clone(), + mode: AiContentProtectionMode::DatabaseManaged, + key_policy_reference: None, + version: 1, + ready: true, + }) + } +} + +struct StaticResolver { + principal: AuthPrincipal, + active: Arc, + clock: Arc, +} + +#[async_trait] +impl CurrentPrincipalResolver for StaticResolver { + async fn resolve( + &self, + reference: &PrincipalReference, + ) -> agql_auth::AuthResult { + if !self.active.load(Ordering::SeqCst) { + return Err(agql_auth::AuthError::Forbidden); + } + ResolvedPrincipal::new(reference.clone(), self.principal.clone(), self.clock.now()) + } +} + +fn principal(subject: &str) -> AuthPrincipal { + AuthPrincipal::User(AuthUser { + user_id: subject.to_owned(), + session_id: Uuid::new_v4(), + roles: vec![], + scopes: vec![], + session: SessionContext::default(), + token_claims: AccessTokenMetadata { + tenant_id: Some("tenant-disposition".to_owned()), + ..AccessTokenMetadata::default() + }, + }) +} + +struct Fixture { + sessions: Arc, + runs: OrmAiRunService, + dispositions: OrmAiRunDispositionService, + owner: AuthPrincipal, + active: Arc, +} + +async fn fixture_on(database: Database, migrate: bool) -> Fixture { + if migrate { + let module = AiSchemaModule; + let plan = database + .schema() + .plan_migration_to_entities( + "ai-run-disposition-test-v1", + "AI run disposition test", + module.entities(), + ) + .await + .expect("AI schema migration should plan"); + database + .schema() + .apply_migration(&plan, ApplyOptions::default()) + .await + .expect("AI schema migration should apply"); + } + let owner = principal("disposition-owner"); + // The session service stamps `next_attempt_at` from the real clock while + // `claim_next` compares against this fixed one, so a fixture pinned to + // "now" only claims when both land in the same second. Lead the real clock + // so the queued run is eligible regardless of where the boundary falls. + let clock = Arc::new(FixedClock::new( + OffsetDateTime::now_utc() + Duration::seconds(2), + )); + let active = Arc::new(AtomicBool::new(true)); + let access_policy: Arc = Arc::new(AllowAll); + let protection_policy: Arc = Arc::new(ProtectionPolicy); + let content_protector: Arc = Arc::new(DatabaseManagedContentProtector); + let principal_resolver: Arc = Arc::new(StaticResolver { + principal: owner.clone(), + active: active.clone(), + clock: clock.clone(), + }); + let sessions = Arc::new(OrmAiSessionService::new( + database.clone(), + access_policy.clone(), + protection_policy.clone(), + content_protector.clone(), + )); + let runs = OrmAiRunService::new( + database.clone(), + clock.clone(), + AiRunServiceLimits::new(Duration::minutes(1), Duration::minutes(1), 16, 3, 3) + .expect("run limits should validate"), + ); + let dispositions = OrmAiRunDispositionService::new( + database, + access_policy, + protection_policy, + content_protector, + principal_resolver, + clock, + AiRunDispositionLimits::default(), + ); + Fixture { + sessions, + runs, + dispositions, + owner, + active, + } +} + +async fn fixture() -> Fixture { + let database = Database::::connect_sqlite("sqlite::memory:") + .await + .expect("in-memory SQLite should open"); + fixture_on(database, true).await +} + +async fn session(fixture: &Fixture) -> AiSessionView { + fixture + .sessions + .create_session( + &fixture.owner, + CreateAiSessionInput { + scope: AiScopeInput { + kind: "workspace".to_owned(), + id: "workspace-disposition".to_owned(), + tenant_id: Some("tenant-disposition".to_owned()), + }, + title: None, + }, + ) + .await + .expect("session should create") +} + +/// Drives one run to a terminal state with the supplied outcome/error code. +async fn failed_run( + fixture: &Fixture, + session_id: Uuid, + final_state: AiRunState, + outcome_code: &str, + error_code: Option<&str>, +) -> SendAiMessagePayload { + let sent = fixture + .sessions + .send_message( + &fixture.owner, + SendAiMessageInput { + session_id, + text: "Count my records".to_owned(), + attachment_ids: vec![], + client_message_id: Uuid::new_v4(), + }, + ) + .await + .expect("message should enqueue a run"); + let claimed = fixture + .runs + .claim_next("run-disposition-test-worker") + .await + .expect("claim should succeed") + .expect("queued run should exist"); + let running = fixture + .runs + .start(&claimed) + .await + .expect("run should start"); + fixture + .runs + .finish( + &running, + AiRunCompletion::new( + final_state, + outcome_code, + error_code.map(str::to_owned), + None, + ) + .expect("completion should validate"), + ) + .await + .expect("terminal write should commit"); + sent +} + +fn failure_record(page: &AiSessionEventPage, event_type: &str) -> serde_json::Value { + let event = page + .events + .iter() + .find(|event| event.event_type == event_type) + .unwrap_or_else(|| panic!("{event_type} should be durable")); + event.payload.0["failure"].clone() +} + +#[tokio::test] +async fn retry_authors_a_new_run_over_the_same_message_and_is_idempotent() { + let fixture = fixture().await; + let session = session(&fixture).await; + let sent = failed_run( + &fixture, + session.id, + AiRunState::Failed, + "agent_rule_budget_exceeded", + Some("agent_rule_budget_exceeded"), + ) + .await; + + let client_request_id = Uuid::new_v4(); + let input = RetryAiRunInput { + session_id: session.id, + run_id: sent.run_id, + client_request_id, + }; + let first = fixture + .dispositions + .retry_run(&fixture.owner, input.clone()) + .await + .expect("a proven-clean failure should admit a retry"); + assert_eq!(first.disposition, AiRunDisposition::Retried); + assert_eq!(first.input_message_id, sent.message_id); + let retry_run_id = first.retry_run_id.expect("retry should author a new run"); + assert_ne!(retry_run_id, sent.run_id); + + // Replaying the same key must not author a second run. + let replay = fixture + .dispositions + .retry_run(&fixture.owner, input) + .await + .expect("the same idempotency key should replay"); + assert_eq!(replay.retry_run_id, Some(retry_run_id)); + assert_eq!(replay.decided_at, first.decided_at); + + // A different key for the same already-disposed run is refused. + assert!(matches!( + fixture + .dispositions + .retry_run( + &fixture.owner, + RetryAiRunInput { + session_id: session.id, + run_id: sent.run_id, + client_request_id: Uuid::new_v4(), + }, + ) + .await, + Err(AiError::Conflict) + )); + + // The new run is queued over the same durable user message and is claimable. + let bootstrap = fixture + .sessions + .conversation_bootstrap(&fixture.owner, AiSessionId(session.id), 20, 20, 100) + .await + .expect("bootstrap should succeed"); + let queued = bootstrap + .active_runs + .iter() + .find(|run| run.id == retry_run_id) + .expect("the retry run should be active"); + assert_eq!(queued.state, "queued"); + assert_eq!(queued.input_message_id, sent.message_id); + assert_eq!( + bootstrap.messages.len(), + 1, + "retry must not duplicate the prompt" + ); + + // The source run stays terminal: retry never resurrects it. + let source = bootstrap + .terminal_runs + .iter() + .find(|run| run.id == sent.run_id) + .expect("the source run should stay terminal"); + assert_eq!(source.state, "failed"); +} + +#[tokio::test] +async fn recovery_required_refuses_retry_but_still_admits_acknowledgement() { + let fixture = fixture().await; + let session = session(&fixture).await; + let sent = failed_run( + &fixture, + session.id, + AiRunState::RecoveryRequired, + "provider_turn_uncertain", + Some("provider_turn_uncertain"), + ) + .await; + + assert!( + matches!( + fixture + .dispositions + .retry_run( + &fixture.owner, + RetryAiRunInput { + session_id: session.id, + run_id: sent.run_id, + client_request_id: Uuid::new_v4(), + }, + ) + .await, + Err(AiError::Conflict) + ), + "an unproven external effect must never be re-executed" + ); + + let acknowledged = fixture + .dispositions + .acknowledge_run_failure( + &fixture.owner, + AcknowledgeAiRunFailureInput { + session_id: session.id, + run_id: sent.run_id, + client_request_id: Uuid::new_v4(), + }, + ) + .await + .expect("dismissing a failure asserts nothing about re-execution safety"); + assert_eq!(acknowledged.disposition, AiRunDisposition::Acknowledged); + assert!(acknowledged.retry_run_id.is_none()); + + // Audit history survives the dismissal. + let page = fixture + .sessions + .session_event_page(&fixture.owner, AiSessionId(session.id), 0, 500) + .await + .expect("events should replay"); + assert!( + page.events + .iter() + .any(|event| event.event_type == "run_recovery_required") + ); + assert!( + page.events + .iter() + .any(|event| event.event_type == "run_failure_acknowledged") + ); +} + +#[tokio::test] +async fn an_unclassified_failure_is_not_retryable() { + let fixture = fixture().await; + let session = session(&fixture).await; + let sent = failed_run( + &fixture, + session.id, + AiRunState::Failed, + "worker_stopped", + None, + ) + .await; + + let page = fixture + .sessions + .session_event_page(&fixture.owner, AiSessionId(session.id), 0, 500) + .await + .expect("events should replay"); + let failure = failure_record(&page, "run_failed"); + assert_eq!(failure["retryable"], serde_json::json!(false)); + assert_eq!(failure["admission"], serde_json::json!("refused_uncertain")); + assert_eq!(failure["code"], serde_json::Value::Null); + + assert!(matches!( + fixture + .dispositions + .retry_run( + &fixture.owner, + RetryAiRunInput { + session_id: session.id, + run_id: sent.run_id, + client_request_id: Uuid::new_v4(), + }, + ) + .await, + Err(AiError::Conflict) + )); +} + +#[tokio::test] +async fn a_revoked_principal_cannot_dispose_of_a_failure() { + let fixture = fixture().await; + let session = session(&fixture).await; + let sent = failed_run( + &fixture, + session.id, + AiRunState::Failed, + "agent_rule_budget_exceeded", + Some("agent_rule_budget_exceeded"), + ) + .await; + fixture.active.store(false, Ordering::SeqCst); + assert!(matches!( + fixture + .dispositions + .retry_run( + &fixture.owner, + RetryAiRunInput { + session_id: session.id, + run_id: sent.run_id, + client_request_id: Uuid::new_v4(), + }, + ) + .await, + Err(AiError::ReauthorizationFailed) + )); +} + +/// Work item 3: a failed run's terminal event and its bounded failure record +/// must survive a host restart and replay to a reconnecting client. +#[tokio::test] +async fn failed_run_events_replay_with_their_failure_record_after_restart() { + let directory = tempfile::tempdir().expect("temporary directory"); + let path = directory.path().join("run-disposition.sqlite"); + let url = format!("sqlite://{}?mode=rwc", path.display()); + + let session_id; + let run_id; + { + let database = Database::::connect_sqlite(&url) + .await + .expect("database opens"); + let fixture = fixture_on(database, true).await; + let view = session(&fixture).await; + session_id = view.id; + run_id = failed_run( + &fixture, + session_id, + AiRunState::Failed, + "agent_rule_budget_exceeded", + Some("agent_rule_budget_exceeded"), + ) + .await + .run_id; + } + + // Fresh process: new handles, no in-memory state carried over. + let database = Database::::connect_sqlite(&url) + .await + .expect("database reopens"); + let fixture = fixture_on(database, false).await; + let page = fixture + .sessions + .session_event_page(&fixture.owner, AiSessionId(session_id), 0, 500) + .await + .expect("durable events should replay after a restart"); + assert!(!page.reset_required); + let failure = failure_record(&page, "run_failed"); + assert_eq!(failure["version"], serde_json::json!(1)); + assert_eq!(failure["ok"], serde_json::json!(false)); + assert_eq!(failure["retryable"], serde_json::json!(true)); + assert_eq!(failure["admission"], serde_json::json!("allowed")); + assert_eq!( + failure["code"], + serde_json::json!("agent_rule_budget_exceeded") + ); + + // The flag is authoritative: the retry it advertises is actually admitted. + let disposition = fixture + .dispositions + .retry_run( + &fixture.owner, + RetryAiRunInput { + session_id, + run_id, + client_request_id: Uuid::new_v4(), + }, + ) + .await + .expect("an advertised retryable failure must be retryable"); + assert_eq!(disposition.disposition, AiRunDisposition::Retried); +} diff --git a/crates/graphql-orm-ai/tests/orm_subscriptions.rs b/crates/graphql-orm-ai/tests/orm_subscriptions.rs index 6d009f59..197c67cf 100644 --- a/crates/graphql-orm-ai/tests/orm_subscriptions.rs +++ b/crates/graphql-orm-ai/tests/orm_subscriptions.rs @@ -1,7 +1,7 @@ #![cfg(feature = "sqlite")] use std::sync::Arc; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::time::Duration; use agql_auth::{ @@ -94,6 +94,20 @@ fn principal() -> AuthPrincipal { }) } +async fn apply_schema(database: &Database, name: &str) { + let module = AiSchemaModule; + let plan = database + .schema() + .plan_migration_to_entities(name, "AI subscription service test", module.entities()) + .await + .expect("schema plans"); + database + .schema() + .apply_migration(&plan, ApplyOptions::default()) + .await + .expect("schema applies"); +} + async fn services( reauthorization_interval: Duration, ) -> ( @@ -256,10 +270,23 @@ async fn replay_is_paged_to_a_watermark_and_revocation_closes_stream() { assert_eq!(second.event.expect("event").sequence, 2); active.store(false, Ordering::SeqCst); - let revoked = tokio::time::timeout(Duration::from_secs(1), stream.next()) + // An authoritative denial closes immediately: the typed envelope names the + // reason, and the error still follows so existing clients keep working. + let close = tokio::time::timeout(Duration::from_secs(1), stream.next()) .await .expect("reauthorization runs") - .expect("terminal error item"); + .expect("terminal close item") + .expect("close envelope is not an error"); + assert_eq!( + close.closed, + Some(AiSessionStreamClose::AuthorizationRevoked) + ); + assert!(close.event.is_none()); + assert!(!close.reset_required); + let revoked = stream + .next() + .await + .expect("terminal error item after the close envelope"); assert!(matches!(revoked, Err(AiError::ReauthorizationFailed))); assert!(stream.next().await.is_none()); } @@ -308,3 +335,202 @@ async fn maximum_sized_replay_pages_are_drained_before_live_delivery() { assert_eq!(live.sequence, 102); assert_eq!(live.event_type, "session_title_changed"); } + +/// Resolver that can be made temporarily unavailable, distinguishing a +/// dependency restart from an authoritative denial. +struct BlipResolver { + principal: AuthPrincipal, + unavailable: Arc, + denied: Arc, + attempts: Arc, +} + +#[async_trait] +impl CurrentPrincipalResolver for BlipResolver { + async fn resolve( + &self, + reference: &PrincipalReference, + ) -> agql_auth::AuthResult { + self.attempts.fetch_add(1, Ordering::SeqCst); + if self.denied.load(Ordering::SeqCst) { + return Err(agql_auth::AuthError::Forbidden); + } + if self.unavailable.load(Ordering::SeqCst) { + return Err(agql_auth::AuthError::AuthServiceUnavailable); + } + ResolvedPrincipal::new( + reference.clone(), + self.principal.clone(), + OffsetDateTime::now_utc(), + ) + } +} + +/// Work item 2: a brief authorization-service restart must not drop the +/// stream, but an authoritative denial must still close it immediately. +#[tokio::test] +async fn reauthorization_blip_is_survived_and_denial_still_fails_fast() { + let database = Database::::connect_sqlite("sqlite::memory:") + .await + .expect("in-memory SQLite opens"); + apply_schema(&database, "ai-subscription-blip-v1").await; + let sessions = Arc::new(OrmAiSessionService::new( + database, + Arc::new(AllowAll), + Arc::new(ProtectionPolicy), + Arc::new(DatabaseManagedContentProtector), + )); + let principal = principal(); + let unavailable = Arc::new(AtomicBool::new(false)); + let denied = Arc::new(AtomicBool::new(false)); + let attempts = Arc::new(AtomicUsize::new(0)); + let subscriptions = OrmAiSubscriptionService::new( + sessions.clone(), + Arc::new(BlipResolver { + principal: principal.clone(), + unavailable: unavailable.clone(), + denied: denied.clone(), + attempts: attempts.clone(), + }), + ) + // Fast enough to observe several reauthorization attempts quickly, but not + // so fast that the durable head check starves a concurrent writer of the + // single in-memory SQLite write lock when the whole suite runs in parallel. + .with_reauthorization_interval(Duration::from_millis(50)) + .with_reauthorization_grace(Duration::from_secs(30)) + .with_replay_check_interval(Duration::from_millis(500)) + .with_replay_page_size(50); + + let session = create_session(&sessions, &principal).await; + send(&sessions, &principal, session.id, "first").await; + let mut stream = subscriptions + .session_events(principal.clone(), AiSessionId(session.id), 0) + .await + .expect("subscription opens"); + let first = stream + .next() + .await + .expect("first item") + .expect("first event"); + assert_eq!(first.event.expect("event").sequence, 1); + + // The dependency goes away for long enough to cover several ticks. Drive + // the stream while waiting so its select loop keeps running, and require + // that reauthorization was actually retried rather than abandoned. + unavailable.store(true, Ordering::SeqCst); + let before = attempts.load(Ordering::SeqCst); + let outage = tokio::time::timeout(Duration::from_secs(5), async { + loop { + if attempts.load(Ordering::SeqCst) >= before + 3 { + return Ok::<(), AiError>(()); + } + tokio::select! { + item = stream.next() => { + let item = item.expect("stream must stay open during the outage")?; + assert!( + item.closed.is_none(), + "an unavailable dependency must not close the stream inside the grace window" + ); + } + () = tokio::time::sleep(Duration::from_millis(20)) => {} + } + } + }) + .await + .expect("reauthorization must be retried during the outage"); + outage.expect("the stream must not error inside the grace window"); + + // Still inside the grace window, durable delivery continues. + send(&sessions, &principal, session.id, "during outage").await; + let during = tokio::time::timeout(Duration::from_secs(5), stream.next()) + .await + .expect("delivery continues during the outage") + .expect("item") + .expect("event delivered while reauthorization is unavailable"); + assert!(during.closed.is_none()); + assert!(during.event.is_some()); + + // Recovery, then an authoritative denial closes immediately. + unavailable.store(false, Ordering::SeqCst); + denied.store(true, Ordering::SeqCst); + let close = tokio::time::timeout(Duration::from_secs(5), async { + loop { + let item = stream.next().await.expect("stream item")?; + if item.closed.is_some() { + return Ok::<_, AiError>(item); + } + } + }) + .await + .expect("denial closes the stream") + .expect("close envelope is not an error"); + assert_eq!( + close.closed, + Some(AiSessionStreamClose::AuthorizationRevoked) + ); +} + +/// Work item 2: single-replica delivery must not depend solely on the +/// in-process wakeup channel. +/// +/// Two `Database` handles over one SQLite file have independent in-process +/// wakeup channels, so a commit through the writer handle never produces a +/// wakeup on the subscriber handle. That is exactly the shape of a dropped or +/// missed hint, and the bounded durable head check is the only path that can +/// deliver it. +#[tokio::test] +async fn durable_replay_fallback_delivers_without_an_in_process_wakeup() { + let directory = tempfile::tempdir().expect("temporary directory"); + let path = directory.path().join("subscription-fallback.sqlite"); + let url = format!("sqlite://{}?mode=rwc", path.display()); + + let subscriber_database = Database::::connect_sqlite(&url) + .await + .expect("subscriber handle opens"); + apply_schema(&subscriber_database, "ai-subscription-fallback-v1").await; + let writer_database = Database::::connect_sqlite(&url) + .await + .expect("writer handle opens"); + + let subscriber_sessions = Arc::new(OrmAiSessionService::new( + subscriber_database, + Arc::new(AllowAll), + Arc::new(ProtectionPolicy), + Arc::new(DatabaseManagedContentProtector), + )); + let writer_sessions = OrmAiSessionService::new( + writer_database, + Arc::new(AllowAll), + Arc::new(ProtectionPolicy), + Arc::new(DatabaseManagedContentProtector), + ); + + let principal = principal(); + let subscriptions = OrmAiSubscriptionService::new( + subscriber_sessions.clone(), + Arc::new(ToggleResolver { + principal: principal.clone(), + active: Arc::new(AtomicBool::new(true)), + }), + ) + .with_reauthorization_interval(Duration::from_secs(600)) + .with_replay_check_interval(Duration::from_millis(20)) + .with_replay_page_size(50); + + let session = create_session(&subscriber_sessions, &principal).await; + let mut stream = subscriptions + .session_events(principal.clone(), AiSessionId(session.id), 0) + .await + .expect("subscription opens"); + + // Committed through the other handle: no wakeup reaches this subscriber. + send(&writer_sessions, &principal, session.id, "no wakeup").await; + + let delivered = tokio::time::timeout(Duration::from_secs(5), stream.next()) + .await + .expect("the durable fallback must deliver without an in-process wakeup") + .expect("item") + .expect("event"); + assert!(delivered.closed.is_none()); + assert_eq!(delivered.event.expect("event").sequence, 1); +} diff --git a/crates/graphql-orm-ai/tests/provider_sessions.rs b/crates/graphql-orm-ai/tests/provider_sessions.rs index bb72a9f0..5eecd51d 100644 --- a/crates/graphql-orm-ai/tests/provider_sessions.rs +++ b/crates/graphql-orm-ai/tests/provider_sessions.rs @@ -758,3 +758,94 @@ async fn exact_absence_authorizes_one_fenced_rebind_with_a_fresh_cursor() { "deleted-thread" ); } + +/// Work item 4.2: every retained-thread invalidation must be disclosed on the +/// ordinary sequenced session stream, so a host can tell the user the model's +/// context was reset even though the durable transcript reads as continuous. +#[tokio::test] +async fn provider_session_invalidation_is_disclosed_on_the_session_stream() { + let fixture = provider_session_fixture().await; + let run = active_run(&fixture, &fixture.owner, "workspace-disclosure").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 claim = fixture + .provider_sessions + .bind_for_run( + &run, + AiProviderSessionBindRequest::new( + descriptor, + AiProviderSessionCursor::new("codex.thread", "retained-thread") + .expect("cursor should validate"), + "c".repeat(64), + None, + ) + .expect("bind request should validate"), + ) + .await + .expect("provider session should bind"); + + let before = fixture + .sessions + .session_event_page(&fixture.owner, run.session_id(), 0, 500) + .await + .expect("events should replay"); + assert!( + !before + .events + .iter() + .any(|event| event.event_type == "provider_session_reset"), + "binding alone must not disclose a reset" + ); + + // Pressing Stop after a completed turn is ordinary user behaviour, and it + // currently costs the retained thread. That must be visible. + fixture + .provider_sessions + .require_cleanup(&claim, "provider_session_cancelled_after_turn") + .await + .expect("invalidation should record cleanup"); + + let after = fixture + .sessions + .session_event_page(&fixture.owner, run.session_id(), 0, 500) + .await + .expect("events should replay"); + let disclosure = after + .events + .iter() + .find(|event| event.event_type == "provider_session_reset") + .expect("invalidation must be disclosed"); + assert_eq!( + disclosure.payload.0["reason"], + serde_json::json!("provider_session_cancelled_after_turn") + ); + assert_eq!( + disclosure.payload.0.as_object().map(serde_json::Map::len), + Some(2), + "the disclosure carries only its tagged format and reason class" + ); + assert!(disclosure.sequence > 0); + assert_eq!(after.watermark, disclosure.sequence); + + // It participates in ordinary replay: a client resuming from the previous + // watermark receives exactly this event. + let resumed = fixture + .sessions + .session_event_page(&fixture.owner, run.session_id(), before.watermark, 500) + .await + .expect("replay from the earlier watermark should succeed"); + assert!(!resumed.reset_required); + assert!( + resumed + .events + .iter() + .any(|event| event.event_type == "provider_session_reset") + ); +} diff --git a/crates/graphql-orm-ai/tests/schema_module.rs b/crates/graphql-orm-ai/tests/schema_module.rs index fa1c414e..6fd17444 100644 --- a/crates/graphql-orm-ai/tests/schema_module.rs +++ b/crates/graphql-orm-ai/tests/schema_module.rs @@ -8,8 +8,8 @@ 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.60.0"); - assert_eq!(catalog.entities().len(), 46); + assert_eq!(AI_SCHEMA_MODULE_VERSION, "0.61.0"); + assert_eq!(catalog.entities().len(), 47); assert!( catalog .entities() diff --git a/crates/graphql-orm/tests/fixtures/backend-coexistence/Cargo.lock b/crates/graphql-orm/tests/fixtures/backend-coexistence/Cargo.lock index d0645eef..f400a5e7 100644 --- a/crates/graphql-orm/tests/fixtures/backend-coexistence/Cargo.lock +++ b/crates/graphql-orm/tests/fixtures/backend-coexistence/Cargo.lock @@ -1329,7 +1329,7 @@ dependencies = [ [[package]] name = "graphql-orm-ai" -version = "0.81.0" +version = "0.82.0" dependencies = [ "agql-auth", "async-graphql", diff --git a/docs/reference/workspace-packages.md b/docs/reference/workspace-packages.md index a15c42ff..2c7626ce 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.81.0` | `crates/graphql-orm-ai` | `sqlite` | `graphql-orm`, `graphql-orm-ai-tool-profiles`, `graphql-orm-storage` | +| `graphql-orm-ai` | `0.82.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 |