Repository navigation
refactor(agent): refactor agent architecture - #2961
Conversation
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
A steer sent mid-turn waited up to one drain interval (1s) to be noticed: `AgentHarness.steer()` is a method on an in-memory object, the turn runs inside a workflow step, and a workflow hook cannot reach into a step that is executing rather than parked. The row in `agent_pending_message` is the only channel, and the step could only poll it. Add a payload-free notification on `agent:pending:<sessionId>` so the running turn drains immediately. It is strictly an accelerator: the message is durable in Postgres before the notification is published, so a dropped notification costs latency and nothing else — which is why the poller stays at its full rate and why the channel carries no payload. Non-Redis cache providers get a null notifier and behave exactly as before. Two things fall out of this: - The drain's coalescing had to change. A notification landing *after* the in-flight drain already claimed its rows cannot be folded into it, or the new message waits for the next poll and the channel buys nothing. Such a request now sets a flag and the drain re-runs when it settles; teardown follows that chain to its end before disposing the engine. - `claim` marks rows consumed *before* delivery, so a throw lost them outright. The common throw is pi's "Cannot steer while idle", which fires when the turn finishes between the claim and the hand-off. Undelivered messages are now released back to the queue and surface in the next turn instead of vanishing. Covered against a real AgentHarness in agent-writing's faux-provider tests. `resolveCacheProvider()` is extracted from `createKeyv` so pub/sub can ask what the backend is without a second copy of the protocol sniffing. Pub/sub gets its own connections rather than reusing the Keyv client: a node-redis connection in subscriber mode refuses ordinary commands. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`pubsub.ts` reached for `resolveCacheProvider()` in `clients.ts`, and
`clients.ts` statically imports every adapter — including `@keyv/valkey`,
which pulls in `iovalkey`. iovalkey uses a dynamic `require` of a file path,
so once `apps/service` imported `@chia/kv/pubsub` the bundled dev server
threw at request time:
Dynamic require of ".../iovalkey/built/Redis.js" is not supported
`nitro build` did not surface it: rollup tree-shakes the unreachable adapter
away because nothing in the service calls `createKeyv()`. The dev pipeline
transforms modules individually and keeps the import, so only the running
server failed. Reproduced directly with esbuild — bundling a module that
imports `@chia/kv/pubsub` pulled in 69 iovalkey references before this
change and none after.
Move the provider resolution to its own module that depends on nothing but
`env`, so asking *which* backend is configured never drags in a client for
it. `clients.ts` re-exports it, so the seam is unchanged for callers.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (2)
📝 WalkthroughWalkthroughThe change adds dedicated maintenance engines, automatic post-turn compaction, Redis-backed pending-message notifications, lossless message release, cache-provider resolution, and expanded runtime coverage. ChangesAgent runtime maintenance and pending delivery
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant AgentTurn
participant PendingMessageStore
participant PendingMessageNotifier
participant RedisPubSub
AgentTurn->>PendingMessageStore: claim pending messages
AgentTurn->>PendingMessageStore: release undelivered IDs on failure
AgentTurn->>PendingMessageNotifier: subscribe to session notifications
PendingMessageNotifier->>RedisPubSub: subscribe to session channel
RedisPubSub-->>PendingMessageNotifier: wake-up notification
PendingMessageNotifier-->>AgentTurn: request pending-message drain
AgentTurn->>PendingMessageStore: claim and deliver queued messages
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 inconclusive)
✅ 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: 4
🧹 Nitpick comments (5)
packages/agent-writing/__tests__/runtime.test.ts (1)
338-358: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winThe test does not cover the partial tail it is named for.
The first phase delivers both messages, so nothing returns. The second phase fails on "third", the first message of the batch, so both "third" and "fourth" return. No case delivers one message and then fails on the next. The
undelivered.shift()logic inpackages/agent-runtime/src/adapters/pi.tsis what produces a partial tail, and it stays unexercised. Add a case where the first message is delivered and the second one fails, or rename the test to match what it asserts.🤖 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-writing/__tests__/runtime.test.ts` around lines 338 - 358, Update the test named “keeps only the undelivered tail when a later message fails” to exercise a partial tail: configure the pending-message delivery so “third” succeeds and “fourth” fails, then assert that only “fourth” remains. Preserve the rejection assertion and use the existing fixture controls to trigger the failure after the first message is delivered.packages/agent-runtime/src/runtime.ts (1)
210-213: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider recording the compaction failure.
The empty
catchis the correct control flow. It also makes a permanently failing compaction invisible: the session grows past its threshold on every turn and nothing reports it. Pass an optional diagnostic sink, or emit a debug-level record, so the condition is observable.🤖 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 210 - 213, Add observability to the compaction failure catch in the runtime turn-boundary flow without changing its swallow-and-retry behavior. Reuse an existing diagnostic sink or debug logger available to the surrounding runtime code, and record the caught error with compaction context while preserving the current control flow.packages/agent-runtime/__tests__/runtime.test.ts (1)
355-356: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winRun
@chia/agent-runtimetests on Node.js 22 or later. Its inherited TypeScript config setslibtoES2024. The root Vitest config includespackages/*, but the current Node.js 20 CI job runs only theserviceproject.🤖 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/__tests__/runtime.test.ts` around lines 355 - 356, Update the agent-runtime test execution configuration so its tests run under Node.js 22 or later, matching the ES2024 library requirement used by the inherited TypeScript configuration. Ensure the relevant CI or Vitest project includes the agent-runtime tests rather than relying only on the Node.js 20 service job.apps/service/src/services/agent-runtime.service.ts (1)
619-629: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueConsider logging when the pending-message notification fails.
The
catcharoundgetAgentPendingNotifier()?.publish(...)swallows all errors with no log. This is fine functionally, since delivery already succeeded and polling remains as a fallback, but a persistently broken Redis pub/sub path would degrade silently with no signal to operators. A debug/warn log on catch would help detect a stuck notifier without changing behavior.🤖 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/service/src/services/agent-runtime.service.ts` around lines 619 - 629, Update the catch block around getAgentPendingNotifier()?.publish(input.sessionId) to emit a debug or warning log containing the notification failure details and session context, while continuing to swallow the error and preserve the existing polling fallback behavior.packages/agent-runtime/src/adapters/pi.ts (1)
191-215: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExtract the duplicated
compact/navigateimplementations.The
compactclosure (lines 191-196 and 289-294) and thenavigatemethod (lines 212-215 and 306-309) are implemented identically increatePiAgentEngineandcreatePiMaintenanceEngine. The file's comment on the two harness constructions justifies keeping the harness setup separate, but it does not apply to these two small result-normalization functions. Extract them into shared helpers keyed byharnessso future changes to the normalization shape (for example, a new field onAgentCompactionResult) do not need to be made twice.♻️ Proposed extraction
+const compactWith = + (harness: AgentHarness) => + async (customInstructions?: string): Promise<AgentCompactionResult> => { + const result = await harness.compact(customInstructions); + return { summary: result.summary, tokensBefore: result.tokensBefore }; + }; + +const navigateWith = + (harness: AgentHarness) => + async ( + entryId: string, + navigationOptions: AgentNavigationOptions + ): Promise<AgentNavigationResult> => { + const result = await harness.navigateTree(entryId, navigationOptions); + return { cancelled: result.cancelled }; + };Then use
compactWith(harness)andnavigateWith(harness)in bothcreatePiAgentEngineandcreatePiMaintenanceEngine. Confirmcompact/navigateTreeare declared on the non-genericAgentHarnessbase type so the helper does not need to be parameterized byTContext.Also applies to: 277-314
🤖 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/pi.ts` around lines 191 - 215, Extract the duplicated result-normalization closures into shared compactWith(harness) and navigateWith(harness) helpers in packages/agent-runtime/src/adapters/pi.ts, using the non-generic AgentHarness base type. Replace the local compact and navigate implementations in both createPiAgentEngine and createPiMaintenanceEngine with these helpers while preserving their existing return shapes and behavior.
🤖 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 `@packages/kv/src/provider.ts`:
- Around line 17-40: Update resolveCacheProvider to distinguish an unset
CACHE_URI from an invalid or unsupported URI: throw a missing-configuration
error when CACHE_URI is absent, and a separate invalid/unsupported-URI error
when it is present but cannot be parsed or uses an unsupported protocol.
Preserve the existing provider mapping for valid redis, valkey, and postgres
protocols and continue using isUrl with allowedProtocols.
In `@packages/kv/src/pubsub.ts`:
- Around line 89-93: Update getRedisPubSub to catch failures from
resolveCacheProvider and return/cache null when provider resolution throws,
preserving the documented degradation contract. Keep successful Redis
initialization and non-Redis caching behavior unchanged so
getAgentPendingNotifier remains safe for unsupported CACHE_URI values.
- Around line 46-60: Update publisherReady and subscriberReady in
createRedisPubSub to clear their respective memoized connection variables when
the awaited ready promise rejects, then propagate the original rejection.
Preserve successful connection reuse while allowing subsequent publish or
subscribe calls to create a fresh connection after an initial failure.
- Around line 38-44: Update the Redis client configuration in connect to set an
explicit socketTimeout and disableOfflineQueue: true, preserving the existing
connection behavior. Add bounded AbortSignal-based waits to publish and
subscribe commands, and ensure stopDraining also bounds its notifierSubscription
wait so connection, command, and shutdown operations cannot remain pending
indefinitely.
---
Nitpick comments:
In `@apps/service/src/services/agent-runtime.service.ts`:
- Around line 619-629: Update the catch block around
getAgentPendingNotifier()?.publish(input.sessionId) to emit a debug or warning
log containing the notification failure details and session context, while
continuing to swallow the error and preserve the existing polling fallback
behavior.
In `@packages/agent-runtime/__tests__/runtime.test.ts`:
- Around line 355-356: Update the agent-runtime test execution configuration so
its tests run under Node.js 22 or later, matching the ES2024 library requirement
used by the inherited TypeScript configuration. Ensure the relevant CI or Vitest
project includes the agent-runtime tests rather than relying only on the Node.js
20 service job.
In `@packages/agent-runtime/src/adapters/pi.ts`:
- Around line 191-215: Extract the duplicated result-normalization closures into
shared compactWith(harness) and navigateWith(harness) helpers in
packages/agent-runtime/src/adapters/pi.ts, using the non-generic AgentHarness
base type. Replace the local compact and navigate implementations in both
createPiAgentEngine and createPiMaintenanceEngine with these helpers while
preserving their existing return shapes and behavior.
In `@packages/agent-runtime/src/runtime.ts`:
- Around line 210-213: Add observability to the compaction failure catch in the
runtime turn-boundary flow without changing its swallow-and-retry behavior.
Reuse an existing diagnostic sink or debug logger available to the surrounding
runtime code, and record the caught error with compaction context while
preserving the current control flow.
In `@packages/agent-writing/__tests__/runtime.test.ts`:
- Around line 338-358: Update the test named “keeps only the undelivered tail
when a later message fails” to exercise a partial tail: configure the
pending-message delivery so “third” succeeds and “fourth” fails, then assert
that only “fourth” remains. Preserve the rejection assertion and use the
existing fixture controls to trigger the failure after the first message is
delivered.
🪄 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: 9f1ebe98-c711-4cc3-9e4b-67ad19e3a544
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (23)
apps/service/package.jsonapps/service/src/services/agent-pending-notifier.tsapps/service/src/services/agent-runtime.service.tsapps/service/src/steps/agent-turn.step.tsdocs/agent-architecture.mddocs/agent-architecture.zh.mdpackages/agent-core/__tests__/pending-messages.test.tspackages/agent-core/src/ports.tspackages/agent-core/src/session/pg-pending-messages.tspackages/agent-runtime/__tests__/pi-compaction.test.tspackages/agent-runtime/__tests__/runtime.test.tspackages/agent-runtime/src/adapters/pi.tspackages/agent-runtime/src/engine.tspackages/agent-runtime/src/index.tspackages/agent-runtime/src/runtime.tspackages/agent-writing/__tests__/runtime.test.tspackages/agent-writing/src/index.tspackages/agent-writing/src/runtime.tspackages/db/src/libs/agent/index.tspackages/kv/package.jsonpackages/kv/src/clients.tspackages/kv/src/provider.tspackages/kv/src/pubsub.ts
| const connect = ( | ||
| uri: string | ||
| ): { client: RedisClientType; ready: Promise<void> } => { | ||
| const client = createClient({ url: uri }) as RedisClientType; | ||
| client.on("error", () => undefined); | ||
| return { client, ready: client.connect().then(() => undefined) }; | ||
| }; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
@redis/client v5 createClient socket connectTimeout reconnectStrategy disableOfflineQueue options
💡 Result:
In @redis/client (node-redis) v5, configuration options are passed to the createClient function. The settings you inquired about are organized under the socket object or as top-level properties within the client configuration object [1][2][3]. socket.connectTimeout This property defines the connection timeout in milliseconds [1][2][4]. Default value: 5000 (5 seconds) [1][2][4]. socket.reconnectStrategy This property determines how the client behaves when the socket closes unexpectedly (without a manual.quit or.disconnect call) [1][2][4]. It accepts one of three types [1][2][4]: 1. false: Do not attempt to reconnect; the client closes and flushes the command queue [1][2][4]. 2. number: Wait for the specified number of milliseconds before attempting to reconnect [1][2][4]. 3. function (retries: number, cause: Error) => false | number | Error: A custom strategy function [1][2][4]. Returning a number specifies the delay for the next attempt, while returning false or an Error object stops reconnection attempts [1][2][4]. disableOfflineQueue This is a top-level configuration option [3]. Default value: false [1][2]. Behavior: When set to false, commands sent while the client is reconnecting are queued in memory and executed once the connection is re-established [5][3]. When set to true, commands are rejected immediately if the client is not connected [5][3]. Example Configuration const client = createClient({ socket: { connectTimeout: 10000, // 10 seconds reconnectStrategy: (retries) => { if (retries > 10) return false; // Stop after 10 attempts return Math.min(retries * 100, 3000); // Exponential backoff } }, disableOfflineQueue: true });
Citations:
- 1: https://github.com/redis/node-redis/blob/HEAD/docs/client-configuration.md
- 2: https://github.com/redis/node-redis/blob/master/docs/client-configuration.md
- 3: https://github.com/redis/node-redis/blob/master/packages/client/lib/client/index.ts
- 4: https://github.com/redis/node-redis/blob/master/packages/client/lib/client/socket.ts
- 5: https://github.com/redis/node-redis/blob/master/docs/FAQ.md
🏁 Script executed:
#!/bin/bash
printf '%s\n' '--- candidate files ---'
git ls-files 'packages/kv/src/pubsub.ts' 'packages/agent-runtime/src/runtime.ts' '*package.json' '*lock*' | sed -n '1,120p'
printf '%s\n' '--- pubsub.ts ---'
cat -n packages/kv/src/pubsub.ts | sed -n '1,130p'
printf '%s\n' '--- runtime teardown references ---'
rg -n -C 8 'notifierSubscription|stopDraining|subscribe|publish' packages/agent-runtime/src/runtime.ts
printf '%s\n' '--- redis dependency declarations ---'
rg -n -C 3 '`@redis/client`|redis' --glob 'package.json' --glob '*lock*' --glob 'pnpm-workspace.yaml' --glob 'yarn.lock' --glob 'package-lock.json' .Repository: Chia1104/chia1104.dev
Length of output: 50377
🌐 Web query:
node-redis @redis/client 5.12.1 command timeout socket timeout connectTimeout disableOfflineQueue subscribe publish
💡 Result:
In node-redis (@redis/client) version 5.12.1, timeouts and queue management are configured through the client options, while Pub/Sub requires a dedicated connection. Socket and Command Timeouts - connectTimeout: Configured via the socket option, this sets the connection timeout in milliseconds. The default is 5000ms [1][2]. createClient({ socket: { connectTimeout: 10000 } }); - socketTimeout: Also configured within the socket object, this defines the maximum idle time (in milliseconds) before the socket is automatically closed [1][2]. - Individual Command Timeouts: The client does not have a global commandTimeout option. Instead, you must use an AbortController signal with individual commands [3]. const ac = new AbortController; const t = setTimeout( => ac.abort, 1000); try { await client.get(commandOptions({ signal: ac.signal }), 'key'); } finally { clearTimeout(t); } Offline Queue - disableOfflineQueue: When set to true, commands sent while the client is not connected (e.g., during reconnection) will be rejected immediately rather than queued in memory [4][5][6]. createClient({ disableOfflineQueue: true }); Publish and Subscribe - Pub/Sub Behavior: In node-redis 5.x, a client that initiates a subscription enters a subscribe-only mode for the lifetime of those bindings. Executing regular commands (like GET or SET) on a client in this state will throw an error [7]. - Best Practice: Use a dedicated client for Pub/Sub by calling.duplicate on your primary client [8][7]. const subscriber = client.duplicate; await subscriber.connect; await subscriber.subscribe('channel', (message) => {... }); await client.publish('channel', 'message'); // Use the original client to publish [8][7]
Citations:
- 1: https://github.com/redis/node-redis/blob/HEAD/docs/client-configuration.md
- 2: https://github.com/redis/node-redis/blob/master/packages/client/lib/client/socket.ts
- 3: https://redis.io/docs/latest/develop/clients/nodejs/produsage/
- 4: https://github.com/redis/node-redis/blob/master/packages/client/lib/client/index.ts
- 5: https://github.com/redis/node-redis/blob/master/docs/FAQ.md
- 6: https://github.com/redis/node-redis/blob/2014e44a/packages/client/lib/client/index.spec.ts
- 7: https://redis.io/docs/latest/develop/use-cases/pub-sub/nodejs/
- 8: https://github.com/redis/node-redis/blob/HEAD/docs/pub-sub.md
🏁 Script executed:
#!/bin/bash
set -eu
base='https://raw.githubusercontent.com/redis/node-redis/v5.12.1'
for path in \
'packages/client/lib/client/socket.ts' \
'packages/client/lib/client/index.ts' \
'docs/client-configuration.md' \
'docs/FAQ.md'; do
echo "--- $path ---"
curl -fsSL "$base/$path" | rg -n -C 5 'connectTimeout|socketTimeout|disableOfflineQueue|commandTimeout|AbortSignal|commandOptions|setTimeout|connect\(' | sed -n '1,220p'
doneRepository: Chia1104/chia1104.dev
Length of output: 501
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- matching upstream tags ---'
git ls-remote --tags https://github.com/redis/node-redis.git '*5.12.1*' | tail -20
printf '%s\n' '--- package metadata ---'
curl -fsSL https://registry.npmjs.org/@redis%2fclient/5.12.1 |
python3 -c 'import json,sys; d=json.load(sys.stdin); print(d["dist"]["tarball"])'Repository: Chia1104/chia1104.dev
Length of output: 1206
🏁 Script executed:
#!/bin/bash
set -eu
base='https://raw.githubusercontent.com/redis/node-redis/838c28d9f55d13a7933e650b9f7782fdb5af4f71'
for path in \
'packages/client/lib/client/socket.ts' \
'packages/client/lib/client/index.ts' \
'docs/client-configuration.md' \
'docs/FAQ.md'; do
echo "--- $path ---"
curl -fsSL "$base/$path" |
rg -n -C 6 'connectTimeout|socketTimeout|disableOfflineQueue|commandTimeout|AbortSignal|commandOptions|setTimeout|connect\(' |
sed -n '1,260p'
doneRepository: Chia1104/chia1104.dev
Length of output: 28436
Bound Redis connection and command waits.
socket.connectTimeout already defaults to 5 seconds and only covers TCP connection. If Redis accepts the connection but does not complete the handshake, connect() can remain pending because socket.socketTimeout is unset. publish and subscribe can also remain pending, and stopDraining awaits notifierSubscription.
Configure an explicit socketTimeout, bound command waits with AbortSignal, and set disableOfflineQueue: true to reject commands during reconnects.
🤖 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/kv/src/pubsub.ts` around lines 38 - 44, Update the Redis client
configuration in connect to set an explicit socketTimeout and
disableOfflineQueue: true, preserving the existing connection behavior. Add
bounded AbortSignal-based waits to publish and subscribe commands, and ensure
stopDraining also bounds its notifierSubscription wait so connection, command,
and shutdown operations cannot remain pending indefinitely.
Chia1104
changed the title
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Chia1104
changed the title
Summary by CodeRabbit