Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(goal): persist state with domain events

Browse Source
This commit is contained in:
_Kerman
2026-07-31 22:52:18 +08:00
parent b6cf9298e3
commit 1a09174987
48 changed files with 354 additions and 650 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
packages/client/connection/src/client/fixture.ts
View File

@@ -901,13 +901,9 @@ function projectionFramesOf(id: SessionId, log: readonly SessionEvent[], event:
if (!Object.hasOwn(values, 'title')) return []
return [{ type: 'session/projection', sessionId: id, key: 'title', value: values['title'], seq: event.seq }]
}
// Goal fold: inserting a round-zero goal change durably advances the unit;
// later admission of the same message must not advance it again.
if (type === 'agent/inbox/spliced') {
const inserted = (event as unknown as { data: { inserted: UserMessage[] } }).data.inserted
if (inserted.some(message => goalChangeOf(message) !== undefined)) {
return [{ type: 'session/projection', sessionId: id, key: 'goal', value: backscanGoal(log), seq: event.seq }]
}
// The goal domain's own durable change advances its projection.
if (type === 'goal/change') {
return [{ type: 'session/projection', sessionId: id, key: 'goal', value: backscanGoal(log), seq: event.seq }]
}
// Standing-plan fold: writes replace the list; turn/start clears it (null).
if (type === 'todo/write' || type === 'turn/start') {
@@ -1140,7 +1136,7 @@ interface FxGoalProjection {
updatedAt: number
}
/** One durable goal change riding a round-zero goal-sourced inbox insertion. */
/** One durable goal change. */
type FxGoalChange =
| { kind: 'goal/change'; version: 1; operation: 'clear'; cleared: { id: string; revision: number }; clearedAt: number }
| {
@@ -1153,14 +1149,6 @@ type FxGoalChange =
updatedAt: number
}
/** Decode a fixture goal change from its durable inbox message. */
function goalChangeOf(message: UserMessage): FxGoalChange | undefined {
const source = message.source as unknown as { kind?: string; round?: number; change?: FxGoalChange }
if (source.kind !== 'goal' || source.round !== 0) return undefined
const change = source.change
return change?.kind === 'goal/change' ? change : undefined
}
/**
* Current goal projection over the full log (host parallel: the GoalService
* unit's last-wins fold of goal/change whole values; clear returns null).
@@ -1169,18 +1157,12 @@ function backscanGoal(log: readonly SessionEvent[]): FxGoalProjection | null {
for (let i = log.length - 1; i >= 0; i--) {
const event = log[i] as unknown as {
type: string
data?: { inserted?: UserMessage[] }
data?: FxGoalChange
} | undefined
if (event === undefined || event.type !== 'agent/inbox/spliced') continue
const inserted = event.data?.inserted ?? []
for (let j = inserted.length - 1; j >= 0; j--) {
const message = inserted[j]
if (message === undefined) continue
const change = goalChangeOf(message)
if (change === undefined) continue
if (change.operation === 'clear') return null
return { goal: change.goal, roundsStarted: change.roundsStarted, createdAt: change.createdAt, updatedAt: change.updatedAt }
}
if (event === undefined || event.type !== 'goal/change' || event.data === undefined) continue
const change = event.data
if (change.operation === 'clear') return null
return { goal: change.goal, roundsStarted: change.roundsStarted, createdAt: change.createdAt, updatedAt: change.updatedAt }
}
return null
}
@@ -1419,33 +1401,12 @@ export function createFixtureApi(options: FixtureOptions = {}): ApiProxy {
for (const frame of projectionFramesOf(id, log, event)) emitMux(frame)
}
/** Append one goal/change as its round-zero goal-sourced inbox insertion (host GoalService parallel). */
/** Append one durable goal/change (host GoalService parallel). */
const appendGoalChange = (id: SessionId, change: FxGoalChange): FxGoalProjection => {
const ref = change.operation === 'clear' ? change.cleared : change.goal
const payload = change.operation === 'clear'
? { cleared: change.cleared, clearedAt: change.clearedAt }
: { goal: change.goal, roundsStarted: change.roundsStarted, createdAt: change.createdAt, updatedAt: change.updatedAt }
const log = logOf(id)
const pendingNextStep = log.reduce((count, event) => {
const inboxEvent = event as unknown as {
type: string
data: { target: string; removedCount?: number; inserted: UserMessage[] }
}
if (inboxEvent.type !== 'agent/inbox/spliced' || inboxEvent.data.target !== 'next-step') return count
return count - (inboxEvent.data.removedCount ?? 0) + inboxEvent.data.inserted.length
}, 0)
append(id, {
type: 'agent/inbox/spliced',
data: {
target: 'next-step',
start: pendingNextStep,
inserted: [
userMessage(
text(`<goal_state>${JSON.stringify(payload)}</goal_state>`),
{ kind: 'goal', goalId: ref.id, revision: ref.revision, round: 0, change } as unknown as MessageSource,
),
],
},
type: 'goal/change',
data: change,
})
return backscanGoal(log) as FxGoalProjection
}