diff --git a/CHANGELOG.md b/CHANGELOG.md index 6ecbfe8..0387ffb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,13 @@ named `D-*` row in the ledger at `PLAN.md` section 18. boundary, even when both API families share the same pending request. V2 permission creation also rejects fields and values outside the frozen request schema before storing them. +- V2 interrupts now cancel registered provider drains and report whether a + turn was interrupted. Drain capacity and reservation waiters are bounded; + session deletion fences new work and waits for admitted turns before removing + session rows. The delete-side wait is bounded by + `OCSERVE_DRAIN_CANCEL_SECS` (default 30 s) so a drain stuck past the provider + watchdog cannot hold a deletion fence indefinitely. Admission checks run on + the blocking pool, not on a tokio worker. ## 0.7.1 - 2026-11-10 diff --git a/MEMORY.md b/MEMORY.md index 74dbad9..3195c47 100644 --- a/MEMORY.md +++ b/MEMORY.md @@ -107,6 +107,23 @@ tokio::runtime::Builder::new_multi_thread() borrowed `&str`/`Cow` fields on hot read paths (session list, event encode). - Strings: `Arc` for shared session/project ids; `internment` only for genuinely repeated identifiers (tool names, agent names) with a hard intern-table cap. +- V2 drain task registry: at most 128 active session entries, enforced by the + slot semaphore created with `MAX_V2_DRAIN_TASKS`; overflow admission fails + before persisting inbox work (`V2-PLAN.md` P3, `D-V2-DRAIN-CAP`). This is a + COUNT bound, not a memory bound: each active drain can hold up to + `PROMPT_HISTORY_MAX_BYTES`, and the shared prompt-lock map caps concurrent + turns at `MAX_PROMPT_LOCKS` (64), so the worst case is bounded by + min(128, 64) x history, which is why the budget is stated in both places. +- V2 slot reservations can queue at most 128 additional waiters, enforced by + `MAX_V2_DRAIN_WAITERS`; a full waiter cap returns bounded backpressure rather + than retaining another request or background task. +- V2 session deletion fences are capped at 128 concurrent sessions by + `MAX_V2_SESSION_DELETES`; a full cap returns bounded backpressure. +- V2 drains waiting on the shared prompt-lock map are bounded by the same 128 + drain slots. The map itself is capped at `MAX_PROMPT_LOCKS` (64); its + `Notify` keeps no per-waiter message queue. Waiters enable their notification + before checking capacity, so each release wakes one registered waiter instead + of coalescing releases into a single stored permit. - Collections: `SmallVec` for ≤4-element hot vectors; bounded `LruCache` (never `HashMap` that only grows); caches hold `Weak` where ownership allows (no `Arc` cycles). - `clippy::unwrap_used` denied in `ocserve-http`/`ocserve-core` (panic=abort ⇒ a panic is an diff --git a/PLAN.md b/PLAN.md index e0dc474..945c9f4 100644 --- a/PLAN.md +++ b/PLAN.md @@ -603,6 +603,7 @@ current anywhere. | `D-V2-PERM-EVAL` | `permission.evaluate` is not bridged as a v2 hook; the server's own gate is authoritative. | accepted | | `D-V2-RPC-EVENTS` | `rpc.events.emit` is not bridged: there is no v2 RPC event SSE surface. | accepted | | `D-V2-SESSION-LIFECYCLE` | Transport-level v2 hooks (`model.request`, `http.*`, `retry`, `title`, `generate`) are not bridged. | accepted | +| `D-V2-DRAIN-CAP` | At most 128 v2 sessions may hold an active drain slot. A request that needs a new slot receives `ServiceUnavailableError` before its inbox row is written; background wake sources wait for a slot. This bounds the per-session drain task registry and preserves durable admitted work. | accepted | | `D-V2-STORAGE-MEM` | `ctx.storage` is in-memory for the life of the sidecar; it is not disk-backed. | accepted | | `D-V2-SESSION-GENERATE` | The transient generate path runs, but the v2 `session.generate` hook is not dispatched. | accepted | | `D-V2-TOOL-EDITOR` | `tool.transform` registers and records, but the tool registry is static, so it is not applied; `tool.list` returns `[]`. | closed (the callback runs against a live editor; added tools are advertised with their JSON Schema and dispatched; a disposed registration removes its names) | @@ -617,6 +618,7 @@ current anywhere. | `D-TOOLS-TASK` | The `task` subagent tool was advertised by the frozen provider but not implemented. | closed (the tool runs on the shared executor with a depth guard) | | `D-PERM-SOURCE-KEY` | The permission `source` object uses different keys per contract line: the v1 (1.19.5) SDK type `PermissionV2Source` is `{type,messageID,callID}` while v2 (2.2.5) `Permission.Source` is `{type,messageID,id}`. ocserve emits `id` on the v2 wire and `callID` on the v1 wire; the gate stores the call id under `callID` and each projector renames it. A body cannot satisfy both schemas because both are `additionalProperties:false`. | open | | `D-PERM-EVENT-NAME` | Upstream 1.19.5 names the v2 permission event `permission.v2.asked`; 2.2.5 names it `permission.asked`. ocserve keeps the v1 name `permission.asked` on every stream and projects the v2 *payload* on `/api/event`, so a 1.19.5-era consumer that dispatches on `permission.v2.asked` will not see it. No shipped 1.19.5 client dispatches on that event: the TUI refreshes permissions over REST, and the web app's adapter maps `permission.v2.asked` only when the server sends it. The 1.19.5/2.2.5 web app also probes `/global/health` first and treats any `{healthy:true}` answer as a v1 server (`server-protocol.ts`), so against ocserve it subscribes to `/global/event` and consumes the v1 payload — the v2 permission frame is reachable only from a client that calls `/api/event` directly. | open | +| `D-V2-INTERRUPT-STATUS` | `/api/session/{id}/interrupt` is in BOTH frozen contracts: v1 declares 204, v2 declares a 200 `SessionInterruptResponse`. One handler cannot return both, and the route is not in the v2-only table, so a v2-disabled server now answers 200 where its own v1 contract says 204. No shipped v1 client calls it (the 1.19.5 v1 SDK uses `/session/{id}/abort`), so the impact is contract-visible only. | open | Historical rows kept for the tokens' own history, all closed: `D-GENERATE-EPHEMERAL`, `D-REVERT-NOSNAP`, `D-VCS-NOAPPLY`, `D-HOOK-*` items are in §16.9 and diff --git a/TESTING.md b/TESTING.md index 10f8eff..d104c00 100644 --- a/TESTING.md +++ b/TESTING.md @@ -98,6 +98,7 @@ traceable test, every test has a traceable requirement. | event retention pruned with unindexed full scans inside the writer transaction (2026-10-11: age DELETE = EQP `SCAN event` on a 166k-row/317 MB table every 1024 appends; the ROW_NUMBER cap window built a temp b-tree over ALL rows even when no session was near the cap) | source: schema v12 `idx_event_time` (age delete = seek); cap window gated on a covering-idx existence check (`HAVING count(*) > cap LIMIT 1`) — early-exits when over, index-only scan otherwise | migration test (v1 fixture without event table stays composable), `prune_events` still returns aged+capped counts; live boot ran the v11→v12 migration (index build) | | boot parity checks counted two full tables on EVERY start (2026-10-11: msg_part + part_search `count(*)` = 4-8 s cold on the live DB; backfill_compaction's two scans 0.4-10 s unconditional) | source: manifest stamps (`import_manifest.parts_indexed` validated against fts docsize + trigger presence; `compaction_backfilled_v1`) — healthy boots become single key lookups, missing/mismatched stamps fall through to the full check; stamps written by import finalize, backfill, self-heal; backfill_compaction batches its anchor writes (was one txn per anchor) | store backfill tests (repair paths unchanged, crash-window test still green), importer post-import stamp write; healthy-boot path returns in one query | | per-request work on hot async paths (2026-10-11 dive): middleware formatted a label + took the global metrics mutex on EVERY request; embedded UI `to_vec()` copied the whole bundle per hit + re-hashed the CSP; /api/agent + /api/command re-projected per request; `load_messages` N+1 (one part query per message) and ran inline on a worker; /file/content read inline; sync_tick ran inline in an async task; v2 set_session_field/session_agent did blocking store IO inline | source: metrics `Dur` uses atomics after first sight (no lock/no alloc on the hot path); `Bytes::from_static` + OnceLock CSP; wire entries for api_agent/api_command; 64-chunk part prefetch in load_messages (F3 pattern); spawn_blocking at history/compaction/file/sync/session-field call sites | metrics tests green; `load_messages_chunk_boundary_and_blob_parts` (65 msgs, parts at 1/64/65 + blob; negative control: pad removed → red); `v2_body_parity` pinned wire entries; webui_perf 6/6 | +| **v1 admission ran a synchronous store query on every prompt** (2026-11-13 review of PR #41): `check_v1_session_admission` called `ocserve_store::session_exists` (an `open_reader` + `SELECT count(*)`) directly in the async handler, on `/message`, `/prompt_async`, `/command`, `/summarize` and the v2 create — and it had REPLACED an existing `run_blocking` call in `prompt_async`, so the change was a net loss of offload. Two more members of the same class in the same PR: the delete path's drain-cancel loop awaited a `watch` with no deadline, and a deferred goal review was dropped with no re-arm. | source: `check_v1_session_admission` is `async` and routes the query through `run_blocking`; `cancel_drain` and `cancel_drains_for_delete` share one `drain_cancel_timeout_secs()` deadline (inner AND outer loop) and record `ocserve_v2_drain_cancel_timeout_total`; capacity is detected by status/name, not the error message string. | `delete_returns_when_a_registered_drain_never_confirms_cancellation` — a fabricated registry entry whose completion channel never fires must still let DELETE return; negative control removed the outer deadline and the request hung until the 90 s test timeout. `v2_admission` 23/23. | | unbounded provider/client-controlled collections (2026-10-11 RAM dive): ToolCallAssembler slot index (provider-supplied `index` → GB-class slot vec), v2 persistent permission asks (no drop-guard path), page-memo env budget unbounded above + single-entry-over-budget retention, gate `always`/`hydrated` never shrinking on session delete, dead `AppState.sessions` map | source: `MAX_TOOL_CALL_SLOTS`=256 early-return; `register_persistent` cap 512 oldest-by-time eviction + metric; page memo budget clamped to the 64 MiB const and oversize-alone refused; `PermissionGate::forget_session` wired into DELETE /session; dead field deleted | `tool_call_assembler_caps_hostile_index` (negative control: cap removed → 1,000,001 slots red); `persistent_registrations_are_capped` (eviction neutered → 700 red); `forget_session_releases_state` (neuter → red); page-memo budget test | | v2 metadata routes re-read the 5.4 MB catalog + deep-cloned the provider tree per request (2026-10-11 perf dive: `/api/model` 96 ms p50 / 104 ms CPU single-thread; conc-6 mean 8.1 s; +374 MB RSS over 15 calls — every web-UI model picker + TUI boot hit this; invisible because the k6 mix had zero v2 routes) | source: the parsed catalog now travels in `Payloads.catalog` (swap-on-reload, same window as F5; `models.json` is already a watch path) and the three v2 handlers project from it by reference (no `provider.all` clone); the response bodies are pure builders (`v2::{models_body,providers_body,integrations_body}`) so `rebuild_wire` precomputes identity+gzip+br+ETag per reload exactly like v1 (`wire_json_h`); `load` re-derives nothing per request. | `f5_wire::v2_metadata_wire_matches_builders` (wire bytes == pure builder, with/without catalog — planted control: handler drift would diverge); `v2_surface.rs` (routes still JSON); live probes: p50 96→<1 ms, conc-6 8.1 s→<20 ms, RSS flat; mix.js gains the three routes behind `LOAD_INCLUDE_V2` (thresholds stay v1-derived) | | session/message token accounting (2026-10-11: the web UI showed a ~800% context meter — the meter sums the LAST assistant message's components over `model.limit.context`, session-context-metrics.ts:28-59; ocserve stored the whole-TURN step sum on the final message AND overwrote the session row with it, and the auto-compaction overflow check compared that same sum, firing compaction ~N× early) | source: `message_tokens(&Usage)` is the single per-step projection (input excludes cache; output excludes reasoning; total = provider raw — Session.getUsage, session.ts:368-377); the final message stores the LAST step's usage (processor.ts:459 REPLACE) with that step's cost (probed live against freeze: every assistant message carries per-step cost); every step-finish ACCUMULATES into the session row via `ocserve_store::accumulate_session_usage` (upstream applyUsage, core/session/projector.ts:89-108,326); `is_overflow` counts the STEP (`total` when present else input+output+cache.read+cache.write — upstream overflow.ts omits reasoning); `finalize_session_prompt` no longer writes cost/tokens (agent/model/time only); mid-turn and final `session.updated` carry the in-memory lifetime aggregate seeded once from the preflight read. | `tests/token_accounting.rs`: a REAL two-step stub turn (todowrite step + final) asserts per-step message tokens (600/10/400 then 1200/15/5 split reasoning/800), session accumulation (1800/25/1200/5), and the shared checker's planted pre-fix control (turn-sum shape) must flag → red→green; store tests `finalize_writes_agent_and_model_columns` + `finalize_does_not_overwrite_accumulated_usage`; llm `parse_usage` reasoning split. Report-only history audit (live DB, read-only): 76 sessions, 1 with an inflated last message (>1M), 20 row-vs-Σ(messages) mismatches from historical imports/compaction — values are display-only now and repair forward. Named divergences: **D-TOKENS-PARTDELETE** (DELETE/PATCH part does not subtract session usage — upstream applyUsage(-1) on PartRemoved), **D-TOKENS-FORK** (fork copies the messages but the forked session row starts at 0 — upstream nets the cloned step usage). | @@ -231,6 +232,8 @@ event types), the two antagonism ledgers (every finding = at least one test). | a built-in command is advertised but not invocable (2026-11-05, reported by a real client: `/goal ` failed with "Command failed"). The built-in entry is advertised with `template: ""` because the v1 `Command` schema requires the field, and the dispatcher tested `template.is_none()` — true only in the test fixture's EMPTY command list, never in production, so the call fell through to `expand_command("")`, sent the model an EMPTY prompt and 500'd; the v2 path had a sibling bug (`command disappeared`) | the predicate is "missing OR empty" and the v2 dispatch matches the built-in by name before its template lookup; guard rule 69 fails on `template.is_none()` as a dispatch test | `tests/goal_e2e.rs` fixtures are built with the PRODUCTION payload shape (`template: ""` advertised) and the regression test asserts the response is `SessionV1.WithParts`; planted negative control: restore `is_none()` → 5 red; live proof on the release binary + stub provider: HTTP 200 in 24 ms, `info.role=assistant`, 3 parts, goal armed | | model-driven reviewer text treated as instructions (2026-11-05 security review H1: the continuation template interpolated `${feedback}` RAW — only the objective was escaped — so a hostile repository could steer the judge into emitting `{"verdict":"continue","reason":""}`, which was persisted as a user-role message and replayed with tools enabled, surviving compaction) | `continuation()` escapes and wraps the verdict in ``; reviewer prose is capped (`MAX_FEEDBACK`); `judge_prompt` escapes shell commands too | `reviewer_feedback_is_escaped_and_wrapped`, `reviewer_feedback_is_bounded`, `judge_escapes_shell_commands` (a `` in the reason must not close the block) | | an autonomous loop that keeps spending after the user stopped it (2026-11-05 reviews M1/M2/M3: `/abort`, v2 `interrupt` and v2 session delete aborted only `prompt_tasks`; the goal driver is a bare task, so the turn finished, the reviewer said continue, and another turn ran; `/abort` also left the row ACTIVE so the next idle hook restarted the loop) | v1 abort, v2 interrupt and v2 delete all `cancel_driver` + pause the goal with "Stopped by the user" (upstream pauses on a user interrupt); drivers re-check status AND objective before every turn; a `Drop` guard removes the registry entry on every exit path; a failed context/lock start PAUSES instead of leaving ACTIVE-with-no-driver | `abort_pauses_the_goal_and_stops_the_loop` (deterministic: the provider holds the kickoff open on a gate so the abort lands mid-turn; asserts status paused, rounds frozen, zero further reviewer calls) with a proven negative control (remove the pause → red) | +| v2 interrupts could not stop the active provider drain, returned the wrong response, and cancellation could strand durable inbox work or reorder terminal events and later turns | Registered drain tasks now have cooperative cancellation and shared completion; bounded active slots and reservation waiters reject overload before admission; drains retain claims and locks through terminal publication, clear coordinator state, and resume only the permitted work after cancellation | `v2_admission.rs::v2_interrupt_cancels_registered_drain_and_reports_status`, `v2_interrupt_waiting_for_session_lock_reports_not_interrupted`, `v2_drain_capacity_rejects_before_inbox_admission`, `v2_interrupt_resume_rejects_when_drain_waiter_cap_is_full`, `v2_interrupt_resume_runs_steer_and_leaves_queued_prompt_parked`, `v2_drain_reports_provider_failure_as_terminal_failure`; `v2_drain_outcome_tests::cancelled_pending_input_does_not_strand_a_later_prompt`, `cancel_finalization_serializes_a_new_prompt_after_interrupted_event`, `late_cancel_does_not_reclassify_a_terminal_success`, `duplicate_wake_queues_rerun_without_cancelling_registered_drain`, `cancel_serializes_coordinator_reset_with_drain_registry`, `v2_drain_waits_for_prompt_lock_capacity`, `prompt_lock_capacity_notify_wakes_multiple_waiters`; `drain_lock::drain_waits_while_a_prompt_holds_the_lock`; `coordinator.rs::explicit_steer_continuation_does_not_widen_to_input`. Negative controls remove task registration/cancellation, terminal retirement, bounded admission, prompt-lock waiting, or coordinator cleanup | +| deleting a session could race drain retirement or admit new v1/v2 work after the final drain check, leaving provider work writing into deleted state | Deletion holds a bounded per-session fence through drain cancellation and row removal; prompt and inbox admissions recheck under the shared lock, and deletes wait for already-admitted turns before removing rows | `v2_drain_outcome_tests::cancelled_drain_is_shared_by_interrupt_and_delete`, `delete_fences_prompt_admitted_after_drain_retirement_check`, `delete_fences_v1_prompt_async_after_drain_retirement_check`, `delete_fences_v1_sync_message_after_drain_retirement_check`, `delete_fences_v1_command_after_drain_retirement_check`, `delete_fences_v2_session_generate_after_drain_retirement_check`. Negative controls remove the admission fence, shared cancellation completion, or session-lock wait | | bundled v2 permission UI crashed on `request.resources.length` because pending asks created by v1 tool calls and v2 permission routes crossed the shared gate without boundary projection; malformed v2 fields could also create requests outside the frozen schema | v1 and v2 permission projectors map gate values into each frozen request shape, filter string arrays, and only project a tool source; v2 create validates required string arrays plus optional metadata/source and rejects unknown keys (including `message`) before persistence; the shared always-grant/cascade logic accepts `always/patterns` and `save/resources`; `/api/event` and v2 reads use v2, while `/permission` uses v1 | `v2_surface.rs::v2_permission_requests_project_v1_shape_for_reads`, `permission_create_evaluates_ruleset` (checks both read shapes, then replies always and verifies grant/cascade), `permission_create_rejects_malformed_request_fields_before_persisting`, `event_envelope_unit::v2_permission_event_projects_the_permission_request_shape`, `event::tests::permission_request_projects_v2_gate_values_into_the_v1_schema`, `event::tests::v2_permission_projection_drops_malformed_fields_and_source_types`; negative controls bypass projectors, pass malformed entries/source types, accept an unknown create key, or read only v1 gate fields | | goal state writes swallowed (2026-11-05 SRE review: `set_goal` results discarded with `let _ =` in the boot scan, the round increment and `settle`; a persistently failing round write silently disables the 50-round cap, and a failed pause leaves ACTIVE-with-no-driver) | every goal write propagates or logs+counts (`ocserve_goal_total{event="persist_failed"}`); boot-scan failure skips the row loudly | `goal_state` suite; the store's single-writer discipline unchanged (all writes go through `Writer`) | | an autonomous loop the operator cannot see or stop (2026-11-05 SRE review: no reviewer budget, no cap on concurrent drivers, reviewer spend invisible, failures not surfaced to clients) | `OCSERVE_GOAL_REVIEW_SECS` (default 300 s) bounds each reviewer call so a chatty provider cannot hold the wake gate; `MAX_CONCURRENT_DRIVERS` (=4) bounds loops process-wide and PAUSES the ones that cannot start; reviewer tokens land in `ocserve_transient_tokens_total{owner="session"}` via `RunOpts::attribution_session`; a failed goal turn emits the same durable `session.error` shape `run_prompt` does | `review_budget` + env contract row (regenerated `docs/ENVIRONMENT.md`); the abort e2e asserts frozen rounds; `ocserve doctor` prints the switch and active goals | diff --git a/TRACEABILITY.md b/TRACEABILITY.md index 938db35..70afb50 100644 --- a/TRACEABILITY.md +++ b/TRACEABILITY.md @@ -260,11 +260,15 @@ no-inflation correct), PASS cost (−58.9% weighted) + score (median 1.0 both) | K-MEDIA | attachments must reach the provider, not just the database | A user turn with an image/PDF was persisted as a `file` part and then SILENTLY flattened to text by `to_provider_messages` (which only read `type=="text"`), so the model never saw the media and nothing errored. Fixed: user file parts encode as OpenAI content parts `{type:file,url,mediaType,filename}` exactly where upstream's AI-SDK conversion does (`message-v2.ts:222-239`), `text/plain` + `application/x-directory` stay dropped as upstream drops them, and a media-free turn remains a plain STRING so text-only provider bytes are byte-identical. `ChatMessage.content` became `Content::{Text,Parts}` (untagged serde) so the choice is explicit at construction. `GET /api/experimental/media/{hash}` serves content-addressed attachments (hash shape validated first, FileNotFoundError on a miss). | `media_encoding_tests` (5: image→parts, text-only stays a string, text/plain+directory excluded, media-only turn sent, assistant ignores file parts); `media_corpus_tests` decodes EVERY file part from a real DB (13 real browser screenshots via blob decompression) and asserts each encodes as media — plus a planted `OCSERVE_MEDIA_PLANT=drop` control proven RED; wired into nightly | green | | K-V2-225 | the v2 contract must track upstream without hand-maintenance, and every new operation must be bound or honestly refused (never shape-without-mutation) | Regenerated the 2.2.5 contract from published npm tarballs through `bump.sh v2`. 169 operations (was 142): 27 new, 0 removed, 0 conflicts. All 27 bound: schedules are a real subsystem (schema v16 `schedule` table: create/list/get/pause/resume/delete with `in`/`at`/`every` due-time computation, `experimental.schedules` gating matching the reference's ForbiddenError); `generate/text` and `generate/decisions` reuse the transient executor; `media/{hash}` serves content-addressed blobs with the reference's hash-shape decode first; `fs/watch` records registrations; `usage/limits`, `voice/models`, `debug/heap-dump`, `plugin/gui`, `permission/saved/{id}` are bound with probe-pinned shapes. Seven honest refusals are named `D-*` rather than stubbed (artifacts, client-mcp, image generation, voice transport, heap dump, fs-watch events, cron firing). Slot renaming landed: `contract-diff --v2`, `pair-v2-check` and the embedded-UI entry test all resolve by SLOT/tag now, so a bump cannot leave them silently testing the old binary (that exact class: `pair-v2` hardcoded a 2.0.28 path). | `tests/v2_225.rs` (18 tests, each citing the live probe it encodes, incl. every-route-answers-JSON); `check-coverage.py` on 2.2.5 → 169/169 implemented, 0 cited, 0 gap; `gen-v2-gate.py` → 123 v2-only ops; `pair-v2-check.sh` vs the REAL 2.2.5 binary → 0 divergences (and it now fails loudly if the arm reports the wrong major); pair-check 18/0 + both arms spec-validate 0 violations; replay 26/0; guards 107/107 | green | | K-FREEZE-1195 | the v1 freeze must be adopted through the pipeline, not by hand, and every artifact that embeds the tag must move with it | Moved 1.18.31 -> 1.19.5 with `bump.sh v1`. Contract: 188 operations over the same 162 paths (byte-diff: 2 schema descriptions, 1 additive `reasoning_options` property, `x-codeSamples` added to every operation). Prompt/tool text for everything the server sends is byte-identical (`tool/shell/shell.txt` unchanged, `bashCommandSection` unchanged); `plan-enter.txt` was DELETED upstream and is removed here. ARITY rules identical. Two live capture timings were also fixed: `acquire-freeze` printed success while `cp` failed on the root-owned vendored path (step 2 then booted the OLD binary), and `extract-prompts` had no remote fallback. | `capdiff` GREEN with the provider-header oracle enabled: it found the two real deltas (composite UA runtime segment `bun/1.3.14 -> 1.4.4`; `x-opencode-session-id` now sent to EVERY provider, not just `opencode*`), both fixed; replay 26/0; pair 18 compared / 0 divergent, 52 schema checks / 0 violations; `render_check` (prompt parity) green; `global_health` golden now `1.19.5`; test `composite_ua_matches_the_captured_artifact` pins the UA to a captured file, so a hand-written literal cannot drift again | green | +| K-V2-DRAIN-CANCEL | a v2 interrupt reaches the active drain, releases its claim and lock, reports whether execution was interrupted, and retires without stranding work or overtaking later turns | Active drains use 128 slots and a separate 128-waiter reservation cap. Cancellation keeps the drain registered through terminal-event publication; callers share completion. Drains wait cancellably for prompt-lock capacity and the session lock, clear coordinator state, and hand later admitted work to a fresh drain only after the interrupted event. Resume processes steer work while queued prompts remain parked. | `v2_admission.rs::v2_interrupt_cancels_registered_drain_and_reports_status`, `v2_interrupt_waiting_for_session_lock_reports_not_interrupted`, `v2_drain_capacity_rejects_before_inbox_admission`, `v2_interrupt_resume_rejects_when_drain_waiter_cap_is_full`, `v2_interrupt_resume_runs_steer_and_leaves_queued_prompt_parked`, `v2_drain_reports_provider_failure_as_terminal_failure`; `v2_drain_outcome_tests::cancelled_pending_input_does_not_strand_a_later_prompt`, `cancel_finalization_serializes_a_new_prompt_after_interrupted_event`, `late_cancel_does_not_reclassify_a_terminal_success`, `duplicate_wake_queues_rerun_without_cancelling_registered_drain`, `cancel_serializes_coordinator_reset_with_drain_registry`, `v2_drain_waits_for_prompt_lock_capacity`, `prompt_lock_capacity_notify_wakes_multiple_waiters`; `drain_lock::drain_waits_while_a_prompt_holds_the_lock`; `coordinator.rs::explicit_steer_continuation_does_not_widen_to_input`. Negative controls remove task registration/cancellation, terminal retirement, bounded admission, lock waiting, or coordinator cleanup | green | +| K-SESSION-DELETE-FENCE | deleting a session prevents new prompt or inbox work from starting and waits for admitted work and drains to retire before removing rows | A bounded per-session fence closes v1/v2 prompt and inbox admission before cancellation; the fence remains active through row removal. Deletes wait for turns already holding the shared prompt lock. | `v2_drain_outcome_tests::cancelled_drain_is_shared_by_interrupt_and_delete`, `delete_fences_prompt_admitted_after_drain_retirement_check`, `delete_fences_v1_prompt_async_after_drain_retirement_check`, `delete_fences_v1_sync_message_after_drain_retirement_check`, `delete_fences_v1_command_after_drain_retirement_check`, `delete_fences_v2_session_generate_after_drain_retirement_check`. Negative controls remove the fence, shared cancellation completion, or prompt-lock wait | green | + | K-MUTANTS | mutation baseline + gate (Phase 2) | `snapshot::tests::non_git_directory_is_inactive` made hermetic (fixture plants a bogus `.git` boundary — the test falsely passed only when TMPDIR sat outside any repo; the mutants harness sets TMPDIR inside the repo, aborting the baseline). **Full-scope baseline (i9, 2330 mutants, 34 min): 939 caught / 1221 missed / 24 timeout / 146 unviable = 43.5 % score** — `bench/mutants/BASELINE.md` + `baseline-outcomes.json`; zero env-gated skips verified (1 intentional ignore). `scripts/check-mutants.py` gates missed count against ceiling 1221 (ratchet-down-only; selftest red/green proven); `scripts/merge-mutants.py` combines `--shard` files with a duplicate guard (selftest red/green proven). Method corrected on evidence: `additional_cargo_test_args=["--lib"]` removed (drops store integration binaries → 933 false survivors, and lands `--lib` on the test binary). Acceleration: `.cargo/mutants.toml` profile=mutants/debug=none + 45s timeout floor; runs on the i9 (`-j8`, mold, sccache, tmpfs) | green | | K-V2-PERM-SHAPE | pending permission requests use each frozen grammar on reads and events, regardless of which API family created them, while shared always-grant behavior accepts both grammars | The permission gate is shared: v1 tool calls store `permission`/`patterns`/`always`/`tool`, while v2 creates store `action`/`resources`/`save`/`source`. V2 create validates string-array members plus metadata/source shapes, and rejects unknown fields such as `message` because the frozen request schema has `additionalProperties: false`. Bidirectional projectors whitelist fields, filter string arrays, and emit `source` only for the frozen `tool` variant. The gate's always-reply and cascade logic reads either vocabulary so v2 asks grant and resolve matching pending requests without leaking fields across wire shapes. | `v2_surface.rs::v2_permission_requests_project_v1_shape_for_reads` covers global/session/detail and isolation; `v2_surface.rs::v2_created_ask_is_readable_in_both_grammars` drives the REAL POST /api/session/{id}/permission create path and asserts the same stored ask reads back with `source.id` on the v2 route and `tool.callID` on the v1 route; `permission_create_evaluates_ruleset` creates v2 asks, checks both list shapes, replies `always`, and verifies the session grant plus matching-request cascade; `permission_create_rejects_malformed_request_fields_before_persisting` covers malformed resources/save entries, optional metadata/source fields, and rejects the non-schema `message` field; `event_envelope_unit::v2_permission_event_projects_the_permission_request_shape` checks v2 event projection and unchanged v1 event; `event::tests::permission_request_projects_v2_gate_values_into_the_v1_schema` pins reverse projection; `event::tests::v2_permission_projection_drops_malformed_fields_and_source_types` covers malformed shared values. Negative controls: reading only v1 `always`/`patterns` makes the v2 grant and cascade assertions red; bypass either projector, allow bad nested payload fields, pass a non-tool source type, or accept an unknown permission-create key breaks the shape assertions. | green | | K-CI-PIN | CI must fail for repository changes, not for a Rust release rolling onto the stable channel | On 2026-11-13 every workflow on `main` and on all fork PRs failed within two minutes with zero tests run. rustc 1.100.0 had landed on the runner and introduced cargo manifest lints (`unused_dependencies`, `unused_workspace_dependencies`, `redundant_homepage`); `CARGO_BUILD_WARNINGS=deny` turned them fatal. Reproduced locally under 1.100.0 with the same env (exit 101) and green under 1.99.0, proving the trigger was the compiler, not the code. Fixed by pinning `rust-toolchain.toml` to 1.100.0 (a floating channel makes `main` hostage to a release) and removing the declarations the lint proved unused (workspace `hyper`, `metrics`, `metrics-exporter-prometheus`, `thiserror`; per-crate `thiserror`, `rustc-hash`, `metrics`) plus the duplicated `homepage`. Also replaced the `current_exe()` binary walk in three process tests with `CARGO_BIN_EXE_ocserve`, which is what CI and remote builds actually need; the walk only worked when the test exe sat beside the bin. | `rust-toolchain.toml` pin; `cargo +1.100.0 clippy --workspace --all-targets -- -D warnings` under `CARGO_BUILD_WARNINGS=deny` clean; 907 tests pass on the build box; guards 110/110, matrix, docs green | green | +| K-OFFLOAD-ADMISSION | admission, cancellation and wake code must not block a tokio worker or a serial loop, and a bounded wait must bound the WHOLE operation | The PR #41 review found the same class three times in one change: a synchronous `session_exists` on every v1 prompt (replacing an existing `run_blocking` call), an unbounded drain-cancel wait on DELETE, and a scheduler that awaited drain capacity inside its single serial tick loop. The first attempt at the delete bound bounded only the inner `watch` wait; the outer re-cancel loop still spun forever, which the regression test caught. | `delete_returns_when_a_registered_drain_never_confirms_cancellation` (fabricated non-confirming registry entry must not hold DELETE; negative control hung to the 90 s timeout); `check_v1_session_admission` async via `run_blocking`; `scheduler::step` uses `reserve_drain_slot` (try) and returns without marking the row so the next pass retries; `ocserve_v2_drain_cancel_timeout_total`; v2_admission 23/23, 929 workspace tests. **Measurement honesty:** the indexed point lookup measures 2.1 us in-process, so this change does NOT produce a measurable p50 improvement and none is claimed. The defect is the class (a worker parked by any blocking call is a convoy for unrelated routes, the recorded F1 signature), and the behavioural proof is the delete test, not a latency delta. | green | ## Test design techniques coverage (TESTING §4) | Technique | Applied | Where | diff --git a/V2-PLAN.md b/V2-PLAN.md index 143fd61..aaac8f7 100644 --- a/V2-PLAN.md +++ b/V2-PLAN.md @@ -101,6 +101,29 @@ requirement land here as edits, never silently in code. a decoded-row loop. Instruction `observe/commit/apply/advanceEpoch/reset` wired; rendering feeds `estimate_request_bytes`; epoch memo designed, built on measurement only. +- Live drains are capped at 128 sessions by a semaphore. A request that needs a + new drain slot gets 503 before inbox admission when the cap is full; already + admitted durable work is resumed by its existing drain. Background wake + sources can wait for active capacity, with their waiters capped at 128 by a + separate semaphore; a full waiter cap returns 503 or defers the background + retry without creating another waiting future. Drains also wait for the + bounded 64-entry shared prompt-lock map and the per-session lock, with + cancellation covering both waits. A canceled drain stays registered through + terminal-event publication; wakes arriving during retirement are handed to a + fresh drain afterward. Every terminal path clears coordinator state when no + doorbell remains, including when a pending Input item was canceled after + scope selection but before promotion. Production wake + registration and + cancellation's coordinator reset share the drain-registry lock, so a new + doorbell cannot be erased in the cancellation window. This named + drain-capacity divergence is `D-V2-DRAIN-CAP`. +- Both session-delete routes acquire one of 128 bounded deletion fences before + canceling drains. The fence excludes new v1/v2 prompt and inbox admission + until the session rows are deleted; wake registration also rejects fenced + sessions. Direct v1 message, async prompt, command, shell, summarize, and + built-in goal kickoff routes, plus v2 prompt and session-generate/compact, + recheck admission after taking the per-session prompt lock. A delete waits + for already-admitted work to finish before removing rows. This closes the + empty-registry-to-delete race without blocking admissions for other sessions. - Named divergences: `D-REVERT-NODELETE` (mark-only kept), compaction anchors kept. - Kill switches: `OCSERVE_V2=1` gates runtime. Tests: E2E ordering, cancel/update, interrupt/resume, kill -9 mid-turn recovery, planted controls. diff --git a/crates/ocserve-core/src/coordinator.rs b/crates/ocserve-core/src/coordinator.rs index 7fb0d97..e46069f 100644 --- a/crates/ocserve-core/src/coordinator.rs +++ b/crates/ocserve-core/src/coordinator.rs @@ -103,13 +103,24 @@ impl CoordinatorSet { /// on work does NOT clear `draining`, so no second drain can spawn /// mid-loop (the bug a naive take-then-recheck would introduce). pub fn next_scope(&self, session_id: &str, work_remains: bool) -> Option { + self.next_scope_for(session_id, work_remains.then_some(WakeScope::Input)) + } + + /// Take the pending scope or continue the caller's specific promotable + /// scope. A steer-only wake can continue through more steers without + /// widening itself into queued input. + pub fn next_scope_for( + &self, + session_id: &str, + continuation: Option, + ) -> Option { let mut inner = self.inner.lock().unwrap_or_else(|e| e.into_inner()); let coord = inner.get_mut(session_id)?; if let Some(scope) = coord.pending.take() { return Some(scope); } - if work_remains { - return Some(WakeScope::Input); + if let Some(scope) = continuation { + return Some(scope); } coord.draining = false; if !claim_would_hold(session_id) { @@ -127,6 +138,17 @@ impl CoordinatorSet { } } + /// Observe a pending doorbell while a registered drain's task registry + /// lock is held. Production callers use this to decide whether a failed + /// terminal turn must rerun before releasing its per-session task entry. + pub fn pending_scope(&self, session_id: &str) -> Option { + self.inner + .lock() + .unwrap_or_else(|e| e.into_inner()) + .get(session_id) + .and_then(|coord| coord.pending) + } + #[cfg(test)] fn len(&self) -> usize { self.inner.lock().unwrap_or_else(|e| e.into_inner()).len() @@ -232,4 +254,21 @@ mod tests { // CONTROL: clear draining on work-continue → the mid-test wake // returns true (second drain) → red. } + + #[test] + fn explicit_steer_continuation_does_not_widen_to_input() { + let set = CoordinatorSet::new(); + assert!(set.wake("e", WakeScope::Steer)); + assert_eq!( + set.next_scope_for("e", Some(WakeScope::Steer)), + Some(WakeScope::Steer) + ); + assert_eq!( + set.next_scope_for("e", Some(WakeScope::Steer)), + Some(WakeScope::Steer), + "additional steer work keeps the same scope" + ); + assert_eq!(set.next_scope_for("e", None), None); + // CONTROL: a boolean work-remains continuation widens this to Input. + } } diff --git a/crates/ocserve-core/src/env_contract.rs b/crates/ocserve-core/src/env_contract.rs index 0f80feb..551c4c4 100644 --- a/crates/ocserve-core/src/env_contract.rs +++ b/crates/ocserve-core/src/env_contract.rs @@ -191,6 +191,15 @@ pub const CONTRACT: &[Var] = &[ "0 | 1", "1 reads the whole session every turn instead of the proven suffix. Debug only; the sent messages are identical.", ), + op( + "Cgroups and lifecycle", + "OCSERVE_DRAIN_CANCEL_SECS", + "30", + "seconds (0 = unbounded)", + "How long DELETE waits for a cancelled v2 drain to confirm before \ + proceeding with teardown. 0 restores the old unbounded wait, which \ + can hold a deletion fence for the provider stall budget.", + ), op( "Prompts and providers", "OCSERVE_PROVIDER_STALL_SECS", diff --git a/crates/ocserve-http/src/goal_loop.rs b/crates/ocserve-http/src/goal_loop.rs index f6b23a9..cba3a27 100644 --- a/crates/ocserve-http/src/goal_loop.rs +++ b/crates/ocserve-http/src/goal_loop.rs @@ -472,6 +472,13 @@ pub(crate) async fn run_turn_now( } }; let guard = release.arc().lock().await; + { + let _admission = st.v2_session_admission_gate.read().await; + if let Err(e) = crate::check_v1_session_admission(st, sid).await { + drop(guard); + return Err(anyhow::anyhow!("{}", e.message)); + } + } // With attachments the kickoff rides as the user's own message (upstream's // `attached` branch); a bare kickoff is a synthetic notice. let attached = !extra_parts.is_empty(); diff --git a/crates/ocserve-http/src/lib.rs b/crates/ocserve-http/src/lib.rs index f3a2c78..24f05c1 100644 --- a/crates/ocserve-http/src/lib.rs +++ b/crates/ocserve-http/src/lib.rs @@ -106,6 +106,18 @@ where .map_err(|e| anyhow::anyhow!("blocking task join: {e}"))? } +/// How long DELETE waits for a cancelled v2 drain to confirm. Default 30 s: +/// short enough that an API client does not hang behind a provider stall, +/// long enough that a drain finishing its teardown event normally confirms. +/// `OCSERVE_DRAIN_CANCEL_SECS=0` restores the old unbounded wait (documented +/// kill switch, not a recommendation). +fn drain_cancel_timeout_secs() -> u64 { + std::env::var("OCSERVE_DRAIN_CANCEL_SECS") + .ok() + .and_then(|v| v.parse::().ok()) + .unwrap_or(30) +} + mod artifact_store; pub mod compress; pub mod doc; @@ -270,6 +282,30 @@ impl IntoResponse for HttpError { /// session → (generation, background prompt task) for POST /abort. pub type PromptTask = (u64, tokio::task::JoinHandle<()>); +/// Tracked v2 drain task. Cooperative cancellation lets it release its claim +/// and publish one terminal execution event before the caller returns. +pub struct V2DrainTask { + pub generation: u64, + pub cancel: tokio::sync::watch::Sender, + /// Retained with the registry entry for task lifecycle and diagnostics. + pub handle: tokio::task::JoinHandle, + /// Shared terminal result so cancellation does not depend on one caller + /// retaining and awaiting the JoinHandle. + pub completion: tokio::sync::watch::Receiver>, + pub slot: std::sync::Arc, + pub rerun: bool, +} + +/// Bound the number of live v2 drain sessions. The semaphore is the admission +/// cap for the registry; durable inbox rows remain the source of queued work. +pub const MAX_V2_DRAIN_TASKS: usize = 128; +/// Maximum background/request reservations allowed to wait for a drain slot. +pub const MAX_V2_DRAIN_WAITERS: usize = 128; +/// Maximum concurrent session-delete admission fences. +pub const MAX_V2_SESSION_DELETES: usize = 128; +/// Maximum prompt-session lock entries, including active holders and waiters. +pub(crate) const MAX_PROMPT_LOCKS: usize = 64; + /// App state shared by handlers. pub struct AppState { /// Config-derived route payloads (W4: swappable after PATCH /config — @@ -345,6 +381,8 @@ pub struct AppState { std::collections::HashMap>>, >, >, + /// Wakes bounded v2 drains waiting for a prompt-lock registry slot. + pub prompt_lock_available: std::sync::Arc, /// When each held prompt lock was first requested. Purely observational: /// the sampler turns the oldest into `ocserve_prompt_lock_age_seconds` and /// warns past the watchdog. A wedged blocking task cannot be cancelled, so @@ -375,6 +413,47 @@ pub struct AppState { /// v2 runtime stays dark until acceptance; the set exists so construction /// is uniform, but no v2 route touches it while disabled). pub coordinators: ocserve_core::coordinator::SharedCoordinators, + /// Active v2 drains, one per in-flight session and capped process-wide by + /// `v2_drain_slots` (`MAX_V2_DRAIN_TASKS`). + pub v2_drain_tasks: + std::sync::Arc>>, + /// Fixed active-session cap for v2 drains (see `MAX_V2_DRAIN_TASKS`). + pub v2_drain_slots: std::sync::Arc, + /// Fixed cap on futures queued while waiting for an active drain slot. + pub v2_drain_waiters: std::sync::Arc, + /// Shared admission readers; delete writers fence a session through + /// cancellation and row deletion without serializing other sessions. + pub(crate) v2_session_admission_gate: tokio::sync::RwLock<()>, + /// Session ids currently fenced by a delete; bounded by delete slots. + pub(crate) v2_deleting_sessions: parking_lot::Mutex>, + /// Bounds concurrent session deletion fences. + pub(crate) v2_session_delete_slots: std::sync::Arc, + /// Monotonic generation source for v2 drain task cleanup. + pub v2_drain_gen: std::sync::Arc, + /// Deterministic interleaving seam for the drain cancellation race test. + #[cfg(test)] + pub v2_cancel_transition_hook: parking_lot::Mutex>>, + /// Signals the delete boundary after its drain-cancellation wait. + #[cfg(test)] + pub v2_session_remove_after_cancel_hook: parking_lot::Mutex>>, + /// Holds a delete exactly after drain cancellation to test late admission. + #[cfg(test)] + pub v2_session_delete_gap_test_barriers: + parking_lot::Mutex>>, + /// Deterministic seam for waking multiple lock-capacity waiters at once. + #[cfg(test)] + pub v2_lock_wait_test_barriers: + parking_lot::Mutex>>, + /// Deterministic seam for cancelling a queued Input item after the drain + /// selected its scope but before it checks the inbox row. + #[cfg(test)] + pub v2_input_scope_test_barriers: + parking_lot::Mutex>>, + /// Deterministic seam for admitting a new prompt after an interrupted + /// drain released its session lock but before its terminal event publishes. + #[cfg(test)] + pub v2_drain_terminal_test_barriers: + parking_lot::Mutex>>, /// /tui/control rendezvous queues (v1 shared/tui-control.ts) pub tui: crate::tui::TuiControl, /// Web-UI reverse proxy (Phase 7; upstream serveUIEffect parity) @@ -721,6 +800,7 @@ impl AppState { prompt_locks: std::sync::Arc::new(parking_lot::Mutex::new( std::collections::HashMap::new(), )), + prompt_lock_available: std::sync::Arc::new(tokio::sync::Notify::new()), lock_ages: std::sync::Arc::new(parking_lot::Mutex::new( std::collections::HashMap::new(), )), @@ -734,6 +814,31 @@ impl AppState { shells: ocserve_core::shell::ShellRegistry::new(), v2_enabled: v2, coordinators: std::sync::Arc::new(ocserve_core::coordinator::CoordinatorSet::new()), + v2_drain_tasks: std::sync::Arc::new(parking_lot::Mutex::new( + std::collections::HashMap::new(), + )), + v2_drain_slots: std::sync::Arc::new(tokio::sync::Semaphore::new(MAX_V2_DRAIN_TASKS)), + v2_drain_waiters: std::sync::Arc::new(tokio::sync::Semaphore::new( + MAX_V2_DRAIN_WAITERS, + )), + v2_session_admission_gate: tokio::sync::RwLock::new(()), + v2_deleting_sessions: parking_lot::Mutex::new(std::collections::HashSet::new()), + v2_session_delete_slots: std::sync::Arc::new(tokio::sync::Semaphore::new( + MAX_V2_SESSION_DELETES, + )), + v2_drain_gen: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(1)), + #[cfg(test)] + v2_cancel_transition_hook: parking_lot::Mutex::new(None), + #[cfg(test)] + v2_session_remove_after_cancel_hook: parking_lot::Mutex::new(None), + #[cfg(test)] + v2_session_delete_gap_test_barriers: parking_lot::Mutex::new(None), + #[cfg(test)] + v2_lock_wait_test_barriers: parking_lot::Mutex::new(None), + #[cfg(test)] + v2_input_scope_test_barriers: parking_lot::Mutex::new(None), + #[cfg(test)] + v2_drain_terminal_test_barriers: parking_lot::Mutex::new(None), tui: crate::tui::TuiControl::default(), ui: crate::ui::UiProxy::from_env(), pty, @@ -2638,6 +2743,10 @@ async fn post_summarize( let Ok(_guard) = release.arc().try_lock() else { return Err(session_busy(&sid)); }; + { + let _admission = st.v2_session_admission_gate.read().await; + check_v1_session_admission(&st, &sid).await?; + } // C5 retrofit (M6, COMPACTION.md §3): manual summarize = v1 source // (handlers/session.ts:273-293 at the FREEZE TAG v1.18.31, verified): @@ -3876,6 +3985,7 @@ struct LockRelease { >, >, ages: std::sync::Arc>>, + available: std::sync::Arc, sid: String, lock: std::sync::Arc>, } @@ -3898,6 +4008,7 @@ impl Drop for LockRelease { { map.remove(&self.sid); self.ages.lock().remove(&self.sid); + self.available.notify_one(); } } } @@ -3910,7 +4021,7 @@ impl Drop for LockRelease { pub(crate) async fn lock_session(st: &Arc, sid: &str) -> Result { let lock = { let mut map = st.prompt_locks.lock(); - if map.len() >= 64 && !map.contains_key(sid) { + if map.len() >= MAX_PROMPT_LOCKS && !map.contains_key(sid) { return Err(ApiError { status: StatusCode::SERVICE_UNAVAILABLE, name: "ServiceUnavailableError", @@ -3933,18 +4044,64 @@ pub(crate) async fn lock_session(st: &Arc, sid: &str) -> Result, sid: &str) -> LockRelease { + loop { + // Enable the notification before checking the map. Merely constructing + // `Notified` does not register it; multiple capacity releases before + // first poll would otherwise collapse into Notify's single permit. + let mut available = Box::pin(st.prompt_lock_available.notified()); + available.as_mut().enable(); + match lock_session(st, sid).await { + Ok(release) => return release, + // Capacity is the only refusal lock_session returns today, and it + // is identified by status/name rather than the message text. The + // old string match meant a wording change silently fell into the + // 1 Hz retry branch while holding a drain slot. + Err(error) + if error.status == StatusCode::SERVICE_UNAVAILABLE + && error.name == "ServiceUnavailableError" => + { + #[cfg(test)] + let test_barriers = st.v2_lock_wait_test_barriers.lock().clone(); + #[cfg(test)] + if let Some(barriers) = test_barriers { + barriers.0.wait().await; + barriers.1.wait().await; + } + available.await; + } + Err(error) => { + // The current lock API only fails for its explicit capacity + // bound. Keep the durable v2 item pending if a future failure + // class is added, retrying with cancellation handled by the + // caller's select instead of silently abandoning the wake. + tracing::warn!( + "v2 prompt-lock acquisition deferred for {sid}: {} {}", + error.name, + error.message + ); + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + } + } + } +} + /// Re-arm the v2 coordinator after a v1 turn releases the session lock. /// -/// The v2 drain now declines to run while a v1 prompt holds the lock (one -/// mutex, both grammars) and leaves the inbox row alone rather than dropping -/// the user's message. Without this wake, a message queued by a v2 client -/// during a v1 turn would sit until some *later* v2 request happened to wake -/// the coordinator — a silent stall, which is worse than the race it fixes. +/// V2 drains wait on the shared per-session lock while retaining the durable +/// inbox row. This completion wake is a fallback for work whose prior drain +/// was cancelled while a v1 prompt held the lock; without it that row could +/// sit until some later request happened to wake the coordinator. /// /// Cheap by construction: one indexed `EXISTS`-shaped query, and only when the /// session actually has queued work. A v1-only deployment pays one query per @@ -3970,16 +4127,26 @@ pub(crate) async fn wake_inbox_after_turn(st: &Arc, sid: &str) { }) .await .unwrap_or(0); - if pending > 0 - && st - .coordinators - .wake(sid, ocserve_core::coordinator::WakeScope::Input) - { - let st2 = st.clone(); - let sid2 = sid.to_string(); - tokio::spawn(async move { - crate::v2::drain_session(st2, sid2).await; - }); + if pending > 0 { + let slot = match crate::v2::reserve_drain_slot_wait(st, sid).await { + Ok(slot) => slot, + Err(_) => { + tracing::warn!("v2 drain capacity unavailable after v1 turn for {sid}"); + return; + } + }; + let _admission = st.v2_session_admission_gate.read().await; + if crate::v2::session_is_deleting(st, sid) + || !ocserve_store::session_exists(&st.db, sid).unwrap_or(false) + { + return; + } + crate::v2::wake_drain( + st.clone(), + sid.to_string(), + ocserve_core::coordinator::WakeScope::Input, + slot, + ); } } @@ -3996,6 +4163,32 @@ pub(crate) fn prompt_err(e: anyhow::Error) -> ApiError { } } +/// v1 prompt admission uses the frozen v1 NotFoundError envelope while +/// sharing the delete fence with v2 inbox/drain admission. +async fn check_v1_session_admission(st: &AppState, sid: &str) -> Result<(), ApiError> { + // The delete-fence read is in-memory and cheap; the existence query is + // SQLite and runs on the blocking pool (F1 class: a sync store call on a + // tokio worker parks it for the query's duration). Returning a TYPED error + // matters: `lock_session_wait` used to branch on this message string. + if crate::v2::session_is_deleting(st, sid) { + return Err(ApiError::not_found(format!("Session not found: {sid}"))); + } + let sid_owned = sid.to_string(); + let exists = run_blocking(&st.db, move |db| { + ocserve_store::session_exists(db, &sid_owned) + }) + .await + .map_err(|e| ApiError { + status: StatusCode::INTERNAL_SERVER_ERROR, + name: "InternalError", + message: format!("{e:#}"), + })?; + if !exists { + return Err(ApiError::not_found(format!("Session not found: {sid}"))); + } + Ok(()) +} + async fn post_message( State(st): State>, axum::extract::Path(id): axum::extract::Path, @@ -4007,6 +4200,10 @@ async fn post_message( let ctx = build_prompt_context(&st, &req, &id)?; let _release = lock_session(&st, &id).await?; let _guard = _release.arc().lock().await; + { + let _admission = st.v2_session_admission_gate.read().await; + check_v1_session_admission(&st, &id).await?; + } let writer = st.writer.clone(); let result = ocserve_core::prompt::run_prompt(&ctx, &writer, &id, &req) .await @@ -4030,23 +4227,17 @@ async fn post_prompt_async( // decode FIRST (probe: `prompt_async {}` → 400 Missing key, never 204; // HttpApi payload decode precedes both 204 and the session check). let req = decode_v1(ocserve_core::wire::PromptRequest::decode(&payload))?; - let exists_id = id.clone(); - if !run_blocking(&st.db, move |db| { - ocserve_store::session_exists(db, &exists_id) - }) - .await - .map_err(|e| ApiError { - status: StatusCode::INTERNAL_SERVER_ERROR, - name: "InternalError", - message: format!("{e:#}"), - })? { - return Err(ApiError::not_found(format!("Session not found: {id}")).into()); + { + let _admission = st.v2_session_admission_gate.read().await; + check_v1_session_admission(&st, &id).await?; } let ctx = build_prompt_context(&st, &req, &id)?; // release MOVES INTO the task: the handler returns 204 immediately, so a // handler-scoped guard would drop (and evict the lock entry) while the // prompt is still running (antagonism A2 — ownership, not just RAII). let release = lock_session(&st, &id).await?; + let _admission = st.v2_session_admission_gate.read().await; + check_v1_session_admission(&st, &id).await?; let writer = st.writer.clone(); let sid = id.clone(); let generation = st @@ -4265,17 +4456,25 @@ async fn patch_session( Ok(Json(info)) } -/// DELETE /session/{id} → true (v1 Schema.Boolean). Aborts any in-flight -/// async prompt first (it would write into a deleted session otherwise). +/// DELETE /session/{id} → true (v1 Schema.Boolean). Stops in-flight v1/v2 +/// work before the session rows are deleted. async fn delete_session_route( State(st): State>, axum::extract::Path(id): axum::extract::Path, ) -> Result, ApiError> { + let delete_fence = crate::v2::begin_session_delete(&st, &id).await?; crate::goal_loop::cancel_driver(&st, &id); + crate::v2::cancel_drains_for_delete(&st, &id).await; if let Some((_, handle)) = st.prompt_tasks.lock().remove(&id) { handle.abort(); } - let found = ocserve_store::delete_session(&st.writer, &st.db, &id).map_err(|e| ApiError { + let delete_result = { + let release = lock_session(&st, &id).await?; + let _session_guard = release.arc().lock().await; + ocserve_store::delete_session(&st.writer, &st.db, &id) + }; + delete_fence.finish().await; + let found = delete_result.map_err(|e| ApiError { status: StatusCode::INTERNAL_SERVER_ERROR, name: "InternalError", message: format!("{e:#}"), @@ -4305,6 +4504,7 @@ async fn post_abort( // A goal loop re-spawns turns, so aborting the current prompt alone would // let it continue: cancel the driver FIRST, then the turn. crate::goal_loop::cancel_driver(&st, &id); + crate::v2::cancel_drain(&st, &id).await; if let Some((_, handle)) = st.prompt_tasks.lock().remove(&id) { handle.abort(); tracing::info!("abort: cancelled prompt task for {id}"); @@ -5110,7 +5310,12 @@ async fn post_command( && st.goals_enabled && template.as_deref().map(str::is_empty).unwrap_or(true); if is_builtin_goal { - return match crate::goal_loop::arm(&st, &id, arguments).await { + let arm = { + let _admission = st.v2_session_admission_gate.read().await; + check_v1_session_admission(&st, &id).await?; + crate::goal_loop::arm(&st, &id, arguments).await + }; + return match arm { // Control words change state and start no turn. The response is // still a valid `SessionV1.WithParts`: a persisted assistant // notice, so a strict Message parser cannot choke on a goal-state @@ -5218,6 +5423,10 @@ async fn post_command( let Ok(_guard) = _release.arc().try_lock() else { return Err(session_busy(&id).into()); }; + { + let _admission = st.v2_session_admission_gate.read().await; + check_v1_session_admission(&st, &id).await?; + } let writer = st.writer.clone(); let (info, parts) = ocserve_core::prompt::run_prompt_with( &ctx, @@ -5268,6 +5477,10 @@ async fn post_shell( let Ok(_guard) = _release.arc().try_lock() else { return Err(session_busy(&id).into()); }; + { + let _admission = st.v2_session_admission_gate.read().await; + check_v1_session_admission(&st, &id).await?; + } let dir = st.paths["directory"].as_str().unwrap_or("/").to_string(); // busy → run → idle (client spinner parity) @@ -6982,9 +7195,9 @@ mod drain_lock { use super::*; - /// Holding the session lock makes the drain decline the turn. + /// A drain waits on the session lock, then consumes its durable inbox item. #[tokio::test] - async fn drain_defers_while_a_prompt_holds_the_lock() { + async fn drain_waits_while_a_prompt_holds_the_lock() { let st = AppState::with_payloads_and_v2(None, Default::default(), true); let sid = "ses_drain_lock"; st.writer @@ -7021,17 +7234,32 @@ mod drain_lock { // which made the drain actually do work). let guard = release.arc().lock().await; - v2::drain_session(st.clone(), sid.to_string()).await; + let drain_st = st.clone(); + let drain_sid = sid.to_string(); + let mut drain = tokio::spawn(async move { v2::drain_session(drain_st, drain_sid).await }); + assert!( + tokio::time::timeout(std::time::Duration::from_millis(20), &mut drain) + .await + .is_err(), + "drain must wait while the v1 prompt owns the shared lock" + ); let remaining = ocserve_store::inbox::list(&st.db, sid).expect("list"); assert_eq!( remaining.len(), 1, - "drain consumed queued work while a prompt held the session lock" + "drain consumed queued work before acquiring the session lock" ); drop(guard); drop(release); + drain.await.unwrap(); + assert!( + ocserve_store::inbox::list(&st.db, sid) + .expect("list after lock release") + .is_empty(), + "waiting drain must resume the admitted inbox item" + ); } } diff --git a/crates/ocserve-http/src/scheduler.rs b/crates/ocserve-http/src/scheduler.rs index f88ace9..dc82150 100644 --- a/crates/ocserve-http/src/scheduler.rs +++ b/crates/ocserve-http/src/scheduler.rs @@ -167,8 +167,28 @@ async fn step(st: &Arc, row: &Value, now: i64) -> anyhow::Result<()> { && !prev.is_empty() && ocserve_store::inbox::get(&st.db, prev)?.is_some() { - st.coordinators - .wake(sid, ocserve_core::coordinator::WakeScope::Input); + // Never await capacity in the single serial tick loop: one slow drain + // would stall every other schedule. A full table returns without + // marking the row, so it stays due and is retried on the next pass; + // the tick interval is the floor on that retry (MIN_SLEEP_MS, 1 s). + let Ok(slot) = crate::v2::reserve_drain_slot(st, sid) else { + ocserve_metrics::labeled_counter( + "ocserve_schedule_fired_total", + "outcome=\"deferred_capacity\"", + 1, + ); + return Ok(()); + }; + let _admission = st.v2_session_admission_gate.read().await; + if crate::v2::session_is_deleting(st, sid) || !ocserve_store::session_exists(&st.db, sid)? { + return Ok(()); + } + crate::v2::wake_drain( + st.clone(), + sid.to_string(), + ocserve_core::coordinator::WakeScope::Input, + slot, + ); let next = following(row, now); let done = finished(row, next, row["remaining_runs"].as_i64()); crate::v2::schedule_mark( @@ -212,6 +232,18 @@ async fn step(st: &Arc, row: &Value, now: i64) -> anyhow::Result<()> { } }, }); + let Ok(drain_slot) = crate::v2::reserve_drain_slot(st, sid) else { + ocserve_metrics::labeled_counter( + "ocserve_schedule_fired_total", + "outcome=\"deferred_capacity\"", + 1, + ); + return Ok(()); + }; + let _admission = st.v2_session_admission_gate.read().await; + if crate::v2::session_is_deleting(st, sid) || !ocserve_store::session_exists(&st.db, sid)? { + return Ok(()); + } let (seq, _evt) = ocserve_store::emit_event( &st.writer, Some(sid), @@ -221,13 +253,12 @@ async fn step(st: &Arc, row: &Value, now: i64) -> anyhow::Result<()> { let fresh = ocserve_store::inbox::admit(&st.writer, sid, &msg_id, "user", &payload, "steer", seq)?; if fresh { - st.coordinators - .wake(sid, ocserve_core::coordinator::WakeScope::Steer); - let st2 = st.clone(); - let sid2 = sid.to_string(); - tokio::spawn(async move { - crate::v2::drain_session(st2, sid2).await; - }); + crate::v2::wake_drain( + st.clone(), + sid.to_string(), + ocserve_core::coordinator::WakeScope::Steer, + drain_slot, + ); ocserve_metrics::labeled_counter("ocserve_schedule_fired_total", "outcome=\"fired\"", 1); } else { // Already admitted by an earlier pass: record the run anyway so the diff --git a/crates/ocserve-http/src/v2.rs b/crates/ocserve-http/src/v2.rs index 147545f..695f353 100644 --- a/crates/ocserve-http/src/v2.rs +++ b/crates/ocserve-http/src/v2.rs @@ -1891,14 +1891,37 @@ pub async fn session_remove_api( Path(id): Path, ) -> Result { // Flag authority is `v2_gate` (generated v2-only table); no handler self-gate. - // The goal driver is a bare task: aborting prompt_tasks does not stop it, - // so the loop could keep spending against a row that is being deleted + let delete_fence = begin_session_delete(&st, &id) + .await + .map_err(crate::HttpError::Api)?; + // The goal driver is separate from both prompt_tasks and tracked v2 + // drains; cancel it before stopping execution and deleting its session // (2026-11-05 review M3). crate::goal_loop::cancel_driver(&st, &id); + cancel_drains_for_delete(&st, &id).await; + #[cfg(test)] + let delete_gap_barriers = { st.v2_session_delete_gap_test_barriers.lock().take() }; + #[cfg(test)] + if let Some(barriers) = delete_gap_barriers { + barriers.0.wait().await; + barriers.1.wait().await; + } + #[cfg(test)] + if let Some(hook) = st.v2_session_remove_after_cancel_hook.lock().take() { + hook(); + } if let Some((_, handle)) = st.prompt_tasks.lock().remove(&id) { handle.abort(); } - let found = ocserve_store::delete_session(&st.writer, &st.db, &id).map_err(internal)?; + let delete_result = { + let release = crate::lock_session(&st, &id) + .await + .map_err(crate::HttpError::Api)?; + let _session_guard = release.arc().lock().await; + ocserve_store::delete_session(&st.writer, &st.db, &id) + }; + delete_fence.finish().await; + let found = delete_result.map_err(internal)?; if !found { return Err(session_not_found(&id)); } @@ -2690,9 +2713,13 @@ async fn admit_typed( delivery: &str, resume: bool, ) -> Result { - if !ocserve_store::session_exists(&st.db, sid).map_err(internal)? { - return Err(session_not_found(sid)); - } + let _admission = st.v2_session_admission_gate.read().await; + check_session_admission(st, sid)?; + let drain_slot = if resume { + Some(reserve_drain_slot(st, sid)?) + } else { + None + }; let msg_id = msg_id.unwrap_or_else(new_v2_msg_id); let (seq, _evt) = ocserve_store::emit_event( &st.writer, @@ -2709,13 +2736,10 @@ async fn admit_typed( } else { ocserve_core::coordinator::WakeScope::Steer }; - if st.coordinators.wake(sid, scope) { - let st2 = st.clone(); - let sid2 = sid.to_string(); - tokio::spawn(async move { - drain_session(st2, sid2).await; - }); - } + let Some(slot) = drain_slot else { + return Err(drain_capacity_error()); + }; + wake_drain(st.clone(), sid.to_string(), scope, slot); } Ok(StatusCode::NO_CONTENT) } @@ -2937,6 +2961,12 @@ pub async fn session_generate_api( if !ocserve_store::session_exists(&st.db, &id).map_err(internal)? { return Err(session_not_found(&id)); } + let release = crate::lock_session(&st, &id).await?; + let _session_guard = release.arc().lock().await; + { + let _admission = st.v2_session_admission_gate.read().await; + check_session_admission(&st, &id)?; + } let req = ocserve_core::wire::PromptRequest { message_id: None, parts: vec![json!({ "type": "text", "text": prompt })], @@ -3890,11 +3920,18 @@ pub async fn prompt( } } } + let _admission = st.v2_session_admission_gate.read().await; + check_session_admission(&st, &id)?; + let msg_id = input.id.clone().unwrap_or_else(new_v2_msg_id); + let delivery = input.delivery.clone().unwrap_or_else(|| "steer".into()); + let drain_slot = if input.resume != Some(false) { + Some(reserve_drain_slot(&st, &id)?) + } else { + None + }; // Revert-commit cleanup before admission (upstream session.ts:165-166: // staged revert commits after prepare, before admit). ocserve_store::revert_cleanup(&st.writer, &st.db, &id).map_err(internal)?; - let msg_id = input.id.clone().unwrap_or_else(new_v2_msg_id); - let delivery = input.delivery.clone().unwrap_or_else(|| "steer".into()); let payload = json!({ "text": input.prompt_text, "files": input.prompt_files, @@ -3935,13 +3972,10 @@ pub async fn prompt( } else { ocserve_core::coordinator::WakeScope::Steer }; - if st.coordinators.wake(&id, scope) { - let st2 = st.clone(); - let sid = id.clone(); - tokio::spawn(async move { - drain_session(st2, sid).await; - }); - } + let Some(slot) = drain_slot else { + return Err(drain_capacity_error()); + }; + wake_drain(st.clone(), id.clone(), scope, slot); } let now_ms = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -3988,12 +4022,458 @@ fn new_v2_msg_id() -> String { out } +fn drain_capacity_error() -> HttpError { + HttpError::TaggedData { + status: StatusCode::SERVICE_UNAVAILABLE, + tag: "ServiceUnavailableError", + fields: json!({"message": "too many active v2 sessions; retry shortly"}), + } +} + +/// Reserve one of the fixed v2 drain slots before admitting durable work. +/// Reuse the current session's slot when a drain already owns it. +pub(crate) fn reserve_drain_slot( + st: &AppState, + sid: &str, +) -> Result, HttpError> { + if let Some(slot) = st + .v2_drain_tasks + .lock() + .get(sid) + .map(|task| task.slot.clone()) + { + return Ok(slot); + } + match st.v2_drain_slots.clone().try_acquire_owned() { + Ok(slot) => Ok(Arc::new(slot)), + Err(_) => st + .v2_drain_tasks + .lock() + .get(sid) + .map(|task| task.slot.clone()) + .ok_or_else(drain_capacity_error), + } +} + +/// Background wake sources may wait for active capacity, but the reservation +/// queue itself is capped. A full waiter cap fails immediately so floods of +/// interrupts/scheduler wakes cannot retain unbounded request or task state. +pub(crate) async fn reserve_drain_slot_wait( + st: &AppState, + sid: &str, +) -> Result, HttpError> { + if let Some(slot) = st + .v2_drain_tasks + .lock() + .get(sid) + .map(|task| task.slot.clone()) + { + return Ok(slot); + } + if let Ok(slot) = st.v2_drain_slots.clone().try_acquire_owned() { + return Ok(Arc::new(slot)); + } + let waiter = st + .v2_drain_waiters + .clone() + .try_acquire_owned() + .map_err(|_| drain_capacity_error())?; + let slot = Arc::new( + st.v2_drain_slots + .clone() + .acquire_owned() + .await + .map_err(|_| drain_capacity_error())?, + ); + drop(waiter); + if let Some(existing) = st + .v2_drain_tasks + .lock() + .get(sid) + .map(|task| task.slot.clone()) + { + return Ok(existing); + } + Ok(slot) +} + +/// Make the coordinator wake and registry transition atomic with cancellation. +/// Without the shared registry lock, a new wake can set `pending` after cancel +/// removes the old drain but before it aborts coordinator state, losing the +/// newly admitted work. +pub(crate) fn wake_drain( + st: Arc, + sid: String, + scope: ocserve_core::coordinator::WakeScope, + slot: Arc, +) -> bool { + let mut tasks = st.v2_drain_tasks.lock(); + if session_is_deleting(&st, &sid) { + return false; + } + let should_spawn = st.coordinators.wake(&sid, scope); + if should_spawn { + register_drain_locked(&st, sid, slot, &mut tasks); + } + should_spawn +} + +/// Test helper for a caller that has already armed the coordinator. Production +/// wake paths use `wake_drain` so wake, registration, and cancellation share +/// one registry critical section. +#[cfg(test)] +pub(crate) fn spawn_drain( + st: Arc, + sid: String, + slot: Arc, +) { + let mut tasks = st.v2_drain_tasks.lock(); + register_drain_locked(&st, sid, slot, &mut tasks); +} + +fn register_drain_locked( + st: &Arc, + sid: String, + slot: Arc, + tasks: &mut std::collections::HashMap, +) { + if session_is_deleting(st, &sid) { + st.coordinators.abort(&sid); + return; + } + let generation = st + .v2_drain_gen + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + if let Some(task) = tasks.get_mut(&sid) { + // The coordinator was re-armed after its prior loop found no work, but + // before that task emitted its terminal event and removed the entry. + // Let it complete that event, then run the newly pending scope using + // the existing permit and registry entry. + task.rerun = true; + return; + } + // watch stores only the latest cancellation bit; the oneshot carries one + // start signal so the registry entry is visible before the drain runs. + let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(false); + let (completion_tx, completion_rx) = tokio::sync::watch::channel(None); + let (start_tx, start_rx) = tokio::sync::oneshot::channel(); + let st_task = st.clone(); + let sid_task = sid.clone(); + let task_cancel = cancel_rx.clone(); + let handle = tokio::spawn(async move { + if start_rx.await.is_err() { + let _ = completion_tx.send(Some(false)); + return false; + } + loop { + let outcome = + drain_session_inner(st_task.clone(), sid_task.clone(), Some(task_cancel.clone())) + .await; + #[cfg(test)] + if outcome.was_interrupted() { + let barriers = st_task.v2_drain_terminal_test_barriers.lock().take(); + if let Some(barriers) = barriers { + barriers.0.wait().await; + barriers.1.wait().await; + } + } + let interrupted = terminal_interrupt_result(&outcome, *cancel_rx.borrow()); + let completed = finish_drain_events(&st_task, &sid_task, outcome); + // Do not promote a late cancel bit into an interruption: the inner + // loop records interruption only when cancellation wins before the + // active turn reaches a terminal state. + let mut tasks = st_task.v2_drain_tasks.lock(); + let registered = tasks + .get_mut(&sid_task) + .filter(|task| task.generation == generation) + .map(|task| (task.rerun, task.slot.clone())); + let rerun = if let Some((queued_rerun, slot)) = registered { + let pending = st_task.coordinators.pending_scope(&sid_task).is_some(); + let cancelled = *cancel_rx.borrow(); + if (queued_rerun || pending) && !cancelled { + if let Some(task) = tasks.get_mut(&sid_task) { + task.rerun = false; + } + true + } else if (queued_rerun || pending) && cancelled { + // A prompt admitted after interrupt began must wait until + // the interrupted event is published and the old turn has + // released its session lock and claim. Hand its doorbell + // to a fresh task while the registry lock still prevents + // a second concurrent replacement. + tasks.remove(&sid_task); + register_drain_locked(&st_task, sid_task.clone(), slot, &mut tasks); + false + } else { + // Every terminal path clears any coordinator state left by + // a consumed-but-cancelled scope (or a terminal error). + // This remains serialized against production wake paths. + st_task.coordinators.abort(&sid_task); + tasks.remove(&sid_task); + false + } + } else { + false + }; + drop(tasks); + if completed { + crate::goal_loop::spawn_wake_public(&st_task, &sid_task); + } + if !rerun { + let _ = completion_tx.send(Some(interrupted)); + return interrupted; + } + } + }); + tasks.insert( + sid.clone(), + crate::V2DrainTask { + generation, + cancel: cancel_tx, + handle, + completion: completion_rx, + slot, + rerun: false, + }, + ); + if start_tx.send(()).is_err() { + tracing::warn!("v2 drain task could not be started for {sid}"); + } +} + +/// Cooperatively stop a registered v2 drain. The task owns claim release and +/// terminal-event publication; this transition resets coordinator state while +/// serialized against every production wake. Concurrent callers observe the +/// same completion signal, so a delete cannot outrun the first interrupter. +pub(crate) async fn cancel_drain(st: &Arc, sid: &str) -> bool { + let completion_rx = { + let mut tasks = st.v2_drain_tasks.lock(); + let completion_rx = if let Some(task) = tasks.get_mut(sid) { + task.cancel.send_replace(true); + task.rerun = false; + Some(task.completion.clone()) + } else { + None + }; + #[cfg(test)] + if let Some(hook) = st.v2_cancel_transition_hook.lock().take() { + hook(); + } + // Keep the registry lock through coordinator reset. Every production + // wake obtains this same lock before setting its coordinator doorbell, + // so no post-cancel wake can be erased by this abort. The registered + // task remains as a retirement marker until its terminal event is + // published; a later wake is handed off by its completion path. + st.coordinators.abort(sid); + completion_rx + }; + let Some(mut completion_rx) = completion_rx else { + return false; + }; + // Bounded wait. A drain can be parked in the provider stall watchdog + // (300 s TTFB + 120 s stream by default), and DELETE must not hold its + // deletion fence — one of a fixed number of slots — for that long with no + // ceiling. On timeout we report "cancellation requested" and let the + // fence/turn teardown finish; the drain's own generation check still + // removes its registry entry. + let budget = std::time::Duration::from_secs(crate::drain_cancel_timeout_secs()); + loop { + if let Some(interrupted) = *completion_rx.borrow() { + return interrupted; + } + match tokio::time::timeout(budget, completion_rx.changed()).await { + Ok(Ok(())) => {} + Ok(Err(_)) => { + tracing::warn!("v2 drain completion signal dropped for {sid}"); + return false; + } + Err(_) => { + ocserve_metrics::labeled_counter( + "ocserve_v2_drain_cancel_timeout_total", + "outcome=\"deferred\"", + 1, + ); + tracing::warn!( + "v2 drain cancellation not confirmed within {}s for {sid}; continuing teardown", + budget.as_secs() + ); + return true; + } + } + } +} + +/// Session deletion must also stop a fresh drain handed off for work that was +/// admitted while the previous cancelled drain was retiring. +pub(crate) async fn cancel_drains_for_delete(st: &Arc, sid: &str) -> bool { + // BOTH loops must be bounded. Bounding only the inner wait is not enough: + // a drain that ignores cancellation stays registered, so this loop would + // simply call `cancel_drain` again and pay the budget repeatedly. The + // deadline is the whole teardown, not one attempt. Proven by + // `delete_returns_when_a_registered_drain_never_confirms_cancellation`, + // which hangs the full request without this cap. + let deadline = std::time::Instant::now() + + std::time::Duration::from_secs(crate::drain_cancel_timeout_secs()); + let mut interrupted = false; + loop { + if !st.v2_drain_tasks.lock().contains_key(sid) { + return interrupted; + } + if std::time::Instant::now() >= deadline { + ocserve_metrics::labeled_counter( + "ocserve_v2_drain_cancel_timeout_total", + "outcome=\"deferred\"", + 1, + ); + tracing::warn!( + "v2 drain still registered after the cancellation budget for {sid}; \ + proceeding with the delete (the drain keeps its slot until it exits)" + ); + return true; + } + interrupted |= cancel_drain(st, sid).await; + } +} + +/// Prevent prompt/inbox admission from registering a new drain while a +/// session is being deleted. The read side is held only over the admission +/// writes; delete holds the write side when opening and closing its fence. +pub(crate) async fn begin_session_delete( + st: &Arc, + sid: &str, +) -> Result { + let permit = st + .v2_session_delete_slots + .clone() + .try_acquire_owned() + .map_err(|_| crate::ApiError { + status: StatusCode::SERVICE_UNAVAILABLE, + name: "ServiceUnavailableError", + message: "too many session deletions; retry shortly".into(), + })?; + let _admission = st.v2_session_admission_gate.write().await; + let _tasks = st.v2_drain_tasks.lock(); + let mut deleting = st.v2_deleting_sessions.lock(); + if !deleting.insert(sid.to_string()) { + return Err(crate::ApiError::not_found(format!( + "Session not found: {sid}" + ))); + } + drop(deleting); + drop(_tasks); + Ok(SessionDeleteFence { + st: st.clone(), + sid: sid.to_string(), + _permit: permit, + }) +} + +/// True while the delete fence is active. `wake_drain` checks this under the +/// same registry lock used when the fence opens and closes. +pub(crate) fn session_is_deleting(st: &AppState, sid: &str) -> bool { + st.v2_deleting_sessions.lock().contains(sid) +} + +/// Revalidate the session after acquiring the admission read lock. A request +/// that passed an earlier existence check may have waited through deletion. +pub(crate) fn check_session_admission(st: &AppState, sid: &str) -> Result<(), HttpError> { + if session_is_deleting(st, sid) + || !ocserve_store::session_exists(&st.db, sid).map_err(internal)? + { + return Err(session_not_found(sid)); + } + Ok(()) +} + +pub(crate) struct SessionDeleteFence { + st: Arc, + sid: String, + _permit: tokio::sync::OwnedSemaphorePermit, +} + +impl SessionDeleteFence { + pub(crate) async fn finish(self) { + { + let _admission = self.st.v2_session_admission_gate.write().await; + let _tasks = self.st.v2_drain_tasks.lock(); + self.st.v2_deleting_sessions.lock().remove(&self.sid); + } + } +} + +impl Drop for SessionDeleteFence { + fn drop(&mut self) { + let _tasks = self.st.v2_drain_tasks.lock(); + self.st.v2_deleting_sessions.lock().remove(&self.sid); + } +} + +#[derive(Debug)] +enum DrainOutcome { + NotStarted, + Terminal { + failure: Option, + interrupted: bool, + }, +} + +impl DrainOutcome { + fn was_interrupted(&self) -> bool { + matches!( + self, + Self::Terminal { + interrupted: true, + .. + } + ) + } +} + +/// The inner `select!` latches whether cancellation won before the turn +/// completed. A later watch update is only a request against an already +/// terminal drain and must not rewrite success/failure as interruption. +fn terminal_interrupt_result( + outcome: &DrainOutcome, + _cancel_requested_after_terminal: bool, +) -> bool { + outcome.was_interrupted() +} + +async fn wait_drain_cancel(cancel: &mut Option>) { + let Some(cancel) = cancel else { + std::future::pending::<()>().await; + return; + }; + loop { + if *cancel.borrow() || cancel.changed().await.is_err() { + return; + } + } +} + +fn drain_cancelled(cancel: &Option>) -> bool { + cancel.as_ref().is_some_and(|cancel| *cancel.borrow()) +} + /// v2 drain loop for one session: promote-one + run-one-turn via the shared /// `run_prompt` executor, until nothing is promotable. Claims wrap the drain /// (write-ahead on start, released on terminal states). /// -/// `pub(crate)` so inbox-update wakes (steer flips) share the exact drain. +/// Keep the direct form for focused unit tests; production wakes go through +/// `spawn_drain` so their JoinHandle is registered before execution starts. +#[cfg(test)] pub(crate) async fn drain_session(st: Arc, sid: String) { + let outcome = drain_session_inner(st.clone(), sid.clone(), None).await; + if finish_drain_events(&st, &sid, outcome) { + crate::goal_loop::spawn_wake_public(&st, &sid); + } +} + +async fn drain_session_inner( + st: Arc, + sid: String, + mut cancel: Option>, +) -> DrainOutcome { use ocserve_core::coordinator::WakeScope; let now_ms = || { std::time::SystemTime::now() @@ -4014,38 +4494,62 @@ pub(crate) async fn drain_session(st: Arc, sid: String) { // busy class already fixed once). Claims stay as the durable status record // that `/api/session/active` reads. // - // Busy means "leave the work queued and come back": consuming an inbox row - // here would drop the user's message. The wake that ends the v1 turn - // re-arms the drain (see `wake_inbox_after_turn`). - let Ok(release) = crate::lock_session(&st, &sid).await else { - return; + // A drain waits for both prompt-lock registry capacity and the per-session + // lock. The process drain semaphore bounds those waiters, cancellation + // drops them, and admitted inbox rows cannot be stranded by a failed + // startup race with a concurrent wake. + let release = tokio::select! { + biased; + _ = wait_drain_cancel(&mut cancel) => { + return DrainOutcome::NotStarted; + } + release = crate::lock_session_wait(&st, &sid) => release, }; - let Ok(guard) = release.arc().try_lock() else { - // A v1 prompt holds the session: leave everything queued, release the - // coordinator so a later wake can start a fresh drain. - st.coordinators.abort(&sid); - return; + let guard = tokio::select! { + biased; + _ = wait_drain_cancel(&mut cancel) => { + return DrainOutcome::NotStarted; + } + guard = release.arc().lock() => guard, }; // Write-ahead claim in the same batch as the started event. let _ = ocserve_store::claim::claim(&st.writer, &sid, now_ms()); let _ = crate::emit_durable_http(&st, &sid, "session.execution.started", json!({})); + let mut failure = None; + let mut interrupted = false; + let mut input_scope_enabled = false; loop { - // Doorbell first, work-continuation second (upstream `advanceToStep` - // loops on promotability, not doorbells — otherwise a wake that - // arrives mid-turn strands the rest of the queue). `next_scope` - // keeps `draining` armed while work remains, so no second drain can - // spawn mid-loop. - let work_remains = ocserve_core::coordinator::next_for_scope( - &st.db, - &sid, - ocserve_core::coordinator::WakeScope::Input, - ) - .ok() - .flatten() - .is_some(); - let Some(scope) = st.coordinators.next_scope(&sid, work_remains) else { + if drain_cancelled(&cancel) { + interrupted = true; break; + } + // Doorbell first, work-continuation second. A steer-only wake keeps + // draining steers; Input continuation is enabled only after an Input + // wake, so interrupt resume cannot fall through into queued prompts. + let continuation = if input_scope_enabled { + ocserve_core::coordinator::WakeScope::Input + } else { + ocserve_core::coordinator::WakeScope::Steer }; + let work_remains = ocserve_core::coordinator::next_for_scope(&st.db, &sid, continuation) + .ok() + .flatten() + .is_some() + .then_some(continuation); + let Some(scope) = st.coordinators.next_scope_for(&sid, work_remains) else { + break; + }; + #[cfg(test)] + if scope == WakeScope::Input { + let barriers = st.v2_input_scope_test_barriers.lock().take(); + if let Some(barriers) = barriers { + barriers.0.wait().await; + barriers.1.wait().await; + } + } + if scope == WakeScope::Input { + input_scope_enabled = true; + } let next = ocserve_core::coordinator::next_for_scope(&st.db, &sid, scope); let Ok(Some(item_id)) = next else { // Nothing promotable in this scope: idle-scope falls through to @@ -4055,21 +4559,86 @@ pub(crate) async fn drain_session(st: Arc, sid: String) { } continue; }; - let turn = run_v2_turn(&st, &sid, &item_id).await; + let turn = tokio::select! { + biased; + _ = wait_drain_cancel(&mut cancel) => { + interrupted = true; + break; + } + turn = run_v2_turn(&st, &sid, &item_id) => turn, + }; if let Err(e) = turn { tracing::warn!("v2 drain turn failed for {sid}: {e:#}"); + let message: String = format!("{e:#}").chars().take(2000).collect(); + failure = Some(json!({ + "sessionID": sid, + "error": {"name": "UnknownError", "data": {"message": message}}, + })); break; } } + if drain_cancelled(&cancel) { + interrupted = true; + } drop(guard); drop(release); let _ = ocserve_store::claim::release(&st.writer, &sid); - let _ = crate::emit_durable_http(&st, &sid, "session.execution.succeeded", json!({})); - // The idle hook must fire for the v2 grammar too: a goal armed through - // `/api/session/{id}/command` was advanced by the v1 prompt path only, so - // a v2-only deployment could arm a goal that never reviews (2026-11-05 - // review). Deduplicated + non-blocking inside the hook. - crate::goal_loop::spawn_wake_public(&st, &sid); + DrainOutcome::Terminal { + failure, + interrupted, + } +} + +fn finish_drain_events(st: &Arc, sid: &str, outcome: DrainOutcome) -> bool { + match outcome { + DrainOutcome::NotStarted => false, + DrainOutcome::Terminal { + failure: _, + interrupted: true, + } => { + if let Err(e) = crate::emit_durable_http( + st, + sid, + "session.execution.interrupted", + json!({"sessionID": sid, "reason": "user"}), + ) { + tracing::warn!( + "v2 execution interruption event emit failed for {sid}: {} {}", + e.name, + e.message + ); + } + true + } + DrainOutcome::Terminal { + failure: Some(error), + interrupted: false, + } => { + if let Err(e) = crate::emit_durable_http(st, sid, "session.execution.failed", error) { + tracing::warn!( + "v2 execution failure event emit failed for {sid}: {} {}", + e.name, + e.message + ); + } + true + } + DrainOutcome::Terminal { + failure: None, + interrupted: false, + } => { + if let Err(e) = + crate::emit_durable_http(st, sid, "session.execution.succeeded", json!({})) + { + tracing::warn!( + "v2 execution success event emit failed for {sid}: {} {}", + e.name, + e.message + ); + } + true + } + } } /// Run one admitted inbox item through the shared executor: deliver (delete @@ -4948,26 +5517,30 @@ pub async fn inbox_update( fields: json!({"message": "delivery must be \"steer\" | \"queue\""}), }); }; - if !ocserve_store::session_exists(&st.db, &id).map_err(internal)? { - return Err(session_not_found(&id)); - } + let _admission = st.v2_session_admission_gate.read().await; + check_session_admission(&st, &id)?; match ocserve_store::inbox::get(&st.db, &inbox_id).map_err(internal)? { Some((sid, _, _, _, _)) if sid == id => { + let drain_slot = if delivery == "steer" { + Some(reserve_drain_slot(&st, &id)?) + } else { + None + }; if !ocserve_store::inbox::set_delivery(&st.writer, &inbox_id, delivery) .map_err(internal)? { return Err(session_not_found(&id)); } - if delivery == "steer" - && st - .coordinators - .wake(&id, ocserve_core::coordinator::WakeScope::Steer) - { - let st2 = st.clone(); - let sid2 = id.clone(); - tokio::spawn(async move { - drain_session(st2, sid2).await; - }); + if delivery == "steer" { + let Some(slot) = drain_slot else { + return Err(drain_capacity_error()); + }; + wake_drain( + st.clone(), + id.clone(), + ocserve_core::coordinator::WakeScope::Steer, + slot, + ); } } _ => return Err(session_not_found(&id)), @@ -4975,32 +5548,89 @@ pub async fn inbox_update( Ok(StatusCode::NO_CONTENT) } -/// POST /api/session/{sessionID}/interrupt — NoContent (abort the task). +/// POST /api/session/{sessionID}/interrupt — `{interrupted:boolean}`. +#[derive(serde::Deserialize, Default)] +pub struct InterruptParams { + resume: Option, +} + pub async fn interrupt( State(st): State>, Path(id): Path, -) -> Result { + Query(params): Query, +) -> Result, HttpError> { // Stop the LOOP too, and pause the goal. Upstream pauses an active goal on // `session.execution.interrupted` with reason "user" (goal.ts:296-305): the // interrupt itself must not be undone by the next reviewer "continue". - // Aborting only the current prompt let the driver finish the turn and - // start another (2026-11-05 review M1). + // Stopping only the current turn let the driver finish and start another + // (2026-11-05 review M1); pause it before either v1 or v2 work unwinds. crate::goal_loop::cancel_driver(&st, &id); + let mut interrupted = false; if let Some((_, handle)) = st.prompt_tasks.lock().remove(&id) { handle.abort(); + interrupted = true; } + interrupted |= cancel_drain(&st, &id).await; crate::goal_loop::pause_if_active(&st, &id, "Stopped by the user").await; - Ok(StatusCode::NO_CONTENT) + + // Upstream resumes steer/control items on interrupt by default, leaving + // queued prompts parked. The interrupted drain clears its in-memory + // doorbell, so re-arm only if promotable steer/control work remains. + let steer_pending = ocserve_core::coordinator::next_for_scope( + &st.db, + &id, + ocserve_core::coordinator::WakeScope::Steer, + ) + .ok() + .flatten() + .is_some(); + if params.resume.unwrap_or(true) && steer_pending { + let slot = reserve_drain_slot_wait(&st, &id).await?; + let _admission = st.v2_session_admission_gate.read().await; + // A read error must not be read as "session gone": that silently drops + // the re-armed steer work the operator just asked for. Propagate. + if !session_is_deleting(&st, &id) + && ocserve_store::session_exists(&st.db, &id).map_err(|e| crate::ApiError { + status: StatusCode::INTERNAL_SERVER_ERROR, + name: "InternalError", + message: format!("{e:#}"), + })? + && ocserve_core::coordinator::next_for_scope( + &st.db, + &id, + ocserve_core::coordinator::WakeScope::Steer, + ) + .ok() + .flatten() + .is_some() + { + wake_drain( + st.clone(), + id.clone(), + ocserve_core::coordinator::WakeScope::Steer, + slot, + ); + } + } + Ok(Json(json!({"interrupted": interrupted}))) } /// POST /api/session/{sessionID}/compact — NoContent (triggers summarize). pub async fn compact( State(st): State>, Path(id): Path, -) -> Result { +) -> Result { // Freeze: ServiceUnavailable when no model is available; ocserve maps to // the same when the session has no resolvable model. Otherwise summarize. let payload = ocserve_core::wire::PromptRequest::default(); + let release = crate::lock_session(&st, &id).await?; + let Ok(_guard) = release.arc().try_lock() else { + return Err(crate::session_busy(&id).into()); + }; + { + let _admission = st.v2_session_admission_gate.read().await; + check_session_admission(&st, &id)?; + } let ctx = match crate::build_prompt_context(&st, &payload, &id) { Ok(c) => c, Err(e) => { @@ -5008,13 +5638,10 @@ pub async fn compact( status: StatusCode::SERVICE_UNAVAILABLE, name: "ServiceUnavailableError", message: e.message, - }); + } + .into()); } }; - let release = crate::lock_session(&st, &id).await?; - let Ok(_guard) = release.arc().try_lock() else { - return Err(crate::session_busy(&id)); - }; let writer = st.writer.clone(); let _ = crate::run_compact(&st, &ctx, &writer, &id).await; drop(_guard); @@ -8199,6 +8826,904 @@ fn client_mcp_refused() -> crate::HttpError { } } +#[cfg(test)] +mod v2_drain_outcome_tests { + use super::{ + DrainOutcome, InterruptParams, cancel_drain, interrupt, session_remove_api, spawn_drain, + terminal_interrupt_result, + }; + use crate::{AppState, MAX_PROMPT_LOCKS, V2DrainTask}; + use axum::body::Body; + use axum::http::{Request, StatusCode}; + use serde_json::json; + use std::sync::Arc; + use tower::ServiceExt; + + fn insert_session(st: &Arc, sid: &str) { + ocserve_store::insert_session( + &st.writer, + &json!({ + "id": sid, "projectID": "global", "directory": "/w", "path": sid, + "slug": sid, "title": "drain test", "version": "1", + "time": {"created": 1, "updated": 2}, + }), + ) + .unwrap(); + } + + async fn post_text(app: &axum::Router, sid: &str, text: &str) -> StatusCode { + let response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri(format!("/api/session/{sid}/prompt")) + .header("content-type", "application/json") + .body(Body::from(json!({"text": text}).to_string())) + .unwrap(), + ) + .await + .unwrap(); + response.status() + } + + fn event_seqs(st: &Arc, sid: &str, event_type: &str) -> Vec { + let conn = ocserve_store::pragma::open_reader(&st.db).unwrap(); + let mut stmt = conn + .prepare_cached( + "SELECT seq FROM event WHERE session_id = ?1 AND type IN (?2, ?3) ORDER BY seq", + ) + .unwrap(); + stmt.query_map( + rusqlite::params![sid, event_type, format!("{event_type}@v1")], + |row| row.get(0), + ) + .unwrap() + .collect::, _>>() + .unwrap() + } + + fn spawn_holding_provider() -> (std::net::SocketAddr, std::sync::mpsc::Receiver<()>) { + use std::io::Read; + + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let (entered_tx, entered_rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + for socket in listener.incoming().flatten() { + let entered_tx = entered_tx.clone(); + std::thread::spawn(move || { + let mut socket = socket; + let mut buffer = [0u8; 4096]; + if socket.read(&mut buffer).unwrap_or(0) > 0 { + let _ = entered_tx.send(()); + while socket.read(&mut buffer).unwrap_or(0) > 0 {} + } + }); + } + }); + (addr, entered_rx) + } + + fn state_with_holding_provider() -> ( + tempfile::TempDir, + Arc, + std::sync::mpsc::Receiver<()>, + ) { + let (addr, provider_entered) = spawn_holding_provider(); + let dir = tempfile::tempdir().unwrap(); + let db = ocserve_store::writer::db_path(dir.path()); + let writer = Arc::new(ocserve_store::Writer::spawn(db.clone()).unwrap()); + let blobs = Arc::new(ocserve_store::BlobStore::new(dir.path().join("blobs")).unwrap()); + let llm = crate::LlmRegistry { + modalities: Default::default(), + endpoints: [("fake".into(), (format!("http://{addr}"), "test-key".into()))] + .into_iter() + .collect(), + pricing: Default::default(), + limits: Default::default(), + default_model: ("fake".into(), "m".into()), + systems: Default::default(), + default_agent: "build".into(), + }; + let st = AppState::with_wiring_and_v2( + None, + Default::default(), + crate::Wires { + db, + blobs, + writer, + llm, + }, + true, + ); + (dir, st, provider_entered) + } + + async fn wait_for_started(st: &Arc, sid: &str, count: usize) -> Vec { + for _ in 0..200 { + let seqs = event_seqs(st, sid, "session.execution.started"); + if seqs.len() >= count { + return seqs; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + event_seqs(st, sid, "session.execution.started") + } + + #[test] + fn late_cancel_does_not_reclassify_a_terminal_success() { + let outcome = DrainOutcome::Terminal { + failure: None, + interrupted: false, + }; + assert!(!terminal_interrupt_result(&outcome, true)); + // Negative control: making the late cancellation bit override the + // terminal outcome makes this assertion red. + } + + #[test] + fn cancellation_latched_by_the_running_drain_is_reported() { + let outcome = DrainOutcome::Terminal { + failure: Some(json!({"error": "provider stopped"})), + interrupted: true, + }; + assert!(terminal_interrupt_result(&outcome, false)); + } + + #[tokio::test] + async fn duplicate_wake_queues_rerun_without_cancelling_registered_drain() { + let st = AppState::with_payloads_and_v2(None, Default::default(), true); + let sid = "ses_drain_rerun".to_string(); + let slot = Arc::new(st.v2_drain_slots.clone().try_acquire_owned().unwrap()); + let (cancel, cancel_rx) = tokio::sync::watch::channel(false); + let (completion_tx, completion) = tokio::sync::watch::channel(None); + drop(completion_tx); + let handle = tokio::spawn(async { false }); + st.v2_drain_tasks.lock().insert( + sid.clone(), + V2DrainTask { + generation: 17, + cancel, + handle, + completion, + slot, + rerun: false, + }, + ); + + let duplicate_slot = Arc::new(st.v2_drain_slots.clone().try_acquire_owned().unwrap()); + spawn_drain(st.clone(), sid.clone(), duplicate_slot); + + assert!( + !*cancel_rx.borrow(), + "a duplicate wake must not cancel the old drain" + ); + assert!( + st.v2_drain_tasks + .lock() + .get(&sid) + .is_some_and(|task| task.rerun) + ); + let task = st.v2_drain_tasks.lock().remove(&sid).unwrap(); + let _ = task.handle.await; + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cancel_serializes_coordinator_reset_with_drain_registry() { + let st = AppState::with_payloads_and_v2(None, Default::default(), true); + let sid = "ses_drain_cancel_race".to_string(); + let slot = Arc::new(st.v2_drain_slots.clone().try_acquire_owned().unwrap()); + assert!( + st.coordinators + .wake(&sid, ocserve_core::coordinator::WakeScope::Steer) + ); + let (cancel, _) = tokio::sync::watch::channel(false); + let (completion_tx, completion) = tokio::sync::watch::channel(None); + drop(completion_tx); + let handle = tokio::spawn(async { false }); + st.v2_drain_tasks.lock().insert( + sid.clone(), + V2DrainTask { + generation: 23, + cancel, + handle, + completion, + slot, + rerun: false, + }, + ); + + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let barrier = Arc::new(std::sync::Barrier::new(2)); + let hook_barrier = barrier.clone(); + let hook_state = st.clone(); + let registry_was_held = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let hook_registry_was_held = registry_was_held.clone(); + *st.v2_cancel_transition_hook.lock() = Some(Box::new(move || { + hook_registry_was_held.store( + hook_state.v2_drain_tasks.try_lock().is_none(), + std::sync::atomic::Ordering::Release, + ); + let _ = entered_tx.send(()); + hook_barrier.wait(); + })); + + let cancel_st = st.clone(); + let cancel_sid = sid.clone(); + let cancel_task = tokio::spawn(async move { cancel_drain(&cancel_st, &cancel_sid).await }); + entered_rx.await.unwrap(); + // The hook pauses between the cancellation signal and coordinator + // reset. A wake uses this same registry lock, so it cannot set a new + // pending scope that the abort would then erase. + barrier.wait(); + assert!(!cancel_task.await.unwrap()); + assert!(registry_was_held.load(std::sync::atomic::Ordering::Acquire)); + assert!( + st.coordinators + .wake(&sid, ocserve_core::coordinator::WakeScope::Steer) + ); + st.coordinators.abort(&sid); + // Negative control: dropping the registry guard before coordinator + // abort makes the hook observe an unlocked registry and this test red. + } + + #[tokio::test] + async fn v2_drain_waits_for_prompt_lock_capacity() { + let st = AppState::with_payloads_and_v2(None, Default::default(), true); + let mut held = Vec::new(); + for index in 0..MAX_PROMPT_LOCKS { + let lock = match crate::lock_session(&st, &format!("ses_lock_cap_{index}")).await { + Ok(lock) => lock, + Err(error) => panic!( + "lock slot {index} was unexpectedly refused: {}", + error.message + ), + }; + held.push(lock); + } + + let wait_st = st.clone(); + let mut waiter = + tokio::spawn( + async move { crate::lock_session_wait(&wait_st, "ses_lock_cap_waiter").await }, + ); + assert!( + tokio::time::timeout(std::time::Duration::from_millis(20), &mut waiter) + .await + .is_err(), + "a v2 drain must remain registered until a bounded lock slot opens" + ); + + drop(held.pop()); + let acquired = tokio::time::timeout(std::time::Duration::from_secs(1), &mut waiter) + .await + .expect("released lock capacity should wake the v2 drain") + .unwrap(); + drop(acquired); + // Negative control: removing LockRelease's notify_one leaves the + // waiter parked until timeout, exposing a durable inbox stall. + } + + #[tokio::test] + async fn prompt_lock_capacity_notify_wakes_multiple_waiters() { + let st = AppState::with_payloads_and_v2(None, Default::default(), true); + let mut held = Vec::new(); + for index in 0..MAX_PROMPT_LOCKS { + let lock = match crate::lock_session(&st, &format!("ses_multi_cap_{index}")).await { + Ok(lock) => lock, + Err(error) => panic!( + "lock slot {index} was unexpectedly refused: {}", + error.message + ), + }; + held.push(lock); + } + + const WAITERS: usize = 3; + let barriers = Arc::new(( + tokio::sync::Barrier::new(WAITERS + 1), + tokio::sync::Barrier::new(WAITERS + 1), + )); + *st.v2_lock_wait_test_barriers.lock() = Some(barriers.clone()); + let mut waiters = Vec::new(); + for index in 0..WAITERS { + let waiter_st = st.clone(); + waiters.push(tokio::spawn(async move { + crate::lock_session_wait(&waiter_st, &format!("ses_multi_waiter_{index}")).await + })); + } + + // Each waiter has observed the full map and enabled its Notified future + // before any capacity is released. Notify's permits are not a queue; + // without enable(), these three releases coalesce and strand waiters. + barriers.0.wait().await; + for _ in 0..WAITERS { + drop(held.pop()); + } + barriers.1.wait().await; + + for waiter in waiters { + let acquired = tokio::time::timeout(std::time::Duration::from_secs(1), waiter) + .await + .expect("every released map slot should wake a waiter") + .unwrap(); + drop(acquired); + } + // Negative control: constructing Notified without calling enable() + // leaves all but one waiter asleep after simultaneous notifications. + } + + #[tokio::test] + async fn cancelled_pending_input_does_not_strand_a_later_prompt() { + let st = AppState::with_payloads_and_v2(None, Default::default(), true); + let sid = "ses_cancel_input_scope"; + let inbox_id = "msg_cancel_input_scope"; + insert_session(&st, sid); + ocserve_store::inbox::admit( + &st.writer, + sid, + inbox_id, + "prompt", + &json!({"text": "cancel me"}), + "queue", + 1, + ) + .unwrap(); + + let barriers = Arc::new((tokio::sync::Barrier::new(2), tokio::sync::Barrier::new(2))); + *st.v2_input_scope_test_barriers.lock() = Some(barriers.clone()); + assert!( + st.coordinators + .wake(sid, ocserve_core::coordinator::WakeScope::Input) + ); + let slot = Arc::new(st.v2_drain_slots.clone().try_acquire_owned().unwrap()); + spawn_drain(st.clone(), sid.to_string(), slot); + + barriers.0.wait().await; + assert!(ocserve_store::inbox::cancel(&st.writer, inbox_id).unwrap()); + barriers.1.wait().await; + for _ in 0..200 { + if !st.v2_drain_tasks.lock().contains_key(sid) { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + assert!(!st.v2_drain_tasks.lock().contains_key(sid)); + assert_eq!(event_seqs(&st, sid, "session.execution.started").len(), 1); + + let app = crate::router(st.clone()); + assert_eq!( + post_text(&app, sid, "prompt after inbox cancellation").await, + StatusCode::OK + ); + let started = wait_for_started(&st, sid, 2).await; + assert_eq!( + started.len(), + 2, + "a new prompt must start after an empty Input scope is cleaned up" + ); + // Negative control: omitting coordinator cleanup leaves `draining` + // set, so the later prompt is admitted but never starts a drain. + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cancel_finalization_serializes_a_new_prompt_after_interrupted_event() { + let (addr, provider_entered) = spawn_holding_provider(); + let dir = tempfile::tempdir().unwrap(); + let db = ocserve_store::writer::db_path(dir.path()); + let writer = Arc::new(ocserve_store::Writer::spawn(db.clone()).unwrap()); + let blobs = Arc::new(ocserve_store::BlobStore::new(dir.path().join("blobs")).unwrap()); + let llm = crate::LlmRegistry { + modalities: Default::default(), + endpoints: [("fake".into(), (format!("http://{addr}"), String::new()))] + .into_iter() + .collect(), + pricing: Default::default(), + limits: Default::default(), + default_model: ("fake".into(), "m".into()), + systems: Default::default(), + default_agent: "build".into(), + }; + let st = AppState::with_wiring_and_v2( + None, + Default::default(), + crate::Wires { + db, + blobs, + writer, + llm, + }, + true, + ); + let sid = "ses_cancel_finalize"; + insert_session(&st, sid); + let barriers = Arc::new((tokio::sync::Barrier::new(2), tokio::sync::Barrier::new(2))); + *st.v2_drain_terminal_test_barriers.lock() = Some(barriers.clone()); + let app = crate::router(st.clone()); + + assert_eq!( + post_text(&app, sid, "first held turn").await, + StatusCode::OK + ); + tokio::task::spawn_blocking(move || { + provider_entered + .recv_timeout(std::time::Duration::from_secs(5)) + .expect("first prompt must reach the provider") + }) + .await + .unwrap(); + + let cancel_st = st.clone(); + let cancel_sid = sid.to_string(); + let cancel_task = tokio::spawn(async move { cancel_drain(&cancel_st, &cancel_sid).await }); + tokio::time::timeout(std::time::Duration::from_secs(5), barriers.0.wait()) + .await + .expect("cancelled drain must reach terminal publication seam"); + + assert_eq!( + post_text(&app, sid, "prompt during cancel finalization").await, + StatusCode::OK + ); + let before_terminal = wait_for_started(&st, sid, 2).await.len() >= 2; + barriers.1.wait().await; + assert!( + cancel_task.await.unwrap(), + "the held first turn was interrupted" + ); + + let starts = wait_for_started(&st, sid, 2).await; + assert_eq!( + starts.len(), + 2, + "the prompt queued during cancellation must run" + ); + let interrupted = event_seqs(&st, sid, "session.execution.interrupted") + .into_iter() + .next() + .expect("the first drain must publish its interruption"); + assert!( + !before_terminal, + "a replacement drain started while the previous interruption was unpublished" + ); + assert!( + interrupted < starts[1], + "the old terminal event must precede the replacement drain start" + ); + // Negative control: removing the retiring task before awaiting it lets + // the replacement start first and this ordering assertion reds. + + assert!( + cancel_drain(&st, sid).await, + "clean up the held replacement turn" + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cancelled_drain_is_shared_by_interrupt_and_delete() { + let (addr, provider_entered) = spawn_holding_provider(); + let dir = tempfile::tempdir().unwrap(); + let db = ocserve_store::writer::db_path(dir.path()); + let writer = Arc::new(ocserve_store::Writer::spawn(db.clone()).unwrap()); + let blobs = Arc::new(ocserve_store::BlobStore::new(dir.path().join("blobs")).unwrap()); + let llm = crate::LlmRegistry { + modalities: Default::default(), + endpoints: [("fake".into(), (format!("http://{addr}"), String::new()))] + .into_iter() + .collect(), + pricing: Default::default(), + limits: Default::default(), + default_model: ("fake".into(), "m".into()), + systems: Default::default(), + default_agent: "build".into(), + }; + let st = AppState::with_wiring_and_v2( + None, + Default::default(), + crate::Wires { + db, + blobs, + writer, + llm, + }, + true, + ); + let sid = "ses_cancel_shared"; + insert_session(&st, sid); + let barriers = Arc::new((tokio::sync::Barrier::new(2), tokio::sync::Barrier::new(2))); + *st.v2_drain_terminal_test_barriers.lock() = Some(barriers.clone()); + let app = crate::router(st.clone()); + + assert_eq!( + post_text(&app, sid, "first held turn").await, + StatusCode::OK + ); + tokio::task::spawn_blocking(move || { + provider_entered + .recv_timeout(std::time::Duration::from_secs(5)) + .expect("first prompt must reach the provider") + }) + .await + .unwrap(); + + let interrupt_st = st.clone(); + let interrupt_sid = sid.to_string(); + let interrupt_task = tokio::spawn(async move { + interrupt( + axum::extract::State(interrupt_st), + axum::extract::Path(interrupt_sid), + axum::extract::Query(InterruptParams::default()), + ) + .await + }); + tokio::time::timeout(std::time::Duration::from_secs(5), barriers.0.wait()) + .await + .expect("interrupted drain must reach terminal publication seam"); + // The first request may disconnect while its drain is still retiring. + // That must not consume the only completion handle or release a later + // delete caller early. + interrupt_task.abort(); + match interrupt_task.await { + Err(error) => assert!(error.is_cancelled()), + Ok(_) => panic!("interrupted request should still be waiting for drain retirement"), + } + + let (second_cancel_entered_tx, second_cancel_entered_rx) = tokio::sync::oneshot::channel(); + *st.v2_cancel_transition_hook.lock() = Some(Box::new(move || { + let _ = second_cancel_entered_tx.send(()); + })); + let (after_cancel_tx, mut after_cancel_rx) = tokio::sync::oneshot::channel(); + *st.v2_session_remove_after_cancel_hook.lock() = Some(Box::new(move || { + let _ = after_cancel_tx.send(()); + })); + let delete_st = st.clone(); + let delete_sid = sid.to_string(); + let delete_task = tokio::spawn(async move { + session_remove_api( + axum::extract::State(delete_st), + axum::extract::Path(delete_sid), + ) + .await + }); + tokio::time::timeout(std::time::Duration::from_secs(5), second_cancel_entered_rx) + .await + .expect("delete must enter drain cancellation") + .expect("test hook sender remains available"); + + assert!( + tokio::time::timeout(std::time::Duration::from_millis(50), &mut after_cancel_rx) + .await + .is_err(), + "delete must await the shared drain retirement before deleting the session" + ); + barriers.1.wait().await; + assert_eq!( + delete_task + .await + .unwrap() + .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR), + StatusCode::NO_CONTENT + ); + assert!( + after_cancel_rx.await.is_ok(), + "delete reached its boundary after the shared wait" + ); + // Negative control: a second cancellation caller with no JoinHandle + // used to return early and reach the delete boundary before the old + // drain published its terminal event. + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn delete_fences_prompt_admitted_after_drain_retirement_check() { + let st = AppState::with_payloads_and_v2(None, Default::default(), true); + let sid = "ses_delete_admission_gap"; + insert_session(&st, sid); + let barriers = Arc::new((tokio::sync::Barrier::new(2), tokio::sync::Barrier::new(2))); + *st.v2_session_delete_gap_test_barriers.lock() = Some(barriers.clone()); + + let delete_st = st.clone(); + let delete_sid = sid.to_string(); + let delete_task = tokio::spawn(async move { + session_remove_api( + axum::extract::State(delete_st), + axum::extract::Path(delete_sid), + ) + .await + }); + tokio::time::timeout(std::time::Duration::from_secs(5), barriers.0.wait()) + .await + .expect("delete must reach the post-cancellation admission gap"); + + let app = crate::router(st.clone()); + let late_prompt = post_text(&app, sid, "prompt while deleting").await; + barriers.1.wait().await; + assert_eq!( + delete_task + .await + .unwrap() + .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR), + StatusCode::NO_CONTENT + ); + assert_eq!( + late_prompt, + StatusCode::NOT_FOUND, + "a prompt entering after the drain check must not be admitted or start a drain" + ); + // Negative control: without the admission fence, this prompt receives + // 200 and can register a fresh drain after delete's final empty check. + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn delete_fences_v1_prompt_async_after_drain_retirement_check() { + let (_dir, st, _provider_entered) = state_with_holding_provider(); + let sid = "ses_delete_v1_prompt_async"; + insert_session(&st, sid); + let barriers = Arc::new((tokio::sync::Barrier::new(2), tokio::sync::Barrier::new(2))); + *st.v2_session_delete_gap_test_barriers.lock() = Some(barriers.clone()); + + let delete_st = st.clone(); + let delete_sid = sid.to_string(); + let delete_task = tokio::spawn(async move { + session_remove_api( + axum::extract::State(delete_st), + axum::extract::Path(delete_sid), + ) + .await + }); + tokio::time::timeout(std::time::Duration::from_secs(5), barriers.0.wait()) + .await + .expect("delete must reach the post-cancellation admission gap"); + + let response = crate::router(st.clone()) + .oneshot( + Request::builder() + .method("POST") + .uri(format!("/session/{sid}/prompt_async")) + .header("content-type", "application/json") + .body(Body::from( + json!({ + "parts": [{"type": "text", "text": "late v1 prompt"}], + "model": {"providerID": "fake", "modelID": "m"}, + "agent": "build" + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + let admitted_task = st.prompt_tasks.lock().remove(sid); + let task_was_registered = admitted_task.is_some(); + if let Some((_, handle)) = admitted_task { + handle.abort(); + } + + barriers.1.wait().await; + assert_eq!( + delete_task + .await + .unwrap() + .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR), + StatusCode::NO_CONTENT + ); + assert_eq!( + response.status(), + StatusCode::NOT_FOUND, + "v1 async prompt must not be admitted during session deletion" + ); + assert!( + !task_was_registered, + "a rejected v1 prompt must not leave a background task" + ); + // Negative control: without a shared admission fence, this route + // returns 204 and registers a prompt task after delete canceled tasks. + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn delete_fences_v1_sync_message_after_drain_retirement_check() { + let (_dir, st, _provider_entered) = state_with_holding_provider(); + let sid = "ses_delete_v1_message"; + insert_session(&st, sid); + let barriers = Arc::new((tokio::sync::Barrier::new(2), tokio::sync::Barrier::new(2))); + *st.v2_session_delete_gap_test_barriers.lock() = Some(barriers.clone()); + + let delete_st = st.clone(); + let delete_sid = sid.to_string(); + let delete_task = tokio::spawn(async move { + session_remove_api( + axum::extract::State(delete_st), + axum::extract::Path(delete_sid), + ) + .await + }); + tokio::time::timeout(std::time::Duration::from_secs(5), barriers.0.wait()) + .await + .expect("delete must reach the post-cancellation admission gap"); + + let request_app = crate::router(st.clone()); + let mut request = tokio::spawn(async move { + request_app + .oneshot( + Request::builder() + .method("POST") + .uri(format!("/session/{sid}/message")) + .header("content-type", "application/json") + .body(Body::from( + json!({ + "parts": [{"type": "text", "text": "late v1 message"}], + "model": {"providerID": "fake", "modelID": "m"}, + "agent": "build" + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap() + .status() + }); + let status = tokio::time::timeout(std::time::Duration::from_millis(500), &mut request) + .await + .ok() + .and_then(Result::ok); + if status.is_none() { + request.abort(); + } + + barriers.1.wait().await; + assert_eq!( + delete_task + .await + .unwrap() + .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR), + StatusCode::NO_CONTENT + ); + assert_eq!( + status, + Some(StatusCode::NOT_FOUND), + "v1 sync message must be rejected before it can enter the provider" + ); + // Negative control: without the fence the sync request reaches the + // holding provider and cannot complete while deletion is paused. + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn delete_fences_v1_command_after_drain_retirement_check() { + let (_dir, st, provider_entered) = state_with_holding_provider(); + let sid = "ses_delete_v1_command"; + insert_session(&st, sid); + st.payloads.write().command = vec![json!({ + "name": "test-command", + "template": "late command prompt", + })]; + let barriers = Arc::new((tokio::sync::Barrier::new(2), tokio::sync::Barrier::new(2))); + *st.v2_session_delete_gap_test_barriers.lock() = Some(barriers.clone()); + + let delete_st = st.clone(); + let delete_sid = sid.to_string(); + let delete_task = tokio::spawn(async move { + session_remove_api( + axum::extract::State(delete_st), + axum::extract::Path(delete_sid), + ) + .await + }); + tokio::time::timeout(std::time::Duration::from_secs(5), barriers.0.wait()) + .await + .expect("delete must reach the post-cancellation admission gap"); + + let request_app = crate::router(st.clone()); + let mut request = tokio::spawn(async move { + request_app + .oneshot( + Request::builder() + .method("POST") + .uri(format!("/session/{sid}/command")) + .header("content-type", "application/json") + .body(Body::from( + json!({"command": "test-command", "arguments": ""}).to_string(), + )) + .unwrap(), + ) + .await + .unwrap() + }); + let response = + tokio::time::timeout(std::time::Duration::from_millis(500), &mut request).await; + let reached_provider = provider_entered + .recv_timeout(std::time::Duration::from_millis(50)) + .is_ok(); + if response.is_err() { + request.abort(); + let _ = request.await; + } + barriers.1.wait().await; + assert_eq!( + delete_task + .await + .unwrap() + .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR), + StatusCode::NO_CONTENT + ); + assert!( + response.is_ok(), + "a v1 command admitted after deletion began must be rejected before provider execution" + ); + assert!( + !reached_provider, + "a command rejected by the delete fence must not start a provider request" + ); + assert_eq!(response.unwrap().unwrap().status(), StatusCode::NOT_FOUND); + // Negative control: without a post-lock admission check, this request + // reaches the held provider and times out while deletion waits on it. + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn delete_fences_v2_session_generate_after_drain_retirement_check() { + let (_dir, st, provider_entered) = state_with_holding_provider(); + let sid = "ses_delete_v2_generate"; + insert_session(&st, sid); + let barriers = Arc::new((tokio::sync::Barrier::new(2), tokio::sync::Barrier::new(2))); + *st.v2_session_delete_gap_test_barriers.lock() = Some(barriers.clone()); + + let delete_st = st.clone(); + let delete_sid = sid.to_string(); + let delete_task = tokio::spawn(async move { + session_remove_api( + axum::extract::State(delete_st), + axum::extract::Path(delete_sid), + ) + .await + }); + tokio::time::timeout(std::time::Duration::from_secs(5), barriers.0.wait()) + .await + .expect("delete must reach the post-cancellation admission gap"); + + let request_app = crate::router(st.clone()); + let mut request = tokio::spawn(async move { + request_app + .oneshot( + Request::builder() + .method("POST") + .uri(format!("/api/session/{sid}/generate")) + .header("content-type", "application/json") + .body(Body::from(json!({"prompt": "late generate"}).to_string())) + .unwrap(), + ) + .await + .unwrap() + }); + let response = + tokio::time::timeout(std::time::Duration::from_millis(500), &mut request).await; + let reached_provider = provider_entered + .recv_timeout(std::time::Duration::from_millis(50)) + .is_ok(); + if response.is_err() { + request.abort(); + let _ = request.await; + } + barriers.1.wait().await; + assert_eq!( + delete_task + .await + .unwrap() + .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR), + StatusCode::NO_CONTENT + ); + assert!( + response.is_ok(), + "session generate admitted after deletion began must be rejected before provider execution" + ); + assert!( + !reached_provider, + "session generate rejected by the delete fence must not start a provider request" + ); + assert_eq!(response.unwrap().unwrap().status(), StatusCode::NOT_FOUND); + // Negative control: the old handler reached the provider and could + // persist its assistant response after the session rows were removed. + } +} + #[cfg(test)] mod workspace_adapter_tests { use super::workspace_adapter_refusal; diff --git a/crates/ocserve-http/tests/v2_admission.rs b/crates/ocserve-http/tests/v2_admission.rs index eebfccf..5b9ef58 100644 --- a/crates/ocserve-http/tests/v2_admission.rs +++ b/crates/ocserve-http/tests/v2_admission.rs @@ -10,6 +10,7 @@ use axum::body::Body; use axum::http::{Request, StatusCode}; use http_body_util::BodyExt; use ocserve_http::{AppState, LlmRegistry, Payloads, Wires}; +use rusqlite::OptionalExtension; use serde_json::{Value, json}; use std::io::Read as _; use std::sync::Arc; @@ -67,6 +68,28 @@ fn spawn_stub_provider() -> std::net::SocketAddr { addr } +fn spawn_holding_provider() -> (std::net::SocketAddr, std::sync::mpsc::Receiver<()>) { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let (entered_tx, entered_rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + for sock in listener.incoming().flatten() { + let entered_tx = entered_tx.clone(); + std::thread::spawn(move || { + let mut sock = sock; + let mut buf = [0u8; 4096]; + if sock.read(&mut buf).unwrap_or(0) > 0 { + let _ = entered_tx.send(()); + // Hold the provider response open until the request future + // is cancelled and closes its socket. + while sock.read(&mut buf).unwrap_or(0) > 0 {} + } + }); + } + }); + (addr, entered_rx) +} + fn setup_with_v2(addr: std::net::SocketAddr, v2: bool) -> Arc { let dir = std::env::temp_dir().join(format!( "ocserve-v2adm-{}-{}", @@ -145,9 +168,23 @@ async fn wait_drain(st: &Arc) { let inbox_empty = ocserve_store::inbox::list(&st.db, "ses_v2a") .map(|v| v.is_empty()) .unwrap_or(false); - if inbox_empty && !ocserve_store::claim::held(&st.db, "ses_v2a").unwrap_or(true) { - // Give the turn body a beat to land its message rows. - tokio::time::sleep(std::time::Duration::from_millis(200)).await; + let terminal_published = ocserve_store::pragma::open_reader(&st.db) + .and_then(|conn| { + let count = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM event WHERE session_id='ses_v2a' \ + AND type IN ('session.execution.succeeded@v1', \ + 'session.execution.failed@v1', \ + 'session.execution.interrupted@v1'))", + [], + |row| row.get::<_, i64>(0), + )?; + Ok(count > 0) + }) + .unwrap_or(false); + if inbox_empty + && terminal_published + && !ocserve_store::claim::held(&st.db, "ses_v2a").unwrap_or(true) + { return; } tokio::time::sleep(std::time::Duration::from_millis(25)).await; @@ -1388,6 +1425,384 @@ async fn v2_drain_runs_turn_and_releases_claim() { // CONTROL: drain that skips claim release → held stays true, red. } +#[tokio::test] +async fn v2_drain_reports_provider_failure_as_terminal_failure() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + drop(listener); + let st = setup_with_v2(addr, true); + let app = ocserve_http::router(st.clone()); + + let (status, _) = post_prompt(&app, json!({"text": "hello v2"})).await; + assert_eq!(status, StatusCode::OK); + wait_drain(&st).await; + + assert!(!ocserve_store::claim::held(&st.db, "ses_v2a").unwrap()); + let conn = ocserve_store::pragma::open_reader(&st.db).unwrap(); + let failed_payload: Option = conn + .query_row( + "SELECT payload FROM event WHERE session_id='ses_v2a' \ + AND type IN ('session.execution.failed', 'session.execution.failed@v1')", + [], + |r| r.get(0), + ) + .optional() + .unwrap(); + let succeeded: i64 = conn + .query_row( + "SELECT count(*) FROM event WHERE session_id='ses_v2a' \ + AND type IN ('session.execution.succeeded', 'session.execution.succeeded@v1')", + [], + |r| r.get(0), + ) + .unwrap(); + let failed_payload = failed_payload.expect("failed drain must emit a terminal failure event"); + let props: Value = serde_json::from_str(&failed_payload).unwrap(); + assert_eq!(props["sessionID"], "ses_v2a"); + assert!( + props["error"].is_object(), + "failure event needs its error: {props}" + ); + assert_eq!(succeeded, 0, "a failed drain must not also report success"); + // V2 consumers receive the structured terminal error on + // session.execution.failed; this direct provider failure does not emit + // the legacy session.error event used by the v1 prompt surface. + + // A failed terminal turn must release the coordinator so a later prompt + // can register and run a fresh drain instead of leaving `draining=true`. + let (status, _) = post_prompt(&app, json!({"text": "prompt after provider failure"})).await; + assert_eq!(status, StatusCode::OK); + let mut attempts = 0; + let failed_again = loop { + let conn = ocserve_store::pragma::open_reader(&st.db).unwrap(); + let failed_again: i64 = conn + .query_row( + "SELECT count(*) FROM event WHERE session_id='ses_v2a' \ + AND type IN ('session.execution.failed', 'session.execution.failed@v1')", + [], + |r| r.get(0), + ) + .unwrap(); + attempts += 1; + if failed_again == 2 || attempts == 80 { + break failed_again; + } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + }; + assert_eq!(failed_again, 2, "a later prompt must start a second drain"); + assert!(!ocserve_store::claim::held(&st.db, "ses_v2a").unwrap()); +} + +#[tokio::test] +async fn v2_interrupt_cancels_registered_drain_and_reports_status() { + let (addr, provider_entered) = spawn_holding_provider(); + let st = setup_with_v2(addr, true); + let app = ocserve_http::router(st.clone()); + + let (status, admitted) = post_prompt(&app, json!({"text": "hold this turn"})).await; + assert_eq!(status, StatusCode::OK); + assert!(admitted["data"]["id"].as_str().unwrap().starts_with("msg_")); + tokio::task::spawn_blocking(move || { + provider_entered + .recv_timeout(std::time::Duration::from_secs(5)) + .expect("drain must reach the held provider") + }) + .await + .unwrap(); + let response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/api/session/ses_v2a/interrupt?resume=false") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = response.into_body().collect().await.unwrap().to_bytes(); + let body: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(body, json!({"interrupted": true})); + assert!(!ocserve_store::claim::held(&st.db, "ses_v2a").unwrap()); + let conn = ocserve_store::pragma::open_reader(&st.db).unwrap(); + let interrupted: i64 = conn + .query_row( + "SELECT count(*) FROM event WHERE session_id='ses_v2a' \ + AND type IN ('session.execution.interrupted', 'session.execution.interrupted@v1')", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!( + interrupted, 1, + "one terminal interruption event is required" + ); + + let idle = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/api/session/ses_v2a/interrupt?resume=false") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(idle.status(), StatusCode::OK); + let body = idle.into_body().collect().await.unwrap().to_bytes(); + assert_eq!( + serde_json::from_slice::(&body).unwrap(), + json!({"interrupted": false}) + ); + + // The interrupted turn releases the shared prompt lock so a later prompt + // can enter the session normally. + if let Some(lock) = st.prompt_locks.lock().get("ses_v2a").cloned() { + let guard = lock.try_lock(); + assert!( + guard.is_ok(), + "interrupted drain must release the session lock" + ); + drop(guard); + } +} + +#[tokio::test] +async fn v2_interrupt_waiting_for_session_lock_reports_not_interrupted() { + let addr = spawn_stub_provider(); + let st = setup_with_v2(addr, true); + let app = ocserve_http::router(st.clone()); + let session_lock = st + .prompt_locks + .lock() + .entry("ses_v2a".to_string()) + .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(()))) + .clone(); + let guard = session_lock.lock().await; + + let (status, _) = post_prompt(&app, json!({"text": "wait for the session lock"})).await; + assert_eq!(status, StatusCode::OK); + for _ in 0..100 { + if st.v2_drain_tasks.lock().contains_key("ses_v2a") { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + assert!(st.v2_drain_tasks.lock().contains_key("ses_v2a")); + + let response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/api/session/ses_v2a/interrupt?resume=false") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = response.into_body().collect().await.unwrap().to_bytes(); + assert_eq!( + serde_json::from_slice::(&body).unwrap(), + json!({"interrupted": false}), + "a drain that never acquired the session lock did not interrupt execution" + ); + + let conn = ocserve_store::pragma::open_reader(&st.db).unwrap(); + let starts: i64 = conn + .query_row( + "SELECT count(*) FROM event WHERE session_id='ses_v2a' \ + AND type IN ('session.execution.started', 'session.execution.started@v1')", + [], + |row| row.get(0), + ) + .unwrap(); + let interruptions: i64 = conn + .query_row( + "SELECT count(*) FROM event WHERE session_id='ses_v2a' \ + AND type IN ('session.execution.interrupted', 'session.execution.interrupted@v1')", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(starts, 0); + assert_eq!(interruptions, 0); + drop(guard); +} + +#[tokio::test] +async fn v2_drain_capacity_rejects_before_inbox_admission() { + let addr = spawn_stub_provider(); + let st = setup_with_v2(addr, true); + let app = ocserve_http::router(st.clone()); + let mut held_slots = Vec::with_capacity(ocserve_http::MAX_V2_DRAIN_TASKS); + for _ in 0..ocserve_http::MAX_V2_DRAIN_TASKS { + held_slots.push(st.v2_drain_slots.clone().try_acquire_owned().unwrap()); + } + + let (status, body) = post_prompt( + &app, + json!({"id": "msg_capacity_probe0000001", "text": "must not be admitted"}), + ) + .await; + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(body["_tag"], "ServiceUnavailableError"); + assert!( + ocserve_store::inbox::list(&st.db, "ses_v2a") + .unwrap() + .is_empty(), + "over-cap request must not leave durable work with no drain slot" + ); + let conn = ocserve_store::pragma::open_reader(&st.db).unwrap(); + let admitted_events: i64 = conn + .query_row( + "SELECT count(*) FROM event WHERE session_id='ses_v2a' \ + AND type IN ('session.inbox.enqueued', 'session.inbox.enqueued@v1') \ + AND json_extract(payload, '$.inboxID')='msg_capacity_probe0000001'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!( + admitted_events, 0, + "capacity rejection must precede the event" + ); + drop(held_slots); + // Negative control: bypassing the reservation admits the inbox row even + // though the registry has no slot, so this assertion is red. +} + +#[tokio::test] +async fn v2_interrupt_resume_rejects_when_drain_waiter_cap_is_full() { + let addr = spawn_stub_provider(); + let st = setup_with_v2(addr, true); + let app = ocserve_http::router(st.clone()); + let held_slots: Vec<_> = (0..ocserve_http::MAX_V2_DRAIN_TASKS) + .map(|_| st.v2_drain_slots.clone().try_acquire_owned().unwrap()) + .collect(); + let held_waiters: Vec<_> = (0..ocserve_http::MAX_V2_DRAIN_WAITERS) + .map(|_| st.v2_drain_waiters.clone().try_acquire_owned().unwrap()) + .collect(); + + let (status, _) = post_prompt( + &app, + json!({ + "id": "msg_interrupt_capacity0001", + "text": "park until resume", + "resume": false + }), + ) + .await; + assert_eq!(status, StatusCode::OK); + + let response = tokio::time::timeout( + std::time::Duration::from_millis(250), + app.clone().oneshot( + Request::builder() + .method("POST") + .uri("/api/session/ses_v2a/interrupt?resume=true") + .body(Body::empty()) + .unwrap(), + ), + ) + .await + .expect("a full waiter cap must return bounded backpressure") + .unwrap(); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + assert!(!ocserve_store::claim::held(&st.db, "ses_v2a").unwrap()); + assert_eq!( + ocserve_store::inbox::list(&st.db, "ses_v2a").unwrap().len(), + 1 + ); + drop(held_slots); + drop(held_waiters); + // Negative control: omitting the bounded waiter reservation parks this + // request past the timeout despite all waiter permits being held. +} + +#[tokio::test] +async fn v2_interrupt_resume_runs_steer_and_leaves_queued_prompt_parked() { + let addr = spawn_stub_provider(); + let st = setup_with_v2(addr, true); + let app = ocserve_http::router(st.clone()); + + let (queued_status, _) = post_prompt( + &app, + json!({ + "id": "msg_interrupt_queue00000001", + "text": "keep parked", + "delivery": "queue", + "resume": false + }), + ) + .await; + let (steer_status, _) = post_prompt( + &app, + json!({ + "id": "msg_interrupt_steer0000001", + "text": "resume this steer", + "delivery": "steer", + "resume": false + }), + ) + .await; + assert_eq!( + (queued_status, steer_status), + (StatusCode::OK, StatusCode::OK) + ); + + // An idle interrupt reports false but its default resume=true still runs + // steer/control work. The queued prompt must remain parked. + let response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/api/session/ses_v2a/interrupt") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = response.into_body().collect().await.unwrap().to_bytes(); + assert_eq!( + serde_json::from_slice::(&body).unwrap(), + json!({"interrupted": false}) + ); + + for _ in 0..200 { + let pending = ocserve_store::inbox::list(&st.db, "ses_v2a").unwrap(); + if pending.len() == 1 + && pending[0].0 == "msg_interrupt_queue00000001" + && !ocserve_store::claim::held(&st.db, "ses_v2a").unwrap() + { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + let pending = ocserve_store::inbox::list(&st.db, "ses_v2a").unwrap(); + assert_eq!(pending.len(), 1, "resume must leave queued prompts parked"); + assert_eq!(pending[0].0, "msg_interrupt_queue00000001"); + assert_eq!(pending[0].3, "queue"); + assert!(!ocserve_store::claim::held(&st.db, "ses_v2a").unwrap()); + + let conn = ocserve_store::pragma::open_reader(&st.db).unwrap(); + let assistants: i64 = conn + .query_row( + "SELECT count(*) FROM msg WHERE session_id='ses_v2a' AND role='assistant'", + [], + |row| row.get(0), + ) + .unwrap(); + assert!( + assistants > 0, + "resume=true must run the pending steer item" + ); +} // --------------------------------------------------------------------------- // Gap closure e2e: environment PUT → turn injection → provider cleanliness. // --------------------------------------------------------------------------- @@ -1869,3 +2284,59 @@ async fn v2_prompt_hook_mutates_the_admitted_row() { rows[0].2 ); } + +/// DELETE must not hold its deletion fence indefinitely on a drain that has +/// registered but never reports completion. This reproduces the worst case +/// directly: a registry entry whose completion channel is never written, +/// which is what a drain stuck past the provider watchdog would look like. +/// Without the bounded wait, `cancel_drains_for_delete` loops forever and the +/// request never returns. +#[tokio::test] +async fn delete_returns_when_a_registered_drain_never_confirms_cancellation() { + let addr = spawn_stub_provider(); + let st = setup_with_v2(addr, true); + let app = ocserve_http::router(st.clone()); + + // One real session row so the delete reaches the drain-cancel stage. + let (status, _) = post_prompt(&app, json!({"text": "seed"})).await; + assert_eq!(status, StatusCode::OK); + wait_drain(&st).await; + + // Fabricate the stalled entry: a task that never sends completion. + let (_completion_tx, completion_rx) = tokio::sync::watch::channel(None); + let (cancel_tx, _cancel_rx) = tokio::sync::watch::channel(false); + let slot = std::sync::Arc::new(st.v2_drain_slots.clone().try_acquire_owned().unwrap()); + let handle = tokio::spawn(async { std::future::pending::().await }); + st.v2_drain_tasks.lock().insert( + "ses_v2a".to_string(), + ocserve_http::V2DrainTask { + generation: u64::MAX, + cancel: cancel_tx, + handle, + completion: completion_rx, + slot, + rerun: false, + }, + ); + + let started = std::time::Instant::now(); + let response = tokio::time::timeout( + std::time::Duration::from_secs(90), + app.clone().oneshot( + Request::builder() + .method("DELETE") + .uri("/session/ses_v2a") + .body(Body::empty()) + .unwrap(), + ), + ) + .await + .expect("DELETE must not block forever on a non-confirming drain") + .unwrap(); + let elapsed = started.elapsed(); + assert_eq!(response.status(), StatusCode::OK); + assert!( + elapsed < std::time::Duration::from_secs(60), + "bounded wait (30 s default, 30 s here) took {elapsed:?}" + ); +} diff --git a/docs/ENVIRONMENT.md b/docs/ENVIRONMENT.md index 741bc95..353e5e5 100644 --- a/docs/ENVIRONMENT.md +++ b/docs/ENVIRONMENT.md @@ -78,6 +78,7 @@ for the normal behaviour. | Variable | Default | Accepted | Meaning | |---|---|---|---| +| `OCSERVE_DRAIN_CANCEL_SECS` | 30 | seconds (0 = unbounded) | How long DELETE waits for a cancelled v2 drain to confirm before proceeding with teardown. 0 restores the old unbounded wait, which can hold a deletion fence for the provider stall budget. | | `OCSERVE_CGROUP_PARTITION` | auto (only under systemd) | 0 \| 1 \| unset | Kids/ cgroup partition for child processes. | | `OCSERVE_KIDS_CGROUP` | (internal) | a cgroup directory | Set by the partition dance for child processes; do not set it yourself. | | `OCSERVE_KIDS_MEMORY_MAX_MB` | 700 | 32-1048576 | Aggregate memory cap for the kids/ cgroup. |