fix(goal): make publication reentrancy-safe

packages/goal/goal/README.md

@@ -21,7 +21,7 @@ At most one goal is current. Creation produces an active revision-one goal and a
 
 Every non-clear mutation appends a complete versioned snapshot through `agent.inject()`; clear appends a revisioned tombstone. The raw `context/message`, its `{ kind: 'goal' }` source, and its metadata must agree exactly. Replay rejects malformed shapes, source/content drift, discontinuous revisions, illegal lifecycle transitions, non-monotonic per-goal timestamps, and non-sequential goal rounds. Mutation timestamps clamp against the preceding goal update when wall time moves backward.
 
-Injection may append immediately or wait in an active tool-batch FIFO. The service overlays accepted pending changes in memory and reconciles each exact payload when it enters the log, so consecutive model-tool mutations see their own latest revisions without treating an unlogged cache as durable state. `goal/changed` fires after the append or enqueue succeeds; listener failures are contained.
+Injection may append immediately or wait in an active tool-batch FIFO. The service overlays accepted pending changes in memory and reconciles each exact payload when it enters the log, so consecutive model-tool mutations see their own latest revisions without treating an unlogged cache as durable state. Reentrant append observers see each accepted mutation exactly once, and incremental replay retains its cursor at the first corrupt event. `goal/changed` fires after the append or enqueue succeeds; listener failures are contained.
 
 Activation is never persisted. A fresh cache and every `agent/session-start` edge disarm it even when replay finds an active durable phase. Session resume and fork therefore retain the objective, phase, revisions, and admitted-round count without initiating work; a later explicit resume mutation must arm continuation.
 
@@ -51,3 +51,4 @@ Append-only within an epoch: each mutation follows the reusable request prefix a
 - **Round-count budget only** — `maxGoalRounds` does not meter tokens, currency, wall time, or provider quotas.
 - **No independent evaluator** — the caller that records completion or blocking is authoritative; evaluator-backed certification is deferred to a separate policy layer.
 - **One current goal** — parallel objectives and a separate goal database are intentionally absent; history remains available in the session log after replacement or clear.
+- **Trusted in-process producers** — a plugin with direct `Session` access can append counterfeit goal metadata. Strict replay detects malformed or inconsistent records and leaves goal access failed at that record until the log is repaired; this is integrity detection, not plugin isolation.

packages/goal/goal/src/index.ts

@@ -64,12 +64,19 @@ export interface ResolvedConfig {
   defaultMaxGoalRounds: number
 }
 
+/** One accepted mutation waiting to enter or be observed in the session log. */
+interface PendingGoalChange {
+  readonly change: GoalChangeMeta
+  readonly activation: GoalActivation
+  applied: boolean
+}
+
 /** Process-local cache plus mutations waiting in the active tool-batch FIFO. */
 interface GoalCache {
   readonly state: GoalFoldState
   activation: GoalActivation
   observedSeq: number
-  readonly pending: GoalChangeMeta[]
+  readonly pending: PendingGoalChange[]
 }
 
 /** Validate a caller-visible positive safe-integer round cap. */
@@ -358,15 +365,21 @@ export class GoalService extends Service {
         const change = decodeGoalEvent(event)
         if (change !== undefined) {
           const pending = cache.pending[0]
-          if (pending !== undefined && sameChange(pending, change)) {
+          if (pending !== undefined && sameChange(pending.change, change)) {
+            if (!pending.applied) {
+              applyGoalChange(cache.state, change)
+              cache.activation = pending.activation
+              pending.applied = true
+            }
             cache.pending.shift()
+            cache.observedSeq += 1
             continue
           }
         }
       }
       applyGoalEvent(cache.state, event)
+      cache.observedSeq += 1
     }
-    cache.observedSeq = session.seq
   }
 
   /** Build a new revision with one replacement phase. */
