From d995e6d9351d17c9502478e67b9f45099031ff35 Mon Sep 17 00:00:00 2001 From: "J.R. Murray" Date: Fri, 13 Nov 2026 01:40:34 -0500 Subject: [PATCH 1/4] fix(v2): cancel active drains and fence session deletes Keep registered drains cancellable through claim release and terminal publication, and fence session deletion against new work. Bound drains, reservation waiters, prompt-lock waiters, and delete fences. Closes K-V2-DRAIN-CANCEL and K-SESSION-DELETE-FENCE (V2-PLAN P3). Focused v2 admission (22), drain and delete races (15), coordinator (5), fmt, clippy, guards, and matrix passed. Full workspace tests have macOS host failures in Linux cgroup and /proc assumptions; the ocserve-http suite also has an unrelated symlink-containment failure on this host. --- CHANGELOG.md | 8 + MEMORY.md | 13 + PLAN.md | 1 + TESTING.md | 2 + TRACEABILITY.md | 3 + V2-PLAN.md | 23 + crates/ocserve-core/src/coordinator.rs | 43 +- crates/ocserve-http/src/goal_loop.rs | 7 + crates/ocserve-http/src/lib.rs | 268 +++- crates/ocserve-http/src/scheduler.rs | 35 +- crates/ocserve-http/src/v2.rs | 1633 ++++++++++++++++++++- crates/ocserve-http/tests/v2_admission.rs | 421 +++++- 12 files changed, 2330 insertions(+), 127 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 44bb90d3..7f580963 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,14 @@ All notable changes to ocserve. The wire contract targets the frozen opencode 1.19.5 release and the v2 surface the 2.2.5 client; anything that differs is a named `D-*` row in the ledger at `PLAN.md` section 18. +## Unreleased + +### Fixed +- 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. + ## 0.7.1 - 2026-11-10 ### Added diff --git a/MEMORY.md b/MEMORY.md index 74dbad90..b8c3d2fa 100644 --- a/MEMORY.md +++ b/MEMORY.md @@ -107,6 +107,19 @@ 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`). +- 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 f0fca14b..5db3f066 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) | diff --git a/TESTING.md b/TESTING.md index 44f17938..4da3558e 100644 --- a/TESTING.md +++ b/TESTING.md @@ -231,6 +231,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 | | 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 | ## 3. Test levels (ISTQB) mapped to ocserve diff --git a/TRACEABILITY.md b/TRACEABILITY.md index 1748264a..8b2ee450 100644 --- a/TRACEABILITY.md +++ b/TRACEABILITY.md @@ -260,6 +260,9 @@ 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 | ## Test design techniques coverage (TESTING §4) diff --git a/V2-PLAN.md b/V2-PLAN.md index 143fd612..aaac8f7b 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 7fb0d978..e46069f8 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-http/src/goal_loop.rs b/crates/ocserve-http/src/goal_loop.rs index f6b23a92..19e1fa96 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) { + 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 f49c8ce2..e8bab5cf 100644 --- a/crates/ocserve-http/src/lib.rs +++ b/crates/ocserve-http/src/lib.rs @@ -270,6 +270,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 +369,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 +401,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 +788,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 +802,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, @@ -2632,6 +2725,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)?; + } // C5 retrofit (M6, COMPACTION.md §3): manual summarize = v1 source // (handlers/session.ts:273-293 at the FREEZE TAG v1.18.31, verified): @@ -3870,6 +3967,7 @@ struct LockRelease { >, >, ages: std::sync::Arc>>, + available: std::sync::Arc, sid: String, lock: std::sync::Arc>, } @@ -3892,6 +3990,7 @@ impl Drop for LockRelease { { map.remove(&self.sid); self.ages.lock().remove(&self.sid); + self.available.notify_one(); } } } @@ -3904,7 +4003,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", @@ -3927,18 +4026,57 @@ 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, + Err(error) if error.message == "too many concurrent prompts" => { + #[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 @@ -3964,16 +4102,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, + ); } } @@ -3990,6 +4138,21 @@ 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. +fn check_v1_session_admission(st: &AppState, sid: &str) -> Result<(), ApiError> { + if crate::v2::session_is_deleting(st, sid) + || !ocserve_store::session_exists(&st.db, sid).map_err(|e| ApiError { + status: StatusCode::INTERNAL_SERVER_ERROR, + name: "InternalError", + message: format!("{e:#}"), + })? + { + 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, @@ -4001,6 +4164,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)?; + } let writer = st.writer.clone(); let result = ocserve_core::prompt::run_prompt(&ctx, &writer, &id, &req) .await @@ -4024,23 +4191,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)?; } 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)?; let writer = st.writer.clone(); let sid = id.clone(); let generation = st @@ -4259,17 +4420,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:#}"), @@ -4299,6 +4468,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}"); @@ -5104,7 +5274,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)?; + 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 @@ -5212,6 +5387,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)?; + } let writer = st.writer.clone(); let (info, parts) = ocserve_core::prompt::run_prompt_with( &ctx, @@ -5262,6 +5441,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)?; + } let dir = st.paths["directory"].as_str().unwrap_or("/").to_string(); // busy → run → idle (client spinner parity) @@ -6971,9 +7154,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 @@ -7010,17 +7193,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 f88ace9c..209c70ee 100644 --- a/crates/ocserve-http/src/scheduler.rs +++ b/crates/ocserve-http/src/scheduler.rs @@ -167,8 +167,19 @@ 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); + let slot = crate::v2::reserve_drain_slot_wait(st, sid) + .await + .map_err(|_| anyhow::anyhow!("v2 drain capacity reservation failed"))?; + 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 +223,13 @@ async fn step(st: &Arc, row: &Value, now: i64) -> anyhow::Result<()> { } }, }); + let drain_slot = crate::v2::reserve_drain_slot_wait(st, sid) + .await + .map_err(|_| anyhow::anyhow!("v2 drain capacity reservation failed"))?; + 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 +239,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 2097d660..30596b96 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 })], @@ -3878,11 +3908,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, @@ -3923,13 +3960,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) @@ -3976,12 +4010,416 @@ 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; + }; + loop { + if let Some(interrupted) = *completion_rx.borrow() { + return interrupted; + } + if completion_rx.changed().await.is_err() { + tracing::warn!("v2 drain completion signal dropped for {sid}"); + return false; + } + } +} + +/// 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 { + let mut interrupted = false; + loop { + if !st.v2_drain_tasks.lock().contains_key(sid) { + return interrupted; + } + 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() @@ -4002,38 +4440,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 @@ -4043,21 +4505,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 @@ -4936,26 +5463,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)), @@ -4963,32 +5494,83 @@ 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; + if !session_is_deleting(&st, &id) + && ocserve_store::session_exists(&st.db, &id).unwrap_or(false) + && 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) => { @@ -4996,13 +5578,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); @@ -8094,6 +8673,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 eebfccf9..d976f9c8 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. // --------------------------------------------------------------------------- From f8713d8c39c835cd3ce5242e2229f1cd6c098ab5 Mon Sep 17 00:00:00 2001 From: alderpath Date: Sat, 14 Nov 2026 02:51:23 +0000 Subject: [PATCH 2/4] fix(v2): take store IO off async workers and bound drain teardown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Amendment to the contributor's branch (credited @jrzmurray), from the review. Three members of one class (blocking work on an executor that must stay free), plus two waits with no ceiling: * check_v1_session_admission ran session_exists (open_reader + SELECT) inline on EVERY v1 prompt — /message, /prompt_async, /command, /summarize, and the v2 create. It had also REPLACED the run_blocking call prompt_async already had, so the change was a net loss of offload. It is now async and goes through run_blocking. (F1 class, PERF-10X.) * lock_session_wait branched on the error MESSAGE string ("too many concurrent prompts") to detect capacity. A wording change would silently fall into the 1 Hz retry branch while holding a drain slot. Now matched on status + name. * scheduler::step awaited reserve_drain_slot_wait inside the single serial tick loop, so one slow drain stalled every other schedule. It now uses the non-blocking reservation and returns without marking the row, so the next pass retries. * cancel_drain bounded its watch wait; cancel_drains_for_delete bounded the WHOLE teardown. The first attempt bounded only the inner loop, and the outer re-cancel loop still spun forever — the regression test caught it. Both record ocserve_v2_drain_cancel_timeout_total. * the interrupt re-arm read session_exists(...).unwrap_or(false), so a read error was treated as "session gone" and silently dropped the work. It propagates. Docs: MEMORY states the drain cap is a COUNT bound and gives the real worst case (min(128 slots, 64 prompt locks) x history); PLAN gains D-V2-INTERRUPT-STATUS (one handler cannot return v1 204 and v2 200); TESTING and TRACEABILITY gain the class and its negative control. Verified: 929 workspace tests, clippy -D under CARGO_BUILD_WARNINGS=deny, guards, matrix, docs. Negative control for the delete bound: delete the outer deadline and the request hangs to the 90 s test timeout. --- MEMORY.md | 6 ++- PLAN.md | 1 + TESTING.md | 1 + TRACEABILITY.md | 1 + crates/ocserve-http/src/goal_loop.rs | 2 +- crates/ocserve-http/src/lib.rs | 62 +++++++++++++++++------ crates/ocserve-http/src/scheduler.rs | 26 +++++++--- crates/ocserve-http/src/v2.rs | 56 ++++++++++++++++++-- crates/ocserve-http/tests/v2_admission.rs | 56 ++++++++++++++++++++ 9 files changed, 183 insertions(+), 28 deletions(-) diff --git a/MEMORY.md b/MEMORY.md index b8c3d2fa..3195c47c 100644 --- a/MEMORY.md +++ b/MEMORY.md @@ -109,7 +109,11 @@ tokio::runtime::Builder::new_multi_thread() 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`). + 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. diff --git a/PLAN.md b/PLAN.md index 5db3f066..a628f9fa 100644 --- a/PLAN.md +++ b/PLAN.md @@ -616,6 +616,7 @@ current anywhere. | `D-WIN-WAL-NETWORK` | A data directory on a network filesystem is refused at boot rather than degraded. | accepted | | `D-MACOS-TRIM` | `trim_heap()` is a no-op off glibc unless measurement justifies otherwise. | open | | `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-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 4da3558e..d0fba999 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). | diff --git a/TRACEABILITY.md b/TRACEABILITY.md index d33d84b3..2b297278 100644 --- a/TRACEABILITY.md +++ b/TRACEABILITY.md @@ -266,6 +266,7 @@ no-inflation correct), PASS cost (−58.9% weighted) + score (median 1.0 both) | 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-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 | green | ## Test design techniques coverage (TESTING §4) | Technique | Applied | Where | diff --git a/crates/ocserve-http/src/goal_loop.rs b/crates/ocserve-http/src/goal_loop.rs index 19e1fa96..cba3a277 100644 --- a/crates/ocserve-http/src/goal_loop.rs +++ b/crates/ocserve-http/src/goal_loop.rs @@ -474,7 +474,7 @@ 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) { + if let Err(e) = crate::check_v1_session_admission(st, sid).await { drop(guard); return Err(anyhow::anyhow!("{}", e.message)); } diff --git a/crates/ocserve-http/src/lib.rs b/crates/ocserve-http/src/lib.rs index e8bab5cf..a225be67 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; @@ -2727,7 +2739,7 @@ async fn post_summarize( }; { let _admission = st.v2_session_admission_gate.read().await; - check_v1_session_admission(&st, &sid)?; + check_v1_session_admission(&st, &sid).await?; } // C5 retrofit (M6, COMPACTION.md §3): manual summarize = v1 source @@ -4045,7 +4057,14 @@ pub(crate) async fn lock_session_wait(st: &Arc, sid: &str) -> LockRele available.as_mut().enable(); match lock_session(st, sid).await { Ok(release) => return release, - Err(error) if error.message == "too many concurrent prompts" => { + // 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)] @@ -4140,14 +4159,25 @@ 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. -fn check_v1_session_admission(st: &AppState, sid: &str) -> Result<(), ApiError> { - if crate::v2::session_is_deleting(st, sid) - || !ocserve_store::session_exists(&st.db, sid).map_err(|e| ApiError { - status: StatusCode::INTERNAL_SERVER_ERROR, - name: "InternalError", - message: format!("{e:#}"), - })? - { +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(()) @@ -4166,7 +4196,7 @@ async fn post_message( let _guard = _release.arc().lock().await; { let _admission = st.v2_session_admission_gate.read().await; - check_v1_session_admission(&st, &id)?; + check_v1_session_admission(&st, &id).await?; } let writer = st.writer.clone(); let result = ocserve_core::prompt::run_prompt(&ctx, &writer, &id, &req) @@ -4193,7 +4223,7 @@ async fn post_prompt_async( let req = decode_v1(ocserve_core::wire::PromptRequest::decode(&payload))?; { let _admission = st.v2_session_admission_gate.read().await; - check_v1_session_admission(&st, &id)?; + 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 @@ -4201,7 +4231,7 @@ async fn post_prompt_async( // 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)?; + check_v1_session_admission(&st, &id).await?; let writer = st.writer.clone(); let sid = id.clone(); let generation = st @@ -5276,7 +5306,7 @@ async fn post_command( if is_builtin_goal { let arm = { let _admission = st.v2_session_admission_gate.read().await; - check_v1_session_admission(&st, &id)?; + check_v1_session_admission(&st, &id).await?; crate::goal_loop::arm(&st, &id, arguments).await }; return match arm { @@ -5389,7 +5419,7 @@ async fn post_command( }; { let _admission = st.v2_session_admission_gate.read().await; - check_v1_session_admission(&st, &id)?; + check_v1_session_admission(&st, &id).await?; } let writer = st.writer.clone(); let (info, parts) = ocserve_core::prompt::run_prompt_with( @@ -5443,7 +5473,7 @@ async fn post_shell( }; { let _admission = st.v2_session_admission_gate.read().await; - check_v1_session_admission(&st, &id)?; + check_v1_session_admission(&st, &id).await?; } let dir = st.paths["directory"].as_str().unwrap_or("/").to_string(); diff --git a/crates/ocserve-http/src/scheduler.rs b/crates/ocserve-http/src/scheduler.rs index 209c70ee..dc821504 100644 --- a/crates/ocserve-http/src/scheduler.rs +++ b/crates/ocserve-http/src/scheduler.rs @@ -167,9 +167,18 @@ async fn step(st: &Arc, row: &Value, now: i64) -> anyhow::Result<()> { && !prev.is_empty() && ocserve_store::inbox::get(&st.db, prev)?.is_some() { - let slot = crate::v2::reserve_drain_slot_wait(st, sid) - .await - .map_err(|_| anyhow::anyhow!("v2 drain capacity reservation failed"))?; + // 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(()); @@ -223,9 +232,14 @@ async fn step(st: &Arc, row: &Value, now: i64) -> anyhow::Result<()> { } }, }); - let drain_slot = crate::v2::reserve_drain_slot_wait(st, sid) - .await - .map_err(|_| anyhow::anyhow!("v2 drain capacity reservation failed"))?; + 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(()); diff --git a/crates/ocserve-http/src/v2.rs b/crates/ocserve-http/src/v2.rs index 30596b96..6393d54c 100644 --- a/crates/ocserve-http/src/v2.rs +++ b/crates/ocserve-http/src/v2.rs @@ -4258,13 +4258,35 @@ pub(crate) async fn cancel_drain(st: &Arc, sid: &str) -> bool { 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; } - if completion_rx.changed().await.is_err() { - tracing::warn!("v2 drain completion signal dropped for {sid}"); - return false; + 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; + } } } } @@ -4272,11 +4294,31 @@ pub(crate) async fn cancel_drain(st: &Arc, sid: &str) -> bool { /// 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; } } @@ -5533,8 +5575,14 @@ pub async fn interrupt( 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).unwrap_or(false) + && 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, diff --git a/crates/ocserve-http/tests/v2_admission.rs b/crates/ocserve-http/tests/v2_admission.rs index d976f9c8..5b9ef58c 100644 --- a/crates/ocserve-http/tests/v2_admission.rs +++ b/crates/ocserve-http/tests/v2_admission.rs @@ -2284,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:?}" + ); +} From ae51d39b0fcb2e2bebd9c287f3c1c34e6763e925 Mon Sep 17 00:00:00 2001 From: alderpath Date: Sat, 14 Nov 2026 05:50:13 +0000 Subject: [PATCH 3/4] chore(env): contract OCSERVE_DRAIN_CANCEL_SECS (rule 58) The new knob was read in code but absent from the env contract, so the guard that exists to catch exactly this failed CI rather than the build passing with an undocumented knob. Group is 'Cgroups and lifecycle' (drain teardown), default 30 s, 0 = the old unbounded wait. docs/ENVIRONMENT.md regenerated. Already covered by K-OFFLOAD-ADMISSION in TRACEABILITY.md; this is the documentation half of the same change. --- crates/ocserve-core/src/env_contract.rs | 9 +++++++++ docs/ENVIRONMENT.md | 1 + 2 files changed, 10 insertions(+) diff --git a/crates/ocserve-core/src/env_contract.rs b/crates/ocserve-core/src/env_contract.rs index 0f80febf..551c4c40 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/docs/ENVIRONMENT.md b/docs/ENVIRONMENT.md index 741bc95d..353e5e50 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. | From 8b54fe97c57403dba4805dfa09b5ac675a65276f Mon Sep 17 00:00:00 2001 From: alderpath Date: Sat, 14 Nov 2026 06:01:45 +0000 Subject: [PATCH 4/4] docs(traceability): record that K-OFFLOAD-ADMISSION is a class fix, not a latency delta The point lookup measures 2.1 us, so no p50 improvement is claimed; the evidence is behavioural (the delete regression test). Recording this stops a future reader from inferring a perf win that was never measured. --- TRACEABILITY.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/TRACEABILITY.md b/TRACEABILITY.md index 2b297278..fe7966aa 100644 --- a/TRACEABILITY.md +++ b/TRACEABILITY.md @@ -266,7 +266,7 @@ no-inflation correct), PASS cost (−58.9% weighted) + score (median 1.0 both) | 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-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 | 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 |