feat(agent-core-v2): event-source the human agent state store - #3691
Conversation
|
commit: |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 3e3d75aa5b
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| const store: AgentEventStore = createEventStoreSync({ journal: memoryJournal(), slices: agentSlices }); | ||
| const initialTurnId = options.initialTurnId ?? 0; | ||
| if (initialTurnId > 0) { | ||
| void store.dispatch(turnEnded({ turnId: initialTurnId - 1, outcome: 'done' })); |
There was a problem hiding this comment.
When initialTurnId > 0, dispatch schedules its fold through this.tail.then(...), so this fire-and-forget call has not updated the store when actor.start() synchronously receives store.ready. Because the agent intentionally ignores subsequent store.changed events, a resumed LoopService starts its machine turn counter at 0 rather than turnKey.nextTurnId, producing duplicate machine turn IDs. Seed the journal before createEventStoreSync, or await initialization before starting the actor.
Useful? React with 👍 / 👎.
| const branch = existed | ||
| ? this.tree.openBranch(agentId) | ||
| : this.tree.createBranch(agentId, opts?.from !== undefined ? { from: opts.from } : undefined); |
There was a problem hiding this comment.
After undo('main', ...), the live store is reset to a branch such as main~2 and the session roster records that mapping, but a later open('main') always reopens the original main branch. This occurs after dispose/process restart, or after closing and reopening the agent, and resurrects the turns that were undone; resolve the agent's current branch through the session roster before opening it.
Useful? React with 👍 / 👎.
| type: turnEnded.type, | ||
| kind: 'event', | ||
| data: turnEnded({ turnId: lastTurnId, outcome: 'done' }), |
There was a problem hiding this comment.
For a migrated v2 agent, this writes only one turn.ended event and no turn.started events. The new turnIndexSlice creates undo boundaries exclusively on turn.started, so stores.undo(agentId, 1) immediately after migration reports insufficient even when the migrated history contains multiple user turns; the previous implementation could undo those turns by scanning their user-message anchors. Migration needs to emit corresponding start boundaries for the retained turns.
Useful? React with 👍 / 👎.
| for (const event of events) { | ||
| const ts = typeof event.time === 'number' ? event.time : Date.now(); | ||
| const ref: BranchRef = { branch: this.journal.ref.branch, seq: this.journal.nextSeq() }; | ||
| const { raised, effects } = this.applyEvent(event, ref, ts, false); |
There was a problem hiding this comment.
When a reducer drain or journal.append fails—for example because the disk write fails or the branch is degraded—applyEvent has already replaced this.state, and queued effects may already have run. The returned dispatch rejects and no event is durable, yet later operations continue from the unpersisted state, so reopening produces a different result; stage or roll back the fold until the append succeeds.
Useful? React with 👍 / 👎.
| void this.journal | ||
| .append({ type: SNAPSHOT_ENTRY_TYPE, kind: SNAPSHOT_ENTRY_KIND, data: { slices } }) | ||
| .catch((error) => this.report(error)); |
There was a problem hiding this comment.
On an asynchronous backend, this fire-and-forget snapshot can remain queued in Branch.append after the triggering dispatch resolves. A following dispatch then reads journal.nextSeq() before the snapshot has entered the branch, while its actual append is queued after that snapshot, making the reducer's FoldContext.ref.seq one less than the persisted event's sequence. In particular, turnIndex can record a snapshot as the turn start and undo will fork at the wrong boundary; serialize the snapshot append through the store tail.
Useful? React with 👍 / 👎.
| name in saved | ||
| ? slice.deserialize !== undefined | ||
| ? slice.deserialize(saved[name]) | ||
| : saved[name] | ||
| : slice.initialState(); |
There was a problem hiding this comment.
When a slice is registered after a snapshot that does not contain it, this seeds the slice with initialState() but replay begins only after the snapshot. Therefore any matching events before the snapshot are permanently invisible to the late-joined slice—for example a plugin loaded on a long-lived session cannot reconstruct its earlier state.updated events. If the selected snapshot lacks a registered slice, that slice must fold from the beginning or from a snapshot that includes it.
Useful? React with 👍 / 👎.
7d7cdeb to
8b97368
Compare
…ent/session human/eventStore core (schema-first defineEvent factories, serial fold-first dispatch with snapshotting, late-join slices, journal reset), agent events/slices/historySchema, machine rewrite around a store actor with mirror consumption, session stores with turnIndex-boundary undo, todo slice, v2 migration to the new event types, and the loop engine bridge.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 8b97368ae1
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| [turnStarted.type]: (draft, event: TurnStarted, ctx) => { | ||
| draft.turns.push({ turnId: event.turnId, start: ctx.ref }); | ||
| }, |
There was a problem hiding this comment.
If the process exits after turn.started is durable but before turn.ended, reopening folds an incomplete index entry while leaving nextTurnId unchanged. The restored agent therefore reuses the interrupted turn's ID for its next turn, creating duplicate turn IDs and ambiguous undo entries. Advance nextTurnId from turn.started as the previous replay logic did.
Useful? React with 👍 / 👎.
| if (!existed) { | ||
| await (await this.session()).dispatch(agentOpened({ agentId, branch: branch.name })); |
There was a problem hiding this comment.
After close(agentId) removes the agent from the session roster, calling open(agentId) again finds the retained physical branch and skips agentOpened. The agent works for the current process but remains absent from the durable roster, so the next session restart does not restore it. Append agentOpened when reopening a closed branch, not only when creating a new branch.
Useful? React with 👍 / 👎.
| ...mirrorPatch(event.state), | ||
| turnTools: {}, | ||
| background: {}, | ||
| scope: createAbortScope(), |
There was a problem hiding this comment.
After any store.reset, this replaces context.scope, but the root controllerGuard invocation at machine.ts:340-344 remains bound to the original scope because it is never restarted. Subsequent LLM requests and tools inherit the replacement scope, so stopping the agent no longer aborts their signals and asynchronous requests or side-effecting tools can continue after shutdown.
Useful? React with 👍 / 👎.
| target.on('turn.started', (event) => { | ||
| if (event.type !== 'turn.started') return; | ||
| state.currentTurn += 1; | ||
| if (state.todos.length === 0) return; | ||
| if (state.todos.every((todo) => todo.status === 'done')) return; | ||
| if (state.currentTurn - state.lastWriteTurn !== STALE_TURNS) return; | ||
| const { todos, currentTurn, lastWriteTurn } = readTodoState(store); |
There was a problem hiding this comment.
For real agent actors, the turn.started notification is emitted before the corresponding store append is sent (agent/machine.ts:556-564), and dispatch folds asynchronously. This handler consequently reads the previous currentTurn, unlike the old implementation that incremented it before the check, so an unfinished todo list is reminded one turn later than STALE_TURNS specifies. The tests mask this by dispatching turnStarted before invoking the handler.
Useful? React with 👍 / 👎.
…he store actor static Phantom-turn root cause: local patches plus the async store.changed mirror were two consistency models for the same data. The machine now owns its state (every transition is a local assign plus a store.append dual write) and consumes only store.ready/reset/error. - store actor becomes static storeActor logic with the engine passed via invoke input; the session machine registers agentActor and spawns it by name - SessionStores.open/fork return the engine directly - store.reset aborts the old abort scope (cascades to in-flight turn/tools/background) - no-op drain events are no longer journaled - engine bridge seeds the initial turn id into the journal before folding, and resetHistory is awaited before the retry notify - tests pass the store via machine input and wait for the log to catch up before store assertions; drop debug residue
… event sourcing Snapshots serialized the whole slice state every 500 events and, on an asynchronous backend, their fire-and-forget append could still be queued when the next dispatch read journal.nextSeq(), making the recorded FoldContext.ref.seq one less than the persisted event's sequence and shifting turnIndex undo boundaries. Removing the mechanism eliminates the race and the full-state rewrite; refold now always seeds from initial state and replays every event record (legacy snapshot entries are ignored).
8b97368 to
2fc619b
Compare
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 2fc619b5f4
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| sendTo('store', ({ context }) => ({ | ||
| type: 'store.append' as const, | ||
| event: turnStarted({ turnId: context.turnId, queueItemId: context.drainedId }), |
There was a problem hiding this comment.
When undoing a completed turn, the prompt has already been persisted by the idle transition (message.appended and queue.drained) before this turn.started event records the undo boundary. SessionStores.undo() forks at start.seq - 1, so it retains the undone user prompt while removing its response; the next turn then sends the model that orphaned prompt followed by the new prompt. Persist the turn boundary before the prompt is appended, or index the prompt's entry as the start.
Useful? React with 👍 / 👎.
| const sourceBranch = this.tree.openBranch(source.ref.branch); | ||
| const head = sourceBranch.head; |
There was a problem hiding this comment.
When a fork is requested as soon as a turn completes, this reads the physical branch head without waiting for the source store's queued appends. In the human agent flow, turn.done is emitted before the completion batch is sent to the store, so a listener that forks from that event can create a child at the previous head and silently omit the just-completed turn. Ensure all source events are enqueued and flushed before choosing the fork ref.
Useful? React with 👍 / 👎.
| return this.open( | ||
| agentId, | ||
| head === null ? undefined : { from: { branch: sourceBranch.name, seq: head } }, | ||
| ); |
There was a problem hiding this comment.
When the source is itself a newly created fork or undo branch with inherited history but no local entries, Branch.head is null, so this passes no from ref and creates an empty target branch. Forking that agent before its first new event therefore loses all inherited conversation state; use the source branch's parent ref when its local head is null.
Useful? React with 👍 / 👎.
Resolve the overlap with the protocol trait refactor (#3641) and the event-sourced agent store (#3691): - engine.ts: keep both additions at the machine wiring site — the turn-aware credential provider feeding input.request.credentials, and the journal-backed AgentEventStore now required by AgentInput - docs/{en,zh}/llm.md: unify the request lifecycle paragraph — the credential resolution flow alongside the requester plan* composition, and the turn-side emptyResponseError / recovery-chain wording that matches the merged code
1 participant
Related Issue
Internal task (no tracking issue). See Problem below.
Problem
The human layer's agent state had no durable, unified store. Persistence mirrored machine snapshots (
persist/agent.ts,persist/session.ts), undo meant scanning journal entries for anchors, and the TreeStore (branch/fork/seq semantics) was only exercised by tests. There was no event-sourced path where agent state is folded from events with proper branch and undo semantics.What changed
human/eventStorecore library: schema-firstdefineEventfactories (zod-validated plain-object events with a registry as the replay contract), serial fold-first dispatch (dispatch resolves after the journal append), internal raised events (not persisted) with a drain limit, snapshots every 500 events, late-join slices that refold history,resetto swap journals with subscriber continuity, plusmemoryJournal/createEventStoreSyncfor sync paths, and a staticstoreActor(fromCallback) that receives the engine via invoke input.store.appenddual write, kept equivalent to the slice folds. It consumes onlystore.ready(restore),store.reset(undo/branch switch, aborting the old scope first) andstore.error; the oldstore.changedmirror was removed because the dual-read race produced phantom turns.AgentInputnow carries the store engine.SessionStores(open/fork/close/undo) over TreeStore branches with a_sessionledger (roster/sessionMeta slices). Undo is an O(1) turnIndex boundary lookup + branch fork +store.reset. The session machine registers a staticagentActorand spawns agents by name with the engine in spawn input.persist/v2/migrate.tsrewrites old wire records into the new event types;persist/open.tsreturns TreeStore + SessionStores. Deletedpersist/agent.ts,persist/session.ts,agent/replay.ts,session/undo.ts,todo/state.ts.memoryJournal), behavior-neutral. Wiring the disk path (kap-server session lifecycle, replacing wire.jsonl) is a follow-up task.Checklist
/approve).gen-changesetsskill, or this PR needs no changeset.gen-docsskill, or this PR needs no doc update.Tests:
src/human/test— 27 files / 306 tests pass; package typecheck passes. No changeset: internal architecture change, not user-perceivable (the loop bridge is behavior-neutral). No doc update: no user-facing behavior change.