diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 553c4d2..c6984ad 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -113,7 +113,7 @@ jobs: run: | set -euo pipefail dex_log="$RUNNER_TEMP/dexcli.log" - PATH="$PWD/.cache/temporal-v1.8.2:$PATH" .cache/dexcli-v0.7.0 dev \ + PATH="$PWD/.cache/temporal-v1.8.2:$PATH" .cache/dexcli-v0.9.0 dev \ -open=false \ -blob-store-dir "$RUNNER_TEMP/dex-blobs" \ -sqlite-db-filename "$RUNNER_TEMP/dex.sqlite.db" \ diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index e9da8d7..80ee5b4 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -172,10 +172,14 @@ bubbles. Replays are idempotent and cannot remove messages that arrived later. A later authoritative Snapshot at a real waiting boundary reopens the gate. Stream loss is corrected by Snapshot. -Approval and Timer setup emit a hidden `snapshot_required` Activity control -event after their durable payload is committed. It requests one non-blocking -Snapshot so waits that intentionally do not advance `WaitingInputRound` remain -visible without polling. +Consumed user bubbles are anchored after the latest explicit message sequence +seen in the Activity stream. This preserves causal ordering while Snapshot is +temporarily behind the Stream. + +The approval and Timer target Steps emit a hidden `snapshot_required` Activity +control from `WaitFor`, after `RouteTool` commits their durable payload. It +requests one non-blocking Snapshot so waits that intentionally do not advance +`WaitingInputRound` remain visible without polling. Every mutation result immediately requests another Snapshot. Mutation controls remain disabled until that Snapshot succeeds. Server errors and explicit @@ -300,8 +304,9 @@ Runtime metadata therefore remains stable for the logical call. `internal/app` owns every long-lived resource. Startup validates configuration, discovers MCP, constructs providers, opens BlobCache, starts the Worker, waits -for its listener, marks readiness, and then serves the API. Any startup failure -closes everything already constructed. +for its listener, marks readiness, and then serves the API. The Dex `v0.9.0` +Worker negotiates a compatible Server protocol before synchronizing indexes or +binding. Any startup failure closes everything already constructed. On cancellation or an unexpected Worker/HTTP exit, readiness is cleared. The HTTP server and Worker receive bounded shutdown contexts, their goroutines are diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index cf2f332..849b90f 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -12,9 +12,11 @@ against the installed released SDK and a version-matched runnable example or real-server compile-contract test. Snapshot, Stream, Channel size snapshot, and Attribute wait code target Dex Go -SDK `v0.7.1` and Server `v0.7.0`. Recheck the installed SDK source and the -installed skill before changing resource projection or errors. Never infer an -API from a design screenshot or unreleased branch. +SDK and Server `v0.9.0`. Version `v0.9.0` Workers negotiate the Server protocol +before binding, so deployments upgrade the Server before the Worker. Recheck +the installed SDK source and the installed skill before changing resource +projection or errors. Never infer an API from a design screenshot or unreleased +branch. ## Deployment boundary diff --git a/Makefile b/Makefile index b9a3865..e4b121c 100644 --- a/Makefile +++ b/Makefile @@ -6,7 +6,7 @@ GO_BUILD_CACHE := $(CURDIR)/.cache/go-build GO_PACKAGES := ./agent/... ./cmd/... ./internal/... ./model/... -DEXCLI_VERSION := v0.7.0 +DEXCLI_VERSION := v0.9.0 DEXCLI_BINARY := $(CURDIR)/.cache/dexcli-$(DEXCLI_VERSION) OSV_SCANNER_VERSION := v2.5.1 OSV_SCANNER_BINARY := $(CURDIR)/.cache/osv-scanner-$(OSV_SCANNER_VERSION) diff --git a/README.md b/README.md index edf2a11..1ceaadc 100644 --- a/README.md +++ b/README.md @@ -76,7 +76,7 @@ and resource model. - Go matching [`go.mod`](go.mod) - Node.js and npm compatible with [`web/package-lock.json`](web/package-lock.json) -- A Dex `v0.7.0` server +- A Dex `v0.9.0` server - A writable directory for disposable Dex BlobCache data ## Quick start @@ -89,7 +89,9 @@ make build-api make build-web ``` -Start a compatible Dex server. Then run the API and Worker: +Start a compatible Dex server. Dex `v0.9.0` Workers require the Server +compatibility RPC, so upgrade the Server before the Worker. Then run the API and +Worker: ```bash SUPERAGENT_HTTP_ALLOWED_ORIGINS=http://127.0.0.1:3000 ./bin/superagent diff --git a/docs/adr/0012-watermark-and-input-consumption-reconciliation.md b/docs/adr/0012-watermark-and-input-consumption-reconciliation.md index bc41226..e1890a6 100644 --- a/docs/adr/0012-watermark-and-input-consumption-reconciliation.md +++ b/docs/adr/0012-watermark-and-input-consumption-reconciliation.md @@ -49,10 +49,10 @@ request, writes an `input_consumed` Activity event. The structured payload contains exact message IDs and nullable Plan revision, while display text contains only counts or revision. The frontend applies it idempotently and closes Plan actions until Snapshot confirms the next real waiting boundary. -Approval and Timer setup write a hidden `snapshot_required` Activity control -after their durable payload is committed. The browser uses it for one -non-blocking Snapshot because those waits intentionally do not advance the -watermark. +The approval and Timer target `WaitFor` methods write a hidden +`snapshot_required` Activity control after `RouteTool` commits their durable +payload. The browser uses it for one non-blocking Snapshot because those waits +intentionally do not advance the watermark. `AnswerQuestions` durably publishes a validated answer to its dedicated Channel. `AwaitUser.Execute` appends it to application history and writes a @@ -74,7 +74,8 @@ Consumed IDs suppress stale queue data until durable history replaces the temporary projection. Snapshot remains the only authoritative durable reconciliation model. -Deployments must use Dex Server `v0.7.0`, Dex Go SDK `v0.7.1`, and the matching -Worker and browser behavior together. They must stop or clear Agent Flows -created with the removed schema before rollout; there is no old-Attribute or -Runtime Lease compatibility shim. +Deployments must use Dex Server and Go SDK `v0.9.0`, and the matching Worker and +browser behavior together. The Server must be upgraded first because `v0.9.0` +Workers reject Servers without protocol negotiation. Deployments must stop or +clear Agent Flows created with the removed schema before rollout; there is no +old-Attribute or Runtime Lease compatibility shim. diff --git a/docs/flow-model.md b/docs/flow-model.md index 3a9e009..8d5d2f3 100644 --- a/docs/flow-model.md +++ b/docs/flow-model.md @@ -14,11 +14,16 @@ `GetArchivedMessages` - Browser synchronization Attribute: `WaitingInputRound` -The implementation requires Dex Go SDK `v0.7.1` and Server `v0.7.0`. Each +The implementation requires Dex Go SDK and Server `v0.9.0`. Each `WaitFor`, `Execute`, and RPC invocation is an independent Dex atomic commit. Provider and MCP calls are external effects and are not part of a Dex transaction. +The `v0.9.0` Worker negotiates the highest common protocol with the Server before +Attribute index synchronization or Worker binding. Deploy the Server before the +Worker. Startup fails when `GetServerInfo` is missing, either interval is +invalid, or the intervals do not overlap. + Renewable sandbox credentials are not an Agent Flow resource. A future, separately designed `SandboxLifecycleFlow` will own that lifecycle. @@ -86,8 +91,8 @@ history, and makes the model replan. | `RecoverToolExecution` | none | Record one unknown result after exhausted Dex retries, then let the Agent continue | | `DurableWait` | Timer or steering | Persist waiting status; record completion or interruption and continue | -Dex Server `v0.7.0` supplies Channel size metadata to the Worker, and Dex Go SDK -`v0.7.1` exposes it in `WaitFor` and `Execute`. `AwaitUser.WaitFor` reads the +Dex Server and Go SDK `v0.9.0` expose Channel size metadata in `WaitFor` and +`Execute`. `AwaitUser.WaitFor` reads the sizes of `SteeredUserMessages`, `QueuedUserMessages`, and the current `PlanExecutions` instance without loading message payloads. It increments `WaitingInputRound` only when the selected Channel set is empty and the Step @@ -169,9 +174,10 @@ sequence, but not the answer text. The browser correlates it with the locally submitted answer for immediate display and requests Snapshot for authoritative reconciliation. -Approval and Timer setup write a `snapshot_required` control event after their -durable payload is committed. The browser hides this event and requests one -non-blocking Snapshot. These waits do not advance `WaitingInputRound`. +The approval and Timer target `WaitFor` methods write a `snapshot_required` +control event after `RouteTool` commits their durable payload. The browser hides +this event and requests one non-blocking Snapshot. These waits do not advance +`WaitingInputRound`. ## Snapshot and history diff --git a/go.mod b/go.mod index ffc725c..822d198 100644 --- a/go.mod +++ b/go.mod @@ -12,7 +12,7 @@ require ( github.com/ogen-go/ogen v1.24.0 github.com/openai/openai-go/v3 v3.55.0 github.com/superdurable/dex/blob-cache-go v0.1.0 - github.com/superdurable/dex/sdk-go v0.7.1 + github.com/superdurable/dex/sdk-go v0.9.0 golang.org/x/net v0.58.0 gopkg.in/yaml.v3 v3.0.1 ) diff --git a/go.sum b/go.sum index c9006fe..ef24fa7 100644 --- a/go.sum +++ b/go.sum @@ -62,8 +62,8 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/superdurable/dex/blob-cache-go v0.1.0 h1:+c3H5YBWG3DlICOHbgT9IUM5vTlfLYGP5Vd5sWr7WTY= github.com/superdurable/dex/blob-cache-go v0.1.0/go.mod h1:Atepb7+sztvDCztVKmlvEKCSKFCkHKtDhoFYjaFmtEw= -github.com/superdurable/dex/sdk-go v0.7.1 h1:BL+VPCnmTgYC+2Tog2dEjncgVRSQjpRXdAiA06dQiJU= -github.com/superdurable/dex/sdk-go v0.7.1/go.mod h1:8Wj5wPf9dyb7hDnA40j8xISR/zjhX57NrUDVcXgf5x8= +github.com/superdurable/dex/sdk-go v0.9.0 h1:F1kJnQMGPMR6pXWiB3ZJk2FPdqpaQ8PQ0+ukfrJEvw4= +github.com/superdurable/dex/sdk-go v0.9.0/go.mod h1:8Wj5wPf9dyb7hDnA40j8xISR/zjhX57NrUDVcXgf5x8= github.com/tidwall/gjson v1.14.2/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk= github.com/tidwall/gjson v1.19.0 h1:xwxm7n691Uf3u5OFjzngavjGTh55KX5q/9w9xHW88JU= github.com/tidwall/gjson v1.19.0/go.mod h1:V37/opeE/JbLUOfH0QTXiNez2l0RUjYUhpT4szFQAfc= diff --git a/internal/agent/flow.go b/internal/agent/flow.go index 2c39a3a..c70fba2 100644 --- a/internal/agent/flow.go +++ b/internal/agent/flow.go @@ -2076,9 +2076,6 @@ func (step routeToolStep) Execute(ctx dex.Context, _ dex.None) (*dex.StepDecisio }); err != nil { return nil, err } - if err := step.flow.writeSnapshotRequired(ctx); err != nil { - return nil, err - } return dex.GoTo(checkSteeredStep{flow: step.flow}, continueDurableWait), nil } if call.Name == ToolNameRequestUserInput { @@ -2155,9 +2152,6 @@ func (step routeToolStep) Execute(ctx dex.Context, _ dex.None) (*dex.StepDecisio }); err != nil { return nil, err } - if err := step.flow.writeSnapshotRequired(ctx); err != nil { - return nil, err - } return dex.GoTo(checkSteeredStep{flow: step.flow}, continueAwaitToolApproval), nil } return dex.GoTo(checkSteeredStep{flow: step.flow}, continueExecuteToolRetry), nil @@ -2182,6 +2176,9 @@ func (step awaitToolApprovalStep) WaitFor(ctx dex.Context, _ dex.None) (*dex.Wai if err := step.flow.updateStatus(ctx, AgentStatusWaitingForToolApproval); err != nil { return nil, err } + if err := step.flow.writeSnapshotRequired(ctx); err != nil { + return nil, err + } return dex.AnyOf( steeredUserMessagesChannel.AtLeastAtMost(1, maximumSteeringMessageCount), toolApprovalsChannel.ForOne(string(call.ID)), @@ -2479,6 +2476,9 @@ func (step durableWaitStep) WaitFor(ctx dex.Context, _ dex.None) (*dex.Wait, err if err := step.flow.updateStatus(ctx, AgentStatusWaitingForTimer); err != nil { return nil, err } + if err := step.flow.writeSnapshotRequired(ctx); err != nil { + return nil, err + } return dex.AnyOf( dex.Timer(time.Duration(timer.DurationSeconds)*time.Second), steeredUserMessagesChannel.AtLeastAtMost(1, maximumSteeringMessageCount), diff --git a/internal/agent/flow_integration_test.go b/internal/agent/flow_integration_test.go index b96fa7e..981a6ac 100644 --- a/internal/agent/flow_integration_test.go +++ b/internal/agent/flow_integration_test.go @@ -166,13 +166,17 @@ func TestAgentFlowDurabilityIntegration(t *testing.T) { if err := environment.agent.SendMessage(t.Context(), flowID, UserMessage{Content: "/tool"}); err != nil { t.Fatal(err) } - approval := waitForPendingApproval(t, environment, flowID) - if approval.ToolName != integrationToolName { - t.Fatalf("pending tool = %q", approval.ToolName) - } snapshotRequired := readActivityUntil(t, environment.agent, flowID, func(event AgentEvent) bool { return event.Kind == EventKindSnapshotRequired }) + approvalSnapshot := readSnapshot(t, environment, flowID) + if approvalSnapshot.Description == nil || approvalSnapshot.Description.PendingApproval == nil { + t.Fatalf("Snapshot after approval control = %#v", approvalSnapshot) + } + approval := *approvalSnapshot.Description.PendingApproval + if approval.ToolName != integrationToolName { + t.Fatalf("pending tool = %q", approval.ToolName) + } if snapshotRequired.Activity.Message != "Durable interaction state changed." || snapshotRequired.Activity.InputConsumption != nil { t.Fatalf("Snapshot control Activity = %#v", snapshotRequired.Activity) @@ -345,6 +349,27 @@ func TestAgentFlowDurabilityIntegration(t *testing.T) { } } +func TestAgentTimerSnapshotNotificationIntegration(t *testing.T) { + environment := newAgentIntegrationEnvironment(t, integrationModel{}, newIntegrationToolRegistry()) + flowID := FlowID("agent-timer-snapshot-" + randomLocalID(t)) + if _, err := environment.agent.Start(t.Context(), flowID, StartRequest{Config: NewAgentConfig()}); err != nil { + t.Fatal(err) + } + waitForAgentState(t, environment, flowID, func(state AgentState) bool { + return state.Status == AgentStatusWaitingForMessage + }) + if err := environment.agent.SendMessage(t.Context(), flowID, UserMessage{Content: "/wait"}); err != nil { + t.Fatal(err) + } + readActivityUntil(t, environment.agent, flowID, func(event AgentEvent) bool { + return event.Kind == EventKindSnapshotRequired + }) + snapshot := readSnapshot(t, environment, flowID) + if snapshot.Description == nil || snapshot.Description.PendingTimer == nil { + t.Fatalf("Snapshot after timer control = %#v", snapshot) + } +} + func TestAgentWaitingInputRoundIntegration(t *testing.T) { environment := newAgentIntegrationEnvironment(t, integrationModel{}, newIntegrationToolRegistry()) flowID := FlowID("agent-interaction-" + randomLocalID(t)) diff --git a/script/install-dexcli.sh b/script/install-dexcli.sh index b72d84e..4aef5d3 100755 --- a/script/install-dexcli.sh +++ b/script/install-dexcli.sh @@ -22,10 +22,10 @@ esac archive_name="dexcli_${version}_${operating_system}_${architecture}.tar.gz" case "$archive_name" in - dexcli_v0.7.0_darwin_amd64.tar.gz) checksum=4631236fb3c491b80848534c0bd9ae804d7c2222196fb447dabe2de3759bb66a ;; - dexcli_v0.7.0_darwin_arm64.tar.gz) checksum=72ecd690559c94e97346bd9aacc1ab922104a90a3b57c7a045dd31b437ce2a3f ;; - dexcli_v0.7.0_linux_amd64.tar.gz) checksum=3f3d93c774528dfd5c8c052ebba50f9b21a9db7d9c5812a01c422234ebe4ebb7 ;; - dexcli_v0.7.0_linux_arm64.tar.gz) checksum=cd08d799bf3bc394232823152c2ff7dd8580d438166525f48f1de2bc1ad2bad1 ;; + dexcli_v0.9.0_darwin_amd64.tar.gz) checksum=071f530422e869554b2e2a2dc10ce5d917e1093a38a5af4e1438192e9c532408 ;; + dexcli_v0.9.0_darwin_arm64.tar.gz) checksum=4ee2df39d0218169b5fe0fc581e9cac2c1f40e24a011ac5c5ba441eccdfd1f51 ;; + dexcli_v0.9.0_linux_amd64.tar.gz) checksum=0df459cdde367191e7c962b819a1491073b614da93f5459129f38f90970a7016 ;; + dexcli_v0.9.0_linux_arm64.tar.gz) checksum=68f5771cde6ae4a1cfb8c78efb35881765273d4727d6353de41d6b4252476d67 ;; *) echo "no checksum is pinned for $archive_name" >&2; exit 1 ;; esac diff --git a/web/src/conversation-state.test.ts b/web/src/conversation-state.test.ts index 0f8579a..46f9633 100644 --- a/web/src/conversation-state.test.ts +++ b/web/src/conversation-state.test.ts @@ -353,6 +353,58 @@ describe("conversationReducer", () => { }); }); + it("anchors consumed input after the latest streamed message sequence", () => { + let state = conversationReducer(initialConversationState(), { + type: "snapshot-loaded", + snapshot: snapshot("run-1", "queued-1", "first"), + }); + state = conversationReducer(state, { + type: "stream-update", + update: { + kind: "activity", + resumeToken: "model-completed-1", + source: "model-call-1", + createdAt: "2026-09-03T00:00:01Z", + value: { + kind: EventKind.MODEL_COMPLETED, + message: "Model response completed.", + callId: null, + toolName: null, + messageSequence: 3, + inputConsumption: null, + }, + }, + }); + state = conversationReducer(state, { + type: "stream-update", + update: { + kind: "activity", + resumeToken: "input-consumed-1", + source: "check-steered-1", + createdAt: "2026-09-03T00:00:02Z", + value: { + kind: EventKind.INPUT_CONSUMED, + message: "Consumed 1 queued user message.", + callId: null, + toolName: null, + messageSequence: null, + inputConsumption: { + queuedMessageIds: ["queued-1"], + steeredMessageIds: [], + planExecutionRevision: null, + }, + }, + }, + }); + + expect(state).toMatchObject({ + kind: "ready", + consumedUserMessages: [ + { messageId: "queued-1", consumedAfterSequence: 3 }, + ], + }); + }); + it("applies a matching Plan task hint until Snapshot reconciliation", () => { const initial = withPlan( snapshotWithStatus(AgentStatus.CALLING_MODEL), diff --git a/web/src/conversation-state.ts b/web/src/conversation-state.ts index 71c5e0a..af029f8 100644 --- a/web/src/conversation-state.ts +++ b/web/src/conversation-state.ts @@ -836,6 +836,13 @@ function applyInputConsumption( const projectedIDs = new Set( state.consumedUserMessages.map((message) => message.messageId), ); + const consumedAfterSequence = state.activities.reduce( + (latestSequence, activity) => + activity.value.messageSequence === null + ? latestSequence + : Math.max(latestSequence, activity.value.messageSequence), + state.snapshot.description.lastSequence, + ); const newlyConsumedUserMessages = [...consumedIDs] .filter((messageID) => !projectedIDs.has(messageID)) .flatMap((messageID): ConsumedUserEntry[] => { @@ -847,7 +854,7 @@ function applyInputConsumption( messageId: messageID, value: pending.value, createdAt: update.createdAt, - consumedAfterSequence: state.snapshot.description.lastSequence, + consumedAfterSequence, }, ]; }); diff --git a/web/tests/full-stack.spec.ts b/web/tests/full-stack.spec.ts index c25d3f5..951876a 100644 --- a/web/tests/full-stack.spec.ts +++ b/web/tests/full-stack.spec.ts @@ -124,6 +124,7 @@ test("renders chronological transient activity and durable queue interactions", exact: true, }); await expect(reasoningCard).toBeVisible(); + await expect(composer).toBeFocused(); if (!(await reasoningText.isVisible())) { await reasoningCard.locator("summary").click(); } @@ -140,7 +141,6 @@ test("renders chronological transient activity and durable queue interactions", ).toHaveCount(1); await expect(history.locator(".live-message")).toHaveCount(0); await expect(history.locator(".activity-entry")).toHaveCount(3); - await expect(composer).toBeFocused(); await expect( history .locator(".activity-entry")