Repository navigation
Conversation
073615f to
63e06e0
Compare
There was a problem hiding this comment.
Reviewed the integration locally against real Redis 7 + PostgreSQL 16 (sessionruntime memory+Redis contract suites, runtimefence, handlers contract + multiprocess tests, conversation/flow, userinput, toolapproval, message, acpagent). All green.
The wiring quality is high: text-delta batching (20ms/4KB) addresses the per-token Redis round-trip concern from #800, the legacy/runtime protocol switch is mutex-coordinated so a session never sees mixed frames, ownership is checked three times before durable writes, and PreserveDecision is correctly connected on the WS deferred response paths. No lease leaks found on any error path I traced.
Main finding — side-band decision responses carry no fence.
Channel-inbound replies (internal/channel/inbound/channel.go, e.g. handleUserInputResponseCommand ~:3447) and the HTTP approval endpoints (internal/handlers/tool_approval.go ~:79) call the resolver with a bare context: no runtimefence.WithContext, no allocate/activate. In distributed mode, once a decision has been claimed by a run (runtime_fencing_token non-NULL), a side-band response falls into the SQL's runtime_fencing_token IS NULL branch → ErrNoRows → the user gets "already decided / not found" while the approval silently does nothing — with no log indicating this was a fence rejection. The WS path has the full deferred + PreserveDecision fallback; side-band has none. Same user action, two different semantics depending on entry point.
Also note: #817 (ask_user cross-channel interactions) just merged into main, so this branch now conflicts — including the migration 0111 number collision and resolver_user_input.go. Suggest rebasing onto main first; the side-band fence integration should be designed together with #817's plain-text channel responses, e.g. side-band paths should reuse the Prepare*Response interfaces to obtain a PreservedDecision rather than growing a parallel path. Happy to help think through that integration — it touches code I just wrote in #817.
Non-blocking:
redis.goWATCH/MULTI loops retry unbounded onTxFailedErrwith no backoff or cap (:183, :240, :308, :379) — carried over from #800, but this PR puts live traffic on them. Under contention (concurrent steer/abort on one session) this can spin.- Lease-loss cleanup is soft:
cancelRunControl(control.go:395) only revokes ownership and signalsabortCh; if the runner is blocked on a ctx-unaware call, the run showsrunninguntil lease TTL. A watchdog marking the run lost on lease failure would close the window; at minimum worth a documented note. ErrStale/ErrRuntimeDecisionOwnerUnavailablesurface as raw English strings viawsErrorMessage(resolver_tool_approval.go:224, resolver_user_input.go:117) instead of structured errors — the frontend can't distinguish "superseded by a newer run" from a real failure, so it can't offer actionable guidance like "refresh and retry".
986e671 to
6083e94
Compare
899893f to
dc1be29
Compare
| } | ||
| go func() { | ||
| defer h.releaseRuntimeCommand() | ||
| handled, dispatchErr := h.sessionRuntime.DispatchActiveCommand(baseCtx, botID, sessionID, commandType, targetID, raw) |
There was a problem hiding this comment.
P1：decision 的 runtime fencing 只覆盖了 WebSocket 入口；IM channel 和 HTTP 入口在 decision 被 runtime 认领后会直接失败。
WS 入口的做法是对的：先经 routeWSRuntimeResponse → DispatchActiveCommand 把响应路由给当前 owner；没有 active run 时，tool approval 和 ask_user 都会走 Prepare*Response + PreserveDecision 激活新的 fenced continuation。
但另外两类入口没有这套逻辑，都是拿普通 context 直接调 resolver：
- IM channel：
streamToolApprovalCommand/streamUserInputResponseCommand，经streamContinuationCommand； - HTTP：
ToolApprovalHandler.respond。
decision 一旦被某个 runtime run 认领，runtime_fencing_token 即非空，SQL 层要求 token 匹配（user_input、tool_approval）。裸 context 带不出 token，更新匹配不到行，最终映射为 ErrAlreadyDecided / ErrNotFound 或 HTTP 404/409。
结果：同一个 pending approval / ask_user，Web（WS）上能正常回复，换成 Telegram / Discord / Feishu 或 HTTP API 就报「未找到 / 已处理」——decision 实际没提交，原 run 也继续挂着。
建议所有入口复用同一套与 transport 无关的 decision router：
- 有 active run 时，用稳定的 command ID + payload hash 路由给当前 owner；
- 没有 active owner 时，复用 prepare/preserve 流程激活新的 fenced continuation；
- 所有权建立后才真正执行 decision response。
不建议放宽 SQL fencing 条件来兼容这些入口——那等于重新允许旧 owner 或无权 writer 提交 decision。
验证建议：在现有 PostgreSQL fencing 集成测试基础上，用真实 Redis/Valkey + PostgreSQL 起两个 server 进程：B 持有包含 pending approval / ask_user 的 run，从 A 的 HTTP 和 IM channel 提交响应。断言 decision 只提交一次、原 run 能继续、重复 payload 幂等恢复、不同 payload 返回冲突，并覆盖 owner 在数据库提交后、command result 发布前崩溃的窗口。
| @@ -621,9 +590,68 @@ func (a *Agent) runStream(ctx context.Context, cfg RunConfig, ch chan<- StreamEv | |||
| // a fully-disconnected consumer hanging this goroutine forever. | |||
| deliveryCtx, deliveryCancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) | |||
There was a problem hiding this comment.
P1：terminal hook 会消耗 terminal event 的 5 秒投递预算，可能导致终态事件丢失、历史不落库。
时序问题：runStream 结束时先创建 5 秒的 deliveryCtx，再同步执行 runTerminalHook()，最后才 sendEvent。而 hook 侧单个 action 默认就允许运行 10 秒，多个 hook/action 在 hooks.Service.Run 中串行执行，总时长没有上限——正常配置下 hook 就可能超过 5 秒。
hook 返回时 deliveryCtx 可能已过期。Agent.Stream 用的是无缓冲 channel，sendEvent 此时会走 <-deliveryCtx.Done() 分支，terminal event 没有交给 resolver。
影响不止是客户端收不到结束态：resolver 依赖 terminal snapshot 分支做终态历史持久化并设置 HistoryCommitted；事件丢失后会落入无 snapshot 跳过持久化的分支——已生成的回答不进历史，runtime 无法形成可靠的 canonical terminal，刷新或重连后状态与持久化结果不一致。hook 的 action 数量和超时都可配置且串行累加，配置得长一些还会拖住整个 agent stream goroutine。
建议把 hook 执行和 terminal 投递拆成两份独立预算：
- hook 在受
TerminalHookAuthority约束、且有明确总超时的 context 中执行； - 无论 hook 成功、失败还是超时，结束后重新创建一份完整的 delivery timeout，再发送 terminal event。
这样既保持「hook 先于终态」的顺序，也不让 hook 占用终态事件的投递时间。
验证建议：在 terminal hook authority 测试中增加一个运行超过 5 秒的 hook，断言 terminal event 恰好收到一次、历史持久化成功且 HistoryCommitted=true；同时覆盖 hook 报错、hook 超时、用户取消、执行期间丢失 runtime ownership，确认这些情况都不会阻塞或吞掉终态事件。
dc1be29 to
d664af2
Compare
d664af2 to
6636b7b
Compare
c122dc6 to
1464316
Compare
|
整体看下来方向没啥问题，前面提到的 HTTP/IM 回复和 terminal hook 也都补上了。 我这边还看到两个收尾点：
另外 CI 还有个 |
刚刚重构成了 step 粒度的持久化，现在还在测试修复一些bug |
1464316 to
a4d2011
Compare
Notify session subscribers after atomic retry and edit replacements so other clients refresh the canonical tail.
Expose stable runtime error codes and localized safe fallbacks while retaining private backend causes only in server logs.
Publish new run admissions and committed ask_user outcomes so remote observers stay idempotent across owners.
Keep required runtime message collections as [] across Redis checkpoints so reconnecting strict clients receive a valid wire shape.
Summary
Backend half of the session-runtime cutover. Builds on the merged foundation in
#800 and wires distributed ownership, decision routing, retry/edit replacement,
and the snapshot/delta contract the Web client in #802 consumes.
runtime_snapshot/runtime_deltafor active and terminal runsask_userrouted to the active owner, or a fenced continuationArchitecture
flowchart TB Handler["internal/handlers<br/>LocalChannelHandler (HTTP/WS)"] Runtime["internal/sessionruntime<br/>Manager · ownership · subscriptions"] Decision["internal/decisionruntime<br/>approval / ask_user routing"] Flow["internal/conversation/flow<br/>retry/edit prep · agent stream"] Fence["internal/runtimefence<br/>PostgreSQL write token"] Message["internal/message<br/>StartCanonicalTurn / AppendCanonicalTurn"] Handler --> Runtime Handler --> Decision Runtime --> Flow Decision --> Flow Flow -->|"each step, guarded"| Fence Fence --> Message Runtime -.snapshot/delta.-> Handler Message -.message_created.-> HandlerDistributed ownership
A run is stateful in-process: the server that admits it holds the Go abort
channel, injection channel, and the live agent stream. That server is the
owner. State that must survive a process (status, messages, ownership) lives
in the backend; the live control does not.
Owner server. Ownership is recorded as a
StreamRef{OwnerID, Generation}and enforced by
Manager.ValidateRunOwnership(internal/sessionruntime/ownership.go).Only the owner can drive the run, because only it holds the in-process channels.
Redis command channel. The pub/sub primitive
(
RedisBackend.PublishCommand/SubscribeCommands, keyed per owner ascommands:<ownerID>) was added in the merged foundation feat(runtime): add session runtime foundation #800. This PR wires itup:
Manager.DispatchActiveCommandruns a command locally whenownerID == m.ownerID, otherwisedispatchRemoteCommandpublishes it to theowner and waits for the result (
internal/sessionruntime/routed_command.go).This is how an approval or abort received on server B reaches a run owned by
server A.
Owner lease.
OwnerLeaseExpiresAtis a TTL in Redis, renewed by the ownervia
RenewLease(renewRedisLeaseScriptininternal/sessionruntime/redis.go).It detects a dead or partitioned owner: once the lease expires the run is no
longer owned, another process can reconcile it, and the old owner starts
failing
ValidateRunOwnership(ctrl.leaseIsValidAt). The lease stops twoservers from both believing they own the same run.
Fencing token. A lease is time-based, so a lagging owner can still be
running for a moment after losing it.
runtimefence.Fence{Token}is amonotonic PostgreSQL generation — "a newer run increments Token and
permanently invalidates older fences" (
internal/runtimefence/fence.go).Every canonical write validates the token (
ValidateScope), so a stale owner'sappend or commit is rejected with
ErrStaleeven if its lease check brieflyraced. The lease guards who may run; the token guards which run may write
history, re-checked on every step append.
Restack note
a2697c7fc(replacementmessage_created) and3d5c1e9e6(public errorsanitization) are the review fixes and now live here. #802 is rebased on
3d5c1e9e6; GitHub lists these commits there too only because both PRs targetmain.Validation
go test ./internal/handlers ./internal/message ./internal/sessionruntimemise run runtime-contract-fixtures