@@ -464,14 +477,26 @@ export class GoalService extends Service {
     const meta = snapshotJsonValue(change) as JsonValue | undefined
     /* v8 ignore next -- validated goal changes contain only finite JSON primitives and records */
     if (meta === undefined) throw new Error('goal change is not losslessly JSON-serializable')
-    agent.inject(renderGoalChange(change), {
-      source: { kind: 'goal', goalId: ref.id, revision: ref.revision, round: 0 },
-      envelope: 'raw',
-      meta,
-    })
-    cache.pending.push(change)
-    applyGoalChange(cache.state, change)
-    cache.activation = activation
+    const pending: PendingGoalChange = { change, activation, applied: false }
+    cache.pending.push(pending)
+    try {
+      agent.inject(renderGoalChange(change), {
+        source: { kind: 'goal', goalId: ref.id, revision: ref.revision, round: 0 },
+        envelope: 'raw',
+        meta,
+      })
+    } catch (error: unknown) {
+      const index = cache.pending.indexOf(pending)
+      /* v8 ignore next -- a committed goal append cannot reject after its contained observers run */
+      if (index < 0) throw new Error('goal injection failed after its pending mutation was reconciled', { cause: error })
+      cache.pending.splice(index, 1)
+      throw error
+    }
+    if (!pending.applied) {
+      applyGoalChange(cache.state, change)
+      cache.activation = activation
+      pending.applied = true
+    }
     this.sync(agent.session, cache)
     const goal = this.view(cache)
     const notification: GoalChanged = {

packages/goal/goal/tests/goal.spec.ts

@@ -235,6 +235,25 @@ describe('GoalService creation and replay', () => {
     expect(() => foldGoal(session.events)).not.toThrow()
   })
 
+  it('removes the service and its session-start listener with the providing fiber', async () => {
+    const ctx = new Context()
+    await ctx.plugin(AgentRegistry)
+    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' })
+
+    await fiber.dispose()
+    expect(ctx.get('goals')).toBeUndefined()
+    agentEvents(ctx, stub.agent).emit('agent/session-start', 'resume')
+    expect(first.get(stub.agent)).toMatchObject({ id: goal.id, activation: 'armed' })
+
+    await ctx.plugin(GoalService)
+    expect(ctx.goals).not.toBe(first)
+    expect(ctx.goals.get(stub.agent)).toMatchObject({ id: goal.id, activation: 'disarmed' })
+  })
+
   it('requires the exact live registry instance for reads and mutations', async () => {
     const { ctx, agent } = await harness()
     const impostor = { ...agent, session: new Session(agent.id) }
@@ -404,6 +423,46 @@ describe('GoalService mutations', () => {
     expect(foldGoal(session.events)).toMatchObject({ goal: { revision: 3, phase: 'paused' } })
   })
 
+  it('publishes a mutation consistently to a reentrant session observer', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    await ctx.plugin(AgentRegistry)
+    await ctx.plugin(GoalService)
+    const stub = stubAgentForSession(ctx.sessions.create(SessionId('goal-reentrant-observer')))
+    ctx.agents.register(stub.agent)
+    let observed: ReturnType<GoalService['get']>
+    ctx.on('session/event', (session, event) => {
+      if (session === stub.session && event.type === 'context/message') observed = ctx.goals.get(stub.agent)
+    })
+
+    const created = ctx.goals.create(stub.agent, { objective: 'publish once' })
+
+    expect(observed).toEqual(created)
+    expect(ctx.goals.get(stub.agent)).toEqual(created)
+    expect(foldGoal(stub.session.events)).toMatchObject({ goal: { id: created.id, revision: 1 } })
+  })
+
+  it('rolls back a pending mutation when injection rejects before append', async () => {
+    const ctx = new Context()
+    await ctx.plugin(AgentRegistry)
+    await ctx.plugin(GoalService)
+    const stub = stubAgent('goal-rejected-injection')
+    const append = stub.agent.inject.bind(stub.agent)
+    let reject = true
+    stub.agent.inject = (content, options) => {
+      if (reject) throw new Error('injection rejected')
+      append(content, options)
+    }
+    ctx.agents.register(stub.agent)
+
+    expect(() => ctx.goals.create(stub.agent, { objective: 'first attempt' })).toThrow('injection rejected')
+    reject = false
+    expect(ctx.goals.create(stub.agent, { objective: 'second attempt' })).toMatchObject({
+      objective: 'second attempt',
+      revision: 1,
+    })
+  })
+
   it('rejects deferred goal mutations that enter the log out of FIFO order', async () => {
     const test = await harness()
     test.setDeferred(true)
@@ -447,6 +506,39 @@ describe('GoalService mutations', () => {
       activation: 'disarmed',
     })
   })
+
+  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,
+    }
+    appendInjection(session, renderGoalChange(change), {
+      source: { kind: 'goal', goalId: change.goal.id, revision: 1, round: 0 },
+      envelope: 'raw',
+      meta: change as never,
+    })
+    appendInjection(session, [{ type: 'text', text: 'corrupt' }], {
+      source: { kind: 'goal', goalId: change.goal.id, revision: 2, round: 0 },
+      envelope: 'raw',
+      meta: { ...change, operation: 'edit', extra: true } as never,
+    })
+
+    expect(() => ctx.goals.get(agent)).toThrow('invalid shape')
+    expect(() => ctx.goals.get(agent)).toThrow('invalid shape')
+  })
 })
 
 describe('goal replay validation', () => {