Add stateful agent runtime on LangGraph (#70) - #80
Conversation
Implements a durable, resumable, checkpointed execution graph behind the existing @muster/agent-harness invocation boundary, per the ReceiveInvocation -> ... -> PersistRunResult design in issue #70. PostgreSQL stays authoritative for identity, capabilities, approvals, tool permissions and final run status; LangGraph checkpoints (new agent_runtime_checkpoints/checkpoint_writes tables) hold only execution state, fully organisation-scoped and version-stamped. - Provider-neutral model router (openai-compatible, anthropic, ollama, openrouter, scripted) selected by ModelPolicy, never a hard-coded provider. - Tool authorisation replays the existing agent tool registry/allowlist/ capability/approval gates before any argument is trusted; tool-call reservations are idempotency-keyed so a resumed run never repeats a completed or in-flight external action. - Approval interrupts pause the graph via LangGraph's interrupt() and resume from the exact node once a decision resolves, without creating a second approval record. - The kill switch and cancellation signal are re-checked at every graph step boundary, not only at claim time. - Runtime events are reduced to a closed, schema-validated vocabulary before they can reach a stream, room timeline or Slack, so hidden reasoning never leaks. - apps/agent-gateway gains an opt-in `MUSTER_AGENT_RUNTIME=graph` execution mode; `codex` remains the default and existing behaviour is unchanged. Adds migration 0021 (agent_runtime_checkpoints, agent_runtime_checkpoint_writes, new agent_runs/agent_tool_calls columns), ADR 0005, and 148 tests (132 offline, 16 against a real PostgreSQL instance covering crash recovery, cross-tenant isolation, approval resume and idempotent tool replay). Closes #70 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
📝 WalkthroughWalkthroughAdds a new LangGraph-based stateful agent runtime with resumable PostgreSQL checkpoints, model-provider routing, governed tool execution, approval interrupts, event streaming, graph-version checks, gateway integration, and supporting schema migrations, documentation, and tests. ChangesStateful agent runtime
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Gateway
participant MusterAgentRuntime
participant PostgreSQL
participant ModelRouter
participant ToolExecutor
Gateway->>MusterAgentRuntime: start or resume run
MusterAgentRuntime->>PostgreSQL: load scope and checkpoint state
MusterAgentRuntime->>ModelRouter: generate next step
ModelRouter-->>MusterAgentRuntime: final response or tool proposal
MusterAgentRuntime->>PostgreSQL: authorize, reserve, and persist state
MusterAgentRuntime->>ToolExecutor: execute approved tool
ToolExecutor-->>MusterAgentRuntime: tool outcome
MusterAgentRuntime->>PostgreSQL: persist checkpoint, result, and events
MusterAgentRuntime-->>Gateway: run handle or streamed event
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 10
🧹 Nitpick comments (14)
packages/agent-runtime/src/agent-runtime.integration.test.ts (1)
100-109: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueThe
capabilitiesparameter type is misleading.
(typeof schema.agentDefinitions.$inferSelect)["id"][]is juststring[], but it reads as if capability names were agent definition ids, and it still requires the cast on line 107. Typing the parameter as the element type ofAuthorisationSubject["capabilities"]states the intent and removes the cast.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/agent-runtime/src/agent-runtime.integration.test.ts` around lines 100 - 109, Update subjectFor’s capabilities parameter to use the element type of AuthorisationSubject["capabilities"] rather than the agentDefinitions id array type, then construct the Set directly without the cast while preserving the returned AuthorisationSubject shape.packages/agent-runtime/src/runtime.test.ts (1)
187-189: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
JSON.stringifyon an Error serialises to{}, so this leak assertion is close to vacuous.
AgentRuntimeError'smessage/codeare not own enumerable properties, soJSON.stringify(inspectError)yields{}and the assertion would pass even if the message did embed the foreign run id. Asserting onmessage(andcode) directly makes the intent real.♻️ Suggested tightening
- expect(JSON.stringify(inspectError)).not.toContain(scope.runId); + expect((inspectError as AgentRuntimeError).message).not.toContain(scope.runId); + expect((inspectError as AgentRuntimeError).message).not.toContain(otherRunId);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/agent-runtime/src/runtime.test.ts` around lines 187 - 189, Update the leak assertion in the AgentRuntimeError test to inspect the error’s message directly instead of JSON.stringify(inspectError), and assert that both message and code do not contain scope.runId. Preserve the existing AgentRuntimeError instance and stale_run code assertions.packages/agent-runtime/src/graph/graph.test.ts (1)
302-321: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winThe kill-switch flip is coupled to an exact guard call count.
guardCalls <= 5encodes the current number and order ofguard()invocations across nodes. Adding or removing a single guard check shifts the flip point, and the test can then silently assert something different (e.g. flipping before the tool runs) while still passing onfailureCode. Keying the flip off an observable milestone instead — for example flipping onceports.executions.length === 1— keeps the intent stable.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/agent-runtime/src/graph/graph.test.ts` around lines 302 - 321, Update the kill-switch condition in the fake ports verdict guard to use the observable execution milestone rather than the exact guardCalls count. Flip to the agent_kill_switch response once ports.executions.length indicates the first tool execution has completed, while preserving the runnable response before that milestone and the existing failure assertions.packages/agent-runtime/src/checkpointer/postgres.ts (3)
534-536: 🎯 Functional Correctness | 🔵 Trivial | 💤 Low valueAdd a deterministic tie-breaker to
latestCheckpointId.Ordering only by
createdAtcan return an arbitrary row when two checkpoints share a timestamp, and it disagrees with thecheckpointId-based ordering used bygetTuple/list.♻️ Proposed tie-breaker
- .orderBy(desc(schema.agentRuntimeCheckpoints.createdAt)) + .orderBy( + desc(schema.agentRuntimeCheckpoints.createdAt), + desc(schema.agentRuntimeCheckpoints.checkpointId), + )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/agent-runtime/src/checkpointer/postgres.ts` around lines 534 - 536, Add a deterministic secondary ordering to latestCheckpointId after createdAt, using checkpointId in the same direction as the ordering in getTuple and list. Keep the existing descending createdAt order and limit(1) behavior unchanged.
432-469: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winConsider batching pending writes into one transaction (or two grouped inserts).
Each write is a separate round trip, and the loop is not transactional, so a crash mid-loop leaves a partial write set for the task (replay recovers it, but only because every insert is idempotent). Grouping the
writeIndex >= 0rows into oneonConflictDoNothinginsert and the negative-index rows into oneonConflictDoUpdateinsert inside a singledb.transactionkeeps the same semantics with far fewer round trips on hot graph steps.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/agent-runtime/src/checkpointer/postgres.ts` around lines 432 - 469, Update the write-persistence loop to batch rows by conflict behavior: collect non-negative writeIndex rows for one onConflictDoNothing insert and negative-index rows for one onConflictDoUpdate insert. Execute both grouped inserts inside a single db.transaction, preserving the existing conflict targets and update fields while avoiding per-write round trips and partial persistence.
296-313: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low value
listmaterialises every checkpoint row for the thread before applyinglimit.The
limitis applied in JS (to allow the metadata filter), so a long-lived thread reads its full checkpoint history — including deserialising each row and issuing one pending-writes query per yielded row — even for{ limit: 1 }. Whenoptions.filteris absent, the limit can be pushed into SQL safely.♻️ Push limit into SQL when no metadata filter is present
- const rows = await this.db + const query = this.db .select() .from(schema.agentRuntimeCheckpoints) .where( organisationScopedWhere( schema.agentRuntimeCheckpoints.organisationId, this.organisationId, eq(schema.agentRuntimeCheckpoints.threadId, threadId), eq( schema.agentRuntimeCheckpoints.checkpointNamespace, checkpointNamespace, ), ...(beforeCheckpointId ? [lt(schema.agentRuntimeCheckpoints.checkpointId, beforeCheckpointId)] : []), ), ) .orderBy(desc(schema.agentRuntimeCheckpoints.checkpointId)); + const rows = + filter === undefined && remaining !== undefined + ? await query.limit(remaining) + : await query;🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/agent-runtime/src/checkpointer/postgres.ts` around lines 296 - 313, Update the checkpoint query in list to apply the requested limit at the database level when options.filter is absent, while preserving the existing JavaScript filtering and limit behavior when a metadata filter is provided. Use the query flow around organisationScopedWhere and the subsequent row processing, ensuring beforeCheckpointId and checkpoint ordering remain unchanged.packages/agent-runtime/src/checkpointer/postgres.integration.test.ts (1)
142-152: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueThis assertion doesn't test the upsert.
expect(rows).toBeDefined()passes as long as any row comes back; the intent (one row, not two, after the secondput) isn't checked. Select a count instead.♻️ Assert a single row for the checkpoint id
- const [rows] = await db - .select({ value: schema.agentRuntimeCheckpoints.checkpointId }) + const [rows] = await db + .select({ value: count() }) .from(schema.agentRuntimeCheckpoints) .where( and( eq(schema.agentRuntimeCheckpoints.organisationId, organisationAId), eq(schema.agentRuntimeCheckpoints.threadId, saver.threadId), eq(schema.agentRuntimeCheckpoints.checkpointId, checkpoint.id), ), ); - expect(rows).toBeDefined(); + expect(rows?.value).toBe(1);Requires adding
countto thedrizzle-ormimport.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/agent-runtime/src/checkpointer/postgres.integration.test.ts` around lines 142 - 152, Update the verification query in the postgres checkpoint test to select a count using Drizzle’s count helper, adding count to the drizzle-orm import. Replace the rows-defined assertion with an assertion that exactly one record exists for the checkpoint after the second put, preserving the existing filters.packages/agent-runtime/tsconfig.json (1)
4-8: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winTest files will be emitted into
dist.
include: ["src/**/*.ts"]picks up*.test.ts/*.integration.test.ts, so withoutDir+declarationthe published build ships test compilation output (and pullsvitesttypes into the build graph). Exclude them.♻️ Proposed change
- "include": ["src/**/*.ts"] + "include": ["src/**/*.ts"], + "exclude": ["src/**/*.test.ts", "src/**/*.integration.test.ts"]🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/agent-runtime/tsconfig.json` around lines 4 - 8, Update the tsconfig include/exclude configuration around "include" so test files such as *.test.ts and *.integration.test.ts are excluded from compilation and declaration output, while retaining production TypeScript sources under src for emission to dist.packages/agent-runtime/src/model/router.test.ts (1)
171-194: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a case for a non-transient failure on the retry hop.
router.tsline 122 rethrowssecondErrorunchanged when the second provider fails non-transiently; no test exercises that branch, so a regression that wrapped it asmodel_provider_unavailable(masking e.g. a config error) would pass.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/agent-runtime/src/model/router.test.ts` around lines 171 - 194, Extend the router tests with a case where the primary provider fails transiently, the retry provider fails with a non-transient error, and router.generate rethrows that second error unchanged. Use distinct error details and assert the original error/code is preserved rather than converted to model_provider_unavailable, covering the retry branch in router.ts.packages/agent-runtime/src/adapters/index.ts (1)
1-1: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winWildcard barrels widen the package's public API beyond intent. The root
src/index.tsenumerates exports for every hand-written module but re-exports the adapter/model barrels withexport *, andadapters/index.tsitself is a bareexport *, so every internal Postgres helper becomes a public, semver-relevant symbol.
packages/agent-runtime/src/adapters/index.ts#L1-L1: replaceexport * from "./postgres.ts"with explicit named exports, matchingcheckpointer/index.ts.packages/agent-runtime/src/index.ts#L74-L76: re-export the named symbols from the model/checkpointer/adapters barrels instead ofexport *.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/agent-runtime/src/adapters/index.ts` at line 1, Replace the wildcard export in packages/agent-runtime/src/adapters/index.ts:1 with explicit named exports from the Postgres adapter, matching the pattern used by checkpointer/index.ts. In packages/agent-runtime/src/index.ts:74-76, replace wildcard re-exports for the model, checkpointer, and adapters barrels with their intended explicit named symbols, preserving only the package’s deliberate public API.packages/agent-runtime/src/model/providers/anthropic.ts (1)
26-55: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
classEnvSuffix,modelIdFor, anddigestPayloadare copy-pasted across providers.
openai-compatible.tsalready exportsresolveClassModels(env, envPrefix, defaults)anddigestPayload; this file (andollama.ts) reimplement both plusmodelIdForandwireContentFor. Move the shared helpers into one module and import them so the untrusted-envelope wording and class-fallback rules cannot drift per provider.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/agent-runtime/src/model/providers/anthropic.ts` around lines 26 - 55, Consolidate the duplicated helpers in anthropic.ts with the shared implementations exported by openai-compatible.ts: use the shared resolveClassModels, digestPayload, modelIdFor, and wireContentFor utilities, passing the appropriate Anthropic environment prefix and defaults. Remove the local classEnvSuffix, resolveClassModels, modelIdFor, digestPayload, and wireContentFor implementations, and update ollama.ts similarly so all providers share the same envelope wording and class-fallback behavior.packages/agent-runtime/src/events.ts (1)
114-127: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winPrefer fail-fast validation over silent defaults for authoritative fields.
runId,organisationId,sequence, andoccurredAtare authoritative audit fields, but missing/invalid values silently fall back to""/0/epoch rather than throwing. Current call sites always supply these correctly, but the contract itself doesn't guard against a future caller omitting them, risking a corrupted event silently entering the append-only audit trail.♻️ Proposed fix: throw instead of silently defaulting
- const runId = record["runId"]; - const organisationId = record["organisationId"]; - const sequence = record["sequence"]; - const occurredAt = record["occurredAt"]; - return { - ...parsed, - runId: typeof runId === "string" ? runId : "", - organisationId: typeof organisationId === "string" ? organisationId : "", - sequence: typeof sequence === "number" ? sequence : 0, - occurredAt: - typeof occurredAt === "string" ? occurredAt : new Date(0).toISOString(), - }; + const runId = record["runId"]; + const organisationId = record["organisationId"]; + const sequence = record["sequence"]; + const occurredAt = record["occurredAt"]; + if (typeof runId !== "string" || !runId) { + throw new Error("Runtime event is missing a runId"); + } + if (typeof organisationId !== "string" || !organisationId) { + throw new Error("Runtime event is missing an organisationId"); + } + return { + ...parsed, + runId, + organisationId, + sequence: typeof sequence === "number" ? sequence : 0, + occurredAt: typeof occurredAt === "string" ? occurredAt : new Date().toISOString(), + };🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/agent-runtime/src/events.ts` around lines 114 - 127, Update the event construction logic around AgentRuntimeEventPayloadSchema.parse so runId, organisationId, sequence, and occurredAt are validated as required authoritative fields and invalid or missing values throw instead of falling back to empty, zero, or epoch defaults. Preserve the parsed payload and return shape for valid records.packages/agent-runtime/src/adapters/postgres.ts (1)
780-780: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winSilently returning when the update matches nothing loses the settlement.
A missing or cross-tenant
toolCallRecordIdmakessettlea no-op, leaving the reservation inrunningforever — later reserves then reportreplayed: trueand the run has no recorded outcome. Throwing anAgentRuntimeError(asbindRun/persistResultdo for a stale run) surfaces the inconsistency instead of hiding it.Also applies to: 815-815
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/agent-runtime/src/adapters/postgres.ts` at line 780, The settle update path must not silently return when no row is updated. Replace the !updated early returns in settle with AgentRuntimeError failures, matching the stale-run error behavior used by bindRun and persistResult, so missing or cross-tenant toolCallRecordId values surface an inconsistency.packages/database/src/schema.ts (1)
1500-1584: 📐 Maintainability & Code Quality | 🔵 TrivialConsider retention and delete semantics for the two checkpoint tables.
Both tables hold unbounded execution state keyed to a run and carry FKs to
agent_runs/organisationswith noonDeletebehaviour, andagent_runtime_checkpoint_writeshas no FK back toagent_runtime_checkpoints. Consequences worth deciding now rather than after the first large tenant: deleting a settled run or an organisation will fail on these references, and write rows can outlive their checkpoint if anything other thandeleteThreadremoves a checkpoint row. A pruning job for terminal runs plus explicit cascade/restrict choices would keep growth and deletion predictable.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/database/src/schema.ts` around lines 1500 - 1584, Define explicit retention and deletion behavior for agentRuntimeCheckpoints and agentRuntimeCheckpointWrites. Add the appropriate foreign-key relationship from checkpoint writes to agentRuntimeCheckpoints, choose and configure intentional onDelete behavior for agentRuns, organisations, and checkpoint rows, and ensure terminal-run pruning is supported so execution state does not grow without bound.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@apps/agent-gateway/src/index.ts`:
- Around line 172-175: Update the authenticated readiness calculation in
apps/agent-gateway/src/index.ts:172-175 to use the configured, policy-capable
model routing check for graph mode instead of treating every non-Codex runtime
as ready; retain the Codex authentication path. In
apps/agent-gateway/src/runtime.ts:453, record graph authentication/provider
state as unavailable or unknown until that same readiness check succeeds.
In `@packages/agent-runtime/src/adapters/postgres.ts`:
- Around line 187-201: Add a dedicated integer sequence column to
agent_run_events with a unique constraint on organisationId, runId, and
sequence. In nextSequence, replace the non-atomic count-based allocation with
atomic max-plus-one allocation under a lock or insertion with retry on unique
violations. In packages/agent-runtime/src/adapters/postgres.ts lines 1073-1087,
query using sequence > afterSequence and order by sequence instead of filtering
materialised events by createdAt; both sites require updates.
- Around line 1009-1012: Update the terminal failure persistence path to pass
terminal.error through redactObservationText before truncating it, matching the
sanitization used by other observation paths. Preserve the existing
2,000-character limit and store the redacted result in failureCode’s error
field.
In `@packages/agent-runtime/src/graph/nodes.ts`:
- Around line 575-589: Update the proposeMemories flow so memory proposals occur
only after persistRunResult has successfully validated and persisted the
authoritative output, rather than during the pre-persistence corrective-turn
path. Ensure the proposal is executed at most once per run, using an idempotency
key derived from the run identity and output hash when supported by the memories
port, and avoid creating proposals for runs that ultimately fail validation.
- Around line 404-431: Require awaitApproval to verify the authoritative
persisted approval via ports.approvals.read(scope, approvalId) after validating
the resume payload, granting toolAuthorisation only when that record is
approved. In packages/agent-runtime/src/agent-runtime.integration.test.ts lines
397-408, record a decided approval before resuming and add coverage that an
approved resume while the persisted row is pending does not execute
tawny.endpoint.isolate. In packages/agent-runtime/src/testing/index.ts lines
125-147, update the approvals fake to record decisions so read returns approved
or rejected states.
In `@packages/agent-runtime/src/model/providers/anthropic.ts`:
- Around line 183-194: Add a policy/config-sourced default request deadline to
the outbound model calls, composing it with any caller-provided signal so either
cancellation or timeout aborts the request. Update
packages/agent-runtime/src/model/providers/anthropic.ts:183-194 for the
/messages call, packages/agent-runtime/src/model/providers/ollama.ts:191-198 for
/api/chat, and
packages/agent-runtime/src/model/providers/openai-compatible.ts:238-245 for the
shared /chat/completions path; implement the latter through its shared helper so
openrouter inherits the behavior.
In `@packages/agent-runtime/src/model/providers/ollama.ts`:
- Around line 244-250: The fallback toolCallId in the toolCalls mapping must be
unique across steps or messages, not only within the current call list. Update
the fallback used by the Ollama provider to incorporate a stable step/message
discriminator or deterministic hash while preserving Ollama’s supplied call.id
when present, so toolCallIdempotencyKey receives distinct identifiers for
separate tool calls.
In `@packages/agent-runtime/src/runtime.ts`:
- Around line 285-307: Update the unsettled-result handling in the graph
invocation flow around result.settled and snapshot.next so an unsettled graph
with no pending node returns a failure status instead of falling through to
status: "completed". Preserve the awaiting_approval path when snapshot.next is
non-empty, and keep completed status only for settled results.
- Around line 98-105: Update cancelRun so cancellation only proceeds for
non-terminal runs, preventing persistResult from changing completed or failed
rows and suppressing run.cancelled for terminal or duplicate cancellations.
Reuse the existing run-record lookup or terminal-status predicate if available;
otherwise enforce the guard within persistResult while preserving normal
cancellation and event emission for active runs.
In `@packages/database/migrations/0021_closed_steel_serpent.sql`:
- Around line 49-51: Update the migration containing the agent_runs foreign key
and agent_tool_calls indexes to be rollout-safe: add the constraint with NOT
VALID, move its validation to a later migration step, and create both indexes
concurrently. Ensure the migration is marked non-transactional as required for
concurrent index creation while preserving the existing constraint and index
definitions.
---
Nitpick comments:
In `@packages/agent-runtime/src/adapters/index.ts`:
- Line 1: Replace the wildcard export in
packages/agent-runtime/src/adapters/index.ts:1 with explicit named exports from
the Postgres adapter, matching the pattern used by checkpointer/index.ts. In
packages/agent-runtime/src/index.ts:74-76, replace wildcard re-exports for the
model, checkpointer, and adapters barrels with their intended explicit named
symbols, preserving only the package’s deliberate public API.
In `@packages/agent-runtime/src/adapters/postgres.ts`:
- Line 780: The settle update path must not silently return when no row is
updated. Replace the !updated early returns in settle with AgentRuntimeError
failures, matching the stale-run error behavior used by bindRun and
persistResult, so missing or cross-tenant toolCallRecordId values surface an
inconsistency.
In `@packages/agent-runtime/src/agent-runtime.integration.test.ts`:
- Around line 100-109: Update subjectFor’s capabilities parameter to use the
element type of AuthorisationSubject["capabilities"] rather than the
agentDefinitions id array type, then construct the Set directly without the cast
while preserving the returned AuthorisationSubject shape.
In `@packages/agent-runtime/src/checkpointer/postgres.integration.test.ts`:
- Around line 142-152: Update the verification query in the postgres checkpoint
test to select a count using Drizzle’s count helper, adding count to the
drizzle-orm import. Replace the rows-defined assertion with an assertion that
exactly one record exists for the checkpoint after the second put, preserving
the existing filters.
In `@packages/agent-runtime/src/checkpointer/postgres.ts`:
- Around line 534-536: Add a deterministic secondary ordering to
latestCheckpointId after createdAt, using checkpointId in the same direction as
the ordering in getTuple and list. Keep the existing descending createdAt order
and limit(1) behavior unchanged.
- Around line 432-469: Update the write-persistence loop to batch rows by
conflict behavior: collect non-negative writeIndex rows for one
onConflictDoNothing insert and negative-index rows for one onConflictDoUpdate
insert. Execute both grouped inserts inside a single db.transaction, preserving
the existing conflict targets and update fields while avoiding per-write round
trips and partial persistence.
- Around line 296-313: Update the checkpoint query in list to apply the
requested limit at the database level when options.filter is absent, while
preserving the existing JavaScript filtering and limit behavior when a metadata
filter is provided. Use the query flow around organisationScopedWhere and the
subsequent row processing, ensuring beforeCheckpointId and checkpoint ordering
remain unchanged.
In `@packages/agent-runtime/src/events.ts`:
- Around line 114-127: Update the event construction logic around
AgentRuntimeEventPayloadSchema.parse so runId, organisationId, sequence, and
occurredAt are validated as required authoritative fields and invalid or missing
values throw instead of falling back to empty, zero, or epoch defaults. Preserve
the parsed payload and return shape for valid records.
In `@packages/agent-runtime/src/graph/graph.test.ts`:
- Around line 302-321: Update the kill-switch condition in the fake ports
verdict guard to use the observable execution milestone rather than the exact
guardCalls count. Flip to the agent_kill_switch response once
ports.executions.length indicates the first tool execution has completed, while
preserving the runnable response before that milestone and the existing failure
assertions.
In `@packages/agent-runtime/src/model/providers/anthropic.ts`:
- Around line 26-55: Consolidate the duplicated helpers in anthropic.ts with the
shared implementations exported by openai-compatible.ts: use the shared
resolveClassModels, digestPayload, modelIdFor, and wireContentFor utilities,
passing the appropriate Anthropic environment prefix and defaults. Remove the
local classEnvSuffix, resolveClassModels, modelIdFor, digestPayload, and
wireContentFor implementations, and update ollama.ts similarly so all providers
share the same envelope wording and class-fallback behavior.
In `@packages/agent-runtime/src/model/router.test.ts`:
- Around line 171-194: Extend the router tests with a case where the primary
provider fails transiently, the retry provider fails with a non-transient error,
and router.generate rethrows that second error unchanged. Use distinct error
details and assert the original error/code is preserved rather than converted to
model_provider_unavailable, covering the retry branch in router.ts.
In `@packages/agent-runtime/src/runtime.test.ts`:
- Around line 187-189: Update the leak assertion in the AgentRuntimeError test
to inspect the error’s message directly instead of JSON.stringify(inspectError),
and assert that both message and code do not contain scope.runId. Preserve the
existing AgentRuntimeError instance and stale_run code assertions.
In `@packages/agent-runtime/tsconfig.json`:
- Around line 4-8: Update the tsconfig include/exclude configuration around
"include" so test files such as *.test.ts and *.integration.test.ts are excluded
from compilation and declaration output, while retaining production TypeScript
sources under src for emission to dist.
In `@packages/database/src/schema.ts`:
- Around line 1500-1584: Define explicit retention and deletion behavior for
agentRuntimeCheckpoints and agentRuntimeCheckpointWrites. Add the appropriate
foreign-key relationship from checkpoint writes to agentRuntimeCheckpoints,
choose and configure intentional onDelete behavior for agentRuns, organisations,
and checkpoint rows, and ensure terminal-run pruning is supported so execution
state does not grow without bound.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: c586927e-a73d-4dd9-8765-3d44a15d0a66
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (48)
.env.exampleREADME.mdapps/agent-gateway/package.jsonapps/agent-gateway/src/index.tsapps/agent-gateway/src/runtime.tsdocs/architecture/0005-stateful-agent-runtime.mdpackages/agent-runtime/package.jsonpackages/agent-runtime/src/adapters/index.tspackages/agent-runtime/src/adapters/postgres.test.tspackages/agent-runtime/src/adapters/postgres.tspackages/agent-runtime/src/agent-runtime.integration.test.tspackages/agent-runtime/src/checkpointer/index.tspackages/agent-runtime/src/checkpointer/postgres.integration.test.tspackages/agent-runtime/src/checkpointer/postgres.test.tspackages/agent-runtime/src/checkpointer/postgres.tspackages/agent-runtime/src/errors.tspackages/agent-runtime/src/events.test.tspackages/agent-runtime/src/events.tspackages/agent-runtime/src/graph/build.tspackages/agent-runtime/src/graph/graph.test.tspackages/agent-runtime/src/graph/nodes.tspackages/agent-runtime/src/graph/state.tspackages/agent-runtime/src/identity.test.tspackages/agent-runtime/src/identity.tspackages/agent-runtime/src/index.tspackages/agent-runtime/src/model/index.tspackages/agent-runtime/src/model/providers/anthropic.tspackages/agent-runtime/src/model/providers/ollama.tspackages/agent-runtime/src/model/providers/openai-compatible.tspackages/agent-runtime/src/model/providers/openrouter.tspackages/agent-runtime/src/model/providers/providers.test.tspackages/agent-runtime/src/model/providers/scripted.tspackages/agent-runtime/src/model/router.test.tspackages/agent-runtime/src/model/router.tspackages/agent-runtime/src/model/types.tspackages/agent-runtime/src/ports.tspackages/agent-runtime/src/runtime.test.tspackages/agent-runtime/src/runtime.tspackages/agent-runtime/src/testing/index.tspackages/agent-runtime/src/types.tspackages/agent-runtime/src/version.test.tspackages/agent-runtime/src/version.tspackages/agent-runtime/tsconfig.jsonpackages/database/migrations/0021_closed_steel_serpent.sqlpackages/database/migrations/meta/0021_snapshot.jsonpackages/database/migrations/meta/_journal.jsonpackages/database/src/schema.tspackages/database/src/verify-clean-install.ts
| // Only the Codex subscription runtime depends on a local Codex login; the | ||
| // stateful graph runtime authenticates per model provider instead. | ||
| const authenticated = | ||
| executionRuntime === "mock" || (await codexAuthenticated()); | ||
| executionRuntime !== "codex" || (await codexAuthenticated()); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Do not report graph mode as ready without a usable model provider.
Graph mode is marked authenticated/reported solely because it is not Codex. With the documented empty provider variables, the gateway reports ready while the first graph generation fails because no configured provider can serve the model policy.
apps/agent-gateway/src/index.ts#L172-L175: derive graph readiness from configured, policy-capable model routing rather thanexecutionRuntime !== "codex".apps/agent-gateway/src/runtime.ts#L453-L453: record graph authentication/provider state as unavailable or unknown until that same readiness check succeeds.
📍 Affects 2 files
apps/agent-gateway/src/index.ts#L172-L175(this comment)apps/agent-gateway/src/runtime.ts#L453-L453
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/agent-gateway/src/index.ts` around lines 172 - 175, Update the
authenticated readiness calculation in apps/agent-gateway/src/index.ts:172-175
to use the configured, policy-capable model routing check for graph mode instead
of treating every non-Codex runtime as ready; retain the Codex authentication
path. In apps/agent-gateway/src/runtime.ts:453, record graph
authentication/provider state as unavailable or unknown until that same
readiness check succeeds.
| async function nextSequence( | ||
| tx: Tx, | ||
| scope: Pick<RuntimeScope, "organisationId" | "runId">, | ||
| ): Promise<number> { | ||
| const [row] = await tx | ||
| .select({ total: sql<number>`count(*)::int` }) | ||
| .from(schema.agentRunEvents) | ||
| .where( | ||
| and( | ||
| eq(schema.agentRunEvents.organisationId, scope.organisationId), | ||
| eq(schema.agentRunEvents.runId, scope.runId), | ||
| ), | ||
| ); | ||
| return (row?.total ?? 0) + 1; | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Event sequence has no column or constraint of its own, which breaks both allocation and querying. Because the ordering key lives only inside the payload jsonb, it must be derived with a non-atomic count(*) and cannot be filtered or ordered by in SQL. Adding a sequence integer column on agent_run_events with unique (organisation_id, run_id, sequence) fixes both sites.
packages/agent-runtime/src/adapters/postgres.ts#L187-L201: allocate the sequence atomically (insert-timemax(...)+1under a lock, or the new column plus retry on unique violation) instead of a read-then-writecount(*).packages/agent-runtime/src/adapters/postgres.ts#L1073-L1087: filtersequence > afterSequenceand order bysequencein the query rather than materialising every event and filtering in memory oncreatedAtorder.
📍 Affects 1 file
packages/agent-runtime/src/adapters/postgres.ts#L187-L201(this comment)packages/agent-runtime/src/adapters/postgres.ts#L1073-L1087
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/agent-runtime/src/adapters/postgres.ts` around lines 187 - 201, Add
a dedicated integer sequence column to agent_run_events with a unique constraint
on organisationId, runId, and sequence. In nextSequence, replace the non-atomic
count-based allocation with atomic max-plus-one allocation under a lock or
insertion with retry on unique violations. In
packages/agent-runtime/src/adapters/postgres.ts lines 1073-1087, query using
sequence > afterSequence and order by sequence instead of filtering materialised
events by createdAt; both sites require updates.
| failureCode: terminal.failureCode, | ||
| error: terminal.error.slice(0, 2_000), | ||
| cancellationReason: null, | ||
| }) |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Failure text is persisted unredacted, unlike every other observation path.
Tool errors go through redactObservationText (Line 792) and results through redactForObservation, but terminal.error is only truncated. A provider or connector error string can carry a URL with a token, header dumps, or record contents straight into agent_runs.error.
🛡️ Proposed fix
- error: terminal.error.slice(0, 2_000),
+ error: redactObservationText(terminal.error, {
+ maxStringLength: 2_000,
+ }),📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| failureCode: terminal.failureCode, | |
| error: terminal.error.slice(0, 2_000), | |
| cancellationReason: null, | |
| }) | |
| failureCode: terminal.failureCode, | |
| error: redactObservationText(terminal.error, { | |
| maxStringLength: 2_000, | |
| }), | |
| cancellationReason: null, | |
| }) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/agent-runtime/src/adapters/postgres.ts` around lines 1009 - 1012,
Update the terminal failure persistence path to pass terminal.error through
redactObservationText before truncating it, matching the sanitization used by
other observation paths. Preserve the existing 2,000-character limit and store
the redacted result in failureCode’s error field.
| async function awaitApproval( | ||
| state: RuntimeState, | ||
| ): Promise<RuntimeStateUpdate> { | ||
| const pending = state.pendingToolCall; | ||
| const approvalId = state.pendingApprovalId ?? pending?.approvalId ?? ""; | ||
| const resumed = interrupt<ApprovalInterrupt, unknown>({ | ||
| kind: "approval_required", | ||
| approvalId, | ||
| toolName: pending?.toolName ?? "", | ||
| argumentsHash: pending?.argumentsHash ?? "", | ||
| }); | ||
| const decision = approvalResumeSchema.safeParse(resumed); | ||
| if (!decision.success || decision.data.approvalId !== approvalId) { | ||
| return { | ||
| pendingApprovalId: null, | ||
| toolAuthorisation: "denied", | ||
| denialReason: "The approval resume did not match the pending request.", | ||
| }; | ||
| } | ||
| if (decision.data.decision === "rejected") { | ||
| return { | ||
| pendingApprovalId: null, | ||
| toolAuthorisation: "denied", | ||
| denialReason: "A reviewer rejected the requested action.", | ||
| }; | ||
| } | ||
| return { pendingApprovalId: null, toolAuthorisation: "allowed" }; | ||
| } |
There was a problem hiding this comment.
🔒 Security & Privacy | 🔴 Critical | ⚡ Quick win
Approval enforcement rests on the resume payload, not on the persisted approval record. One root cause: awaitApproval never reads the authoritative approval state, so a caller-supplied decision: "approved" is sufficient to execute an approval-gated tool, and neither the fakes nor the integration test can detect it.
packages/agent-runtime/src/graph/nodes.ts#L404-L431: after validating the resume payload, readports.approvals.read(scope, approvalId)and only settoolAuthorisation: "allowed"when the persisted state is approved.packages/agent-runtime/src/agent-runtime.integration.test.ts#L397-L408: transition the approvals row to a decided state before resuming, and add a case asserting that resuming withapprovedwhile the row is still pending does not executetawny.endpoint.isolate.packages/agent-runtime/src/testing/index.ts#L125-L147: give the approvals fake a way to record a decision soreadcan returnapproved/rejected, mirroring the production port.
As per coding guidelines: "Require server-side capability checks and approval records for dangerous actions."
📍 Affects 3 files
packages/agent-runtime/src/graph/nodes.ts#L404-L431(this comment)packages/agent-runtime/src/agent-runtime.integration.test.ts#L397-L408packages/agent-runtime/src/testing/index.ts#L125-L147
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/agent-runtime/src/graph/nodes.ts` around lines 404 - 431, Require
awaitApproval to verify the authoritative persisted approval via
ports.approvals.read(scope, approvalId) after validating the resume payload,
granting toolAuthorisation only when that record is approved. In
packages/agent-runtime/src/agent-runtime.integration.test.ts lines 397-408,
record a decided approval before resuming and add coverage that an approved
resume while the persisted row is pending does not execute
tawny.endpoint.isolate. In packages/agent-runtime/src/testing/index.ts lines
125-147, update the approvals fake to record decisions so read returns approved
or rejected states.
Source: Coding guidelines
| async function proposeMemories( | ||
| state: RuntimeState, | ||
| ): Promise<RuntimeStateUpdate> { | ||
| if (!state.finalContent || !dependencies.memoryProposer) { | ||
| return { proposedMemories: 0 }; | ||
| } | ||
| const proposals = await dependencies.memoryProposer({ | ||
| scope, | ||
| finalContent: state.finalContent, | ||
| }); | ||
| if (proposals.length === 0) return { proposedMemories: 0 }; | ||
| const count = await ports.memories.propose(scope, proposals); | ||
| if (count > 0) await emit(dependencies, { type: "memory.proposed", count }); | ||
| return { proposedMemories: count }; | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Memories are proposed from unvalidated final content and can be written twice on the corrective turn.
proposeMemories runs before persistRunResult (build.ts line 53), so proposals are derived from finalContent that may still fail JSON/schema validation. On the one allowed corrective turn the graph re-enters planNextStep → proposeMemories → persistRunResult, calling ports.memories.propose a second time with no idempotency key, and a run that ultimately fails can still have left memory proposals behind.
Proposing after successful validation (or keying the proposal on the run + output hash) would keep durable memory tied to authoritative output.
As per coding guidelines: "Use idempotency keys for inbound events, jobs, and external actions."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/agent-runtime/src/graph/nodes.ts` around lines 575 - 589, Update the
proposeMemories flow so memory proposals occur only after persistRunResult has
successfully validated and persisted the authoritative output, rather than
during the pre-persistence corrective-turn path. Ensure the proposal is executed
at most once per run, using an idempotency key derived from the run identity and
output hash when supported by the memories port, and avoid creating proposals
for runs that ultimately fail validation.
Source: Coding guidelines
| let response: Response; | ||
| try { | ||
| response = await fetchImpl(`${baseUrl}/messages`, { | ||
| method: "POST", | ||
| headers: { | ||
| "content-type": "application/json", | ||
| "x-api-key": apiKey ?? "", | ||
| "anthropic-version": ANTHROPIC_API_VERSION, | ||
| }, | ||
| body: JSON.stringify(body), | ||
| ...(request.signal ? { signal: request.signal } : {}), | ||
| }); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
No provider enforces a request deadline on its outbound model call. All three adapters forward request.signal only when the caller supplied one, so a provider that accepts the TCP connection but never responds blocks the graph node forever — in a durable runtime that means a run parked in running with no failure event and no cancellation path other than the caller's signal. Add a default timeout (e.g. AbortSignal.any([AbortSignal.timeout(ms), request.signal])) with the budget sourced from policy/config.
packages/agent-runtime/src/model/providers/anthropic.ts#L183-L194: compose a default deadline withrequest.signalfor the/messagescall.packages/agent-runtime/src/model/providers/ollama.ts#L191-L198: same for the/api/chatcall.packages/agent-runtime/src/model/providers/openai-compatible.ts#L238-L245: same for the shared/chat/completionscall, ideally via the shared helper soopenrouterinherits it.
📍 Affects 3 files
packages/agent-runtime/src/model/providers/anthropic.ts#L183-L194(this comment)packages/agent-runtime/src/model/providers/ollama.ts#L191-L198packages/agent-runtime/src/model/providers/openai-compatible.ts#L238-L245
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/agent-runtime/src/model/providers/anthropic.ts` around lines 183 -
194, Add a policy/config-sourced default request deadline to the outbound model
calls, composing it with any caller-provided signal so either cancellation or
timeout aborts the request. Update
packages/agent-runtime/src/model/providers/anthropic.ts:183-194 for the
/messages call, packages/agent-runtime/src/model/providers/ollama.ts:191-198 for
/api/chat, and
packages/agent-runtime/src/model/providers/openai-compatible.ts:238-245 for the
shared /chat/completions path; implement the latter through its shared helper so
openrouter inherits the behavior.
| const toolCalls: ModelToolProposal[] = (message.tool_calls ?? []).map( | ||
| (call, index) => ({ | ||
| name: call.function.name, | ||
| arguments: normaliseToolArguments(call.function.arguments), | ||
| toolCallId: call.id ?? `ollama-tool-${index}`, | ||
| }), | ||
| ); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
fd -t f 'postgres.ts' packages/agent-runtime/src/adapters --exec rg -n -C6 'toolCallId|idempotenc|reserv' {}
fd -t f 'nodes.ts' packages/agent-runtime/src/graph --exec rg -n -C4 'toolCallId|idempotenc' {}Repository: jusso-dev/Muster
Length of output: 10472
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the Ollama provider around the cited lines.
FILE="packages/agent-runtime/src/model/providers/ollama.ts"
wc -l "$FILE"
sed -n '200,320p' "$FILE"
# Inspect the run loop / state machine that consumes tool proposals.
sed -n '240,620p' packages/agent-runtime/src/graph/nodes.ts
# Re-read the idempotency key construction for tool reservations.
sed -n '115,140p' packages/agent-runtime/src/adapters/postgres.ts
sed -n '646,725p' packages/agent-runtime/src/adapters/postgres.tsRepository: jusso-dev/Muster
Length of output: 18417
Make the Ollama tool-call fallback step-scoped The ollama-tool-${index} fallback becomes ollama-tool-0 on every step when Ollama omits id, so toolCallIdempotencyKey(runId, toolCallId) can collapse distinct tool calls into the same reservation. Include a step/message discriminator or another stable hash in the fallback.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/agent-runtime/src/model/providers/ollama.ts` around lines 244 - 250,
The fallback toolCallId in the toolCalls mapping must be unique across steps or
messages, not only within the current call list. Update the fallback used by the
Ollama provider to incorporate a stable step/message discriminator or
deterministic hash while preserving Ollama’s supplied call.id when present, so
toolCallIdempotencyKey receives distinct identifiers for separate tool calls.
Source: Coding guidelines
| async cancelRun(input: CancelAgentRunInput): Promise<void> { | ||
| const scope = this.scopeFor(input.scope); | ||
| await this.options.ports.runRecords.persistResult(scope, { | ||
| status: "cancelled", | ||
| reason: input.reason, | ||
| }); | ||
| await this.options.ports.runRecords.emit(scope, { type: "run.cancelled" }); | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check whether persistResult guards against overwriting a terminal run status.
fd -t f 'postgres.ts' packages/agent-runtime/src/adapters --exec sh -c 'ast-grep outline "$1" --items all' _ {}
rg -nP --type=ts -C12 'persistResult' packages/agent-runtime/src/adapters/postgres.tsRepository: jusso-dev/Muster
Length of output: 2947
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '930,1045p' packages/agent-runtime/src/adapters/postgres.ts
printf '\n====\n'
sed -n '80,120p' packages/agent-runtime/src/runtime.tsRepository: jusso-dev/Muster
Length of output: 5762
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '1045,1105p' packages/agent-runtime/src/adapters/postgres.tsRepository: jusso-dev/Muster
Length of output: 2414
Prevent cancelRun from overwriting terminal runs. persistResult({ status: "cancelled" }) updates the row unconditionally, so a late or duplicate cancel can change a completed/failed run to cancelled and emit a contradictory run.cancelled event. Add a non-terminal check here or have persistResult reject terminal rows.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/agent-runtime/src/runtime.ts` around lines 98 - 105, Update
cancelRun so cancellation only proceeds for non-terminal runs, preventing
persistResult from changing completed or failed rows and suppressing
run.cancelled for terminal or duplicate cancellations. Reuse the existing
run-record lookup or terminal-status predicate if available; otherwise enforce
the guard within persistResult while preserving normal cancellation and event
emission for active runs.
Source: Coding guidelines
| const result = (await graph.invoke(input, config)) as RuntimeState; | ||
| if (!result.settled) { | ||
| // The graph stopped without settling: the only non-error way that | ||
| // happens is an interrupt awaiting a human decision. | ||
| const snapshot = await graph.getState(config); | ||
| if (snapshot.next.length > 0) { | ||
| return this.handleFor(scope, { | ||
| status: "awaiting_approval", | ||
| ...(result.pendingApprovalId | ||
| ? { pendingApprovalId: result.pendingApprovalId } | ||
| : {}), | ||
| usage: result.usage, | ||
| estimatedCostCents: result.estimatedCostCents, | ||
| stepCount: result.stepCount, | ||
| }); | ||
| } | ||
| } | ||
| return this.handleFor(scope, { | ||
| status: "completed", | ||
| usage: result.usage, | ||
| estimatedCostCents: result.estimatedCostCents, | ||
| stepCount: result.stepCount, | ||
| }); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
An unsettled graph result is reported as completed.
When result.settled is false and snapshot.next is empty, control falls through to the status: "completed" handle even though persistRunResult never ran, so no authoritative output/terminal state was written for the run. The caller then sees a completed run whose agent_runs row is still non-terminal. A graph that halts without settling and without a pending node is an anomaly and should surface as a failure rather than a success.
🐛 Suggested fix
if (!result.settled) {
// The graph stopped without settling: the only non-error way that
// happens is an interrupt awaiting a human decision.
const snapshot = await graph.getState(config);
if (snapshot.next.length > 0) {
return this.handleFor(scope, {
status: "awaiting_approval",
...(result.pendingApprovalId
? { pendingApprovalId: result.pendingApprovalId }
: {}),
usage: result.usage,
estimatedCostCents: result.estimatedCostCents,
stepCount: result.stepCount,
});
}
+ const error = new AgentRuntimeError(
+ "Agent graph halted without settling a result",
+ "runtime_error",
+ );
+ await this.fail(scope, error);
+ return this.handleFor(scope, {
+ status: "failed",
+ failureCode: error.code,
+ error: error.message,
+ usage: result.usage,
+ estimatedCostCents: result.estimatedCostCents,
+ stepCount: result.stepCount,
+ });
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| const result = (await graph.invoke(input, config)) as RuntimeState; | |
| if (!result.settled) { | |
| // The graph stopped without settling: the only non-error way that | |
| // happens is an interrupt awaiting a human decision. | |
| const snapshot = await graph.getState(config); | |
| if (snapshot.next.length > 0) { | |
| return this.handleFor(scope, { | |
| status: "awaiting_approval", | |
| ...(result.pendingApprovalId | |
| ? { pendingApprovalId: result.pendingApprovalId } | |
| : {}), | |
| usage: result.usage, | |
| estimatedCostCents: result.estimatedCostCents, | |
| stepCount: result.stepCount, | |
| }); | |
| } | |
| } | |
| return this.handleFor(scope, { | |
| status: "completed", | |
| usage: result.usage, | |
| estimatedCostCents: result.estimatedCostCents, | |
| stepCount: result.stepCount, | |
| }); | |
| const result = (await graph.invoke(input, config)) as RuntimeState; | |
| if (!result.settled) { | |
| // The graph stopped without settling: the only non-error way that | |
| // happens is an interrupt awaiting a human decision. | |
| const snapshot = await graph.getState(config); | |
| if (snapshot.next.length > 0) { | |
| return this.handleFor(scope, { | |
| status: "awaiting_approval", | |
| ...(result.pendingApprovalId | |
| ? { pendingApprovalId: result.pendingApprovalId } | |
| : {}), | |
| usage: result.usage, | |
| estimatedCostCents: result.estimatedCostCents, | |
| stepCount: result.stepCount, | |
| }); | |
| } | |
| const error = new AgentRuntimeError( | |
| "Agent graph halted without settling a result", | |
| "runtime_error", | |
| ); | |
| await this.fail(scope, error); | |
| return this.handleFor(scope, { | |
| status: "failed", | |
| failureCode: error.code, | |
| error: error.message, | |
| usage: result.usage, | |
| estimatedCostCents: result.estimatedCostCents, | |
| stepCount: result.stepCount, | |
| }); | |
| } | |
| return this.handleFor(scope, { | |
| status: "completed", | |
| usage: result.usage, | |
| estimatedCostCents: result.estimatedCostCents, | |
| stepCount: result.stepCount, | |
| }); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/agent-runtime/src/runtime.ts` around lines 285 - 307, Update the
unsettled-result handling in the graph invocation flow around result.settled and
snapshot.next so an unsettled graph with no pending node returns a failure
status instead of falling through to status: "completed". Preserve the
awaiting_approval path when snapshot.next is non-empty, and keep completed
status only for settled results.
| ALTER TABLE "agent_runs" ADD CONSTRAINT "agent_runs_pending_approval_id_approvals_id_fk" FOREIGN KEY ("pending_approval_id") REFERENCES "public"."approvals"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint | ||
| CREATE INDEX "agent_runs_org_conversation_idx" ON "agent_runs" USING btree ("organisation_id","conversation_id","started_at");--> statement-breakpoint | ||
| CREATE UNIQUE INDEX "agent_tool_calls_org_idempotency_unique" ON "agent_tool_calls" USING btree ("organisation_id","idempotency_key"); No newline at end of file |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Make the populated-table changes non-blocking.
Line 49 validates an FK on agent_runs, and Lines 50–51 build regular indexes on active write tables. This can block run/tool-call writes during deployment. Split this into a rollout-safe migration: add the FK NOT VALID, validate later, and create indexes concurrently in a non-transactional migration stage.
🧰 Tools
🪛 Squawk (2.59.0)
[warning] 49-49: By default new constraints require a table scan and block writes to the table while that scan occurs. Use NOT VALID with a later VALIDATE CONSTRAINT call.
(constraint-missing-not-valid)
[warning] 49-49: Adding a foreign key constraint requires a table scan and a SHARE ROW EXCLUSIVE lock on both tables, which blocks writes to each table. Add NOT VALID to the constraint in one transaction and then VALIDATE the constraint in a separate transaction.
(adding-foreign-key-constraint)
[warning] 50-50: During normal index creation, table updates are blocked, but reads are still allowed. Use concurrently to avoid blocking writes.
(require-concurrent-index-creation)
[warning] 51-51: During normal index creation, table updates are blocked, but reads are still allowed. Use concurrently to avoid blocking writes.
(require-concurrent-index-creation)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/database/migrations/0021_closed_steel_serpent.sql` around lines 49 -
51, Update the migration containing the agent_runs foreign key and
agent_tool_calls indexes to be rollout-safe: add the constraint with NOT VALID,
move its validation to a later migration step, and create both indexes
concurrently. Ensure the migration is marked non-transactional as required for
concurrent index creation while preserving the existing constraint and index
definitions.
Source: Linters/SAST tools
|
Closing without merge because the product pivot retires Muster’s separate LangGraph/model runtime. Hermes is now the harness. The branch is retained for selective salvage of tenant, approval, idempotency and audit patterns into #72. |
1 participant
Summary
Adds
packages/agent-runtime(@muster/agent-runtime): a durable, resumable, checkpointed execution graph for agent runs, operating strictly behind the existing@muster/agent-harnessinvocation boundary (unchanged — no second invocation path). Implements the exact node graph from issue #70 (ReceiveInvocation→ ... →PersistRunResult, withAuthoriseToolbranching to denial/approval-interrupt/execution) using LangGraph v1 (@langchain/langgraph1.4.8).agent_runtime_checkpoints/agent_runtime_checkpoint_writestables) hold only execution state — current node, model messages, tool-call progress, context summary, pending interrupt — and every row carriesorganisation_id/agent_id/conversation_id/run_id/graph_version.MusterPostgresCheckpointSaveris constructed for one organisation and asserts every thread id it's handed belongs to that organisation before touching a row (CheckpointScopeViolationErrorotherwise) — defence in depth on top of organisation-scoped SQL predicates on every port.AGENT_RUNTIME_GRAPH_VERSIONis stamped on every run; a run resumes only against the version it started with, or fails closed withgraph_version_mismatchand an explicit migration requirement.ToolPolicyPortreplays the existing@muster/agentstool registry/allowlist/capability/approval gates before any model-supplied name or argument is trusted.ToolExecutionPortreserves a call (idempotency-keyedagent-runtime.tool:{runId}:{toolCallId}) before executing, so a resumed run replays a completed or in-flight reservation instead of repeating an external action.InterruptForApprovalcallsApprovalPort.require(org-scoped, idempotency-keyed against replay) and pauses the graph via LangGraph'sinterrupt(); resuming with a decision continues from that exact node.RuntimeGuardPort.assertRunnableis re-evaluated at every graph step boundary (model call, tool call, post-tool validation), not only at claim time; an aborted signal or a flipped kill switch stops the run mid-execution.ModelPolicy(capability class + fallback +allowLocal), never a vendor name.ModelRouterresolves it acrossopenai-compatible/anthropic/ollama/openrouterproviders (plus an offlinescriptedprovider for tests), honouring fallback order.AgentRuntimeEvents are reduced to a closed, schema-validated vocabulary (run.*,model.*,tool.*,memory.proposed) before they can reach a stream, a room timeline, or Slack.apps/agent-gatewaygains an opt-inMUSTER_AGENT_RUNTIME=graphmode.codex(Codex subscription runtime) remains the default; existing deployments are unaffected until an operator explicitly opts in.See ADR 0005 for the full design, tenancy enforcement, failure taxonomy, and security boundaries.
Migration
packages/database/migrations/0021_closed_steel_serpent.sql— additive only:agent_runtime_checkpoints,agent_runtime_checkpoint_writes(composite PKs on the full tenant+checkpoint path, FKs toorganisations/agent_definitions/agent_runs).agent_runs(graph_version,conversation_id,checkpoint_thread_id,pending_approval_id) andagent_tool_calls(tool_call_id,idempotency_key,checkpoint_id,result), plus a new unique indexagent_tool_calls_org_idempotency_unique.agent_runsrow with these new columns simply null.Rollback: drop the two new tables and columns (or leave them — they're inert unless
MUSTER_AGENT_RUNTIME=graphis set), and/or setMUSTER_AGENT_RUNTIMEback tocodex/mock. A build that no longer recognises a run's recordedgraph_versionfails that resume closed withgraph_version_mismatchrather than silently reprocessing it.Validation
All commands run from a clean worktree against a scratch PostgreSQL 17.6 container (not shared with any deployment):
Fresh-database gate (migrate → bootstrap → verify-clean), twice, to prove both a clean install and re-runnable integration fixtures:
148 tests in the new package (132 offline, 16 against real PostgreSQL), mapped to issue #70's acceptance criteria:
MusterAgentRuntimeinstance (fresh ports, fresh checkpointer, same PostgreSQL) resumes a run without repeating its tool call — proven for both a plain crash-recovery resume and an approval-interrupt resume.reserve()for the same tool-call id returns the recorded outcome (already_completed) rather than a second row; the unique index enforces this at the database level.stale_run) rather than leaking existence, and a mismatched thread id tripsCheckpointScopeViolationErrorbefore any row is touched.graph_version_mismatch) without invoking the model.agent_run_eventsrows written to PostgreSQL.awaiting_approval.agent_kill_switchand no further tool calls, both offline and against real PostgreSQL.Residual risks
MusterAgentRuntime.resumeRun({ approval })is implemented, tested, and stable, butapps/agent-gateway's poll loop (dispatch()) does not currently selectawaiting_approvalruns, so nothing today automatically re-invokes the graph the moment a human approves. The existing web approval-decision route (ApprovalDomainService.decide) is unchanged. Per the tracker's dependency graph (#72 requires #70), wiring a live trigger belongs to governed tool execution (P0: Ship the first Hermes-native Muster MCP vertical slice #72); extendingclaim()'s CAS eligibility checks to a third status class touches shared, heavily-tested logic for every other execution runtime and was judged too risky to do safely in this PR's scope.fetch, not a live call.packages/agent-runtime'sToolExecutionPort.execute()has no registered tool implementations yet (by design — issue P0: Ship the first Hermes-native Muster MCP vertical slice #72 owns the governed connector/MCP executor); until then, any tool call it does authorise fails safely with "Tool has no registered executor" rather than performing a side effect.Closes #70
Summary by CodeRabbit
New Features
Documentation
Tests