perf(goal): read durable state from session projection

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<GoalProjection | null> = zod.union([
   zod.object({
     goal: zod.object({
@@ -81,7 +77,7 @@ const goalProjectionSchema: ZodType<GoalProjection | null> = zod.union([
 ]) as ZodType<GoalProjection | null>
 
 /**
- * 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<GoalProjection | null> = 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<Config> = z.object({
     defaultMaxGoalRounds: z.number().default(256),
   })
 
   private readonly resolved: ResolvedConfig
-  private readonly caches = new WeakMap<Session, GoalCache>()
+  private readonly runtimeStates = new WeakMap<Session, GoalRuntimeState>()
 
   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<GoalOperation, 'create' | 'clear'>,
     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<GoalOperation, 'clear'>,
     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,
     }
   }
 

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
   }

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', () => {

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] },