From 6c4bbf7300830fd6952a6da1cc30fdaa0a9747ff Mon Sep 17 00:00:00 2001 From: _Kerman Date: Fri, 21 Aug 2026 16:28:50 +0800 Subject: [PATCH] perf(goal): read durable state from session projection --- packages/goal/goal/src/index.ts | 267 +++++++++----------- packages/goal/goal/src/types.ts | 8 +- packages/goal/goal/tests/goal.spec.ts | 60 ++--- packages/goal/goal/tests/projection.spec.ts | 19 +- 4 files changed, 169 insertions(+), 185 deletions(-) diff --git a/packages/goal/goal/src/index.ts b/packages/goal/goal/src/index.ts index baf8c921a9..7a02532e54 100644 --- a/packages/goal/goal/src/index.ts +++ b/packages/goal/goal/src/index.ts @@ -13,15 +13,11 @@ import { agentEvents } from '@deepseek-ai/dsh-agent' import type { Agent } from '@deepseek-ai/dsh-agent' import type { Session, SessionEvent } from '@deepseek-ai/dsh-session' import { TypertRemoteService, Remote } from '@deepseek-ai/dsh-typert-protocol' -// Type-only: resolves ctx.sessionProjections for the optional unit child. import type {} from '@deepseek-ai/dsh-session-projection' import { - applyGoalEvent, decodeGoalChange, - emptyGoalFoldState, goalChangeRef, } from './fold.ts' -import type { GoalFoldState } from './fold.ts' import { GOAL_CHANGE_VERSION, GoalError, @@ -62,7 +58,7 @@ declare module '@deepseek-ai/cordis' { } } -/** Wire payload schema of the `goal` projection (whole current goal or pre-create/cleared null). */ +/** Wire payload schema of the `goal` projection (current goal or pre-create/cleared null). */ const goalProjectionSchema: ZodType = zod.union([ zod.object({ goal: zod.object({ @@ -81,7 +77,7 @@ const goalProjectionSchema: ZodType = zod.union([ ]) as ZodType /** - * Light last-wins fold of the `goal` projection unit. Unlike the strict + * Light fold of the `goal` projection unit. Unlike the strict * replay fold (fold.ts: transition validation, fail-loud on malformed * changes, Set-typed state), this transition is projection-grade: the state * is plain JSON (persisted-cache precondition), any non-goal or malformed @@ -91,25 +87,34 @@ const goalProjectionSchema: ZodType = zod.union([ * invariant rejects a violating stream fail-loud where it is installed). * @param state - the projection covering all prior events. * @param event - the next committed session event. - * @returns the next projection (same reference when the event is not a goal change). + * @returns the next projection (same reference when the event is unrelated). */ export function applyGoalProjection(state: GoalProjection | null, event: SessionEvent): GoalProjection | null { - if (event.type !== 'goal/change') return state - let change: GoalChangeMeta | undefined - try { - change = decodeGoalChange(event.data) - } catch (_invalidPersistedGoalChange) { - return state + if (event.type === 'user/message' && state !== null) { + const source = event.data.source + if (source.kind !== 'goal' + || source.goalId !== state.goal.id + || source.revision !== state.goal.revision) return state + return { ...state, roundsStarted: source.round } } - if (change === undefined) return state - return change.operation === 'clear' - ? null - : { - goal: change.goal, - roundsStarted: change.roundsStarted, - createdAt: change.createdAt, - updatedAt: change.updatedAt, + if (event.type === 'goal/change') { + let change: GoalChangeMeta | undefined + try { + change = decodeGoalChange(event.data) + } catch (_invalidPersistedGoalChange) { + return state } + if (change === undefined) return state + return change.operation === 'clear' + ? null + : { + goal: change.goal, + roundsStarted: change.roundsStarted, + createdAt: change.createdAt, + updatedAt: change.updatedAt, + } + } + return state } /** Deployment defaults for goal creation. */ @@ -124,11 +129,9 @@ export interface ResolvedConfig { defaultMaxGoalRounds: number } -/** Process-local cache plus activation intent crossing the synchronous append boundary. */ -interface GoalCache { - readonly state: GoalFoldState +/** Process-local activation state crossing the synchronous append boundary. */ +interface GoalRuntimeState { activation: GoalActivation - observedSeq: number pendingActivation: { readonly seq: number; readonly activation: GoalActivation } | undefined } @@ -181,14 +184,14 @@ function resolveBlockReason(reason: unknown): GoalBlockReason { /** Goal service (`ctx.goals`) backed exclusively by the owning session log. */ export class GoalService extends TypertRemoteService { - static inject = ['agents'] + static inject = ['agents', 'sessionProjections'] static Config: z = z.object({ defaultMaxGoalRounds: z.number().default(256), }) private readonly resolved: ResolvedConfig - private readonly caches = new WeakMap() + private readonly runtimeStates = new WeakMap() constructor(ctx: Context, config: Config = {}) { super(ctx, 'goals') @@ -196,20 +199,22 @@ export class GoalService extends TypertRemoteService { defaultMaxGoalRounds: resolveMaxGoalRounds(config.defaultMaxGoalRounds ?? 256), } ctx.on('agent/session-start', ({ agent }) => { - this.cache(agent.session).activation = 'disarmed' + this.runtimeState(agent.session).activation = 'disarmed' }) - // The `goal` projection unit: last-wins fold of goal/change whole values - // (see applyGoalProjection). The unit child activates only when a - // projection registry is composed (headless assemblies stay unaffected). - ctx.inject(['sessionProjections'], (projectionCtx) => { - projectionCtx.sessionProjections.register<'goal', GoalProjection | null>({ - key: 'goal', - stateSchema: goalProjectionSchema, - init: () => null, - apply: applyGoalProjection, - wire: { viewSchema: goalProjectionSchema, view: state => state }, - stateVersion: 4, - }) + ctx.sessionProjections.register<'goal', GoalProjection | null>({ + key: 'goal', + stateSchema: goalProjectionSchema, + init: () => null, + apply: applyGoalProjection, + wire: { viewSchema: goalProjectionSchema, view: state => state }, + stateVersion: 5, + }) + ctx.on('session/event', (session, event) => { + if (event.type !== 'goal/change') return + const runtime = this.runtimeState(session) + runtime.activation = runtime.pendingActivation?.seq === event.seq + ? runtime.pendingActivation.activation + : 'disarmed' }) } @@ -221,9 +226,7 @@ export class GoalService extends TypertRemoteService { */ get(agent: Agent): GoalView | undefined { this.assertLive(agent) - const cache = this.cache(agent.session) - this.sync(agent.session, cache) - return this.view(cache) + return this.view(this.state(agent.session), this.runtimeState(agent.session)) } /** @@ -235,10 +238,9 @@ export class GoalService extends TypertRemoteService { */ disarm(agent: Agent): GoalView | undefined { this.assertLive(agent) - const cache = this.cache(agent.session) - this.sync(agent.session, cache) - cache.activation = 'disarmed' - return this.view(cache) + const runtime = this.runtimeState(agent.session) + runtime.activation = 'disarmed' + return this.view(this.state(agent.session), runtime) } /** @@ -250,8 +252,8 @@ export class GoalService extends TypertRemoteService { */ create(agent: Agent, request: CreateGoalRequest): GoalView { const spec = resolveCreateGoal(request, this.resolved.defaultMaxGoalRounds) - const cache = this.prepareMutation(agent) - const current = cache.state.goal + const [state, runtime] = this.prepareMutation(agent) + const current = state?.goal if (current !== undefined && current.phase !== 'complete') { throw new GoalError(`goal "${current.id}" already exists with phase "${current.phase}"`, 'GOAL_ALREADY_EXISTS') } @@ -263,7 +265,7 @@ export class GoalService extends TypertRemoteService { phase: 'active', maxGoalRounds: spec.maxGoalRounds, } - return this.commitSnapshot(agent, cache, 'create', goal, 0, now, now, 'armed') + return this.commitSnapshot(agent, runtime, 'create', goal, 0, now, now, 'armed') } /** @@ -275,8 +277,9 @@ export class GoalService extends TypertRemoteService { */ @Remote('edit') edit(agent: Agent, ref: GoalRef, request: EditGoalRequest): GoalView { - const cache = this.prepareMutation(agent) - const current = this.expectCurrent(cache, ref) + const [state, runtime] = this.prepareMutation(agent) + const currentState = this.expectCurrent(state, ref) + const current = currentState.goal if (request.objective === undefined && request.maxGoalRounds === undefined) { throw new GoalError('goal edit requires objective and/or maxGoalRounds', 'GOAL_INVALID_EDIT') } @@ -286,7 +289,7 @@ export class GoalService extends TypertRemoteService { ...request.objective === undefined ? {} : { objective: resolveObjective(request.objective) }, ...request.maxGoalRounds === undefined ? {} : { maxGoalRounds: resolveMaxGoalRounds(request.maxGoalRounds) }, } - return this.commitCurrent(agent, cache, 'edit', goal, cache.activation) + return this.commitCurrent(agent, currentState, runtime, 'edit', goal, runtime.activation) } /** @@ -309,22 +312,23 @@ export class GoalService extends TypertRemoteService { */ @Remote('resume') resume(agent: Agent, ref: GoalRef): GoalView { - const cache = this.prepareMutation(agent) - const current = this.expectCurrent(cache, ref) + const [state, runtime] = this.prepareMutation(agent) + const currentState = this.expectCurrent(state, ref) + const current = currentState.goal const resumable: readonly GoalPhase[] = ['active', 'paused', 'blocked'] if (!resumable.includes(current.phase)) { throw this.transitionError(current, 'resume', resumable) } - if (current.phase === 'active' && cache.activation === 'armed') { + if (current.phase === 'active' && runtime.activation === 'armed') { throw new GoalError(`goal "${current.id}" is already active and armed`, 'GOAL_INVALID_TRANSITION') } - if (cache.state.roundsStarted >= current.maxGoalRounds) { + if (currentState.roundsStarted >= current.maxGoalRounds) { throw new GoalError( `goal "${current.id}" exhausted ${current.maxGoalRounds} goal rounds; increase maxGoalRounds before resuming`, 'GOAL_INVALID_TRANSITION', ) } - return this.commitCurrent(agent, cache, 'resume', this.withPhase(current, 'active'), 'armed') + return this.commitCurrent(agent, currentState, runtime, 'resume', this.withPhase(current, 'active'), 'armed') } /** @@ -353,14 +357,16 @@ export class GoalService extends TypertRemoteService { * @returns the blocked view with its durable reason. */ block(agent: Agent, ref: GoalRef, reason: GoalBlockReason): GoalView { - const cache = this.prepareMutation(agent) - const current = this.expectCurrent(cache, ref) + const [state, runtime] = this.prepareMutation(agent) + const currentState = this.expectCurrent(state, ref) + const current = currentState.goal if (current.phase !== 'active') { throw this.transitionError(current, 'block', ['active']) } return this.commitCurrent( agent, - cache, + currentState, + runtime, 'block', { ...this.withPhase(current, 'blocked'), blockedReason: resolveBlockReason(reason) }, 'disarmed', @@ -375,39 +381,38 @@ export class GoalService extends TypertRemoteService { */ @Remote('clear') clear(agent: Agent, ref: GoalRef): GoalRef { - const cache = this.prepareMutation(agent) - const current = this.expectCurrent(cache, ref) + const [state, runtime] = this.prepareMutation(agent) + const currentState = this.expectCurrent(state, ref) + const current = currentState.goal const tombstone: GoalRef = { id: current.id, revision: current.revision + 1 } const change: GoalClearChangeMeta = { kind: 'goal/change', version: GOAL_CHANGE_VERSION, operation: 'clear', cleared: tombstone, - clearedAt: this.nextMutationTime(cache), + clearedAt: this.nextMutationTime(currentState), } - this.commit(agent, cache, change, 'disarmed') + this.commit(agent, runtime, change, 'disarmed') return { ...tombstone } } - /** Resolve and validate the cache used by a mutation. */ - private prepareMutation(agent: Agent): GoalCache { + /** Resolve the durable and process-local state used by a mutation. */ + private prepareMutation(agent: Agent): readonly [GoalProjection | null, GoalRuntimeState] { this.assertLive(agent) - const cache = this.cache(agent.session) - this.sync(agent.session, cache) - return cache + return [this.state(agent.session), this.runtimeState(agent.session)] } /** Reject stale or missing current-state refs. */ - private expectCurrent(cache: GoalCache, ref: GoalRef): GoalSnapshot { - const current = cache.state.goal - if (current === undefined) throw new GoalError('no current goal', 'GOAL_NOT_FOUND') + private expectCurrent(state: GoalProjection | null, ref: GoalRef): GoalProjection { + if (state === null) throw new GoalError('no current goal', 'GOAL_NOT_FOUND') + const current = state.goal if (ref.id !== current.id || ref.revision !== current.revision) { throw new GoalError( `stale goal ref "${ref.id}" revision ${ref.revision}; current is "${current.id}" revision ${current.revision}`, 'GOAL_STALE_REVISION', ) } - return current + return state } /** Enforce exact live-agent identity rather than trusting a matching id. */ @@ -417,33 +422,21 @@ export class GoalService extends TypertRemoteService { } } - /** Return the per-session cache, folding a seed once with activation disarmed. */ - private cache(session: Session): GoalCache { - let cache = this.caches.get(session) - if (cache !== undefined) return cache - const state = emptyGoalFoldState() - for (const event of session.events) applyGoalEvent(state, event) - cache = { - state, - activation: 'disarmed', - observedSeq: session.seq, - pendingActivation: undefined, - } - this.caches.set(session, cache) - return cache + /** Read the current durable projection maintained by the registry. */ + private state(session: Session): GoalProjection | null { + return this.ctx.sessionProjections.stateOf(session, 'goal') as GoalProjection | null } - /** Incrementally observe durable events and reconcile local activation intent. */ - private sync(session: Session, cache: GoalCache): void { - for (const event of session.events.slice(cache.observedSeq)) { - applyGoalEvent(cache.state, event) - if (event.type === 'goal/change') { - cache.activation = cache.pendingActivation?.seq === event.seq - ? cache.pendingActivation.activation - : 'disarmed' - } - cache.observedSeq += 1 + /** Return the process-local activation state, initially disarmed. */ + private runtimeState(session: Session): GoalRuntimeState { + let runtime = this.runtimeStates.get(session) + if (runtime !== undefined) return runtime + runtime = { + activation: 'disarmed', + pendingActivation: undefined, } + this.runtimeStates.set(session, runtime) + return runtime } /** Build a new revision with one replacement phase. */ @@ -466,10 +459,11 @@ export class GoalService extends TypertRemoteService { phase: GoalPhase, activation: GoalActivation, ): GoalView { - const cache = this.prepareMutation(agent) - const current = this.expectCurrent(cache, ref) + const [state, runtime] = this.prepareMutation(agent) + const currentState = this.expectCurrent(state, ref) + const current = currentState.goal if (!allowed.includes(current.phase)) throw this.transitionError(current, operation, allowed) - return this.commitCurrent(agent, cache, operation, this.withPhase(current, phase), activation) + return this.commitCurrent(agent, currentState, runtime, operation, this.withPhase(current, phase), activation) } /** Render a stable invalid-transition error. */ @@ -483,38 +477,33 @@ export class GoalService extends TypertRemoteService { /** Commit a mutation that retains the current goal's derived counters/times. */ private commitCurrent( agent: Agent, - cache: GoalCache, + state: GoalProjection, + runtime: GoalRuntimeState, operation: Exclude, goal: GoalSnapshot, activation: GoalActivation, ): GoalView { - const createdAt = cache.state.createdAt - /* v8 ignore next -- strict replay and every snapshot commit set createdAt whenever a current goal exists */ - if (createdAt === undefined) throw new Error('current goal cache lacks createdAt') return this.commitSnapshot( agent, - cache, + runtime, operation, goal, - cache.state.roundsStarted, - createdAt, - this.nextMutationTime(cache), + state.roundsStarted, + state.createdAt, + this.nextMutationTime(state), activation, ) } /** Clamp a current goal's next timestamp across backward wall-clock movement. */ - private nextMutationTime(cache: GoalCache): number { - const updatedAt = cache.state.updatedAt - /* v8 ignore next -- strict replay and every snapshot commit set updatedAt whenever a current goal exists */ - if (updatedAt === undefined) throw new Error('current goal cache lacks updatedAt') - return Math.max(Date.now(), updatedAt) + private nextMutationTime(state: GoalProjection): number { + return Math.max(Date.now(), state.updatedAt) } /** Build and commit one full-snapshot mutation. */ private commitSnapshot( agent: Agent, - cache: GoalCache, + runtime: GoalRuntimeState, operation: Exclude, goal: GoalSnapshot, roundsStarted: number, @@ -531,24 +520,27 @@ export class GoalService extends TypertRemoteService { createdAt, updatedAt, } - this.commit(agent, cache, change, activation) - const view = this.view(cache) - /* v8 ignore next -- the durable goal event installs the snapshot before this read */ - if (view === undefined) throw new Error('snapshot commit cleared the goal unexpectedly') - return view + this.commit(agent, runtime, change, activation) + return { + ...goal, + roundsStarted, + createdAt, + updatedAt, + activation: runtime.activation, + } } - /** Commit one mutation into the goal log, cache, and live event stream. */ - private commit(agent: Agent, cache: GoalCache, change: GoalChangeMeta, activation: GoalActivation): void { + /** Commit one mutation into the goal log and live event stream. */ + private commit(agent: Agent, runtime: GoalRuntimeState, change: GoalChangeMeta, activation: GoalActivation): void { const ref = goalChangeRef(change) - cache.pendingActivation = { seq: agent.session.seq, activation } + runtime.pendingActivation = { seq: agent.session.seq, activation } try { - agent.session.append('goal/change', change) - this.sync(agent.session, cache) + const event = agent.session.append('goal/change', change) + if (runtime.pendingActivation?.seq === event.seq) runtime.activation = activation } finally { - cache.pendingActivation = undefined + runtime.pendingActivation = undefined } - const goal = this.view(cache) + const goal = this.view(this.state(agent.session), runtime) const notification: GoalChanged = { operation: change.operation, ref: { ...ref }, @@ -558,21 +550,14 @@ export class GoalService extends TypertRemoteService { } /** Build a detached current view. */ - private view(cache: GoalCache): GoalView | undefined { - const goal = cache.state.goal - const createdAt = cache.state.createdAt - const updatedAt = cache.state.updatedAt - if (goal === undefined) return undefined - /* v8 ignore next 3 -- strict replay and snapshot commits establish both timestamps with every current goal */ - if (createdAt === undefined || updatedAt === undefined) { - throw new Error(`goal "${goal.id}" cache lacks timestamps`) - } + private view(state: GoalProjection | null, runtime: GoalRuntimeState): GoalView | undefined { + if (state === null) return undefined return { - ...goal, - roundsStarted: cache.state.roundsStarted, - createdAt, - updatedAt, - activation: cache.activation, + ...state.goal, + roundsStarted: state.roundsStarted, + createdAt: state.createdAt, + updatedAt: state.updatedAt, + activation: runtime.activation, } } diff --git a/packages/goal/goal/src/types.ts b/packages/goal/goal/src/types.ts index 402caade5e..0ef17dd1a6 100644 --- a/packages/goal/goal/src/types.ts +++ b/packages/goal/goal/src/types.ts @@ -84,7 +84,7 @@ export interface GoalView extends GoalSnapshot { /** * The `goal` projection value: the current durable goal with its replay - * counters, exactly as the latest `goal/change` event carried them. + * counters, including admitted goal rounds. * Activation is process-local (never persisted) and deliberately absent — * the projection reflects durable phase only. */ @@ -105,10 +105,10 @@ declare module '@deepseek-ai/dsh-session-projection/types' { } interface SessionProjectionMap { /** - * The session's current goal (the latest `goal/change` whole value), or + * The session's current goal and admitted-round count, or * `null` before the first create and after a clear tombstone. - * Whole-value rule: every goal change carries the complete post-change - * state, so the fold is last-wins. + * `goal/change` supplies the whole lifecycle value; matching admitted + * `user/message` events advance `roundsStarted`. */ goal: GoalProjection | null } diff --git a/packages/goal/goal/tests/goal.spec.ts b/packages/goal/goal/tests/goal.spec.ts index e94e0754a0..8718ef4aff 100644 --- a/packages/goal/goal/tests/goal.spec.ts +++ b/packages/goal/goal/tests/goal.spec.ts @@ -4,6 +4,7 @@ import AgentRegistry, { agentEvents, Inbox } from '@deepseek-ai/dsh-agent' import type { Agent } from '@deepseek-ai/dsh-agent' import { createUserMessage, HarnessError } from '@deepseek-ai/dsh-llm' import SessionStore, { Session, SessionId, type UserMessage } from '@deepseek-ai/dsh-session' +import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection' import GoalService, { GoalError, GoalId, @@ -59,9 +60,11 @@ function stubAgent(rawId: string, seed?: readonly import('@deepseek-ai/dsh-sessi async function harness(config: { defaultMaxGoalRounds?: number } = {}) { const ctx = new Context() + await ctx.plugin(SessionStore) await ctx.plugin(AgentRegistry) + await ctx.plugin(SessionProjectionRegistry) await ctx.plugin(GoalService, config) - const stub = stubAgent(`goal-test-${Math.random()}`) + const stub = stubAgentForSession(ctx.sessions.create(SessionId(`goal-test-${Math.random()}`))) ctx.agents.register(stub.agent) return { ctx, ...stub } } @@ -131,6 +134,7 @@ describe('GoalService creation and replay', () => { it('also resolves the default when constructed directly without Cordis config normalization', async () => { const ctx = new Context() await ctx.plugin(AgentRegistry) + await ctx.plugin(SessionProjectionRegistry) const goals = new GoalService(ctx) const stub = stubAgent('goal-direct-construction') ctx.agents.register(stub.agent) @@ -142,6 +146,7 @@ describe('GoalService creation and replay', () => { it('rejects invalid direct configuration', async () => { const ctx = new Context() await ctx.plugin(AgentRegistry) + await ctx.plugin(SessionProjectionRegistry) await expect(ctx.plugin(GoalService, { defaultMaxGoalRounds: -1 })).rejects.toThrow(expect.objectContaining({ code: 'GOAL_INVALID_MAX_ROUNDS', })) @@ -155,6 +160,7 @@ describe('GoalService creation and replay', () => { const ctx = new Context() await ctx.plugin(AgentRegistry) + await ctx.plugin(SessionProjectionRegistry) await ctx.plugin(GoalService) const resumed = stubAgent('seeded-goal', first.session.events) ctx.agents.register(resumed.agent) @@ -169,6 +175,7 @@ describe('GoalService creation and replay', () => { const ctx = new Context() await ctx.plugin(SessionStore) await ctx.plugin(AgentRegistry) + await ctx.plugin(SessionProjectionRegistry) await ctx.plugin(GoalService) const parent = stubAgentForSession(ctx.sessions.create(SessionId('goal-fork-parent'))) ctx.agents.register(parent.agent) @@ -212,19 +219,19 @@ describe('GoalService creation and replay', () => { expect(ctx.goals.resume(agent, goal)).toMatchObject({ revision: 2, activation: 'armed' }) }) - it('removes the service and its session-start listener with the providing fiber', async () => { + it('removes the service and projection with the providing fiber', async () => { const ctx = new Context() await ctx.plugin(AgentRegistry) + await ctx.plugin(SessionProjectionRegistry) const fiber = await ctx.plugin(GoalService) const first = ctx.goals const stub = stubAgent('goal-hmr') ctx.agents.register(stub.agent) - const goal = first.create(stub.agent, { objective: 'survive service reload' }) + const goal = ctx.goals.create(stub.agent, { objective: 'survive service reload' }) await fiber.dispose() expect(ctx.get('goals')).toBeUndefined() - agentEvents(ctx, stub.agent).emit('agent/session-start', { source: 'resume' }) - expect(first.get(stub.agent)).toMatchObject({ id: goal.id, activation: 'armed' }) + expect(ctx.sessionProjections.stateOf(stub.session, 'goal')).toBeUndefined() await ctx.plugin(GoalService) expect(ctx.goals).not.toBe(first) @@ -423,6 +430,7 @@ describe('GoalService mutations', () => { const ctx = new Context() await ctx.plugin(SessionStore) await ctx.plugin(AgentRegistry) + await ctx.plugin(SessionProjectionRegistry) await ctx.plugin(GoalService) const stub = stubAgentForSession(ctx.sessions.create(SessionId('goal-reentrant-observer'))) ctx.agents.register(stub.agent) @@ -441,6 +449,7 @@ describe('GoalService mutations', () => { it('does not delegate goal persistence to agent injection', async () => { const ctx = new Context() await ctx.plugin(AgentRegistry) + await ctx.plugin(SessionProjectionRegistry) await ctx.plugin(GoalService) const stub = stubAgent('goal-independent-injection') stub.agent.inject = () => { throw new Error('injection must not be called') } @@ -454,23 +463,24 @@ describe('GoalService mutations', () => { expect(stub.session.events.map(event => event.type)).toEqual(['goal/change']) }) - it('observes a valid goal snapshot appended after an empty cache was established', async () => { + it('observes an external goal change and disarms local activation', async () => { const { ctx, agent, session } = await harness() - expect(ctx.goals.get(agent)).toBeUndefined() + const created = ctx.goals.create(agent, { objective: 'before external edit', maxGoalRounds: 4 }) + expect(created.activation).toBe('armed') const change: GoalSnapshotChangeMeta = { kind: 'goal/change', version: 1, - operation: 'create', + operation: 'edit', goal: { - id: GoalId('goal-external'), - revision: 1, + id: created.id, + revision: created.revision + 1, objective: 'observe external append', phase: 'active', maxGoalRounds: 4, }, - roundsStarted: 0, - createdAt: 12, - updatedAt: 12, + roundsStarted: created.roundsStarted, + createdAt: created.createdAt, + updatedAt: created.updatedAt, } session.append('goal/change', change) @@ -481,30 +491,6 @@ describe('GoalService mutations', () => { }) }) - it('reports the same corrupt unseen event after committing its valid prefix', async () => { - const { ctx, agent, session } = await harness() - expect(ctx.goals.get(agent)).toBeUndefined() - const change: GoalSnapshotChangeMeta = { - kind: 'goal/change', - version: 1, - operation: 'create', - goal: { - id: GoalId('goal-valid-prefix'), - revision: 1, - objective: 'valid prefix', - phase: 'active', - maxGoalRounds: 4, - }, - roundsStarted: 0, - createdAt: 12, - updatedAt: 12, - } - session.append('goal/change', change) - session.append('goal/change', { ...change, operation: 'edit', extra: true } as never) - - expect(() => ctx.goals.get(agent)).toThrow('snapshot change must have exactly') - expect(() => ctx.goals.get(agent)).toThrow('snapshot change must have exactly') - }) }) describe('goal replay validation', () => { diff --git a/packages/goal/goal/tests/projection.spec.ts b/packages/goal/goal/tests/projection.spec.ts index 6282a6f739..a6b8ab1725 100644 --- a/packages/goal/goal/tests/projection.spec.ts +++ b/packages/goal/goal/tests/projection.spec.ts @@ -17,8 +17,8 @@ import type { UserMessage } from '@deepseek-ai/dsh-session' import SessionStore from '@deepseek-ai/dsh-session' import type { Session } from '@deepseek-ai/dsh-session' import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection' -import GoalService, { applyGoalProjection, foldGoal } from '@deepseek-ai/dsh-goal' -import type { GoalRef } from '@deepseek-ai/dsh-goal' +import GoalService, { GoalId, applyGoalProjection, foldGoal } from '@deepseek-ai/dsh-goal' +import type { GoalProjection, GoalRef } from '@deepseek-ai/dsh-goal' interface Bench { ctx: Context @@ -149,9 +149,22 @@ describe('goal projection unit', () => { source: { kind: 'user' }, }) const user = { type: 'user/message', seq: 0, time: 1, data: plainUser } as never - const state = { goal: { id: 'g1', revision: 1, objective: 'x', phase: 'active', maxGoalRounds: 4 }, roundsStarted: 0, createdAt: 1, updatedAt: 1 } as never + const state: GoalProjection = { + goal: { id: GoalId('g1'), revision: 1, objective: 'x', phase: 'active', maxGoalRounds: 4 }, + roundsStarted: 0, + createdAt: 1, + updatedAt: 1, + } const empty = null expect(applyGoalProjection(empty, user)).toBe(empty) + const admittedRound = { + type: 'user/message', seq: 1, time: 2, + data: createUserMessage({ + content: [{ type: 'text', text: 'continue' }], + source: { kind: 'goal', goalId: 'g1', revision: 1, round: 1 } as never, + }), + } as never + expect(applyGoalProjection(state, admittedRound)).toEqual({ ...state, roundsStarted: 1 }) const queuedUser = { type: 'agent/inbox/spliced', seq: 1, time: 2, data: { target: 'next-step', start: 0, inserted: [plainUser] },