diff --git a/docs/architecture.md b/docs/architecture.md index 4d7cdfb..48af316 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -29,10 +29,22 @@ flowchart LR ### The server owns history -The event log is the source of truth for a session. Before each model turn the -application reads committed events and projects them into a Messages API -conversation. The model endpoint performs inference; it does not own session -state. +The event log is the source of truth for a session, but two different orderings +are read from it: + +- **Public event history** (`GET .../events`, list, and the live SSE stream) is + the immutable receipt/commit sequence. It never reorders or hides events. +- **Model-facing conversation order** is reconstructed per turn from run + causality, not from raw commit order. For each prior completed run, in + admission order, the projection replays that run's trigger event IDs followed + by its persisted output event IDs, then appends the current run's trigger. The + output association is durable state on the run, so this ordering is rebuilt + identically after a restart, and a run never sees a later trigger that was + queued while it was still running. + +Before each model turn the application reconstructs that causal history and +projects it into a Messages API conversation. The model endpoint performs +inference; it does not own session state. ### Wire and domain models are separate @@ -75,9 +87,9 @@ or sandbox dependencies. ## Durable write path Submitting input is not “write an event, then enqueue work.” The store commits -the client events, `session.status_running`, session projection, and queued run -in one transaction. A crash therefore cannot leave accepted input without -corresponding work. +the client events, `session.status_running`, session projection, and one queued +run per processable trigger (in admission order) in one transaction. A crash +therefore cannot leave accepted input without corresponding work. Runtime calls happen outside SQL transactions. At run completion, buffered authoritative output, trigger `processed_at`, the final session status, and run @@ -108,20 +120,21 @@ ownership and duplicate-side-effect risks. The strongest current risks are semantic rather than structural: -1. A batch of multiple triggering input events is admitted as one run, while - each trigger currently reprojects only committed history. Output buffered by - an earlier trigger in the same claim is not visible to the next trigger. -2. Runtime output is buffered until completion even though tools may already +1. Runtime output is buffered until completion even though tools may already have performed side effects. A process crash can therefore repeat a side effect without a durable journal of the prior attempt. -3. Pending client actions are encoded in events and stop reasons rather than a - first-class durable `pending_actions` model. -4. Sandboxes are session-scoped: a session's logical sandbox is provisioned on +2. Pending client actions are encoded in events and stop reasons rather than a + first-class durable `pending_actions` model. A single custom-tool park and + its `user.custom_tool_result` resume are implemented and tested, but there is + no durable pending-action gate: if other trigger runs were already queued + before a run parks with `requires_action`, their gating and correlation + against the parked action are not yet fully modeled. +3. Sandboxes are session-scoped: a session's logical sandbox is provisioned on first tool use, reused across its runs, and released on session deletion. The manager is in-memory, so a process restart does not restore an idle session's workspace, and there is no durable checkpoint, quota, or eviction policy yet. -5. `SessionService` currently combines session CRUD, admission, dispatch, and +4. `SessionService` currently combines session CRUD, admission, dispatch, and completion orchestration. These responsibilities should be separated before introducing multiple workers or richer retry behavior. diff --git a/docs/architecture/session-lifecycle.md b/docs/architecture/session-lifecycle.md index b3adbfe..ab35d19 100644 --- a/docs/architecture/session-lifecycle.md +++ b/docs/architecture/session-lifecycle.md @@ -31,12 +31,12 @@ sequenceDiagram participant M as Messages API participant S as Sandbox - C->>A: POST user.message - A->>DB: input + status_running + queued run + C->>A: POST user.message (batch) + A->>DB: input + status_running + one queued run per trigger DB-->>A: commit A-->>C: accepted input events - A->>DB: claim oldest run for session - A->>DB: read committed event history + A->>DB: claim next queued run (admission order) + A->>DB: reconstruct causal run history A->>R: Run(snapshot, projected messages) R->>M: create streamed message M-->>R: text and/or tool_use @@ -48,20 +48,51 @@ sequenceDiagram R-->>A: buffered authoritative events A->>DB: output + processed_at + final status + completion DB-->>A: commit + Note over A,DB: next run is claimed only after this commit,
so it projects the previous run's committed output ``` ## Admission invariant -The server commits all of the following atomically: +The server commits all of the following atomically, in submitted input order: - submitted client events; - a `session.status_running` event when the session was not already running; - the mutable session status; -- one queued internal run referencing the submitted event IDs. +- one queued internal run **per processable trigger event**, in admission + order. A run references exactly one trigger; multiple triggers are never + grouped into a single run. The client is never told that input was accepted unless the corresponding work item is durable. +## Two orderings of the event log + +The event log is read in two distinct orders: + +- **Public event history** — `GET .../events`, list, and the live SSE stream — + is the immutable receipt/commit sequence. It is never reordered. +- **Model-facing conversation order** is reconstructed per turn from run + causality. For each prior completed (or failed) run, in admission order, the + projection replays that run's trigger event IDs followed by its persisted + output event IDs, then appends the current run's trigger. The run's output + event IDs are durable state committed in the same transaction that closes the + run, so this ordering survives a restart and a run never sees a later trigger + queued while it was still running. Tested by + `TestRunStore_ModelHistorySurvivesReopenInCausalOrder` (file-backed reopen) + and `TestSessionService_BatchedTriplePerRunCausalProjection` (three batched + triggers project as three causal turns). + +## Per-run boundary + +Runs drain one at a time in admission order. A run commits its buffered +authoritative output and stamps its trigger processed *before* the next run is +claimed, so each run's causal history already includes the previous run's +committed output — a later user event in the same batch sees the earlier agent +reply (`TestRunStore_CompletionBeforeNextClaimObservesOutput`, +`TestSessionService_SecondUserEventObservesFirstAgentOutput`). A terminated +session is final: its leftover queued runs are never claimed, and it is never +flipped back to `running`. + ## Per-session ordering A partial unique database index permits at most one `running` item per session. @@ -102,5 +133,11 @@ A custom tool or `always_ask` built-in can park a run: 3. the stop reason names the event IDs the client must answer; 4. a custom tool response starts a new durable run. -Custom-tool resume is implemented. Built-in `user.tool_confirmation` is -accepted by the HTTP API, but its allow/deny resume semantics are not complete. +Custom-tool resume is implemented and tested for a single park/result cycle. +Built-in `user.tool_confirmation` is accepted by the HTTP API, but its +allow/deny resume semantics are not complete. + +There is no first-class durable pending-action gate. If other trigger runs were +already queued before a run parks with `requires_action`, their gating and +correlation against the parked action are not yet fully modeled; this case is +not claimed to work. diff --git a/docs/compatibility.md b/docs/compatibility.md index 683e934..4a39f0a 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -52,7 +52,7 @@ tests in `internal/httpapi/sdk_golden_test.go`. |---|---|---|---| | Event is a flat top-level tagged union (`{id,type,...,processed_at}`) | partial | `TestGolden_EventIsFlatTaggedUnion` | The raw golden covers one `user.message`; the full event union and exact per-variant keys are not covered. | | `user.message` with `content[]` blocks | partial | `TestSDK_EventSendAndList`, `TestGolden_EventIsFlatTaggedUnion` | One text block round-trips; other block variants and validation are incomplete. | -| Send events (`POST .../events`) echoes submitted events | partial | `TestSDK_EventSendAndList`, `TestSessionService_InitialEventBatchProcessesEveryEventInOrder`, `TestSessionService_SendEventBatchProcessesEveryEventInOrder`, `TestRunStore_QueuesRunsPerSessionInAdmissionOrder` | Same-request and cross-request batches enter a durable per-session order and interrupted runs are requeued on restart. All event variants, cancellation, and external-side-effect idempotency remain incomplete. | +| Send events (`POST .../events`) echoes submitted events | partial | `TestSDK_EventSendAndList`, `TestSessionService_InitialEventBatchProcessesEveryEventInOrder`, `TestSessionService_SendEventBatchProcessesEveryEventInOrder`, `TestRunStore_QueuesRunsPerSessionInAdmissionOrder`, `TestRunStore_AdmitBatchCreatesOneRunPerTrigger`, `TestRunStore_CompletionBeforeNextClaimObservesOutput`, `TestSessionService_SecondUserEventObservesFirstAgentOutput`, `TestSessionService_BatchedTriplePerRunCausalProjection`, `TestRunStore_ModelHistorySurvivesReopenInCausalOrder` | A batch is admitted atomically in input order; each processable trigger gets its own durable queued run (never grouped), and runs drain one at a time. Model-facing history is reconstructed from run causality (each prior run's trigger IDs then its persisted output IDs, then the current trigger), so a later trigger's projection observes the earlier run's committed agent output as a separate turn — proven for A/B (`TestSessionService_SecondUserEventObservesFirstAgentOutput`) and A/B/C (`TestSessionService_BatchedTriplePerRunCausalProjection`), and durable across a file-backed reopen (`TestRunStore_ModelHistorySurvivesReopenInCausalOrder`). All event variants, interrupt cancellation, and external-side-effect idempotency remain incomplete. | | Reject unknown / server-only event types and validate client variants | partial | `TestGolden_RejectsServerOnlyEventType`, `TestSendEvents_ValidatesVariantShape` | Type gating, empty-batch rejection, and validation for the currently modeled client variants are exercised; the full client-event union and every field constraint are not. | | List events (`GET .../events`): `types` filter, `limit`, `page`, `next_page` | partial | `TestSDK_EventListPaginationAndTypesFilter`, `TestListEvents_CursorIsBoundToSessionAndFilters` | Cursors are bound to the session and normalized filters. Bounds and `processed_at` ordering/keyset semantics remain incomplete. | | List events: `order`, `created_at[gt|gte|lt|lte]` filters | partial | `TestQuery_CreatedAtNamedFilterUsesProcessedAt`, `TestQuery_ProcessedAtFiltersExactAndFractionalWithinSecond`, `TestListEvents_RejectsInvalidQueryValues` | The public `created_at` query name compares a fixed-width `processed_at` key. Results and cursors are still ordered by internal sequence rather than a `processed_at` keyset, including its null/tie semantics. | @@ -74,7 +74,7 @@ tests in `internal/httpapi/sdk_golden_test.go`. | Custom-tool handoff (`agent.custom_tool_use` → park → `user.custom_tool_result` → resume) | partial | `TestWorkshop_CustomToolHandoff`, `TestSessionService_CustomToolParksAndResumes`, `TestAgentCore_CustomToolParksWithRequiresAction`, `TestFake_CustomToolResultEndsIdle`, `TestProjectMessages_CustomToolResultPairing` | A custom tool call parks the run at `session.status_idle` with `stop_reason.type == "requires_action"` and `event_ids` naming the committed `agent.custom_tool_use`; a `user.custom_tool_result` referencing that id resumes a fresh run to `end_turn`. End-to-end proven; the exact `requires_action` payload shape is unconfirmed against the official wire. | | Built-in tool run end-to-end via the app layer | partial | `TestSessionService_BuiltinToolRunEndToEnd` | A session-driven run provisions a sandbox, executes a built-in tool, and reaches `end_turn` through the durable run path. | | `always_ask` permission policy on built-in tools | partial | `TestAgentCore_AlwaysAskBuiltinParks` | An `always_ask` built-in call parks the run (`agent.tool_use{evaluated_permission:"ask"}` + `requires_action`); the **resume** path (`user.tool_confirmation` → projected `tool_result` + built-in execution) is not wired. `ProjectMessages` drops the dangling `tool_use` so the parked call never poisons a real request (`TestProjectMessages_DropsDanglingToolUse`, `TestProjectMessages_DropsDanglingCustomToolUse`). | -| Session-scoped sandbox lifecycle (`internal/sandbox` `SessionManager`) | partial | `TestSessionManager_ReusesSandboxPerSession`, `TestSessionManager_IsolatesSessions`, `TestSessionManager_ReleaseDestroysExactlyOnce`, `TestSessionService_SandboxPersistsAcrossRuns`, `TestSessionService_SandboxIsolatedBetweenSessions`, `TestSessionService_SandboxProvisionedOncePerSession`, `TestSessionService_IdleDoesNotDestroySandbox`, `TestSessionService_DeleteReleasesSandboxExactlyOnce` | A sandbox is **scoped to the session**, not the run. The first run needing tools provisions one logical sandbox; later runs in the same session reuse it, so tool-produced file state **persists across turns**. Different sessions get distinct sandboxes and stay isolated. Entering idle does not destroy the sandbox; deleting the session releases it exactly once. Ownership lives in a session-scoped manager that wraps the provider inside the `sandbox` package; `AgentRuntime` receives a resolved sandbox and is unaware of the lifecycle. The manager is in-memory: a process restart does not restore an idle session's sandbox, and there is no durable checkpoint, quota, or eviction yet. | +| Session-scoped sandbox lifecycle (`internal/sandbox` `SessionManager`) | partial | `TestSessionManager_ReusesSandboxPerSession`, `TestSessionManager_IsolatesSessions`, `TestSessionManager_ReleaseDestroysExactlyOnce`, `TestSessionService_SandboxPersistsAcrossRuns`, `TestSessionService_SandboxIsolatedBetweenSessions`, `TestSessionService_SandboxProvisionedOncePerSession`, `TestSessionService_IdleDoesNotDestroySandbox`, `TestSessionService_DeleteReleasesSandboxExactlyOnce`, `TestSessionService_DeleteTeardownSurvivesRequestCancellation` | A sandbox is **scoped to the session**, not the run. The first run needing tools provisions one logical sandbox; later runs in the same session reuse it, so tool-produced file state **persists across turns**. Different sessions get distinct sandboxes and stay isolated. Entering idle does not destroy the sandbox; deleting the session releases it exactly once. Delete runs that teardown *after* the durable delete has committed, on a context detached from the request context (`context.WithoutCancel`) and bounded by its own timeout, so a client disconnect cannot cancel an in-flight `Destroy` yet a stuck provider cannot hang forever. Ownership lives in a session-scoped manager that wraps the provider inside the `sandbox` package; `AgentRuntime` receives a resolved sandbox and is unaware of the lifecycle. The manager is in-memory: a process restart does not restore an idle session's sandbox, and there is no durable checkpoint, quota, or eviction yet. | | Local sandbox (`Provider`/`Sandbox`, restricted local process) | partial | `TestLocal_ExecEcho`, `TestLocal_FileRoundTripAndConfinement`, `TestLocal_Timeout` | `internal/sandbox` provides a two-layer interface and a local-process default that confines paths to a work dir, clears the environment, applies a timeout, and caps output. **Dev-grade guardrail, not a security boundary — do not run untrusted code.** Sandboxes are session-scoped (see the row above): provisioned on first tool use and reused across the session's runs. | | Docker sandbox (real-isolation `Provider`, opt-in) | partial | `TestDocker_*` (skipped without a daemon), `TestResolveSandboxProvider_DefaultsToLocal` | The same `Provider`/`Sandbox` interface has a Docker-backed implementation (shells out to the `docker` CLI, no extra module dependency). It gives **real isolation**: each sandbox is a container with its own Linux namespaces/cgroups, a separate filesystem, and `--network none` by default. Selected at startup via `MANAGED_AGENT_SANDBOX=docker` (default is local); image via `MANAGED_AGENT_SANDBOX_IMAGE` (defaults `alpine:latest`). gVisor (`--runtime=runsc`) can layer under the same interface later with no interface change. Not audited for hostile multi-tenant use (shared host kernel). Docker tests are gated on a running daemon and skip in default offline CI. | | MCP toolsets | unsupported | `TestParseTools_BuiltinCustomMCP` | Parsed into `domain.ToolSet` but never resolved or executed. | @@ -105,16 +105,18 @@ tests in `internal/httpapi/sdk_golden_test.go`. - `domain.ProjectMessages` merges adjacent same-role events into one message (concatenating their content blocks in order) so the projected conversation - always alternates roles and forms a legal Messages-API request. This covers - the two real flows that produce consecutive same-role events — several - `user.message` events arriving before `drainRuns` claims them, and a model - turn that emits no text (no `agent.message`) leaving two user turns adjacent - (`TestProjectMessages_MergesConsecutiveUsers`, - `TestProjectMessages_MergesConsecutiveAssistantsAndAlternates`). Within a - single `drainRuns` claim every trigger reuses one projected snapshot: reuse is - legal (the snapshot alternates), but the snapshot can be stale relative to - drafts produced by an earlier trigger in the same claim. Defining and fixing - this batch-turn boundary is the first roadmap item. + always alternates roles and forms a legal Messages-API request. Under the + causal Run history model, multiple `user.message` events queued before + `drainRuns` claims them are **not** collapsed into one projected user message: + each processable trigger runs as its own durable run and projects as a + separate turn (`TestSessionService_SecondUserEventObservesFirstAgentOutput`, + `TestSessionService_BatchedTriplePerRunCausalProjection`), and the causal + ordering survives a restart (`TestRunStore_ModelHistorySurvivesReopenInCausalOrder`). + Merging still fires only when a turn genuinely emits no assistant content — + e.g. a model turn that produces no `agent.message`, leaving two user turns + adjacent (`TestProjectMessages_MergesConsecutiveUsers`, + `TestProjectMessages_MergesConsecutiveAssistantsAndAlternates`). Context + compaction when a session exceeds the projection limit is still deferred. - Server shutdown does not cancel in-flight model calls. `drainRuns` runs on `context.Background()`, so a SIGTERM does not propagate cancellation to an in-progress Messages-API request; cancellation propagation is not implemented. @@ -144,6 +146,12 @@ tests in `internal/httpapi/sdk_golden_test.go`. established. - Opaque `multiagent` input is stored with tested replace/null-clear behavior; roster resolution, reference validation, and multiagent execution are absent. +- There is no first-class durable pending-action gate. A single custom-tool + park and its `user.custom_tool_result` resume are implemented and tested + (`TestSessionService_CustomToolParksAndResumes`), but if other trigger runs + were already queued before a run parks with `requires_action`, their gating + and correlation against the parked action are not yet fully modeled; this case + is not claimed to work. - The durable queue is single-process and restart recovery is at-least-once. A crash after an external side effect but before completion commit may replay it; no lease, fencing, or idempotency-key protocol exists yet. diff --git a/docs/getting-started.md b/docs/getting-started.md index 7a14f5c..ee78626 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -150,4 +150,6 @@ input or in production. The pre-release project does not maintain schema migrations. Stop the server before deleting a disposable local database, then recreate it on the next -start. +start. This applies to schema additions too: the durable run-output association +column (`session_runs.output_event_ids`) follows the same rebuild-the-dev-DB +policy — there is no migration subsystem. diff --git a/docs/roadmap.md b/docs/roadmap.md index fdb387c..fef9fa6 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -10,8 +10,6 @@ surface-area compatibility. ## 1. Execution semantics -- Define batch input as one turn or one durable run per trigger, then make - projection and commit behavior match that definition. - Add a first-class durable pending-action model for custom tools and permission confirmations. - Complete `always_ask` confirmation resume. @@ -66,6 +64,16 @@ When real deployment requirements demand it: - Server-owned, multi-turn Messages API history projection. - Versioned agents and immutable session snapshots. - Atomic input/run admission and single-node restart recovery. +- Per-trigger durable runs: a request batch is admitted atomically in input + order, each processable trigger gets its own durable run and commit boundary, + and each run persists the exact IDs of the output events it committed. The + model-facing conversation is reconstructed durably from run causality — each + prior run's trigger IDs followed by its persisted output IDs, then the current + trigger — so a run projects earlier runs' committed output as separate turns, + the ordering survives a restart, and a run never sees a later queued trigger. + This is our own causal association; it does not claim to mirror any exact + Anthropic-internal ordering. A terminated session never claims its leftover + queued work. - Multi-step model/tool loop. - Local sandbox plus optional Docker provider. - Session-scoped sandbox ownership: reused across a session's runs, isolated diff --git a/internal/app/recovery_test.go b/internal/app/recovery_test.go index 90a1c0e..e2472ea 100644 --- a/internal/app/recovery_test.go +++ b/internal/app/recovery_test.go @@ -34,8 +34,8 @@ func TestRecover_RequeuesAndCompletesInterruptedRun(t *testing.T) { if err != nil { t.Fatal(err) } - if admission.Run == nil { - t.Fatal("admission did not create a run") + if len(admission.Runs) != 1 { + t.Fatalf("admission runs = %d, want 1", len(admission.Runs)) } if _, ok, err := runs.ClaimNext(ctx, "sesn_1"); err != nil || !ok { t.Fatalf("claim before simulated crash: ok=%v err=%v", ok, err) @@ -66,7 +66,7 @@ func TestRecover_RequeuesAndCompletesInterruptedRun(t *testing.T) { if got.Status != domain.StatusIdle { t.Fatalf("expected recovered idle, got %s", got.Status) } - run, err := recoveryRuns.Get(ctx, admission.Run.ID) + run, err := recoveryRuns.Get(ctx, admission.Runs[0].ID) if err != nil { t.Fatal(err) } diff --git a/internal/app/session_service.go b/internal/app/session_service.go index ff2c400..af839b3 100644 --- a/internal/app/session_service.go +++ b/internal/app/session_service.go @@ -55,12 +55,19 @@ const sessionLockShardCount = 256 // historyProjectionLimit bounds how many events are replayed into the model // conversation per turn. Projection uses the NEWEST window of this size (see -// EventService.HistoryTail): an over-limit session carries its most recent +// RunStore.ModelHistory): an over-limit session carries its most recent causal // context rather than the oldest events. Compaction is a later slice; until // then this is a generous ceiling that keeps a single unbounded session from // OOMing a turn. const historyProjectionLimit = 10000 +// sandboxReleaseTimeout bounds the detached sandbox teardown performed during +// session deletion. The durable delete has already committed before teardown +// runs, so teardown executes on a context detached from the request's +// cancellation; this ceiling keeps that detached cleanup finite instead of +// letting a stuck provider Destroy block forever. +const sandboxReleaseTimeout = 30 * time.Second + type SessionService struct { sess *store.SessionRepo agents *store.AgentRepo @@ -158,7 +165,7 @@ func (s *SessionService) Create(ctx context.Context, in CreateSessionInput) (dom return domain.Session{}, err } s.events.PublishCommitted(admission.Events) - if admission.Run != nil { + if len(admission.Runs) > 0 { s.kick(sess.ID) } return admission.Session, nil @@ -210,7 +217,7 @@ func (s *SessionService) SendEvent(ctx context.Context, id string, drafts []doma return nil, err } s.events.PublishCommitted(admission.Events) - if admission.Run != nil { + if len(admission.Runs) > 0 { s.kick(id) } // Send Events echoes only the caller-submitted events, not the status event @@ -278,7 +285,17 @@ func (s *SessionService) Delete(ctx context.Context, id string) error { // and a no-op when the session never provisioned one; when it did, it runs // the provider teardown exactly once. This is an external call made after the // durable delete, outside the store transaction and outside the shard lock. - if err := s.sandbox.Release(ctx, id); err != nil { + // + // Teardown must not inherit cancellation from the request context. The + // durable delete has already committed, so if the caller cancels (client + // disconnects, request deadline elapses) we still have to finish tearing the + // sandbox down or leak a container / host temp dir. context.WithoutCancel + // detaches from the request's cancellation while preserving its context + // values (trace metadata, etc.); the timeout then bounds the detached + // teardown so a stuck provider Destroy cannot hang deletion forever. + cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), sandboxReleaseTimeout) + defer cancel() + if err := s.sandbox.Release(cleanupCtx, id); err != nil { // A teardown failure can leak sandbox resources (a host temp dir or a // container). Log it; the session is already durably deleted. log.Printf("delete: sandbox release failed session_id=%s: %v", id, err) @@ -348,22 +365,25 @@ func (s *SessionService) drainRuns(sessionID string) { } if runErr == nil { - // Process one trigger to completion before the next. Each trigger - // re-reads and re-projects the newest bounded window of the ordered - // event log (HistoryTail returns the most recent historyProjectionLimit - // events in chronological order, not the whole log). Note the limit of - // this within a single claim: the runtime's drafts are held in the - // in-memory bufferedSink and only committed to the store when the run - // completes, so HistoryTail here observes only the user events admission - // already persisted — not the agent output an earlier trigger in this - // same claim produced but has not yet committed. True per-trigger - // isolation (each trigger committed independently before the next - // projects) is deferred. History is read outside any transaction, - // before calling the runtime — the server owns history via the event - // log. ProjectMessages merges adjacent same-role events, so each - // snapshot alternates roles and is a legal Messages-API request. + // Each claim carries exactly one trigger (admission enqueues one durable + // run per processable event, in admission order). The run projects + // history reconstructed from run causality — RunStore.ModelHistory walks + // prior completed/failed runs in admission order, replaying each run's + // trigger events followed by that run's persisted output events, then + // this run's own trigger. That deliberately differs from raw + // receipt/commit order (EventStore.History): a later trigger admitted + // before this run finished is excluded, so a turn never sees a future + // user message. Because run N commits its output and marks its trigger + // processed before run N+1 is claimed, run N+1's causal history already + // includes run N's committed agent reply. History is read outside the + // runtime call — the server owns history via the event log. + // ProjectMessages merges adjacent same-role events and drops a dangling + // tool_use / orphan tool_result, so each snapshot is a legal Messages-API + // request even when the bounded window cut a pair. The loop below + // iterates the claim's single trigger; the requires_action break still + // parks the run so its awaited result is admitted as the next trigger. for _, trigger := range claim.Triggers { - history, histErr := s.events.HistoryTail(ctx, sessionID, historyProjectionLimit) + history, histErr := s.runs.ModelHistory(ctx, claim.Run, historyProjectionLimit) if histErr != nil { runErr = histErr break diff --git a/internal/app/session_service_test.go b/internal/app/session_service_test.go index b44ad98..da36151 100644 --- a/internal/app/session_service_test.go +++ b/internal/app/session_service_test.go @@ -560,6 +560,357 @@ func textBlocks(text string) []any { return []any{map[string]any{"type": "text", "text": text}} } +// projectingRuntime records the projected Messages it receives for each run and +// emits a distinct agent.message per trigger so a later run's projection can be +// asserted to include an earlier run's committed output. +type projectingRuntime struct { + projections chan []domain.Message +} + +func (r projectingRuntime) Run( + ctx context.Context, + req agentruntime.RunRequest, + sink agentruntime.EventSink, +) (agentruntime.RunOutcome, error) { + r.projections <- req.Messages + _, err := sink.Emit(ctx, []domain.EventDraft{ + {Type: domain.EvAgentMessage, Payload: map[string]any{ + "content": []any{map[string]any{ + "type": "text", + "text": "reply-to: " + contentText(req.Trigger.Payload), + }}, + }}, + {Type: domain.EvSessionStatusIdle, Payload: map[string]any{ + "stop_reason": map[string]any{"type": "end_turn"}, + }}, + }) + return agentruntime.RunOutcome{}, err +} + +func contentText(payload map[string]any) string { + blocks, _ := payload["content"].([]any) + if len(blocks) == 0 { + return "" + } + block, _ := blocks[0].(map[string]any) + text, _ := block["text"].(string) + return text +} + +// TestSessionService_SecondUserEventObservesFirstAgentOutput proves the +// completion-before-next-claim guarantee end to end with EXACT projections: two +// user events admitted in one batch produce two runs. Run A's projected Messages +// are exactly user(A). Run B's are exactly user(A), assistant(reply-to:A), +// user(B) — the second run observes the first run's committed reply, in causal +// order, and nothing more. The public event history preserves receipt/commit +// order and is not rewritten. +func TestSessionService_SecondUserEventObservesFirstAgentOutput(t *testing.T) { + db, err := store.OpenMemory() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { db.Close() }) + ctx := context.Background() + ids := domain.NewSeqIDGen() + clk := domain.FixedClock{T: time.Unix(1, 0).UTC()} + events := NewEventService(store.NewEventStore(db, ids, clk), NewHub(64)) + agents := NewAgentService(store.NewAgentRepo(db), ids, clk) + envs := NewEnvironmentService(store.NewEnvironmentRepo(db), ids, clk) + projections := make(chan []domain.Message, 4) + ss := NewSessionService( + store.NewSessionRepo(db), store.NewAgentRepo(db), store.NewEnvironmentRepo(db), + events, store.NewRunStore(db, ids, clk), + projectingRuntime{projections: projections}, sandbox.NewLocalProvider(), ids, clk, + ) + ag, _ := agents.Create(ctx, domain.Agent{Name: "a", Model: domain.Model{ID: "claude-opus-4-8"}}) + env, _ := envs.Create(ctx, domain.Environment{Name: "e", ConfigType: "cloud"}) + + sess, err := ss.Create(ctx, CreateSessionInput{ + AgentID: ag.ID, + EnvironmentID: env.ID, + InitialEvents: []domain.EventDraft{ + {Type: domain.EvUserMessage, Payload: map[string]any{"content": textBlocks("first")}}, + {Type: domain.EvUserMessage, Payload: map[string]any{"content": textBlocks("second")}}, + }, + }) + if err != nil { + t.Fatalf("Create: %v", err) + } + + first := receiveProjection(t, projections) + second := receiveProjection(t, projections) + pollUntilStatus(t, ss, sess.ID, domain.StatusIdle) + + // Run A's projection is exactly the first user message. + assertMessages(t, "run A", first, []wantMessage{ + {domain.RoleUser, "first"}, + }) + // Run B's projection is exactly user(A), assistant(reply-to:A), user(B). + assertMessages(t, "run B", second, []wantMessage{ + {domain.RoleUser, "first"}, + {domain.RoleAssistant, "reply-to: first"}, + {domain.RoleUser, "second"}, + }) + + // The public event history preserves receipt/commit order and is not + // rewritten: both user triggers are committed (with ascending seq) before the + // agent replies they caused. + assertUserTriggersBeforeOutputs(t, ss, sess.ID, []string{"first", "second"}) +} + +// TestSessionService_BatchedTriplePerRunCausalProjection admits A,B,C in one +// batch and asserts each run's projection contains ONLY the completed prior +// trigger/output turns plus its current trigger, in exact role/content order — +// never a later, still-queued trigger. +func TestSessionService_BatchedTriplePerRunCausalProjection(t *testing.T) { + db, err := store.OpenMemory() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { db.Close() }) + ctx := context.Background() + ids := domain.NewSeqIDGen() + clk := domain.FixedClock{T: time.Unix(1, 0).UTC()} + events := NewEventService(store.NewEventStore(db, ids, clk), NewHub(64)) + agents := NewAgentService(store.NewAgentRepo(db), ids, clk) + envs := NewEnvironmentService(store.NewEnvironmentRepo(db), ids, clk) + projections := make(chan []domain.Message, 8) + ss := NewSessionService( + store.NewSessionRepo(db), store.NewAgentRepo(db), store.NewEnvironmentRepo(db), + events, store.NewRunStore(db, ids, clk), + projectingRuntime{projections: projections}, sandbox.NewLocalProvider(), ids, clk, + ) + ag, _ := agents.Create(ctx, domain.Agent{Name: "a", Model: domain.Model{ID: "claude-opus-4-8"}}) + env, _ := envs.Create(ctx, domain.Environment{Name: "e", ConfigType: "cloud"}) + + sess, err := ss.Create(ctx, CreateSessionInput{ + AgentID: ag.ID, + EnvironmentID: env.ID, + InitialEvents: []domain.EventDraft{ + {Type: domain.EvUserMessage, Payload: map[string]any{"content": textBlocks("A")}}, + {Type: domain.EvUserMessage, Payload: map[string]any{"content": textBlocks("B")}}, + {Type: domain.EvUserMessage, Payload: map[string]any{"content": textBlocks("C")}}, + }, + }) + if err != nil { + t.Fatalf("Create: %v", err) + } + + runA := receiveProjection(t, projections) + runB := receiveProjection(t, projections) + runC := receiveProjection(t, projections) + pollUntilStatus(t, ss, sess.ID, domain.StatusIdle) + + assertMessages(t, "run A", runA, []wantMessage{ + {domain.RoleUser, "A"}, + }) + assertMessages(t, "run B", runB, []wantMessage{ + {domain.RoleUser, "A"}, + {domain.RoleAssistant, "reply-to: A"}, + {domain.RoleUser, "B"}, + }) + assertMessages(t, "run C", runC, []wantMessage{ + {domain.RoleUser, "A"}, + {domain.RoleAssistant, "reply-to: A"}, + {domain.RoleUser, "B"}, + {domain.RoleAssistant, "reply-to: B"}, + {domain.RoleUser, "C"}, + }) + + // Public history keeps receipt/commit order: A,B,C triggers are all committed + // before any of the agent replies they produced. + assertUserTriggersBeforeOutputs(t, ss, sess.ID, []string{"A", "B", "C"}) +} + +type wantMessage struct { + role domain.Role + text string +} + +// assertMessages requires msgs to equal want exactly: same length, same role +// and single-text-block content per message, in order. +func assertMessages(t *testing.T, label string, msgs []domain.Message, want []wantMessage) { + t.Helper() + if len(msgs) != len(want) { + t.Fatalf("%s: projection has %d messages, want %d: %#v", label, len(msgs), len(want), msgs) + } + for i, w := range want { + m := msgs[i] + if m.Role != w.role { + t.Fatalf("%s: message[%d] role = %s, want %s: %#v", label, i, m.Role, w.role, msgs) + } + if len(m.Content) != 1 || m.Content[0].Text != w.text { + t.Fatalf("%s: message[%d] content = %#v, want single text %q", label, i, m.Content, w.text) + } + } +} + +// assertUserTriggersBeforeOutputs proves the public event history is authentic +// receipt/commit order: the user.message triggers (in wantUserText order, with +// strictly ascending sequence) all precede the agent.message replies. Nothing +// rewrites event seq to match causal projection. +func assertUserTriggersBeforeOutputs(t *testing.T, ss *SessionService, sessionID string, wantUserText []string) { + t.Helper() + history, err := ss.events.History(context.Background(), sessionID, 0, 1000) + if err != nil { + t.Fatalf("history: %v", err) + } + var userText []string + var lastSeq int64 + var firstAgentSeq int64 = -1 + var lastUserTriggerSeq int64 + for _, e := range history { + if e.Sequence <= lastSeq { + t.Fatalf("event history not in ascending seq order: %d after %d", e.Sequence, lastSeq) + } + lastSeq = e.Sequence + switch e.Type { + case domain.EvUserMessage: + userText = append(userText, contentBlockText(e.Payload["content"])) + lastUserTriggerSeq = e.Sequence + case domain.EvAgentMessage: + if firstAgentSeq < 0 { + firstAgentSeq = e.Sequence + } + } + } + if len(userText) != len(wantUserText) { + t.Fatalf("user triggers = %q, want %q", userText, wantUserText) + } + for i := range wantUserText { + if userText[i] != wantUserText[i] { + t.Fatalf("user trigger order = %q, want %q", userText, wantUserText) + } + } + if firstAgentSeq >= 0 && lastUserTriggerSeq >= firstAgentSeq { + t.Fatalf("public history reordered: a user trigger (seq %d) landed after an agent reply (seq %d)", + lastUserTriggerSeq, firstAgentSeq) + } +} + +func receiveProjection(t *testing.T, ch chan []domain.Message) []domain.Message { + t.Helper() + select { + case msgs := <-ch: + return msgs + case <-time.After(2 * time.Second): + t.Fatal("runtime did not receive a projected turn") + return nil + } +} + +// blockingDestroySandbox is a sandbox whose Destroy blocks until the test +// releases it, and which records the state of the context it was destroyed +// under. It lets a test hold a teardown mid-flight, cancel the original request +// context, and then assert on the context the teardown actually ran with. +type blockingDestroySandbox struct { + started chan struct{} // closed when Destroy begins + release chan struct{} // closed by the test to let Destroy finish + + // Captured inside Destroy, after the test has cancelled the original request + // context. Read only after Destroy has returned (via the delete result + // channel), so no synchronization beyond that happens-before is needed. + errAfterReqCancel error + hasDeadline bool +} + +func (b *blockingDestroySandbox) Exec(context.Context, sandbox.Command) (*sandbox.Result, error) { + return &sandbox.Result{}, nil +} +func (b *blockingDestroySandbox) ReadFile(context.Context, string) ([]byte, error) { return nil, nil } +func (b *blockingDestroySandbox) WriteFile(context.Context, string, []byte) error { return nil } +func (b *blockingDestroySandbox) Root() string { return "" } + +func (b *blockingDestroySandbox) Destroy(ctx context.Context) error { + close(b.started) + <-b.release + // The test cancels the original request context before closing release, so + // this read observes the teardown context after that cancellation. Because + // Delete detaches teardown from the request context, Err must still be nil. + b.errAfterReqCancel = ctx.Err() + _, b.hasDeadline = ctx.Deadline() + return nil +} + +// stubSandboxProvider provisions a single pre-built sandbox, so a test can +// inject a controllable box into a SessionService's SessionManager. +type stubSandboxProvider struct{ box sandbox.Sandbox } + +func (p stubSandboxProvider) Provision(context.Context, sandbox.Spec) (sandbox.Sandbox, error) { + return p.box, nil +} + +// TestSessionService_DeleteTeardownSurvivesRequestCancellation proves that once +// sandbox teardown has started during Delete, cancelling the original request +// context does not cancel that teardown, while the teardown context is still +// bounded by a deadline. Coordination is by channels only — no timing sleeps. +func TestSessionService_DeleteTeardownSurvivesRequestCancellation(t *testing.T) { + db, err := store.OpenMemory() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { db.Close() }) + ctx := context.Background() + ids := domain.NewSeqIDGen() + clk := domain.FixedClock{T: time.Unix(1, 0).UTC()} + events := NewEventService(store.NewEventStore(db, ids, clk), NewHub(64)) + agents := NewAgentService(store.NewAgentRepo(db), ids, clk) + envs := NewEnvironmentService(store.NewEnvironmentRepo(db), ids, clk) + + box := &blockingDestroySandbox{ + started: make(chan struct{}), + release: make(chan struct{}), + } + ss := NewSessionService( + store.NewSessionRepo(db), store.NewAgentRepo(db), store.NewEnvironmentRepo(db), + events, store.NewRunStore(db, ids, clk), + agentruntime.NewFake(), stubSandboxProvider{box: box}, ids, clk, + ) + ag, _ := agents.Create(ctx, domain.Agent{Name: "a", Model: domain.Model{ID: "claude-opus-4-8"}}) + env, _ := envs.Create(ctx, domain.Environment{Name: "e", ConfigType: "cloud"}) + + // An idle session (no initial events) that Delete is allowed to remove. + sess, err := ss.Create(ctx, CreateSessionInput{AgentID: ag.ID, EnvironmentID: env.ID}) + if err != nil { + t.Fatalf("Create: %v", err) + } + // Provision the session's sandbox so Delete's Release has a box to destroy. + if _, err := ss.sandbox.Acquire(ctx, sess.ID, sandbox.Spec{}); err != nil { + t.Fatalf("Acquire: %v", err) + } + + reqCtx, cancelReq := context.WithCancel(context.Background()) + deleteDone := make(chan error, 1) + go func() { deleteDone <- ss.Delete(reqCtx, sess.ID) }() + + // Wait for teardown to start under a timeout so a regression (Delete never + // reaching Release, or Release never invoking Destroy) fails the test instead + // of hanging it forever. + select { + case <-box.started: // teardown is now in flight + case <-time.After(2 * time.Second): + cancelReq() + t.Fatal("sandbox teardown never started during Delete") + } + cancelReq() // cancel the ORIGINAL request context mid-teardown + close(box.release) // let Destroy observe its context and return + + select { + case err := <-deleteDone: + if err != nil { + t.Fatalf("Delete: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Delete did not return after teardown was released") + } + if box.errAfterReqCancel != nil { + t.Fatalf("teardown context was cancelled by request cancellation: %v", box.errAfterReqCancel) + } + if !box.hasDeadline { + t.Fatal("teardown context was not bounded by a deadline") + } +} + func assertBatchProcessedInOrder( t *testing.T, ss *SessionService, diff --git a/internal/domain/run.go b/internal/domain/run.go index ba4c8c5..805b913 100644 --- a/internal/domain/run.go +++ b/internal/domain/run.go @@ -18,8 +18,13 @@ type SessionRun struct { SessionID string AdmissionSeq int64 TriggerEventIDs []string - State RunState - Error *string - CreatedAt time.Time - UpdatedAt time.Time + // OutputEventIDs are the exact committed event ids this run appended when it + // closed (agent output plus the run's terminal/status events), persisted in + // the same transaction that closes the run. Empty until completion. This is + // internal-only durable state and is never serialized onto the public API. + OutputEventIDs []string + State RunState + Error *string + CreatedAt time.Time + UpdatedAt time.Time } diff --git a/internal/store/run_store.go b/internal/store/run_store.go index a255557..802f3d3 100644 --- a/internal/store/run_store.go +++ b/internal/store/run_store.go @@ -25,7 +25,10 @@ func NewRunStore(db *DB, ids domain.IDGenerator, clock domain.Clock) *RunStore { type Admission struct { Session domain.Session Events []domain.Event - Run *domain.SessionRun + // Runs holds one durable queued run per processable trigger event, in the + // same stable order as the admitted events. Multiple trigger IDs are never + // grouped into a single run. + Runs []domain.SessionRun } type RunClaim struct { @@ -42,7 +45,7 @@ type RunCompletion struct { } // CreateSession atomically creates a session and, when initial events are -// present, admits the first durable run. +// present, admits one durable queued run per processable trigger event. func (s *RunStore) CreateSession( ctx context.Context, session domain.Session, @@ -68,7 +71,8 @@ func (s *RunStore) CreateSession( } // Admit atomically persists client events, projects the session to running when -// needed, emits session.status_running, and enqueues one durable run. +// needed, emits session.status_running, and enqueues one durable run per +// processable trigger event, in admission order. func (s *RunStore) Admit( ctx context.Context, sessionID string, @@ -137,11 +141,16 @@ func (s *RunStore) admitTx( admission.Session = session } - run, err := s.insertRunTx(ctx, tx, session.ID, triggerIDs) - if err != nil { - return Admission{}, err + // One durable queued run per trigger, in admission order. Grouping multiple + // triggers into a single run would let a later trigger project history + // before the earlier trigger's agent output was committed. + for _, triggerID := range triggerIDs { + run, err := s.insertRunTx(ctx, tx, session.ID, triggerID) + if err != nil { + return Admission{}, err + } + admission.Runs = append(admission.Runs, run) } - admission.Run = &run return admission, nil } @@ -149,8 +158,9 @@ func (s *RunStore) insertRunTx( ctx context.Context, tx *sql.Tx, sessionID string, - triggerIDs []string, + triggerID string, ) (domain.SessionRun, error) { + triggerIDs := []string{triggerID} var sequence int64 if err := tx.QueryRowContext(ctx, `SELECT COALESCE(MAX(admission_seq), 0) + 1 FROM session_runs WHERE session_id=?`, @@ -166,7 +176,7 @@ func (s *RunStore) insertRunTx( ID: s.ids.NewID(domain.PrefixRun), SessionID: sessionID, AdmissionSeq: sequence, - TriggerEventIDs: append([]string(nil), triggerIDs...), + TriggerEventIDs: triggerIDs, State: domain.RunQueued, CreatedAt: now, UpdatedAt: now, @@ -189,6 +199,19 @@ func (s *RunStore) ClaimNext(ctx context.Context, sessionID string) (RunClaim, b } defer tx.Rollback() + // A terminated session is final: never claim its leftover queued work and + // never flip it back to running. Guard before selecting a run so a session + // that terminated with runs still queued stays terminated. + session, err := getSessionTx(ctx, tx, sessionID) + if err != nil { + // getSessionTx already maps a missing row to domain.NotFound; propagate + // the truthful error rather than silently reporting "nothing to claim". + return RunClaim{}, false, err + } + if session.Status == domain.StatusTerminated { + return RunClaim{}, false, nil + } + run, err := selectNextQueuedRun(ctx, tx, sessionID) if err == sql.ErrNoRows { return RunClaim{}, false, nil @@ -224,10 +247,6 @@ WHERE id=? AND state=? if err != nil { return RunClaim{}, false, err } - session, err := getSessionTx(ctx, tx, sessionID) - if err != nil { - return RunClaim{}, false, err - } var statusEvents []domain.Event if session.Status != domain.StatusRunning { session.Status = domain.StatusRunning @@ -330,11 +349,26 @@ SELECT EXISTS( if runError != nil { run.State = domain.RunFailed } + // Persist the exact committed output event ids on the run in the same + // transaction that closes it. These are precisely the events this run + // appended above (agent output plus the run's terminal/status events), in + // commit order. Writing them here — not in a follow-up statement — keeps the + // invariant that there is never a completed run without its output + // association. ModelHistory later replays these ids to rebuild causal history. + outputIDs := make([]string, len(events)) + for i, event := range events { + outputIDs[i] = event.ID + } + run.OutputEventIDs = outputIDs + outputJSON, err := json.Marshal(outputIDs) + if err != nil { + return RunCompletion{}, err + } _, err = tx.ExecContext(ctx, ` UPDATE session_runs -SET state=?, error=?, updated_at=? +SET state=?, error=?, output_event_ids=?, updated_at=? WHERE id=? AND state=?`, - string(run.State), nullableString(run.Error), timeVal(run.UpdatedAt), + string(run.State), nullableString(run.Error), string(outputJSON), timeVal(run.UpdatedAt), run.ID, string(domain.RunRunning)) if err != nil { return RunCompletion{}, err @@ -345,6 +379,86 @@ WHERE id=? AND state=?`, return RunCompletion{Run: run, Session: session, Events: events}, nil } +// ModelHistory reconstructs the causal conversation history for a claimed/current +// run, to be projected into the model. Public event ordering (History/List/the +// live stream) is authoritative receipt/commit order and is deliberately NOT +// what a turn should replay: a later queued trigger admitted before an earlier +// run finished must not appear in that earlier turn's projection. This method +// rebuilds history from run causality instead of raw sequence: +// +// - It walks the prior completed/failed runs for the same session in admission +// order and, for each, appends that run's trigger events followed by that +// run's persisted output events (the exact events it committed on +// completion). This interleaves user turn / agent reply in the causal order +// they actually resolved. +// - It then appends the current run's own trigger events. +// - Every later queued trigger (admission_seq greater than this run's, or any +// run not yet completed/failed) is excluded, because only prior terminal runs +// and this run's trigger are visited. +// +// The reconstructed history is finally bounded to the newest `limit` events (the +// historyProjectionLimit-equivalent), preserving chronological causal order. A +// window that cuts a tool_use/tool_result pair is left for ProjectMessages' +// existing dangling/orphan filtering to repair. +func (s *RunStore) ModelHistory( + ctx context.Context, + run domain.SessionRun, + limit int, +) ([]domain.Event, error) { + tx, err := s.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true}) + if err != nil { + return nil, err + } + defer tx.Rollback() + + rows, err := tx.QueryContext(ctx, ` +SELECT id +FROM session_runs +WHERE session_id=? AND admission_seq 0 && len(orderedIDs) > limit { + orderedIDs = orderedIDs[len(orderedIDs)-limit:] + } + return loadEventsByIDTx(ctx, tx, run.SessionID, orderedIDs) +} + // UpdateTitle keeps the session row and session.updated event in one commit. func (s *RunStore) UpdateTitle( ctx context.Context, @@ -446,14 +560,15 @@ func getRunTx(ctx context.Context, q runQueryRower, id string) (domain.SessionRu var ( run domain.SessionRun triggerJSON string + outputJSON string state, created, updated string runError sql.NullString ) err := q.QueryRowContext(ctx, ` -SELECT id, session_id, admission_seq, trigger_event_ids, state, error, created_at, updated_at +SELECT id, session_id, admission_seq, trigger_event_ids, output_event_ids, state, error, created_at, updated_at FROM session_runs WHERE id=?`, id).Scan( - &run.ID, &run.SessionID, &run.AdmissionSeq, &triggerJSON, &state, + &run.ID, &run.SessionID, &run.AdmissionSeq, &triggerJSON, &outputJSON, &state, &runError, &created, &updated) if err != nil { return domain.SessionRun{}, err @@ -461,6 +576,9 @@ WHERE id=?`, id).Scan( if err := json.Unmarshal([]byte(triggerJSON), &run.TriggerEventIDs); err != nil { return domain.SessionRun{}, fmt.Errorf("store: decode run triggers: %w", err) } + if err := json.Unmarshal([]byte(outputJSON), &run.OutputEventIDs); err != nil { + return domain.SessionRun{}, fmt.Errorf("store: decode run outputs: %w", err) + } run.State = domain.RunState(state) if runError.Valid { run.Error = &runError.String diff --git a/internal/store/run_store_test.go b/internal/store/run_store_test.go index 52a37a4..65a7617 100644 --- a/internal/store/run_store_test.go +++ b/internal/store/run_store_test.go @@ -2,6 +2,7 @@ package store import ( "context" + "path/filepath" "testing" "time" @@ -55,8 +56,8 @@ func TestRunStore_CreateSessionAdmissionIsOneCommit(t *testing.T) { if admission.Session.Status != domain.StatusRunning { t.Fatalf("session status = %s, want running", admission.Session.Status) } - if admission.Run == nil || admission.Run.State != domain.RunQueued { - t.Fatalf("run = %#v", admission.Run) + if len(admission.Runs) != 1 || admission.Runs[0].State != domain.RunQueued { + t.Fatalf("runs = %#v", admission.Runs) } if len(admission.Events) != 2 || admission.Events[0].Type != domain.EvUserMessage || @@ -68,7 +69,7 @@ func TestRunStore_CreateSessionAdmissionIsOneCommit(t *testing.T) { if err != nil { t.Fatal(err) } - storedRun, err := runs.Get(ctx, admission.Run.ID) + storedRun, err := runs.Get(ctx, admission.Runs[0].ID) if err != nil { t.Fatal(err) } @@ -145,18 +146,21 @@ func TestRunStore_QueuesRunsPerSessionInAdmissionOrder(t *testing.T) { if err != nil { t.Fatal(err) } - if first.Run.AdmissionSeq != 1 || second.Run.AdmissionSeq != 2 { - t.Fatalf("run sequences = %d, %d", first.Run.AdmissionSeq, second.Run.AdmissionSeq) + if len(first.Runs) != 1 || len(second.Runs) != 1 { + t.Fatalf("run counts = %d, %d, want 1, 1", len(first.Runs), len(second.Runs)) + } + if first.Runs[0].AdmissionSeq != 1 || second.Runs[0].AdmissionSeq != 2 { + t.Fatalf("run sequences = %d, %d", first.Runs[0].AdmissionSeq, second.Runs[0].AdmissionSeq) } firstClaim, ok, err := runs.ClaimNext(ctx, session.ID) - if err != nil || !ok || firstClaim.Run.ID != first.Run.ID { + if err != nil || !ok || firstClaim.Run.ID != first.Runs[0].ID { t.Fatalf("first claim = %#v ok=%v err=%v", firstClaim.Run, ok, err) } if _, ok, err := runs.ClaimNext(ctx, session.ID); err != nil || ok { t.Fatalf("second claim while first running: ok=%v err=%v", ok, err) } - firstDone, err := runs.Complete(ctx, first.Run.ID, + firstDone, err := runs.Complete(ctx, first.Runs[0].ID, []domain.EventDraft{{Type: domain.EvSessionStatusIdle}}, domain.StatusIdle, nil) if err != nil { @@ -166,7 +170,288 @@ func TestRunStore_QueuesRunsPerSessionInAdmissionOrder(t *testing.T) { t.Fatalf("queued successor should keep projection running, got %s", firstDone.Session.Status) } secondClaim, ok, err := runs.ClaimNext(ctx, session.ID) - if err != nil || !ok || secondClaim.Run.ID != second.Run.ID { + if err != nil || !ok || secondClaim.Run.ID != second.Runs[0].ID { + t.Fatalf("second claim = %#v ok=%v err=%v", secondClaim.Run, ok, err) + } +} + +// TestRunStore_AdmitBatchCreatesOneRunPerTrigger proves a single atomic +// admission of multiple triggers yields one durable queued run per trigger, in +// admission order, each holding exactly one trigger id — never one grouped run. +func TestRunStore_AdmitBatchCreatesOneRunPerTrigger(t *testing.T) { + _, runs, session := newRunStoreFixture(t) + ctx := context.Background() + admission, err := runs.CreateSession(ctx, session, []domain.EventDraft{ + {Type: domain.EvUserMessage, Payload: map[string]any{"content": []any{ + map[string]any{"type": "text", "text": "first"}, + }}}, + {Type: domain.EvUserMessage, Payload: map[string]any{"content": []any{ + map[string]any{"type": "text", "text": "second"}, + }}}, + }) + if err != nil { + t.Fatal(err) + } + if len(admission.Runs) != 2 { + t.Fatalf("runs = %d, want 2 (one per trigger)", len(admission.Runs)) + } + // The two user events are events[0] and events[1]; events[2] is the status + // event emitted by the same admission. + wantTriggers := []string{admission.Events[0].ID, admission.Events[1].ID} + for i, run := range admission.Runs { + if run.AdmissionSeq != int64(i+1) { + t.Fatalf("run %d admission_seq = %d, want %d", i, run.AdmissionSeq, i+1) + } + if len(run.TriggerEventIDs) != 1 || run.TriggerEventIDs[0] != wantTriggers[i] { + t.Fatalf("run %d triggers = %#v, want [%s]", i, run.TriggerEventIDs, wantTriggers[i]) + } + if run.State != domain.RunQueued { + t.Fatalf("run %d state = %s, want queued", i, run.State) + } + } +} + +// TestRunStore_CompletionBeforeNextClaimObservesOutput proves that after run N +// commits its agent output and marks its trigger processed, run N+1 becomes +// claimable and projects history that includes run N's committed output. +func TestRunStore_CompletionBeforeNextClaimObservesOutput(t *testing.T) { + db, runs, session := newRunStoreFixture(t) + ctx := context.Background() + admission, err := runs.CreateSession(ctx, session, []domain.EventDraft{ + {Type: domain.EvUserMessage, Payload: map[string]any{"content": []any{ + map[string]any{"type": "text", "text": "first"}, + }}}, + {Type: domain.EvUserMessage, Payload: map[string]any{"content": []any{ + map[string]any{"type": "text", "text": "second"}, + }}}, + }) + if err != nil { + t.Fatal(err) + } + if len(admission.Runs) != 2 { + t.Fatalf("runs = %d, want 2", len(admission.Runs)) + } + + firstClaim, ok, err := runs.ClaimNext(ctx, session.ID) + if err != nil || !ok { + t.Fatalf("first claim: ok=%v err=%v", ok, err) + } + // Run N+1 is not claimable while run N is running. + if _, ok, err := runs.ClaimNext(ctx, session.ID); err != nil || ok { + t.Fatalf("second claim before first completes: ok=%v err=%v", ok, err) + } + + // Run N commits agent output and marks its trigger processed. + if _, err := runs.Complete(ctx, firstClaim.Run.ID, []domain.EventDraft{ + {Type: domain.EvAgentMessage, Payload: map[string]any{"content": []any{ + map[string]any{"type": "text", "text": "answer-one"}, + }}}, + {Type: domain.EvSessionStatusIdle}, + }, domain.StatusRunning, nil); err != nil { + t.Fatal(err) + } + + // The first trigger is now processed. + es := NewEventStore(db, domain.NewSeqIDGen(), domain.FixedClock{T: session.CreatedAt}) + history, err := es.History(ctx, session.ID, 0, 100) + if err != nil { + t.Fatal(err) + } + if history[0].ProcessedAt == nil { + t.Fatal("first trigger not marked processed after run N completed") + } + + // Run N+1 is now claimable, and history projected for it includes the agent + // output committed by run N. + secondClaim, ok, err := runs.ClaimNext(ctx, session.ID) + if err != nil || !ok || secondClaim.Run.ID != admission.Runs[1].ID { t.Fatalf("second claim = %#v ok=%v err=%v", secondClaim.Run, ok, err) } + history, err = es.History(ctx, session.ID, 0, 100) + if err != nil { + t.Fatal(err) + } + var sawAgentMessage bool + for _, event := range history { + if event.Type == domain.EvAgentMessage { + sawAgentMessage = true + } + } + if !sawAgentMessage { + t.Fatal("history for run N+1 does not include run N's committed agent output") + } +} + +// TestRunStore_TerminatedSessionNeverClaims proves a terminated session is +// final: ClaimNext must not claim leftover queued work nor reopen the session. +func TestRunStore_TerminatedSessionNeverClaims(t *testing.T) { + db, runs, session := newRunStoreFixture(t) + ctx := context.Background() + admission, err := runs.CreateSession(ctx, session, []domain.EventDraft{ + {Type: domain.EvUserMessage, Payload: map[string]any{"content": []any{ + map[string]any{"type": "text", "text": "first"}, + }}}, + {Type: domain.EvUserMessage, Payload: map[string]any{"content": []any{ + map[string]any{"type": "text", "text": "second"}, + }}}, + }) + if err != nil { + t.Fatal(err) + } + if len(admission.Runs) != 2 { + t.Fatalf("runs = %d, want 2", len(admission.Runs)) + } + + firstClaim, ok, err := runs.ClaimNext(ctx, session.ID) + if err != nil || !ok { + t.Fatalf("first claim: ok=%v err=%v", ok, err) + } + // Run N terminates the session even though run N+1 is still queued. + msg := "boom" + if _, err := runs.Complete(ctx, firstClaim.Run.ID, []domain.EventDraft{ + {Type: domain.EvSessionError, Payload: map[string]any{"error": map[string]any{ + "type": "api_error", "message": msg, + }}}, + {Type: domain.EvSessionStatusTerminated}, + }, domain.StatusTerminated, &msg); err != nil { + t.Fatal(err) + } + + // The leftover queued run must never be claimed, and the session must stay + // terminated. + if claim, ok, err := runs.ClaimNext(ctx, session.ID); err != nil || ok { + t.Fatalf("claim after termination: claim=%#v ok=%v err=%v", claim.Run, ok, err) + } + stored, err := NewSessionRepo(db).Get(ctx, session.ID) + if err != nil { + t.Fatal(err) + } + if stored.Status != domain.StatusTerminated { + t.Fatalf("session status = %s, want terminated", stored.Status) + } +} + +// TestRunStore_ModelHistorySurvivesReopenInCausalOrder proves the durable output +// association: after run A completes (persisting its committed output event ids +// in the same transaction that closes it), the process can close and reopen a +// file-backed database, and RunStore.ModelHistory for run B still reconstructs +// trigger(A), output(A), trigger(B) in causal order — reading the persisted +// column, not any in-memory state. +func TestRunStore_ModelHistorySurvivesReopenInCausalOrder(t *testing.T) { + path := filepath.Join(t.TempDir(), "causal.db") + db, err := Open(path) + if err != nil { + t.Fatal(err) + } + ctx := context.Background() + now := time.Unix(1, 0).UTC() + ids := domain.NewSeqIDGen() + clk := domain.FixedClock{T: now} + if err := NewAgentRepo(db).PutVersion(ctx, domain.Agent{ + ID: "agent_1", Version: 1, Name: "agent", + Model: domain.Model{ID: "model"}, CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatal(err) + } + if err := NewEnvironmentRepo(db).Put(ctx, domain.Environment{ + ID: "env_1", Name: "environment", ConfigType: "cloud", + CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatal(err) + } + session := domain.Session{ + ID: "sesn_1", AgentID: "agent_1", AgentVersion: 1, + EnvironmentID: "env_1", Status: domain.StatusIdle, + CreatedAt: now, UpdatedAt: now, + } + runs := NewRunStore(db, ids, clk) + + // Admit A and B in one batch (two runs), claim and complete A with a distinct + // agent output. Then close the database. + admission, err := runs.CreateSession(ctx, session, []domain.EventDraft{ + {Type: domain.EvUserMessage, Payload: map[string]any{"content": []any{ + map[string]any{"type": "text", "text": "A"}, + }}}, + {Type: domain.EvUserMessage, Payload: map[string]any{"content": []any{ + map[string]any{"type": "text", "text": "B"}, + }}}, + }) + if err != nil { + t.Fatal(err) + } + if len(admission.Runs) != 2 { + t.Fatalf("runs = %d, want 2", len(admission.Runs)) + } + runBID := admission.Runs[1].ID + + claimA, ok, err := runs.ClaimNext(ctx, session.ID) + if err != nil || !ok { + t.Fatalf("claim A: ok=%v err=%v", ok, err) + } + if _, err := runs.Complete(ctx, claimA.Run.ID, []domain.EventDraft{ + {Type: domain.EvAgentMessage, Payload: map[string]any{"content": []any{ + map[string]any{"type": "text", "text": "reply-A"}, + }}}, + {Type: domain.EvSessionStatusIdle}, + }, domain.StatusRunning, nil); err != nil { + t.Fatal(err) + } + if err := db.Close(); err != nil { + t.Fatal(err) + } + + // Reopen the file-backed database in a fresh RunStore and reconstruct history + // for run B purely from persisted state. + reopened, err := Open(path) + if err != nil { + t.Fatal(err) + } + defer reopened.Close() + reopenedRuns := NewRunStore(reopened, domain.NewSeqIDGen(), clk) + + runB, err := reopenedRuns.Get(ctx, runBID) + if err != nil { + t.Fatal(err) + } + history, err := reopenedRuns.ModelHistory(ctx, runB, 10000) + if err != nil { + t.Fatal(err) + } + + // Expect exactly trigger(A), output(A) [agent.message reply-A, status_idle], + // trigger(B), in causal order. + msgs := domain.ProjectMessages(history) + if len(msgs) != 3 { + t.Fatalf("projected messages = %d, want 3: %#v", len(msgs), msgs) + } + if msgs[0].Role != domain.RoleUser || msgs[0].Content[0].Text != "A" { + t.Fatalf("messages[0] = %#v, want user 'A'", msgs[0]) + } + if msgs[1].Role != domain.RoleAssistant || msgs[1].Content[0].Text != "reply-A" { + t.Fatalf("messages[1] = %#v, want assistant 'reply-A'", msgs[1]) + } + if msgs[2].Role != domain.RoleUser || msgs[2].Content[0].Text != "B" { + t.Fatalf("messages[2] = %#v, want user 'B'", msgs[2]) + } + + // And prove the trigger(A), output(A), trigger(B) event ordering directly on + // the raw reconstructed events, independent of projection folding. + if len(history) < 4 { + t.Fatalf("reconstructed history has %d events, want >=4", len(history)) + } + if history[0].ID != claimA.Run.TriggerEventIDs[0] { + t.Fatalf("history[0] is not trigger(A): %#v", history[0]) + } + if history[len(history)-1].ID != runB.TriggerEventIDs[0] { + t.Fatalf("last history event is not trigger(B): %#v", history[len(history)-1]) + } + var sawAgentBetween bool + for _, e := range history[1 : len(history)-1] { + if e.Type == domain.EvAgentMessage { + sawAgentBetween = true + } + } + if !sawAgentBetween { + t.Fatal("output(A) agent.message did not survive reopen between trigger(A) and trigger(B)") + } } diff --git a/internal/store/schema.sql b/internal/store/schema.sql index 1b032c6..84e6586 100644 --- a/internal/store/schema.sql +++ b/internal/store/schema.sql @@ -44,6 +44,11 @@ CREATE TABLE IF NOT EXISTS session_runs ( session_id TEXT NOT NULL, admission_seq INTEGER NOT NULL, trigger_event_ids TEXT NOT NULL, + -- Internal-only. The exact committed output event ids this run appended when + -- it closed, persisted in the same transaction that closes the run so there is + -- never a completed run without its output association. Empty until the run + -- completes. Never serialized onto the public wire. + output_event_ids TEXT NOT NULL DEFAULT '[]', state TEXT NOT NULL, error TEXT, created_at TEXT NOT NULL,