diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c6984ad..b89f2e3 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.9.0 dev \ + PATH="$PWD/.cache/temporal-v1.8.2:$PATH" .cache/dexcli-v0.10.0 dev \ -open=false \ -blob-store-dir "$RUNNER_TEMP/dex-blobs" \ -sqlite-db-filename "$RUNNER_TEMP/dex.sqlite.db" \ diff --git a/.github/workflows/dex-release-upgrade.yml b/.github/workflows/dex-release-upgrade.yml deleted file mode 100644 index a1c8ef7..0000000 --- a/.github/workflows/dex-release-upgrade.yml +++ /dev/null @@ -1,59 +0,0 @@ -name: Prepare Dex release upgrade - -on: - repository_dispatch: - types: [dex-release-published] - -permissions: - contents: read - -concurrency: - group: superagent-dex-${{ github.event.client_payload.version }} - cancel-in-progress: false - -jobs: - upgrade: - if: github.event.sender.type == 'Bot' - runs-on: ubuntu-24.04 - timeout-minutes: 30 - steps: - - name: Create repository-scoped release automation token - id: app-token - uses: actions/create-github-app-token@v2 - with: - app-id: ${{ vars.RELEASE_AUTOMATION_APP_ID }} - private-key: ${{ secrets.RELEASE_AUTOMATION_PRIVATE_KEY }} - owner: superdurable - repositories: superagent - - uses: actions/checkout@v7 - - uses: actions/setup-go@v6 - with: - go-version-file: go.mod - - name: Update immutable Dex release pins - env: - DEX_MANIFEST_SHA256: ${{ github.event.client_payload.manifest_sha256 }} - DEX_MANIFEST_URL: ${{ github.event.client_payload.manifest_url }} - run: | - python3 script/update_dex_release.py \ - --manifest-url "${DEX_MANIFEST_URL}" \ - --manifest-sha256 "${DEX_MANIFEST_SHA256}" - GOWORK=off go mod tidy - python3 -m unittest script/update_dex_release_test.py - python3 script/check_dex_release.py - - name: Open draft upgrade pull request before compilation - uses: peter-evans/create-pull-request@v8 - with: - token: ${{ steps.app-token.outputs.token }} - branch: automation/dex-v${{ github.event.client_payload.version }} - delete-branch: true - draft: true - commit-message: Upgrade SuperAgent to Dex ${{ github.event.client_payload.version }} - title: Upgrade SuperAgent to Dex ${{ github.event.client_payload.version }} - body: | - Automated from the immutable Dex compatibility manifest. - - Review `openFlowsCompatibility` before merging. It defaults to - `cancel-required`; automation never claims open-Flow compatibility. - - Normal pull-request CI owns compilation and tests. When a Dex SDK - API breaks, continue the required migration in this draft PR. diff --git a/.github/workflows/github-release-ui.yml b/.github/workflows/github-release-ui.yml index 1785310..6f6cdef 100644 --- a/.github/workflows/github-release-ui.yml +++ b/.github/workflows/github-release-ui.yml @@ -106,71 +106,3 @@ jobs: install_url="https://github.com/${GITHUB_REPOSITORY}/releases/download/${RELEASE_TAG}/${asset_name}" echo "Install with: npm install ${install_url} react" >> "${GITHUB_STEP_SUMMARY}" - - notify-downstreams: - name: Request IaC and SuperVerse upgrades - if: github.event_name == 'release' - needs: attach - runs-on: ubuntu-24.04 - steps: - - uses: actions/checkout@v7 - with: - ref: ${{ github.event.release.tag_name }} - - name: Validate release compatibility declaration - id: release - env: - RELEASE_TAG: ${{ github.event.release.tag_name }} - run: | - python3 script/check_dex_release.py - python3 - "${RELEASE_TAG}" "${GITHUB_OUTPUT}" <<'PY' - import json - from pathlib import Path - import re - import subprocess - import sys - - tag, output_path = sys.argv[1:] - if re.fullmatch(r"v[0-9]+\.[0-9]+\.[0-9]+", tag) is None: - raise SystemExit("SuperAgent release tag is not stable semver") - lock = json.loads(Path("dex-release.lock.json").read_text(encoding="utf-8")) - commit = subprocess.run( - ["git", "rev-parse", "HEAD"], check=True, capture_output=True, text=True - ).stdout.strip() - with Path(output_path).open("a", encoding="utf-8") as output: - for name, value in { - "dex_version": lock["release"], - "manifest_url": lock["manifest"]["url"], - "manifest_sha256": lock["manifest"]["sha256"], - "open_flows_compatibility": lock["openFlowsCompatibility"], - "superagent_commit": commit, - }.items(): - print(f"{name}={value}", file=output) - PY - - name: Create repository-scoped release automation token - id: app-token - uses: actions/create-github-app-token@v2 - with: - app-id: ${{ vars.RELEASE_AUTOMATION_APP_ID }} - private-key: ${{ secrets.RELEASE_AUTOMATION_PRIVATE_KEY }} - owner: superdurable - repositories: iac,superverse - - name: Dispatch audited downstream upgrades - env: - DEX_MANIFEST_SHA256: ${{ steps.release.outputs.manifest_sha256 }} - DEX_MANIFEST_URL: ${{ steps.release.outputs.manifest_url }} - DEX_VERSION: ${{ steps.release.outputs.dex_version }} - GH_TOKEN: ${{ steps.app-token.outputs.token }} - OPEN_FLOWS_COMPATIBILITY: ${{ steps.release.outputs.open_flows_compatibility }} - SUPERAGENT_COMMIT: ${{ steps.release.outputs.superagent_commit }} - SUPERAGENT_RELEASE: ${{ github.event.release.tag_name }} - run: | - for repository in iac superverse; do - gh api --method POST "repos/superdurable/${repository}/dispatches" \ - -f event_type=superagent-release-published \ - -f "client_payload[dex_version]=${DEX_VERSION}" \ - -f "client_payload[manifest_url]=${DEX_MANIFEST_URL}" \ - -f "client_payload[manifest_sha256]=${DEX_MANIFEST_SHA256}" \ - -f "client_payload[superagent_release]=${SUPERAGENT_RELEASE}" \ - -f "client_payload[superagent_commit]=${SUPERAGENT_COMMIT}" \ - -f "client_payload[open_flows_compatibility]=${OPEN_FLOWS_COMPATIBILITY}" - done diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 6bf6c55..ea1fadc 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -75,8 +75,10 @@ prompts, and cumulative context summaries are separate typed Attributes. Snapshot is the only durable current-interaction and reconciliation read model. Archive paging is an immutable history continuation. Each page uses one -read-only Flow RPC that loads `AgentState` and one exact archive chunk, without -loading current interaction state or pending Channels. +read-only Flow RPC that loads `AgentState` and the retained archive map, then +returns one exact chunk without loading current interaction state or pending +Channels. Dex Go SDK `v0.10.0` fixes selective loads at RPC registration, so an +input-selected AttributeMap instance cannot be loaded independently. Commands follow Dex's transactional RPC model. There is no permanent command receipt, caller request ID, payload fingerprint, global mutation revision, or @@ -328,7 +330,7 @@ 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. The Dex `v0.9.0` +for its listener, marks readiness, and then serves the API. The Dex Go SDK `v0.10.0` Worker negotiates a compatible Server protocol before synchronizing indexes or binding. Any startup failure closes everything already constructed. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 8963fc8..b897ce6 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -12,19 +12,17 @@ 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 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 +SDK `v0.10.0`. Workers negotiate the Server protocol before binding. 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. -`dex-release.lock.json` binds the direct Go SDK requirement to one immutable -Dex manifest. A Dex publication opens an automated upgrade PR with open Flow -compatibility set to `cancel-required` for review. Publishing the subsequent -SuperAgent release dispatches the reviewed IaC and SuperVerse upgrades. -The automation opens the draft after asset validation and mechanical pin -updates, before product compilation. Resolve breaking SDK API migrations in -that draft; normal pull-request CI remains the merge gate. +Dex Go SDK and dexcli are independent direct dependencies. Run +`script/update_dex_versions.py` with explicit component versions, then run +`script/check_dex_versions.py`. The updater reads dexcli's native +`checksums.txt`; SuperAgent does not consume a cross-component compatibility +manifest. Resolve SDK API changes in the same pull request. Normal compilation, +real-Server integration, and browser E2E are the merge gates. ## Deployment boundary diff --git a/Makefile b/Makefile index 1b9c406..b72f2f7 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: audit-web build-api build-web check check-agent-rules check-flow-definition \ +.PHONY: audit-web build-api build-web check check-agent-rules check-dex-versions check-flow-definition \ check-generated copyright-check flow-visualize format-check fuzz generate \ generate-go generate-web governance-check install-dexcli install-osv-scanner install-temporal lint lint-go lint-web lint-workflows \ test test-agent test-api test-app test-config test-dex-integration test-mcp test-model test-openai-live \ @@ -6,7 +6,7 @@ GO_BUILD_CACHE := $(CURDIR)/.cache/go-build GO_PACKAGES := ./agent/... ./cmd/... ./internal/... ./model/... ./toolcontract/... -DEXCLI_VERSION := v0.9.0 +DEXCLI_VERSION := v0.10.0 DEXCLI_BINARY := $(CURDIR)/.cache/dexcli-$(DEXCLI_VERSION) OSV_SCANNER_VERSION := v2.5.1 OSV_SCANNER_BINARY := $(CURDIR)/.cache/osv-scanner-$(OSV_SCANNER_VERSION) @@ -48,6 +48,9 @@ copyright-check: governance-check: check-agent-rules copyright-check +check-dex-versions: + @python3 script/check_dex_versions.py + install-dexcli: $(DEXCLI_BINARY) $(DEXCLI_BINARY): script/install-dexcli.sh @@ -80,6 +83,10 @@ check-flow-definition: install-dexcli sed -n '/"diagnostics"/,$$p' "$${flow_definition}" >&2; \ exit 1; \ fi; \ + if grep -Fq '"name": "ExecuteTool"' "$${flow_definition}"; then \ + echo "Flow definition must not contain the removed ExecuteTool Step" >&2; \ + exit 1; \ + fi; \ for channel in answeredUserInputsChannel queuedUserMessagesChannel steeredUserMessagesChannel toolApprovalsChannel toolRecoveryDecisionsChannel parallelToolResultsChannel planExecutionsChannel; do \ if ! grep -Fq "\"id\": \"resource:channel:$${channel}\"" "$${flow_definition}" || \ ! grep -Fq "\"resourceId\": \"resource:channel:$${channel}\"" "$${flow_definition}"; then \ @@ -164,4 +171,4 @@ test-web: test-openai-live: @GOCACHE=$(GO_BUILD_CACHE) GOWORK=off go test -tags=live -count=1 -run '^TestLiveOpenAIResponses$$' ./internal/model -check: governance-check check-generated format-check build-api build-web vet lint test test-race test-web vulnerability-check audit-web +check: governance-check check-dex-versions check-generated format-check build-api build-web vet lint test test-race test-web vulnerability-check audit-web diff --git a/README.md b/README.md index ee21b3b..0cd111f 100644 --- a/README.md +++ b/README.md @@ -89,7 +89,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.9.0` server +- A Dex Server compatible with Dex Go SDK `v0.10.0` - A writable directory for disposable Dex BlobCache data ## Quick start @@ -102,9 +102,7 @@ make build-api make build-web ``` -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: +Start a compatible Dex server, then run the API and Worker: ```bash SUPERAGENT_HTTP_ALLOWED_ORIGINS=http://127.0.0.1:3000 ./bin/superagent @@ -152,6 +150,12 @@ persisted in Dex state or logged. Copy [`web/mcp-servers.example.yaml`](web/mcp-servers.example.yaml) to configure trusted MCP servers. +Each configured tool defaults to `running_type: short_running` and a 60-second +heartbeat timeout. Use `long_running` when more than half of expected calls +exceed five seconds. This is a Dex placement optimization, not a timeout or +SLA; short-running calls may fall back and complete normally. Keep the +heartbeat default unless a healthy tool can remain silent longer. + For a cross-origin frontend deployment, add its exact origin to `SUPERAGENT_HTTP_ALLOWED_ORIGINS`. Wildcards and credentialed cross-origin requests are intentionally unsupported. Serve `config.json` with diff --git a/agent/agent.go b/agent/agent.go index 7952152..7a84638 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -125,6 +125,9 @@ const ( ToolOutcomeKnownFailure = agentinternal.ToolOutcomeKnownFailure ToolOutcomeUnknown = agentinternal.ToolOutcomeUnknown + ToolRunningTypeShortRunning = agentinternal.ToolRunningTypeShortRunning + ToolRunningTypeLongRunning = agentinternal.ToolRunningTypeLongRunning + ToolRetryExhaustionPolicyManualRecovery = agentinternal.ToolRetryExhaustionPolicyManualRecovery ToolRetryExhaustionPolicyContinueWithUnknown = agentinternal.ToolRetryExhaustionPolicyContinueWithUnknown @@ -194,6 +197,7 @@ type ( EventKind = agentinternal.EventKind Provider = agentinternal.Provider ToolOutcome = agentinternal.ToolOutcome + ToolRunningType = agentinternal.ToolRunningType ToolRetryExhaustionPolicy = agentinternal.ToolRetryExhaustionPolicy ToolRecoveryResolution = agentinternal.ToolRecoveryResolution ToolRecoveryAction = agentinternal.ToolRecoveryAction diff --git a/dex-release.lock.json b/dex-release.lock.json deleted file mode 100644 index b1e74ea..0000000 --- a/dex-release.lock.json +++ /dev/null @@ -1,17 +0,0 @@ -{ - "schemaVersion": 1, - "release": "0.9.0", - "manifest": { - "url": "https://github.com/superdurable/dex/releases/download/server/v0.9.0/dex-compatibility-v0.9.0.json", - "sha256": "dc09203a2d785008d4449e23f70bd3598107e82d5f7934d86f49f1c534310906" - }, - "sourceCommit": "e93b803a829735292af8c81a0cc1c98b12aee7f7", - "sdkGoVersion": "0.9.0", - "protocol": { - "minimum": 1, - "maximum": 1 - }, - "runningFlowsCompatibility": "compatible", - "persistenceCompatibility": "compatible", - "openFlowsCompatibility": "cancel-required" -} diff --git a/docs/adr/0011-dex-owned-tool-retries.md b/docs/adr/0011-dex-owned-tool-retries.md index 577c6d2..17b7943 100644 --- a/docs/adr/0011-dex-owned-tool-retries.md +++ b/docs/adr/0011-dex-owned-tool-retries.md @@ -19,6 +19,11 @@ Go error and use Dex retry. Exhaustion follows the tool definition's recovery policy. It defaults to the manual boundary introduced by ADR 0013; explicitly configured tools may route to `RecoverToolExecution` and continue with unknown. +New Agent Flows default Step durability to ASYNC. Short-running tools inherit +that default and may fall back to regular execution. Long-running tools override +Execute durability to SYNC. Tool policy also supplies heartbeat timeout, while +attempt timeout bounds both Dex execution and the registry child context. + Approval and CallID remain stable across attempts. External effects promise recoverable at-least-once execution, not exactly-once execution. diff --git a/docs/adr/0012-watermark-and-input-consumption-reconciliation.md b/docs/adr/0012-watermark-and-input-consumption-reconciliation.md index e1890a6..80f06a2 100644 --- a/docs/adr/0012-watermark-and-input-consumption-reconciliation.md +++ b/docs/adr/0012-watermark-and-input-consumption-reconciliation.md @@ -74,8 +74,6 @@ 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 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. +Deployments must update the Worker and browser behavior together. The product +has not launched, so removed Agent schemas have no migration path. Runtime +protocol negotiation still rejects an unsupported Server. diff --git a/docs/adr/0014-registered-rpc-options.md b/docs/adr/0014-registered-rpc-options.md new file mode 100644 index 0000000..4b70086 --- /dev/null +++ b/docs/adr/0014-registered-rpc-options.md @@ -0,0 +1,41 @@ +# ADR 0014: Register immutable RPC execution options + +## Status + +Accepted on 2026-09-17. + +## Context + +Dex Go SDK `v0.10.0` replaces reflected RPC discovery with explicit `GetRPCs` +definitions. Timeout, locks, transactional execution, and selective collection +loads belong to the registered RPC definition. Callers can impose a shorter +context deadline, but cannot change those options per invocation. + +Most Agent RPCs always use the same resources. `GetArchivedMessages` is the +exception: its input selects one `ArchivedMessages` instance. The released SDK +cannot derive an AttributeMap instance load from RPC input. + +## Decision + +`AIAgentFlow.GetRPCs` explicitly registers every production RPC and its complete +execution policy. Client calls provide only the Flow ID, registered method, +typed input, and output destination. Snapshot uses a five-second registered +timeout. Commands and archive reads use twenty seconds. Existing Attribute +locks and transactional Channel mutations remain attached to their RPCs. + +`GetArchivedMessages` registers a whole-map `ArchivedMessages` load and returns +only the requested immutable ten-message chunk. It still excludes current +messages and pending Channels. Retention remains the bound on loaded archive +state. Integration-only RPCs use the same explicit registration contract. + +## Consequences + +Worker and Client registries share one visible RPC contract, and invalid loads +or locks fail during registry construction. Call sites cannot accidentally +weaken transactional behavior or select undeclared state. + +An archive page now hydrates every retained archive chunk before returning one +page. This is a known cost of the released `v0.10.0` contract, not an SLA change. +A future bounded design requires a new durable storage boundary or a released +SDK facility for input-derived instance selection; it must not emulate mutable +per-call options in application code. diff --git a/docs/flow-model.md b/docs/flow-model.md index 0469b21..e8f2bc2 100644 --- a/docs/flow-model.md +++ b/docs/flow-model.md @@ -14,15 +14,29 @@ `GetArchivedMessages` - Browser synchronization Attribute: `WaitingInputRound` -The implementation requires Dex Go SDK and Server `v0.9.0`. Each +The implementation requires Dex Go SDK `v0.10.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. +New Agent Flows set `FlowConfig.StepDurability` to ASYNC. Ordinary Steps inherit +that default. `CompactContext`, `CallModel`, and tools declared long-running +override Execute durability to SYNC. A short-running tool may fall back from +local to regular execution; that is an expected optimization path and does not +change its ASYNC durability. Registry policy supplies each tool's attempt, +heartbeat, retry, and recovery settings. Ordinary Step methods use a one-minute +timeout, while model methods retain their explicit ten-minute timeout and +five-minute heartbeat. + +The Worker negotiates the highest common protocol with the Server before +Attribute index synchronization or Worker binding. Startup fails when +`GetServerInfo` is missing, either interval is invalid, or the intervals do not +overlap. Release automation does not duplicate this runtime check. + +Snapshot reads retry inactive-run and server long-poll expiry errors within a +three-attempt budget. Snapshot is read-only, so these retries cannot duplicate +commands or external effects. Caller cancellation and other errors still return +immediately. Renewable sandbox credentials are not an Agent Flow resource. A future, separately designed `SandboxLifecycleFlow` will own that lifecycle. @@ -55,9 +69,6 @@ AwaitToolApproval -> next tool or CompactContext (rejected) -> CompactContext (steered) -ExecuteTool - -> next tool or CompactContext (legacy open executions only) - ExecuteToolWithRetry -> next tool or CompactContext (success or known failure) -> RecoverToolExecution (configured automatic unknown) @@ -104,7 +115,6 @@ history, and makes the model replan. | `CheckSteered` | bounded steered batch | Apply steering at a safe boundary or route the explicit continuation | | `RouteTool` | none | Validate built-in arguments and select approval, MCP execution, timer, input, or next-call path | | `AwaitToolApproval` | exact call-ID approval or steering | Persist waiting status; consume one decision or replan on steering | -| `ExecuteTool` | none | Perform one external MCP effect with stable Flow/call identity, then persist its result | | `ExecuteToolWithRetry` | none | Perform one external tool attempt under dynamically selected Dex timeout and retry policy | | `RecoverToolExecution` | none | Record one unknown result for an explicitly configured automatic recovery, then continue | | `ExecuteParallelTool` | none | Perform one bounded-wave branch effect and publish exactly one typed result without shared-state mutation | @@ -114,7 +124,7 @@ history, and makes the model replan. | `AwaitManualToolRecovery` | exact recovery decision or steering | Persist recovery state; retry selected calls, continue unknowns, stop the sequence, or replan | | `DurableWait` | Timer or steering | Persist waiting status; record completion or interruption and continue | -Dex Server and Go SDK `v0.9.0` expose Channel size metadata in `WaitFor` and +Dex Server `v0.10.0` and Go SDK `v0.10.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 @@ -215,8 +225,10 @@ retained messages. Snapshot is one read-only Flow RPC that loads current history, the interaction description, and pending Channels. It returns `WaitingInputRound` and stable -application message IDs. Archive paging loads exactly one immutable chunk and -the bounded sequence metadata needed for continuation. +application message IDs. Archive paging returns exactly one immutable chunk and +the bounded sequence metadata needed for continuation. Its registered +`v0.10.0` RPC options load the retained archive map because the requested chunk +key is an RPC input and invocation-specific selective loads no longer exist. The browser begins with the Snapshot round, waits for `round > watermark`, uses the actual matched round as the next watermark, then refreshes Snapshot. A @@ -226,6 +238,13 @@ every completed Snapshot read. Hidden pages pause the timer and live reads. ## External effects and recovery - Tool execution policy is copied from `ToolDefinition` into Dex StepOptions. +- `short_running` is the default and inherits Flow ASYNC durability. Use + `long_running` when more than half of expected calls are likely to exceed five + seconds; it overrides Execute durability to SYNC. This classification is an + optimization hint, not a runtime guarantee. +- Tool heartbeat defaults to one minute. Increase it only when healthy regular + execution can remain silent for longer. `AttemptTimeout` also bounds the + registry context because ASYNC local execution ignores Dex method timeouts. - The `mock/dex` model alone exposes `simulate_tool_failure`; `/tool-failure` uses it to verify retry exhaustion and the manual recovery surface locally. - Known business failures return a normal tool result. Transient or ambiguous diff --git a/go.mod b/go.mod index b17e624..5f70fa1 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.9.0 + github.com/superdurable/dex/sdk-go v0.10.0 golang.org/x/net v0.58.0 gopkg.in/yaml.v3 v3.0.1 ) @@ -54,4 +54,5 @@ require ( ) tool github.com/ogen-go/ogen/cmd/ogen + tool github.com/ogen-go/ogen/cmd/jschemagen diff --git a/go.sum b/go.sum index ef24fa7..3266885 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.9.0 h1:F1kJnQMGPMR6pXWiB3ZJk2FPdqpaQ8PQ0+ukfrJEvw4= -github.com/superdurable/dex/sdk-go v0.9.0/go.mod h1:8Wj5wPf9dyb7hDnA40j8xISR/zjhX57NrUDVcXgf5x8= +github.com/superdurable/dex/sdk-go v0.10.0 h1:TtXm17mxRE3ZdXI9hltIWDb3v6UsboHj/K5gEncO4Xg= +github.com/superdurable/dex/sdk-go v0.10.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/client.go b/internal/agent/client.go index e4bf339..0c25fe5 100644 --- a/internal/agent/client.go +++ b/internal/agent/client.go @@ -30,6 +30,9 @@ import ( const ( defaultCommandTimeout = 20 * time.Second defaultEventPoll = 20 * time.Second + // Snapshot reads use a shorter budget so clients can retry across continue-as-new. + defaultSnapshotTimeout = 5 * time.Second + maximumSnapshotAttempts = 3 // MaximumRecentEventLimit matches Dex's default maximum Stream list page size. MaximumRecentEventLimit = 1_000 ) @@ -78,10 +81,13 @@ func (client *Client) Start(ctx context.Context, flowID FlowID, request StartReq if err != nil { return "", fmt.Errorf("encode Agent runtime metadata: %w", err) } - runID, err := client.sdk.StartFlow(ctx, client.flow, string(flowID), request.Config, dex.StartFlowOptions{ - IDReusePolicy: dex.IDReuseDisallow, - Attributes: []dex.InitialAttributeDef{initialMetadata}, - }) + runID, err := client.sdk.StartFlow( + ctx, + client.flow, + string(flowID), + request.Config, + newAgentStartFlowOptions(initialMetadata), + ) if err != nil { return "", err } @@ -91,6 +97,17 @@ func (client *Client) Start(ctx context.Context, flowID FlowID, request StartReq return RunID(runID), nil } +func newAgentStartFlowOptions(initialMetadata dex.InitialAttributeDef) dex.StartFlowOptions { + durability := dex.StepDurabilityAsync + return dex.StartFlowOptions{ + IDReusePolicy: dex.IDReuseDisallow, + Attributes: []dex.InitialAttributeDef{initialMetadata}, + ConfigOverride: &dex.FlowConfig{ + StepDurability: &durability, + }, + } +} + // SendMessage invokes the durable SendMessage command. func (client *Client) SendMessage(ctx context.Context, flowID FlowID, message UserMessage) error { if err := validateFlowID(flowID); err != nil { @@ -101,10 +118,7 @@ func (client *Client) SendMessage(ctx context.Context, flowID FlowID, message Us } pending := PendingUserMessage{MessageID: MessageID(uuid.NewString()), Value: message} accepted, err := invokeLockedCommand(ctx, client.commandTimeout, func(ctx context.Context, accepted *bool) error { - return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.SendMessage, pending, accepted, dex.InvokeOptions{ - Timeout: client.commandTimeout, - LockAttributes: []dex.AttributeLock{dex.LockAttribute(pendingUserInputAttribute)}, - }) + return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.SendMessage, pending, accepted) }) if err != nil { return err @@ -125,10 +139,7 @@ func (client *Client) AnswerQuestions( return err } accepted, err := invokeLockedCommand(ctx, client.commandTimeout, func(ctx context.Context, accepted *bool) error { - return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.AnswerQuestions, request, accepted, dex.InvokeOptions{ - Timeout: client.commandTimeout, - LockAttributes: []dex.AttributeLock{dex.LockAttribute(pendingUserInputAttribute)}, - }) + return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.AnswerQuestions, request, accepted) }) if err != nil { return err @@ -176,12 +187,7 @@ func (client *Client) SteerMessage(ctx context.Context, flowID FlowID, request S return err } accepted, err := invokeLockedCommand(ctx, client.commandTimeout, func(ctx context.Context, accepted *bool) error { - return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.SteerMessage, request, accepted, dex.InvokeOptions{ - Timeout: client.commandTimeout, - IsTransactional: true, - LoadChannels: []dex.ChannelDef{queuedUserMessagesChannel}, - LockAttributes: []dex.AttributeLock{dex.LockAttribute(pendingToolRecoveryAttribute)}, - }) + return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.SteerMessage, request, accepted) }) if err != nil { return err @@ -201,30 +207,48 @@ func (client *Client) GetSnapshot( return AgentSnapshot{}, err } current, statusErr := client.latestAgentRun(ctx, flowID) - if statusErr == nil && current != nil && current.Status != dex.FlowRunning { + if statusErr == nil && current != nil && isTerminalFlowStatus(current.Status) { return client.terminalSnapshot(ctx, flowID, RunID(current.RunID)) } - var snapshot AgentSnapshot - err := client.sdk.InvokeRPC(ctx, string(flowID), client.flow.GetSnapshot, nil, &snapshot, dex.InvokeOptions{ - Timeout: client.commandTimeout, - LoadAttributeMaps: []dex.AttributeDef{currentMessagesAttribute}, - LoadChannels: []dex.ChannelDef{ - queuedUserMessagesChannel, - steeredUserMessagesChannel, - }, - }) - if err == nil { - return snapshot, nil - } - var inactive *dex.FlowNotActiveError - if !errors.As(err, &inactive) { - return AgentSnapshot{}, err + var retryErr error + for range maximumSnapshotAttempts { + snapshot, err := client.invokeSnapshotRPC(ctx, flowID) + if err == nil { + return snapshot, nil + } + var inactive *dex.FlowNotActiveError + var pollTimeout *dex.LongPollTimeoutError + if !errors.As(err, &inactive) && !errors.As(err, &pollTimeout) { + return AgentSnapshot{}, err + } + // Snapshot is read-only, so server long-poll expiry can safely retry within this bounded budget. + retryErr = err + current, statusErr = client.latestAgentRun(ctx, flowID) + if statusErr != nil { + return AgentSnapshot{}, errors.Join(err, statusErr) + } + if current == nil || !isTerminalFlowStatus(current.Status) { + continue + } + return client.terminalSnapshot(ctx, flowID, RunID(current.RunID)) } - terminal, terminalErr := client.terminalSnapshot(ctx, flowID, "") - if terminalErr != nil { - return AgentSnapshot{}, errors.Join(err, terminalErr) + return AgentSnapshot{}, retryErr +} + +func isTerminalFlowStatus(status dex.FlowStatus) bool { + return (dex.FlowResult{Status: status}).IsTerminal() +} + +func (client *Client) invokeSnapshotRPC(ctx context.Context, flowID FlowID) (AgentSnapshot, error) { + timeout := client.commandTimeout + if timeout <= 0 || timeout > defaultSnapshotTimeout { + timeout = defaultSnapshotTimeout } - return terminal, nil + rpcContext, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + var snapshot AgentSnapshot + err := client.sdk.InvokeRPC(rpcContext, string(flowID), client.flow.GetSnapshot, nil, &snapshot) + return snapshot, err } // GetArchivedMessages reads exactly one immutable history chunk before a sequence boundary. @@ -232,17 +256,11 @@ func (client *Client) GetArchivedMessages(ctx context.Context, flowID FlowID, be if err := validateFlowID(flowID); err != nil { return HistoryPage{}, err } - first, isValid := archivedMessageChunkFirst(before) - if !isValid { + if _, isValid := archivedMessageChunkFirst(before); !isValid { return HistoryPage{}, fmt.Errorf("before sequence must identify a %d-message boundary", archiveMessageChunkSize) } var result archivedMessagesRPCOutput - err := client.sdk.InvokeRPC(ctx, string(flowID), client.flow.GetArchivedMessages, before, &result, dex.InvokeOptions{ - Timeout: client.commandTimeout, - LoadAttributeMapInstances: []dex.AttributeMapLoad{ - archivedMessagesAttribute.Load(sequenceKey(first)), - }, - }) + err := client.sdk.InvokeRPC(ctx, string(flowID), client.flow.GetArchivedMessages, before, &result) if err != nil { return HistoryPage{}, err } @@ -414,11 +432,6 @@ func (client *Client) DeleteQueuedMessage(ctx context.Context, flowID FlowID, me client.flow.DeleteQueuedMessage, messageID, &deleted, - dex.InvokeOptions{ - Timeout: client.commandTimeout, - IsTransactional: true, - LoadChannels: []dex.ChannelDef{queuedUserMessagesChannel}, - }, ); err != nil { return err } @@ -437,11 +450,7 @@ func (client *Client) ApproveTool(ctx context.Context, flowID FlowID, request To return errors.New("call ID must not be empty") } var accepted bool - if err := client.sdk.InvokeRPC(ctx, string(flowID), client.flow.ApproveTool, request, &accepted, dex.InvokeOptions{ - Timeout: client.commandTimeout, - IsTransactional: true, - LockAttributes: []dex.AttributeLock{dex.LockAttribute(pendingApprovalAttribute)}, - }); err != nil { + if err := client.sdk.InvokeRPC(ctx, string(flowID), client.flow.ApproveTool, request, &accepted); err != nil { return err } return ensureAccepted(accepted, CommandApproveTool) @@ -477,10 +486,6 @@ func (client *Client) ResolveToolRecovery( client.flow.ResolveToolRecovery, request, accepted, - dex.InvokeOptions{ - Timeout: client.commandTimeout, - LockAttributes: []dex.AttributeLock{dex.LockAttribute(pendingToolRecoveryAttribute)}, - }, ) }) if err != nil { @@ -498,11 +503,7 @@ func (client *Client) ExecutePlan(ctx context.Context, flowID FlowID, request Pl return errors.New("plan revision must be positive") } var accepted bool - if err := client.sdk.InvokeRPC(ctx, string(flowID), client.flow.ExecutePlan, request, &accepted, dex.InvokeOptions{ - Timeout: client.commandTimeout, - IsTransactional: true, - LockAttributes: []dex.AttributeLock{dex.LockAttribute(agentStateAttribute)}, - }); err != nil { + if err := client.sdk.InvokeRPC(ctx, string(flowID), client.flow.ExecutePlan, request, &accepted); err != nil { return err } return ensureAccepted(accepted, CommandExecutePlan) diff --git a/internal/agent/client_test.go b/internal/agent/client_test.go index 37b05b3..6b76b67 100644 --- a/internal/agent/client_test.go +++ b/internal/agent/client_test.go @@ -16,7 +16,22 @@ package agent -import "testing" +import ( + "testing" + + "github.com/superdurable/dex/sdk-go/dex" +) + +func TestAgentStartFlowOptionsDefaultStepsToAsyncDurability(t *testing.T) { + t.Parallel() + options := newAgentStartFlowOptions(nil) + if options.ConfigOverride == nil || options.ConfigOverride.StepDurability == nil { + t.Fatalf("ConfigOverride = %+v", options.ConfigOverride) + } + if got := *options.ConfigOverride.StepDurability; got != dex.StepDurabilityAsync { + t.Fatalf("StepDurability = %v, want %v", got, dex.StepDurabilityAsync) + } +} func TestListRecentEventsRejectsInvalidLimits(t *testing.T) { t.Parallel() @@ -28,3 +43,26 @@ func TestListRecentEventsRejectsInvalidLimits(t *testing.T) { } } } + +func TestIsTerminalFlowStatusTreatsContinueAsNewAsActive(t *testing.T) { + t.Parallel() + for _, test := range []struct { + name string + status dex.FlowStatus + terminal bool + }{ + {name: "running", status: dex.FlowRunning}, + {name: "continued as new", status: dex.FlowContinuedAsNew}, + {name: "completed", status: dex.FlowCompleted, terminal: true}, + {name: "failed", status: dex.FlowFailed, terminal: true}, + {name: "canceled", status: dex.FlowCanceled, terminal: true}, + {name: "terminated", status: dex.FlowTerminated, terminal: true}, + } { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + if got := isTerminalFlowStatus(test.status); got != test.terminal { + t.Fatalf("isTerminalFlowStatus(%v) = %t, want %t", test.status, got, test.terminal) + } + }) + } +} diff --git a/internal/agent/flow.go b/internal/agent/flow.go index 2397d96..f051e9f 100644 --- a/internal/agent/flow.go +++ b/internal/agent/flow.go @@ -17,11 +17,11 @@ package agent import ( + "context" "encoding/json" "errors" "fmt" "math" - "reflect" "slices" "strings" "time" @@ -56,8 +56,9 @@ var ( // Flow is the durable AI Agent state machine. type Flow struct { - modelClient ModelClient - tools ToolRegistry + modelClient ModelClient + tools ToolRegistry + rpcDefinitionsForTestOnly []dex.RPCDef } var _ dex.Flow = (*Flow)(nil) @@ -91,7 +92,6 @@ func (flow *Flow) GetSteps() []dex.StepDef { dex.DefineStep(checkSteeredStep{flow: flow}), dex.DefineStep(routeToolStep{flow: flow}), dex.DefineStep(awaitToolApprovalStep{flow: flow}), - dex.DefineStep(executeToolStep{flow: flow}), dex.DefineStep(executeToolWithRetryStep{flow: flow}), dex.DefineStep(recoverToolExecutionStep{flow: flow}), dex.DefineStep(executeParallelToolStep{flow: flow}), @@ -103,6 +103,58 @@ func (flow *Flow) GetSteps() []dex.StepDef { } } +// GetRPCs registers synchronous Agent reads and commands with immutable execution policy. +func (flow *Flow) GetRPCs() []dex.RPCDef { + definitions := []dex.RPCDef{ + dex.DefineRPC(flow.SendMessage, &dex.RPCOptions{ + Timeout: defaultCommandTimeout, + LockAttributes: []dex.AttributeLock{dex.LockAttribute(pendingUserInputAttribute)}, + }), + dex.DefineRPC(flow.AnswerQuestions, &dex.RPCOptions{ + Timeout: defaultCommandTimeout, + LockAttributes: []dex.AttributeLock{dex.LockAttribute(pendingUserInputAttribute)}, + }), + dex.DefineRPC(flow.SteerMessage, &dex.RPCOptions{ + Timeout: defaultCommandTimeout, + LockAttributes: []dex.AttributeLock{dex.LockAttribute(pendingToolRecoveryAttribute)}, + IsTransactional: true, + LoadChannels: []dex.ChannelDef{queuedUserMessagesChannel}, + }), + dex.DefineRPC(flow.GetSnapshot, &dex.RPCOptions{ + Timeout: defaultSnapshotTimeout, + LoadAttributeMaps: []dex.AttributeDef{currentMessagesAttribute}, + LoadChannels: []dex.ChannelDef{ + queuedUserMessagesChannel, + steeredUserMessagesChannel, + }, + }), + dex.DefineRPC(flow.GetArchivedMessages, &dex.RPCOptions{ + Timeout: defaultCommandTimeout, + LoadAttributeMaps: []dex.AttributeDef{archivedMessagesAttribute}, + }), + dex.DefineRPC(flow.DeleteQueuedMessage, &dex.RPCOptions{ + Timeout: defaultCommandTimeout, + IsTransactional: true, + LoadChannels: []dex.ChannelDef{queuedUserMessagesChannel}, + }), + dex.DefineRPC(flow.ApproveTool, &dex.RPCOptions{ + Timeout: defaultCommandTimeout, + LockAttributes: []dex.AttributeLock{dex.LockAttribute(pendingApprovalAttribute)}, + IsTransactional: true, + }), + dex.DefineRPC(flow.ResolveToolRecovery, &dex.RPCOptions{ + Timeout: defaultCommandTimeout, + LockAttributes: []dex.AttributeLock{dex.LockAttribute(pendingToolRecoveryAttribute)}, + }), + dex.DefineRPC(flow.ExecutePlan, &dex.RPCOptions{ + Timeout: defaultCommandTimeout, + LockAttributes: []dex.AttributeLock{dex.LockAttribute(agentStateAttribute)}, + IsTransactional: true, + }), + } + return append(definitions, flow.rpcDefinitionsForTestOnly...) +} + // GetPersistenceSchema registers every durable value and best-effort stream. func (*Flow) GetPersistenceSchema() dex.PersistenceSchema { return dex.PersistenceSchema{ @@ -556,6 +608,10 @@ func (flow *Flow) validateConfig(config AgentConfig) error { func validateToolExecutionPolicy(definition ToolDefinition) error { policy := definition.RetryExhaustionPolicy.Effective() + runningType := definition.RunningType.Effective() + if err := runningType.Validate(); err != nil { + return err + } switch { case definition.MaximumAttempts <= 0: return errors.New("maximum attempts must be positive") @@ -563,6 +619,8 @@ func validateToolExecutionPolicy(definition ToolDefinition) error { return errors.New("maximum attempts exceeds the Dex limit") case definition.AttemptTimeout < 0: return errors.New("attempt timeout must not be negative") + case definition.HeartbeatTimeout < 0: + return errors.New("heartbeat timeout must not be negative") case definition.RetryTotalDuration < 0: return errors.New("retry total duration must not be negative") default: @@ -606,7 +664,8 @@ func (flow *Flow) toolStepOptions(definition ToolDefinition) *dex.StepOptions { } return &dex.StepOptions{ ExecuteMethodTimeout: definition.AttemptTimeout, - HeartbeatTimeout: toolStepOptions.HeartbeatTimeout, + HeartbeatTimeout: effectiveToolHeartbeatTimeout(definition), + ExecuteDurability: toolExecuteDurability(definition), ExecuteLoadAttributeMaps: toolStepOptions.ExecuteLoadAttributeMaps, ExecuteRetry: &dex.RetryPolicy{ // #nosec G115 -- validateToolExecutionPolicy rejects values outside int32. @@ -620,18 +679,33 @@ func (flow *Flow) toolStepOptions(definition ToolDefinition) *dex.StepOptions { func (flow *Flow) parallelToolStepOptions(definition ToolDefinition) *dex.StepOptions { return &dex.StepOptions{ ExecuteMethodTimeout: definition.AttemptTimeout, - HeartbeatTimeout: toolStepOptions.HeartbeatTimeout, + HeartbeatTimeout: effectiveToolHeartbeatTimeout(definition), + ExecuteDurability: toolExecuteDurability(definition), ExecuteRetry: &dex.RetryPolicy{ MaximumAttempts: int32(definition.MaximumAttempts), // #nosec G115 -- validated before scheduling. TotalDuration: definition.RetryTotalDuration, }, ExecuteFailure: dex.ProceedToOnExecuteFailure( recoverParallelToolExecutionStep{flow: flow}, - nil, + defaultStepOptions, ), } } +func effectiveToolHeartbeatTimeout(definition ToolDefinition) time.Duration { + if definition.HeartbeatTimeout == 0 { + return time.Minute + } + return definition.HeartbeatTimeout +} + +func toolExecuteDurability(definition ToolDefinition) dex.StepDurability { + if definition.RunningType.Effective() == ToolRunningTypeLongRunning { + return dex.StepDurabilitySync + } + return dex.StepDurabilityDefault +} + func (flow *Flow) parallelToolMovements( config AgentConfig, state AgentState, @@ -723,11 +797,27 @@ func (flow *Flow) invocationToolDefinition(config AgentConfig, state AgentState, return ToolDefinition{}, fmt.Errorf("unknown or disabled tool %q", name) } -func (flow *Flow) executeTool(ctx dex.Context, invocation ToolInvocation) (ToolExecutionResult, error) { +func (flow *Flow) executeTool( + ctx dex.Context, + definition ToolDefinition, + invocation ToolInvocation, +) (ToolExecutionResult, error) { if invocation.Name == ToolNameSimulateFailure { return ToolExecutionResult{}, simulatedToolFailureError{} } - return flow.tools.Execute(ctx, invocation) + executionContext, cancel := newToolExecutionContext(ctx, definition.AttemptTimeout) + defer cancel() + return flow.tools.Execute(executionContext, invocation) +} + +func newToolExecutionContext( + ctx context.Context, + attemptTimeout time.Duration, +) (context.Context, context.CancelFunc) { + if attemptTimeout == 0 { + return ctx, func() {} + } + return context.WithTimeout(ctx, attemptTimeout) } func (flow *Flow) beginUserTurn(ctx dex.Context, message UserMessage) (Sequence, error) { @@ -1576,7 +1666,6 @@ const ( continueCompactContext continuation = "compact_context" continueRouteTool continuation = "route_tool" continueAwaitToolApproval continuation = "await_tool_approval" - continueExecuteTool continuation = "execute_tool" continueExecuteToolRetry continuation = "execute_tool_with_retry" continueDurableWait continuation = "durable_wait" @@ -1588,7 +1677,6 @@ const ( stepTypeCheckSteered stepType = "CheckSteered" stepTypeRouteTool stepType = "RouteTool" stepTypeAwaitApproval stepType = "AwaitToolApproval" - stepTypeExecuteTool stepType = "ExecuteTool" stepTypeExecuteRetry stepType = "ExecuteToolWithRetry" stepTypeRecoverTool stepType = "RecoverToolExecution" stepTypeExecuteParallel stepType = "ExecuteParallelTool" @@ -1668,7 +1756,13 @@ type awaitParallelToolResultsInput struct { } var ( + defaultStepOptions = &dex.StepOptions{ + WaitForMethodTimeout: time.Minute, + ExecuteMethodTimeout: time.Minute, + } messageMutationStepOptions = &dex.StepOptions{ + WaitForMethodTimeout: time.Minute, + ExecuteMethodTimeout: time.Minute, ExecuteLoadAttributeMaps: []dex.AttributeDef{currentMessagesAttribute}, ExecuteLockAttributes: []dex.AttributeLock{ dex.LockAttribute(pendingUserInputAttribute), @@ -1677,6 +1771,8 @@ var ( }, } awaitUserStepOptions = &dex.StepOptions{ + WaitForMethodTimeout: time.Minute, + ExecuteMethodTimeout: time.Minute, ExecuteLoadAttributeMaps: []dex.AttributeDef{ currentMessagesAttribute, }, @@ -1687,6 +1783,8 @@ var ( }, } messageContextStepOptions = &dex.StepOptions{ + WaitForMethodTimeout: time.Minute, + ExecuteMethodTimeout: time.Minute, ExecuteLoadAttributeMaps: []dex.AttributeDef{ currentMessagesAttribute, archivedMessagesAttribute, @@ -1699,6 +1797,7 @@ var ( modelStepOptions = &dex.StepOptions{ ExecuteMethodTimeout: 10 * time.Minute, HeartbeatTimeout: 5 * time.Minute, + ExecuteDurability: dex.StepDurabilitySync, ExecuteLoadAttributeMaps: messageContextStepOptions.ExecuteLoadAttributeMaps, ExecuteRetry: &dex.RetryPolicy{ MaximumAttempts: 3, @@ -1707,13 +1806,15 @@ var ( } toolStepOptions = &dex.StepOptions{ ExecuteMethodTimeout: 2 * time.Hour, - HeartbeatTimeout: 5 * time.Minute, + HeartbeatTimeout: time.Minute, ExecuteLoadAttributeMaps: messageMutationStepOptions.ExecuteLoadAttributeMaps, ExecuteRetry: &dex.RetryPolicy{ MaximumAttempts: 1, }, } manualToolRecoveryStepOptions = &dex.StepOptions{ + WaitForMethodTimeout: time.Minute, + ExecuteMethodTimeout: time.Minute, ExecuteLoadAttributeMaps: messageMutationStepOptions.ExecuteLoadAttributeMaps, ExecuteLockAttributes: []dex.AttributeLock{ dex.LockAttribute(pendingToolRecoveryAttribute), @@ -1730,6 +1831,8 @@ var _ dex.Step[AgentConfig] = initStep{} func (initStep) GetStepType() string { return string(stepTypeInit) } +func (initStep) GetStepOptions() *dex.StepOptions { return defaultStepOptions } + func (step initStep) Execute(ctx dex.Context, input AgentConfig) (*dex.StepDecision, error) { if err := step.flow.validateConfig(input); err != nil { return nil, err @@ -2201,8 +2304,6 @@ func (step checkSteeredStep) Execute(ctx dex.Context, input continuation) (*dex. return dex.GoTo(routeToolStep{flow: step.flow}, nil), nil case continueAwaitToolApproval: return dex.GoTo(awaitToolApprovalStep{flow: step.flow}, nil), nil - case continueExecuteTool: - return dex.GoTo(executeToolStep{flow: step.flow}, nil), nil case continueExecuteToolRetry: options, err := step.flow.currentToolStepOptions(ctx) if err != nil { @@ -2519,65 +2620,6 @@ func (step awaitToolApprovalStep) Execute(ctx dex.Context, _ dex.None) (*dex.Ste return dex.GoTo(checkSteeredStep{flow: step.flow}, continueCompactContext), nil } -type executeToolStep struct { - dex.StepDefaultsNoWaitFor[dex.None] - flow *Flow -} - -var _ dex.Step[dex.None] = executeToolStep{} - -func (executeToolStep) GetStepType() string { return string(stepTypeExecuteTool) } - -func (executeToolStep) GetStepOptions() *dex.StepOptions { return toolStepOptions } - -func (step executeToolStep) Execute(ctx dex.Context, _ dex.None) (*dex.StepDecision, error) { - if statusErr := step.flow.updateStatus(ctx, AgentStatusExecutingTool); statusErr != nil { - return nil, statusErr - } - call, callErr := step.flow.currentToolCall(ctx) - if callErr != nil { - return nil, callErr - } - config, configErr := agentConfigAttribute.Get(ctx) - if configErr != nil { - return nil, configErr - } - runtimeMetadata, metadataErr := agentRuntimeMetadataAttribute.Get(ctx) - if isAttributeNotFound(metadataErr) { - runtimeMetadata = MustJSONObject(`{}`) - } else if metadataErr != nil { - return nil, metadataErr - } - progress := toolProgress{ctx: ctx, flow: step.flow, call: call} - result, executeErr := step.flow.executeTool(ctx, ToolInvocation{ - FlowID: FlowID(ctx.FlowID()), - RuntimeMetadata: runtimeMetadata, - Name: call.Name, - Arguments: call.Arguments, - EnabledServers: config.EnabledMCPServers, - WriteProgress: progress.write, - CallID: call.ID, - Attempt: ctx.Attempt(), - FirstAttemptAt: ctx.FirstAttemptAt(), - }) - if executeErr != nil { - failureResult, encodeErr := encodeToolResult(toolResultPayload{ - Status: toolResultStatusFailed, - Outcome: ToolOutcomeUnknown, - ErrorType: errorTypeName(executeErr), - }, ToolOutcomeUnknown, true) - if encodeErr != nil { - return nil, errors.Join(executeErr, encodeErr) - } - result = failureResult - } - next, finishErr := step.flow.finishToolExecution(ctx, call, result) - if finishErr != nil { - return nil, finishErr - } - return dex.GoTo(checkSteeredStep{flow: step.flow}, next), nil -} - func (flow *Flow) finishToolExecution( ctx dex.Context, call ToolCall, @@ -2705,6 +2747,14 @@ func (step executeToolWithRetryStep) Execute(ctx dex.Context, _ dex.None) (*dex. if err != nil { return nil, err } + state, err := agentStateAttribute.Get(ctx) + if err != nil { + return nil, err + } + definition, err := step.flow.invocationToolDefinition(config, state, call.Name) + if err != nil { + return nil, err + } runtimeMetadata, err := agentRuntimeMetadataAttribute.Get(ctx) if isAttributeNotFound(err) { runtimeMetadata = MustJSONObject(`{}`) @@ -2715,7 +2765,7 @@ func (step executeToolWithRetryStep) Execute(ctx dex.Context, _ dex.None) (*dex. if writeErr := progress.write(fmt.Sprintf("Calling %s (attempt %d).", call.Name, ctx.Attempt())); writeErr != nil { return nil, writeErr } - result, err := step.flow.executeTool(ctx, ToolInvocation{ + result, err := step.flow.executeTool(ctx, definition, ToolInvocation{ FlowID: FlowID(ctx.FlowID()), RuntimeMetadata: runtimeMetadata, Name: call.Name, @@ -2732,14 +2782,6 @@ func (step executeToolWithRetryStep) Execute(ctx dex.Context, _ dex.None) (*dex. if validationErr := result.Outcome.Validate(); validationErr != nil { return nil, fmt.Errorf("tool %q outcome: %w", call.Name, validationErr) } - state, err := agentStateAttribute.Get(ctx) - if err != nil { - return nil, err - } - definition, err := step.flow.invocationToolDefinition(config, state, call.Name) - if err != nil { - return nil, err - } if result.Outcome == ToolOutcomeUnknown && definition.RetryExhaustionPolicy.Effective() == ToolRetryExhaustionPolicyManualRecovery { resultCopy := result @@ -2821,7 +2863,7 @@ func (step executeParallelToolStep) GetStepOptions() *dex.StepOptions { ExecuteRetry: toolStepOptions.ExecuteRetry, ExecuteFailure: dex.ProceedToOnExecuteFailure( recoverParallelToolExecutionStep{flow: step.flow}, - nil, + defaultStepOptions, ), } } @@ -2834,6 +2876,14 @@ func (step executeParallelToolStep) Execute( if configErr != nil { return nil, configErr } + state, stateErr := agentStateAttribute.Get(ctx) + if stateErr != nil { + return nil, stateErr + } + definition, definitionErr := step.flow.invocationToolDefinition(config, state, input.Call.Name) + if definitionErr != nil { + return nil, definitionErr + } runtimeMetadata, metadataErr := agentRuntimeMetadataAttribute.Get(ctx) if isAttributeNotFound(metadataErr) { runtimeMetadata = MustJSONObject(`{}`) @@ -2844,7 +2894,7 @@ func (step executeParallelToolStep) Execute( if progressErr := progress.write(fmt.Sprintf("Calling %s (attempt %d).", input.Call.Name, ctx.Attempt())); progressErr != nil { return nil, progressErr } - result, executeErr := step.flow.executeTool(ctx, ToolInvocation{ + result, executeErr := step.flow.executeTool(ctx, definition, ToolInvocation{ FlowID: FlowID(ctx.FlowID()), RuntimeMetadata: runtimeMetadata, Name: input.Call.Name, @@ -2888,7 +2938,9 @@ func (recoverParallelToolExecutionStep) GetStepType() string { return string(stepTypeRecoverParallel) } -func (recoverParallelToolExecutionStep) GetStepOptions() *dex.StepOptions { return nil } +func (recoverParallelToolExecutionStep) GetStepOptions() *dex.StepOptions { + return defaultStepOptions +} func (recoverParallelToolExecutionStep) Execute( ctx dex.Context, @@ -3407,17 +3459,3 @@ func toolProgressMessage(tool ToolName, message string) string { } return condenseActivityMessage(message) } - -func errorTypeName(err error) string { - value := reflect.TypeOf(err) - if value == nil { - return "error" - } - for value.Kind() == reflect.Pointer { - value = value.Elem() - } - if value.Name() == "" { - return "error" - } - return value.Name() -} diff --git a/internal/agent/flow_integration_test.go b/internal/agent/flow_integration_test.go index 2d34b03..1f0db01 100644 --- a/internal/agent/flow_integration_test.go +++ b/internal/agent/flow_integration_test.go @@ -1577,10 +1577,32 @@ type agentIntegrationEnvironment struct { agent *Client } +func registerRPCDefinitionsForTestOnly(flow *Flow) { + flow.rpcDefinitionsForTestOnly = []dex.RPCDef{ + dex.DefineRPC(flow.GetFlowStateForTestOnly, &dex.RPCOptions{ + Timeout: defaultCommandTimeout, + LoadChannels: []dex.ChannelDef{queuedUserMessagesChannel}, + }), + dex.DefineRPC(flow.GetPlanExecutionMessagesForTestOnly, &dex.RPCOptions{ + Timeout: defaultCommandTimeout, + LoadChannelMaps: []dex.ChannelDef{planExecutionsChannel}, + }), + dex.DefineRPC(flow.GetMessagesAfterForTestOnly, &dex.RPCOptions{ + Timeout: defaultCommandTimeout, + LoadAttributeMaps: []dex.AttributeDef{ + currentMessagesAttribute, + archivedMessagesAttribute, + }, + }), + } +} + func newAgentIntegrationEnvironment(t *testing.T, modelClient ModelClient, tools ToolRegistry) *agentIntegrationEnvironment { t.Helper() + flow := NewFlow(modelClient, tools) + registerRPCDefinitionsForTestOnly(flow) environment := &agentIntegrationEnvironment{ - flow: NewFlow(modelClient, tools), + flow: flow, address: availableLocalAddress(t, t.Context()), serverAddress: os.Getenv("DEX_FLOW_SERVICE_ADDRESS"), } diff --git a/internal/agent/history_test.go b/internal/agent/history_test.go index d8bd655..bea54d1 100644 --- a/internal/agent/history_test.go +++ b/internal/agent/history_test.go @@ -105,10 +105,6 @@ func (client *Client) GetFlowStateForTestOnly( client.flow.GetFlowStateForTestOnly, nil, &result, - dex.InvokeOptions{ - Timeout: client.commandTimeout, - LoadChannels: []dex.ChannelDef{queuedUserMessagesChannel}, - }, ) return result, err } @@ -141,12 +137,6 @@ func (client *Client) GetPlanExecutionMessagesForTestOnly( client.flow.GetPlanExecutionMessagesForTestOnly, revision, &messages, - dex.InvokeOptions{ - Timeout: client.commandTimeout, - LoadChannelMapInstances: []dex.ChannelMapLoad{ - planExecutionsChannel.LoadMessages(planRevisionKey(revision)), - }, - }, ) return messages, err } @@ -231,13 +221,6 @@ func (client *Client) GetMessagesAfterForTestOnly( client.flow.GetMessagesAfterForTestOnly, getMessagesAfterInputForTestOnly{After: after, Limit: limit}, &page, - dex.InvokeOptions{ - Timeout: client.commandTimeout, - LoadAttributeMaps: []dex.AttributeDef{ - currentMessagesAttribute, - archivedMessagesAttribute, - }, - }, ) return page, err } diff --git a/internal/agent/tool_recovery_test.go b/internal/agent/tool_recovery_test.go index ae7ce8f..929f1be 100644 --- a/internal/agent/tool_recovery_test.go +++ b/internal/agent/tool_recovery_test.go @@ -18,8 +18,11 @@ package agent import ( "context" + "errors" "testing" "time" + + "github.com/superdurable/dex/sdk-go/dex" ) func TestParallelToolMovementsUseBoundedContiguousSafeWave(t *testing.T) { @@ -95,6 +98,72 @@ func TestToolDefinitionsExposeFailureSimulationOnlyToLocalMock(t *testing.T) { } } +func TestToolStepOptionsMapRunningTypeAndHeartbeat(t *testing.T) { + flow := &Flow{} + short := parallelDefinitionForTestOnly("short") + shortOptions := flow.toolStepOptions(short) + if shortOptions.ExecuteDurability != dex.StepDurabilityDefault || + shortOptions.HeartbeatTimeout != time.Minute { + t.Fatalf("short options = %+v", shortOptions) + } + parallelShortOptions := flow.parallelToolStepOptions(short) + if parallelShortOptions.ExecuteDurability != dex.StepDurabilityDefault || + parallelShortOptions.HeartbeatTimeout != time.Minute { + t.Fatalf("parallel short options = %+v", parallelShortOptions) + } + + long := short + long.RunningType = ToolRunningTypeLongRunning + long.HeartbeatTimeout = 15 * time.Minute + longOptions := flow.toolStepOptions(long) + if longOptions.ExecuteDurability != dex.StepDurabilitySync || + longOptions.HeartbeatTimeout != 15*time.Minute { + t.Fatalf("long options = %+v", longOptions) + } + parallelLongOptions := flow.parallelToolStepOptions(long) + if parallelLongOptions.ExecuteDurability != dex.StepDurabilitySync || + parallelLongOptions.HeartbeatTimeout != 15*time.Minute { + t.Fatalf("parallel long options = %+v", parallelLongOptions) + } +} + +func TestRegisteredStepOptionsUseBoundedTimeoutsAndModelSyncDurability(t *testing.T) { + if defaultStepOptions.WaitForMethodTimeout != time.Minute || + defaultStepOptions.ExecuteMethodTimeout != time.Minute { + t.Fatalf("default Step options = %+v", defaultStepOptions) + } + if modelStepOptions.ExecuteDurability != dex.StepDurabilitySync || + modelStepOptions.ExecuteMethodTimeout != 10*time.Minute || + modelStepOptions.HeartbeatTimeout != 5*time.Minute { + t.Fatalf("model Step options = %+v", modelStepOptions) + } +} + +func TestValidateToolExecutionPolicyRejectsUnknownRunningType(t *testing.T) { + definition := parallelDefinitionForTestOnly("invalid") + definition.RunningType = "sometimes" + err := validateToolExecutionPolicy(definition) + var validationErr *EnumValidationError + if !errors.As(err, &validationErr) || validationErr.Type != "ToolRunningType" { + t.Fatalf("validation error = %T %v", err, err) + } +} + +func TestToolExecutionContextUsesDeclaredAttemptTimeout(t *testing.T) { + started := time.Now() + ctx, cancel := newToolExecutionContext(context.Background(), time.Minute) + defer cancel() + deadline, found := ctx.Deadline() + if !found || deadline.Before(started.Add(59*time.Second)) || deadline.After(started.Add(61*time.Second)) { + t.Fatalf("deadline = %v, found = %t", deadline, found) + } + withoutDeadline, cancelWithoutDeadline := newToolExecutionContext(context.Background(), 0) + defer cancelWithoutDeadline() + if _, found := withoutDeadline.Deadline(); found { + t.Fatal("zero attempt timeout added a deadline") + } +} + func TestValidateToolRecoveryResolutionRequiresAtomicCompleteDecision(t *testing.T) { pending := PendingToolRecovery{ RecoveryID: "recovery-1", diff --git a/internal/agent/types.go b/internal/agent/types.go index 7289ea2..1e07fe8 100644 --- a/internal/agent/types.go +++ b/internal/agent/types.go @@ -443,6 +443,39 @@ func (outcome *ToolOutcome) UnmarshalJSON(data []byte) error { return decodeEnum(data, outcome, ToolOutcome.Validate) } +// ToolRunningType selects the preferred Dex execution path for one tool. +type ToolRunningType string + +const ( + // ToolRunningTypeShortRunning optimizes for ASYNC local execution with regular fallback. + ToolRunningTypeShortRunning ToolRunningType = "short_running" + // ToolRunningTypeLongRunning starts directly as a regular SYNC activity. + ToolRunningTypeLongRunning ToolRunningType = "long_running" +) + +// Validate rejects unknown tool running types. +func (runningType ToolRunningType) Validate() error { + switch runningType { + case ToolRunningTypeShortRunning, ToolRunningTypeLongRunning: + return nil + default: + return newEnumValidationError("ToolRunningType", string(runningType)) + } +} + +// Effective returns the short-running default for an omitted registry policy. +func (runningType ToolRunningType) Effective() ToolRunningType { + if runningType == "" { + return ToolRunningTypeShortRunning + } + return runningType +} + +// UnmarshalJSON decodes and validates a tool running type. +func (runningType *ToolRunningType) UnmarshalJSON(data []byte) error { + return decodeEnum(data, runningType, ToolRunningType.Validate) +} + // ToolRetryExhaustionPolicy controls what happens after a tool's Dex retries are exhausted. type ToolRetryExhaustionPolicy string @@ -1131,7 +1164,9 @@ type ToolDefinition struct { Description string InputSchema JSONObject RequiresApproval bool + RunningType ToolRunningType AttemptTimeout time.Duration + HeartbeatTimeout time.Duration MaximumAttempts int RetryTotalDuration time.Duration SupportsParallelExecution bool diff --git a/internal/mcp/config.go b/internal/mcp/config.go index 0408f6f..de9c56c 100644 --- a/internal/mcp/config.go +++ b/internal/mcp/config.go @@ -34,6 +34,7 @@ import ( const ( maximumToolAttempts = 10 maximumToolTimeout = 24 * 60 * 60 + maximumHeartbeatTimeout = 24 * 60 * 60 maximumToolRetrySeconds = 7 * 24 * 60 * 60 ) @@ -117,12 +118,48 @@ func (policy *RetryExhaustionPolicy) UnmarshalYAML(node *yaml.Node) error { return nil } +// RunningType selects the preferred Dex execution path for one MCP tool. +type RunningType string + +const ( + RunningTypeShortRunning RunningType = "short_running" + RunningTypeLongRunning RunningType = "long_running" +) + +// Validate rejects unknown running types. +func (runningType RunningType) Validate() error { + switch runningType { + case RunningTypeShortRunning, RunningTypeLongRunning: + return nil + default: + return fmt.Errorf("unsupported running type %q", runningType) + } +} + +// UnmarshalYAML decodes and validates one running type. +func (runningType *RunningType) UnmarshalYAML(node *yaml.Node) error { + var value string + if err := node.Decode(&value); err != nil { + return err + } + decoded := RunningType(value) + if err := decoded.Validate(); err != nil { + return err + } + *runningType = decoded + return nil +} + // ToolPolicy configures safety and bounded retries for one tool. type ToolPolicy struct { // ReadOnly overrides the tool annotation; nil means unknown. ReadOnly *bool `yaml:"read_only"` // TimeoutSeconds defaults to 60 and bounds one attempt. TimeoutSeconds float64 `yaml:"timeout_seconds"` + // RunningType defaults to short_running for ASYNC local execution with fallback. + RunningType RunningType `yaml:"running_type"` + // HeartbeatTimeoutSeconds defaults to 60 for regular execution. + HeartbeatTimeoutSeconds *float64 `yaml:"heartbeat_timeout_seconds"` // MaximumAttempts defaults to three for trusted reads and one otherwise. MaximumAttempts *int `yaml:"maximum_attempts"` // RetryTotalSeconds defaults to 300 and bounds all attempts. @@ -217,6 +254,13 @@ func applyDefaults(server *ServerConfig) { if policy.RetryTotalSeconds == 0 { policy.RetryTotalSeconds = 300 } + if policy.RunningType == "" { + policy.RunningType = RunningTypeShortRunning + } + if policy.HeartbeatTimeoutSeconds == nil { + value := float64(60) + policy.HeartbeatTimeoutSeconds = &value + } if policy.RetryExhaustionPolicy == "" { policy.RetryExhaustionPolicy = RetryExhaustionPolicyManualRecovery } @@ -268,6 +312,17 @@ func validateServer(server ServerConfig) error { if !isFinitePositive(policy.TimeoutSeconds) || policy.TimeoutSeconds > maximumToolTimeout { return fmt.Errorf("timeout_seconds for %q must be positive and at most %d", name, maximumToolTimeout) } + if err := policy.RunningType.Validate(); err != nil { + return fmt.Errorf("running_type for %q: %w", name, err) + } + if policy.HeartbeatTimeoutSeconds == nil || !isFinitePositive(*policy.HeartbeatTimeoutSeconds) || + *policy.HeartbeatTimeoutSeconds > maximumHeartbeatTimeout { + return fmt.Errorf( + "heartbeat_timeout_seconds for %q must be positive and at most %d", + name, + maximumHeartbeatTimeout, + ) + } if !isFinitePositive(policy.RetryTotalSeconds) || policy.RetryTotalSeconds > maximumToolRetrySeconds { return fmt.Errorf("retry_total_seconds for %q must be positive and at most %d", name, maximumToolRetrySeconds) } @@ -327,6 +382,10 @@ func cloneToolPolicies(source map[string]ToolPolicy) map[string]ToolPolicy { attempts := *policy.MaximumAttempts policy.MaximumAttempts = &attempts } + if policy.HeartbeatTimeoutSeconds != nil { + heartbeatTimeout := *policy.HeartbeatTimeoutSeconds + policy.HeartbeatTimeoutSeconds = &heartbeatTimeout + } if policy.ReadOnly != nil { readOnly := *policy.ReadOnly policy.ReadOnly = &readOnly diff --git a/internal/mcp/config_test.go b/internal/mcp/config_test.go index 0964d94..8702fa3 100644 --- a/internal/mcp/config_test.go +++ b/internal/mcp/config_test.go @@ -41,12 +41,49 @@ func TestLoadConfigAppliesSafeDefaults(t *testing.T) { t.Fatalf("LoadConfig() error = %v", err) } policy := servers[0].Tools["query"] - if policy.TimeoutSeconds != 60 || policy.RetryTotalSeconds != 300 || + if policy.TimeoutSeconds != 60 || policy.RunningType != RunningTypeShortRunning || + policy.HeartbeatTimeoutSeconds == nil || *policy.HeartbeatTimeoutSeconds != 60 || + policy.RetryTotalSeconds != 300 || policy.RetryExhaustionPolicy != RetryExhaustionPolicyManualRecovery { t.Fatalf("policy defaults = %+v", policy) } } +func TestLoadConfigAcceptsLongRunningToolPolicy(t *testing.T) { + path := writeConfig(t, `servers: + - name: build + transport: stdio + command: build-server + tools: + compile: + running_type: long_running + heartbeat_timeout_seconds: 900 +`) + servers, err := LoadConfig(path) + if err != nil { + t.Fatal(err) + } + policy := servers[0].Tools["compile"] + if policy.RunningType != RunningTypeLongRunning || policy.HeartbeatTimeoutSeconds == nil || + *policy.HeartbeatTimeoutSeconds != 900 { + t.Fatalf("policy = %+v", policy) + } +} + +func TestLoadConfigRejectsUnknownRunningType(t *testing.T) { + path := writeConfig(t, `servers: + - name: build + transport: stdio + command: build-server + tools: + compile: + running_type: sometimes +`) + if _, err := LoadConfig(path); err == nil { + t.Fatal("LoadConfig() error = nil") + } +} + func TestLoadConfigAcceptsAutomaticUnknownRecovery(t *testing.T) { path := writeConfig(t, `servers: - name: search @@ -157,6 +194,22 @@ func TestLoadConfigRejectsUnsafeRetryPolicy(t *testing.T) { } } +func TestLoadConfigRejectsUnsafeHeartbeatTimeout(t *testing.T) { + for _, value := range []string{".nan", "0", "-1"} { + path := writeConfig(t, `servers: + - name: search + transport: stdio + command: search-server + tools: + query: + heartbeat_timeout_seconds: `+value+` +`) + if _, err := LoadConfig(path); err == nil { + t.Fatalf("heartbeat_timeout_seconds %s: LoadConfig() error = nil", value) + } + } +} + func TestResolveEnvironmentRequiresEveryConfiguredSource(t *testing.T) { const missing = "SUPERAGENT_TEST_MISSING_MCP_SECRET" t.Setenv(missing, "present") diff --git a/internal/mcp/registry.go b/internal/mcp/registry.go index d469d8f..3447e54 100644 --- a/internal/mcp/registry.go +++ b/internal/mcp/registry.go @@ -39,9 +39,10 @@ import ( ) const ( - defaultToolTimeout = 60 * time.Second - defaultRetryDuration = 5 * time.Minute - maximumPublicNameSize = 64 + defaultToolTimeout = 60 * time.Second + defaultToolHeartbeatTimeout = time.Minute + defaultRetryDuration = 5 * time.Minute + maximumPublicNameSize = 64 ) var invalidNameCharacter = regexp.MustCompile(`[^A-Za-z0-9_]`) @@ -441,9 +442,11 @@ func registeredTool(server ServerConfig, tool *mcpsdk.Tool) (agent.RegisteredToo policy, configured := server.Tools[tool.Name] if !configured { policy = ToolPolicy{ - TimeoutSeconds: 60, - RetryTotalSeconds: 300, - RetryExhaustionPolicy: RetryExhaustionPolicyManualRecovery, + TimeoutSeconds: 60, + RunningType: RunningTypeShortRunning, + HeartbeatTimeoutSeconds: float64Pointer(60), + RetryTotalSeconds: 300, + RetryExhaustionPolicy: RetryExhaustionPolicyManualRecovery, } } readOnly := policy.ReadOnly @@ -466,13 +469,26 @@ func registeredTool(server ServerConfig, tool *mcpsdk.Tool) (agent.RegisteredToo if attemptTimeout == 0 { attemptTimeout = defaultToolTimeout } + heartbeatTimeout := defaultToolHeartbeatTimeout + if policy.HeartbeatTimeoutSeconds != nil { + heartbeatTimeout = time.Duration(*policy.HeartbeatTimeoutSeconds * float64(time.Second)) + } retryDuration := time.Duration(policy.RetryTotalSeconds * float64(time.Second)) if retryDuration == 0 { retryDuration = defaultRetryDuration } - if maximumAttempts <= 0 || attemptTimeout <= 0 || retryDuration <= 0 { + if maximumAttempts <= 0 || attemptTimeout <= 0 || heartbeatTimeout <= 0 || retryDuration <= 0 { return agent.RegisteredTool{}, fmt.Errorf("invalid policy for %q", publicName) } + var runningType agent.ToolRunningType + switch policy.RunningType { + case "", RunningTypeShortRunning: + runningType = agent.ToolRunningTypeShortRunning + case RunningTypeLongRunning: + runningType = agent.ToolRunningTypeLongRunning + default: + return agent.RegisteredTool{}, fmt.Errorf("invalid running type %q for %q", policy.RunningType, publicName) + } inputSchema, err := schemaObject(tool.InputSchema) if err != nil { return agent.RegisteredTool{}, fmt.Errorf("tool %q input schema: %w", publicName, err) @@ -489,7 +505,9 @@ func registeredTool(server ServerConfig, tool *mcpsdk.Tool) (agent.RegisteredToo Description: description, InputSchema: inputSchema, RequiresApproval: readOnly == nil || !*readOnly, + RunningType: runningType, AttemptTimeout: attemptTimeout, + HeartbeatTimeout: heartbeatTimeout, MaximumAttempts: maximumAttempts, RetryTotalDuration: retryDuration, SupportsParallelExecution: readOnly != nil && *readOnly, @@ -498,6 +516,10 @@ func registeredTool(server ServerConfig, tool *mcpsdk.Tool) (agent.RegisteredToo }, nil } +func float64Pointer(value float64) *float64 { + return &value +} + func schemaObject(value any) (agent.JSONObject, error) { encoded, err := json.Marshal(value) if err != nil { diff --git a/internal/mcp/registry_test.go b/internal/mcp/registry_test.go index baf3351..effb8b3 100644 --- a/internal/mcp/registry_test.go +++ b/internal/mcp/registry_test.go @@ -21,8 +21,10 @@ import ( "log/slog" "strings" "testing" + "time" mcpsdk "github.com/modelcontextprotocol/go-sdk/mcp" + "github.com/superdurable/superagent/internal/agent" ) func TestRegisteredToolDefaultsWritesToOneAttemptAndApproval(t *testing.T) { @@ -39,11 +41,34 @@ func TestRegisteredToolDefaultsWritesToOneAttemptAndApproval(t *testing.T) { if err != nil { t.Fatalf("registeredTool() error = %v", err) } - if !registered.Definition.RequiresApproval || registered.Definition.MaximumAttempts != 1 { + if !registered.Definition.RequiresApproval || registered.Definition.MaximumAttempts != 1 || + registered.Definition.RunningType != agent.ToolRunningTypeShortRunning || + registered.Definition.HeartbeatTimeout != time.Minute { t.Fatalf("unsafe defaults = %+v", registered.Definition) } } +func TestRegisteredToolProjectsLongRunningPolicy(t *testing.T) { + registered, err := registeredTool(ServerConfig{ + Name: "build", + Transport: TransportStdio, + Command: "server", + Tools: map[string]ToolPolicy{ + "compile": { + RunningType: RunningTypeLongRunning, + HeartbeatTimeoutSeconds: float64Pointer(900), + }, + }, + }, &mcpsdk.Tool{Name: "compile", InputSchema: map[string]any{"type": "object"}}) + if err != nil { + t.Fatal(err) + } + if registered.Definition.RunningType != agent.ToolRunningTypeLongRunning || + registered.Definition.HeartbeatTimeout != 15*time.Minute { + t.Fatalf("definition = %+v", registered.Definition) + } +} + func TestRegisteredToolTrustsReadOnlyAnnotationOnlyWhenConfigured(t *testing.T) { readOnly := &mcpsdk.Tool{ Name: "search", diff --git a/script/check_dex_release.py b/script/check_dex_release.py deleted file mode 100644 index 50f47c7..0000000 --- a/script/check_dex_release.py +++ /dev/null @@ -1,79 +0,0 @@ -#!/usr/bin/env python3 -# Copyright (c) 2026 Super Durable, Inc. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# SPDX-License-Identifier: Apache-2.0 - -"""Verify SuperAgent's direct Dex dependency against its immutable release lock.""" - -from __future__ import annotations - -import json -from pathlib import Path -import re - -import update_dex_release - - -ROOT = Path(__file__).resolve().parents[1] - - -def main() -> None: - lock = json.loads((ROOT / "dex-release.lock.json").read_text(encoding="utf-8")) - if set(lock) != { - "schemaVersion", - "release", - "manifest", - "sourceCommit", - "sdkGoVersion", - "protocol", - "runningFlowsCompatibility", - "persistenceCompatibility", - "openFlowsCompatibility", - }: - raise update_dex_release.UpgradeError("Dex release lock has unexpected fields") - content = update_dex_release.download(lock["manifest"]["url"]) - manifest = update_dex_release.validate_manifest( - lock["manifest"]["url"], lock["manifest"]["sha256"], content - ) - requirements = dict( - re.findall(r"(?m)^\s*([^\s()]+)\s+(v[^\s]+)(?:\s+//.*)?$", (ROOT / "go.mod").read_text(encoding="utf-8")) - ) - update_dex_release.require(lock["schemaVersion"] == 1, "unsupported Dex release lock") - update_dex_release.require(lock["release"] == manifest["release"], "Dex release mismatch") - update_dex_release.require( - lock["sourceCommit"] == manifest["sourceCommit"], "Dex source commit mismatch" - ) - update_dex_release.require( - lock["sdkGoVersion"] == manifest["components"]["sdkGo"]["version"], - "Dex Go SDK version mismatch", - ) - update_dex_release.require( - requirements.get("github.com/superdurable/dex/sdk-go") == f'v{lock["sdkGoVersion"]}', - "SuperAgent must directly require the locked Dex Go SDK", - ) - update_dex_release.require( - lock["protocol"] == manifest["protocol"]["clients"]["sdkGo"], - "Dex protocol mismatch", - ) - for field in ("runningFlowsCompatibility", "persistenceCompatibility"): - update_dex_release.require(lock[field] == manifest[field], f"Dex {field} mismatch") - update_dex_release.require( - lock["openFlowsCompatibility"] in {"compatible", "cancel-required"}, - "invalid open Flow compatibility", - ) - print(f'SuperAgent directly requires locked Dex {lock["release"]}') - - -if __name__ == "__main__": - main() diff --git a/script/check_dex_versions.py b/script/check_dex_versions.py new file mode 100644 index 0000000..58179b9 --- /dev/null +++ b/script/check_dex_versions.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +# Copyright (c) 2026 Super Durable, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# SPDX-License-Identifier: Apache-2.0 + +"""Verify SuperAgent's direct Dex Go SDK and dexcli version pins.""" + +from __future__ import annotations + +from pathlib import Path +import re + + +ROOT = Path(__file__).resolve().parents[1] +SEMVER = r"[0-9]+\.[0-9]+\.[0-9]+" + + +class DexVersionError(RuntimeError): + """A Dex component version pin is missing or inconsistent.""" + + +def require_match(pattern: str, content: str, label: str) -> re.Match[str]: + match = re.search(pattern, content, flags=re.MULTILINE) + if match is None: + raise DexVersionError(f"{label} version pin is missing") + return match + + +def read_versions(root: Path) -> tuple[str, str]: + go_mod = (root / "go.mod").read_text(encoding="utf-8") + sdk_version = require_match( + rf"^\s*github\.com/superdurable/dex/sdk-go\s+v({SEMVER})\s*$", + go_mod, + "Dex Go SDK", + ).group(1) + + makefile = (root / "Makefile").read_text(encoding="utf-8") + dexcli_version = require_match( + rf"^DEXCLI_VERSION := v({SEMVER})$", makefile, "dexcli Makefile" + ).group(1) + + installer = (root / "script/install-dexcli.sh").read_text(encoding="utf-8") + archives = re.findall( + r"dexcli_v([0-9]+\.[0-9]+\.[0-9]+)_(?:darwin|linux)_(?:amd64|arm64)\.tar\.gz", + installer, + ) + if len(archives) != 4 or set(archives) != {dexcli_version}: + raise DexVersionError("dexcli installer versions do not match the Makefile") + + workflow = (root / ".github/workflows/ci.yml").read_text(encoding="utf-8") + expected_binary = f".cache/dexcli-v{dexcli_version} dev" + if expected_binary not in workflow: + raise DexVersionError("dexcli CI version does not match the Makefile") + return sdk_version, dexcli_version + + +def main() -> None: + sdk_version, dexcli_version = read_versions(ROOT) + print(f"SuperAgent uses Dex Go SDK {sdk_version} and dexcli {dexcli_version}") + + +if __name__ == "__main__": + main() diff --git a/script/install-dexcli.sh b/script/install-dexcli.sh index 4aef5d3..cfb2c0a 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.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 ;; + dexcli_v0.10.0_darwin_amd64.tar.gz) checksum=927d48d360da5183b4956823e827f890fde6da8756a954e5f647098e0e6c348a ;; + dexcli_v0.10.0_darwin_arm64.tar.gz) checksum=1f9c12be1a8b4c7f65af57b93db2a125ff70be63daaf98e902a2b792b095bf97 ;; + dexcli_v0.10.0_linux_amd64.tar.gz) checksum=0cee3b0795147b581c45d2258b0c2d581cd35ce75c515027e2d9b294ce364e2d ;; + dexcli_v0.10.0_linux_arm64.tar.gz) checksum=6ce2d4cdc8a2d91b6fba0c549bdf238ef8d210de69cd140bf60deef59c1412f4 ;; *) echo "no checksum is pinned for $archive_name" >&2; exit 1 ;; esac diff --git a/script/testdata/public-api-consumer/consumer_test.go b/script/testdata/public-api-consumer/consumer_test.go index 29c30b7..361462d 100644 --- a/script/testdata/public-api-consumer/consumer_test.go +++ b/script/testdata/public-api-consumer/consumer_test.go @@ -20,6 +20,7 @@ import ( "context" "net/http" "testing" + "time" "github.com/superdurable/dex/sdk-go/dex" "github.com/superdurable/superagent/agent" @@ -78,6 +79,7 @@ var ( _ agent.EventKind = agent.EventKindPlanTaskUpdated _ agent.EventKind = agent.EventKindInputConsumed _ agent.EventKind = agent.EventKindSnapshotRequired + _ agent.ToolRunningType = agent.ToolRunningTypeShortRunning _ agent.PlanTaskIndex = 0 ) @@ -105,6 +107,13 @@ func TestExternalModuleCanConstructAndRegisterAgent(t *testing.T) { if agent.MaximumRuntimeMetadataBytes != 16<<10 { t.Fatal("public runtime metadata limit is unavailable") } + definition := agent.ToolDefinition{ + RunningType: agent.ToolRunningTypeLongRunning, + HeartbeatTimeout: time.Minute, + } + if definition.RunningType != agent.ToolRunningTypeLongRunning { + t.Fatal("public tool running policy is unavailable") + } } func TestExternalModuleCanConstructProviderRouter(t *testing.T) { diff --git a/script/update_dex_release.py b/script/update_dex_release.py deleted file mode 100644 index cca51a5..0000000 --- a/script/update_dex_release.py +++ /dev/null @@ -1,153 +0,0 @@ -#!/usr/bin/env python3 -# Copyright (c) 2026 Super Durable, Inc. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# SPDX-License-Identifier: Apache-2.0 - -"""Prepare a reviewed SuperAgent upgrade from one immutable Dex manifest.""" - -from __future__ import annotations - -import argparse -import hashlib -import json -from pathlib import Path -import re -import urllib.request -from typing import Any - - -ROOT = Path(__file__).resolve().parents[1] -SEMVER = re.compile(r"[0-9]+\.[0-9]+\.[0-9]+") -SHA256 = re.compile(r"[0-9a-f]{64}") -DEX_MANIFEST_URL = re.compile( - r"https://github\.com/superdurable/dex/releases/download/server/v" - r"([0-9]+\.[0-9]+\.[0-9]+)/dex-compatibility-v\1\.json" -) - - -class UpgradeError(RuntimeError): - """The requested Dex upgrade is incomplete or inconsistent.""" - - -def require(condition: bool, message: str) -> None: - if not condition: - raise UpgradeError(message) - - -def download(url: str) -> bytes: - request = urllib.request.Request(url, headers={"User-Agent": "superagent-dex-upgrade/1"}) - with urllib.request.urlopen(request, timeout=30) as response: - return response.read() - - -def replace_once(path: Path, pattern: str, replacement: str) -> None: - content = path.read_text(encoding="utf-8") - updated, count = re.subn(pattern, replacement, content, count=1, flags=re.MULTILINE) - require(count == 1, f"expected one version pin in {path}") - path.write_text(updated, encoding="utf-8") - - -def validate_manifest( - manifest_url: str, - manifest_sha256: str, - content: bytes, -) -> dict[str, Any]: - match = DEX_MANIFEST_URL.fullmatch(manifest_url) - require(match is not None, "manifest URL is not an immutable Dex Server release asset") - require(SHA256.fullmatch(manifest_sha256) is not None, "manifest SHA-256 is invalid") - require(hashlib.sha256(content).hexdigest() == manifest_sha256, "manifest SHA-256 mismatch") - manifest = json.loads(content) - version = match.group(1) - require(manifest["release"] == version, "manifest release does not match its URL") - require(manifest["rolloutOrder"] == "server-first", "Dex rollout must be server-first") - require( - manifest["components"]["sdkGo"]["version"] == version, - "Dex Go SDK version does not match the release", - ) - server_protocol = manifest["protocol"]["server"] - go_protocol = manifest["protocol"]["clients"]["sdkGo"] - require( - max(server_protocol["minimum"], go_protocol["minimum"]) - <= min(server_protocol["maximum"], go_protocol["maximum"]), - "Dex Go SDK and Server protocols are incompatible", - ) - require(manifest["persistenceCompatibility"] == "compatible", "Dex persistence is incompatible") - return manifest - - -def update_repository( - root: Path, - manifest_url: str, - manifest_sha256: str, - manifest: dict[str, Any], -) -> None: - version = manifest["release"] - checksums = manifest["components"]["cli"]["checksums"] - archives = tuple( - f"dexcli_v{version}_{platform}_{architecture}.tar.gz" - for platform in ("darwin", "linux") - for architecture in ("amd64", "arm64") - ) - require(set(checksums) == set(archives), "Dex CLI checksums are incomplete") - replace_once( - root / "go.mod", - r"(github\.com/superdurable/dex/sdk-go\s+)v[^\s]+", - rf"\g<1>v{version}", - ) - replace_once(root / "Makefile", r"^DEXCLI_VERSION := v[^\s]+$", f"DEXCLI_VERSION := v{version}") - installer = root / "script/install-dexcli.sh" - installer_content = installer.read_text(encoding="utf-8") - cases = "\n".join( - f" {archive}) checksum={checksums[archive]} ;;" - for archive in archives - ) - updated, count = re.subn( - r" dexcli_v[^\n]+\n dexcli_v[^\n]+\n dexcli_v[^\n]+\n dexcli_v[^\n]+", - cases, - installer_content, - count=1, - ) - require(count == 1, "expected four Dex CLI checksum pins") - installer.write_text(updated, encoding="utf-8") - - lock = { - "schemaVersion": 1, - "release": version, - "manifest": {"url": manifest_url, "sha256": manifest_sha256}, - "sourceCommit": manifest["sourceCommit"], - "sdkGoVersion": manifest["components"]["sdkGo"]["version"], - "protocol": manifest["protocol"]["clients"]["sdkGo"], - "runningFlowsCompatibility": manifest["runningFlowsCompatibility"], - "persistenceCompatibility": manifest["persistenceCompatibility"], - "openFlowsCompatibility": "cancel-required", - } - (root / "dex-release.lock.json").write_text( - json.dumps(lock, indent=2) + "\n", - encoding="utf-8", - ) - - -def main() -> None: - parser = argparse.ArgumentParser() - parser.add_argument("--manifest-url", required=True) - parser.add_argument("--manifest-sha256", required=True) - args = parser.parse_args() - content = download(args.manifest_url) - manifest = validate_manifest(args.manifest_url, args.manifest_sha256, content) - update_repository(ROOT, args.manifest_url, args.manifest_sha256, manifest) - print(f'Prepared SuperAgent for Dex {manifest["release"]}; open Flows require review') - - -if __name__ == "__main__": - main() diff --git a/script/update_dex_release_test.py b/script/update_dex_release_test.py deleted file mode 100644 index 77be304..0000000 --- a/script/update_dex_release_test.py +++ /dev/null @@ -1,129 +0,0 @@ -#!/usr/bin/env python3 -# Copyright (c) 2026 Super Durable, Inc. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# SPDX-License-Identifier: Apache-2.0 - -from __future__ import annotations - -import copy -import hashlib -import importlib.util -import json -from pathlib import Path -import sys -import tempfile -import unittest - - -MODULE_PATH = Path(__file__).with_name("update_dex_release.py") -SPEC = importlib.util.spec_from_file_location("update_dex_release", MODULE_PATH) -assert SPEC and SPEC.loader -MODULE = importlib.util.module_from_spec(SPEC) -sys.modules[SPEC.name] = MODULE -SPEC.loader.exec_module(MODULE) - - -def manifest() -> dict[str, object]: - return { - "release": "1.2.3", - "sourceCommit": "a" * 40, - "rolloutOrder": "server-first", - "runningFlowsCompatibility": "compatible", - "persistenceCompatibility": "compatible", - "protocol": { - "server": {"minimum": 2, "maximum": 3}, - "clients": {"sdkGo": {"minimum": 2, "maximum": 3}}, - }, - "components": { - "sdkGo": {"version": "1.2.3"}, - "cli": { - "checksums": { - "dexcli_v1.2.3_darwin_amd64.tar.gz": "1" * 64, - "dexcli_v1.2.3_darwin_arm64.tar.gz": "2" * 64, - "dexcli_v1.2.3_linux_amd64.tar.gz": "3" * 64, - "dexcli_v1.2.3_linux_arm64.tar.gz": "4" * 64, - } - }, - }, - } - - -class UpdateDexReleaseTests(unittest.TestCase): - def test_validates_and_updates_all_superagent_pins(self) -> None: - value = manifest() - content = (json.dumps(value) + "\n").encode() - digest = hashlib.sha256(content).hexdigest() - url = ( - "https://github.com/superdurable/dex/releases/download/server/v1.2.3/" - "dex-compatibility-v1.2.3.json" - ) - validated = MODULE.validate_manifest(url, digest, content) - with tempfile.TemporaryDirectory() as directory: - root = Path(directory) - (root / "script").mkdir() - (root / "go.mod").write_text( - "require github.com/superdurable/dex/sdk-go v0.9.0\n", encoding="utf-8" - ) - (root / "Makefile").write_text("DEXCLI_VERSION := v0.9.0\n", encoding="utf-8") - (root / "script/install-dexcli.sh").write_text( - "case x in\n" - + "\n".join(f" dexcli_v0.9.0_{name}.tar.gz) checksum=old ;;" for name in ( - "darwin_amd64", "darwin_arm64", "linux_amd64", "linux_arm64" - )) - + "\nesac\n", - encoding="utf-8", - ) - MODULE.update_repository(root, url, digest, validated) - lock = json.loads((root / "dex-release.lock.json").read_text(encoding="utf-8")) - self.assertEqual(lock["release"], "1.2.3") - self.assertEqual(lock["protocol"], {"minimum": 2, "maximum": 3}) - self.assertEqual(lock["openFlowsCompatibility"], "cancel-required") - self.assertIn("sdk-go v1.2.3", (root / "go.mod").read_text(encoding="utf-8")) - self.assertIn("DEXCLI_VERSION := v1.2.3", (root / "Makefile").read_text(encoding="utf-8")) - self.assertIn("checksum=" + "4" * 64, (root / "script/install-dexcli.sh").read_text(encoding="utf-8")) - - def test_rejects_tampering_and_incompatible_protocol(self) -> None: - value = manifest() - content = (json.dumps(value) + "\n").encode() - url = ( - "https://github.com/superdurable/dex/releases/download/server/v1.2.3/" - "dex-compatibility-v1.2.3.json" - ) - with self.assertRaisesRegex(MODULE.UpgradeError, "SHA-256 mismatch"): - MODULE.validate_manifest(url, "0" * 64, content) - incompatible = copy.deepcopy(value) - incompatible["protocol"]["clients"]["sdkGo"] = {"minimum": 4, "maximum": 4} - incompatible_content = (json.dumps(incompatible) + "\n").encode() - with self.assertRaisesRegex(MODULE.UpgradeError, "protocols are incompatible"): - MODULE.validate_manifest( - url, - hashlib.sha256(incompatible_content).hexdigest(), - incompatible_content, - ) - - def test_upgrade_workflow_opens_a_draft_before_product_ci(self) -> None: - workflow = (MODULE.ROOT / ".github/workflows/dex-release-upgrade.yml").read_text( - encoding="utf-8" - ) - self.assertIn("draft: true", workflow) - self.assertIn("Normal pull-request CI owns compilation and tests", workflow) - self.assertNotIn("make governance-check format-check vet test", workflow) - self.assertLess( - workflow.index("Update immutable Dex release pins"), - workflow.index("Open draft upgrade pull request before compilation"), - ) - - -if __name__ == "__main__": - unittest.main() diff --git a/script/update_dex_versions.py b/script/update_dex_versions.py new file mode 100644 index 0000000..d0e486e --- /dev/null +++ b/script/update_dex_versions.py @@ -0,0 +1,117 @@ +#!/usr/bin/env python3 +# Copyright (c) 2026 Super Durable, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# SPDX-License-Identifier: Apache-2.0 + +"""Update SuperAgent's independent Dex Go SDK and dexcli pins.""" + +from __future__ import annotations + +import argparse +from pathlib import Path +import re +import urllib.request + + +ROOT = Path(__file__).resolve().parents[1] +SEMVER = re.compile(r"[0-9]+\.[0-9]+\.[0-9]+") +PLATFORMS = ("darwin_amd64", "darwin_arm64", "linux_amd64", "linux_arm64") + + +class DexVersionError(RuntimeError): + """A requested Dex component version or release asset is invalid.""" + + +def require(condition: bool, message: str) -> None: + if not condition: + raise DexVersionError(message) + + +def replace_once(path: Path, pattern: str, replacement: str) -> None: + content = path.read_text(encoding="utf-8") + updated, count = re.subn(pattern, replacement, content, count=1, flags=re.MULTILINE) + require(count == 1, f"expected one version pin in {path}") + path.write_text(updated, encoding="utf-8") + + +def download_cli_checksums(version: str) -> bytes: + url = f"https://github.com/superdurable/dex/releases/download/cli-v{version}/checksums.txt" + request = urllib.request.Request(url, headers={"User-Agent": "superagent-dexcli-upgrade/1"}) + with urllib.request.urlopen(request, timeout=30) as response: + return response.read() + + +def parse_cli_checksums(version: str, content: bytes) -> dict[str, str]: + checksums: dict[str, str] = {} + for line in content.decode("utf-8").splitlines(): + match = re.fullmatch(r"([0-9a-f]{64})\s+\*?\.?/?(dexcli_v[^/\s]+\.tar\.gz)", line) + if match is not None: + checksums[match.group(2)] = match.group(1) + expected = {f"dexcli_v{version}_{platform}.tar.gz" for platform in PLATFORMS} + require(set(checksums) == expected, "dexcli checksums.txt does not contain the four release archives") + return checksums + + +def update_repository( + root: Path, + sdk_go_version: str, + dexcli_version: str, + cli_checksums: dict[str, str], +) -> None: + require(SEMVER.fullmatch(sdk_go_version) is not None, "invalid Dex Go SDK version") + require(SEMVER.fullmatch(dexcli_version) is not None, "invalid dexcli version") + replace_once( + root / "go.mod", + r"(github\.com/superdurable/dex/sdk-go\s+)v[^\s]+", + rf"\g<1>v{sdk_go_version}", + ) + replace_once( + root / "Makefile", r"^DEXCLI_VERSION := v[^\s]+$", f"DEXCLI_VERSION := v{dexcli_version}" + ) + installer = root / "script/install-dexcli.sh" + cases = "\n".join( + f" {archive}) checksum={cli_checksums[archive]} ;;" + for archive in sorted(cli_checksums) + ) + content = installer.read_text(encoding="utf-8") + updated, count = re.subn( + r" dexcli_v[^\n]+\n dexcli_v[^\n]+\n dexcli_v[^\n]+\n dexcli_v[^\n]+", + cases, + content, + count=1, + ) + require(count == 1, "expected four dexcli checksum pins") + installer.write_text(updated, encoding="utf-8") + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--sdk-go-version", required=True) + parser.add_argument("--dexcli-version", required=True) + parser.add_argument("--checksums-file", type=Path) + arguments = parser.parse_args() + checksum_content = ( + arguments.checksums_file.read_bytes() + if arguments.checksums_file is not None + else download_cli_checksums(arguments.dexcli_version) + ) + checksums = parse_cli_checksums(arguments.dexcli_version, checksum_content) + update_repository(ROOT, arguments.sdk_go_version, arguments.dexcli_version, checksums) + print( + f"Updated Dex Go SDK to {arguments.sdk_go_version} and dexcli to {arguments.dexcli_version}" + ) + + +if __name__ == "__main__": + main() diff --git a/script/update_dex_versions_test.py b/script/update_dex_versions_test.py new file mode 100644 index 0000000..6e5df35 --- /dev/null +++ b/script/update_dex_versions_test.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +# Copyright (c) 2026 Super Durable, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import importlib.util +from pathlib import Path +import sys +import tempfile +import unittest + + +def load_module(name: str): + path = Path(__file__).with_name(f"{name}.py") + spec = importlib.util.spec_from_file_location(name, path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +CHECK = load_module("check_dex_versions") +UPDATE = load_module("update_dex_versions") + + +def checksums(version: str) -> bytes: + return "\n".join( + f"{'1234abcd' * 8} ./dexcli_v{version}_{platform}.tar.gz" + for platform in UPDATE.PLATFORMS + ).encode() + + +class DexVersionTests(unittest.TestCase): + def test_updates_and_checks_independent_component_versions(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "script").mkdir() + (root / ".github/workflows").mkdir(parents=True) + (root / "go.mod").write_text( + "require (\n\tgithub.com/superdurable/dex/sdk-go v0.9.0\n)\n", encoding="utf-8" + ) + (root / "Makefile").write_text("DEXCLI_VERSION := v0.9.0\n", encoding="utf-8") + (root / "script/install-dexcli.sh").write_text( + "\n".join( + f" dexcli_v0.9.0_{platform}.tar.gz) checksum=old ;;" + for platform in UPDATE.PLATFORMS + ) + + "\n", + encoding="utf-8", + ) + (root / ".github/workflows/ci.yml").write_text( + ".cache/dexcli-v1.2.4 dev\n", encoding="utf-8" + ) + parsed = UPDATE.parse_cli_checksums("1.2.4", checksums("1.2.4")) + UPDATE.update_repository(root, "1.2.3", "1.2.4", parsed) + self.assertEqual(CHECK.read_versions(root), ("1.2.3", "1.2.4")) + + def test_rejects_incomplete_checksums_and_version_drift(self) -> None: + with self.assertRaisesRegex(UPDATE.DexVersionError, "four release archives"): + UPDATE.parse_cli_checksums("1.2.3", b"") + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "script").mkdir() + (root / ".github/workflows").mkdir(parents=True) + (root / "go.mod").write_text( + "require (\n\tgithub.com/superdurable/dex/sdk-go v1.2.3\n)\n", encoding="utf-8" + ) + (root / "Makefile").write_text("DEXCLI_VERSION := v1.2.3\n", encoding="utf-8") + (root / "script/install-dexcli.sh").write_text( + "\n".join( + f" dexcli_v1.2.2_{platform}.tar.gz) checksum=old ;;" + for platform in UPDATE.PLATFORMS + ), + encoding="utf-8", + ) + (root / ".github/workflows/ci.yml").write_text( + ".cache/dexcli-v1.2.3 dev\n", encoding="utf-8" + ) + with self.assertRaisesRegex(CHECK.DexVersionError, "installer versions"): + CHECK.read_versions(root) + + +if __name__ == "__main__": + unittest.main() diff --git a/web/mcp-servers.example.yaml b/web/mcp-servers.example.yaml index 6446842..194c9aa 100644 --- a/web/mcp-servers.example.yaml +++ b/web/mcp-servers.example.yaml @@ -9,7 +9,11 @@ servers: tools: brave_web_search: read_only: true + # Defaults to short_running. Use long_running when most calls exceed five seconds. + running_type: short_running timeout_seconds: 30 + # Defaults to 60. Raise only for healthy tools with longer silent intervals. + heartbeat_timeout_seconds: 60 maximum_attempts: 3 retry_total_seconds: 120 # Defaults to manual_recovery. Use this only when automatic progress is safe. diff --git a/web/src/App.test.tsx b/web/src/App.test.tsx index e2bcae0..4c8c4dc 100644 --- a/web/src/App.test.tsx +++ b/web/src/App.test.tsx @@ -291,6 +291,35 @@ describe("App", () => { }); }); + it("reconciles when a recovered Stream tail requires a Snapshot", async () => { + vi.mocked(listRecentEvents).mockImplementation(({ query }) => + Promise.resolve({ + events: + query.stream === EventStream.ACTIVITY + ? [ + activityEvent( + "snapshot-required-recovered", + EventKind.SNAPSHOT_REQUIRED, + "Durable interaction state changed.", + "2026-09-03T00:01:00Z", + ), + ] + : [], + }), + ); + window.history.replaceState({}, "", "/?flowId=flow-existing"); + + render(); + + await screen.findByRole("heading", { name: "SuperAgent" }); + await waitFor(() => { + expect(getAgentSnapshot).toHaveBeenCalledTimes(3); + }); + expect( + screen.queryByText("Durable interaction state changed."), + ).not.toBeInTheDocument(); + }); + it("starts through the generated client and loads one Snapshot", async () => { render(); const button = await screen.findByRole("button", { name: "Start agent" }); diff --git a/web/src/Conversation.tsx b/web/src/Conversation.tsx index 6291208..d9eda7c 100644 --- a/web/src/Conversation.tsx +++ b/web/src/Conversation.tsx @@ -174,10 +174,15 @@ export function Conversation({ const newest = recent.events.at(-1); resumeToken = newest?.resumeToken; resumeTokens.current[stream] = resumeToken; - dispatch({ - type: "stream-recovered", - updates: recent.events.map((event) => liveUpdate(stream, event)), - }); + const updates = recent.events.map((event) => + liveUpdate(stream, event), + ); + dispatch({ type: "stream-recovered", updates }); + if (updates.some(shouldReconcileAfter)) { + requestSnapshot({ blocking: false }); + // Stream visibility can precede the durable wait commit. + requestSnapshot({ blocking: false }); + } } catch (reason: unknown) { if (isAbortError(reason)) return; isCurrent = false;