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

Merge branch 'codex/simp-ui-identity-residue' into codex/simp-hide-concrete-agent-loop

Browse Source 
# Conflicts:
#	docs/config-catalog.md
#	docs/cordis-catalog/events.md
#	docs/cordis-catalog/services.md
#	docs/event-producer-consumer.md
This commit is contained in:
Tianyi Cui
2026-07-14 12:46:55 +08:00
parent 91ad787ef4 faf438df58
commit 6b0de58b1c
21 changed files with 139 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/compact/compact/README.md
View File

@@ -10,7 +10,7 @@ This package is the interface tier of the compaction capability, split so each c
| `@deepseek-ai/dsh-compact-basic` (deferred) | a backend: chars-per-token estimation (`charsPerToken`, default 4) + token-budget retention + `llm.stream()` summarization |
| `@deepseek-ai/dsh-tool-compact` (deferred) | the model-facing `/compact` tool over `ctx.compact` |
Unlike the bash seam, this interface depends on `@deepseek-ai/dsh-session` and `@deepseek-ai/dsh-llm` — the contract's verbs are defined over a `Session` and its output is the `ContentBlock` vocabulary, so they cannot be expressed without naming those packages. That deviation from the "interface depends only on cordis" guidance is intentional and recorded in the [compaction capability-seam RFC](../../../docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md).
Unlike the bash seam, this interface depends on `@deepseek-ai/dsh-session` and `@deepseek-ai/dsh-llm` — the contract's verbs are defined over a `Session`, and its durable `compact/summary` event uses the `ContentBlock` vocabulary, so they cannot be expressed without naming those packages. That deviation from the "interface depends only on cordis" guidance is intentional and recorded in the [compaction capability-seam RFC](../../../docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md).
## Service API (`ctx.compact`)

2
packages/core/agent-loop/README.md
View File

@@ -41,7 +41,7 @@ interface Config {
}
```
Agents listed in config are auto-created at startup. `cwd` seeds a fresh config-created session; a materialized exact `sessionId` remount and an explicit `resumeSessionId` keep the persisted session header. A declarative lookup, resume, setup, or publication failure is contained, logged, and emitted as `agent-loop/config-start-failed(sessionId, error)` because no live `Agent` exists for an `agent/*` signal. Config agents have no per-agent persona field: they use `dsh-system-prompt`'s deployment default, while programmatic factory callers can register an agent-scoped `deployment:persona` shadow in `setup`. The plugin registers the built-in `model`/`cwd` prompt variables on `ctx.systemPrompt`, resolved per step from `assembleContextFor(agent)` — the helper couples the typed agent with its matching scope selector. These are runtime facts of the agents THIS loop drives, unlike the `harness:identity` and default `deployment:persona` sections, which live on `dsh-system-prompt` so they survive a swapped loop plugin.
Agents listed in config are auto-created at startup. `cwd` seeds a fresh config-created session; a materialized exact `sessionId` remount and an explicit `resumeSessionId` keep the persisted session header. While the factory is active, a declarative lookup, resume, setup, or publication failure is contained, logged, and emitted as `agent-loop/config-start-failed(sessionId, error)` because no live `Agent` exists for an `agent/*` signal; cancellation caused by factory teardown is silent. Config agents have no per-agent persona field: they use `dsh-system-prompt`'s deployment default, while programmatic factory callers can register an agent-scoped `deployment:persona` shadow in `setup`. The plugin registers the built-in `model`/`cwd` prompt variables on `ctx.systemPrompt`, resolved per step from `assembleContextFor(agent)` — the helper couples the typed agent with its matching scope selector. These are runtime facts of the agents THIS loop drives, unlike the `harness:identity` and default `deployment:persona` sections, which live on `dsh-system-prompt` so they survive a swapped loop plugin.
### Internal concrete driver

4
packages/core/agent-loop/src/index.ts
View File

@@ -339,7 +339,8 @@ declare module 'cordis' {
/**
* A declarative agent entry failed before it could publish a live agent.
* Consumers that buffer work for the configured identity use this
* transient signal to reject that work instead of waiting forever.
* transient signal to reject that work instead of waiting forever. Normal
* factory teardown suppresses failures from the cancelled startup attempt.
* @param sessionId - exact shared agent/session identity that failed startup.
* @param error - persistence, setup, or publication failure.
* @mode emit
@@ -430,6 +431,7 @@ export class AgentLoop extends Service implements AgentFactory {
sessionId: SessionId,
error: unknown,
): void {
if (!this.ownership.isActive()) return
this.ctx.logger.warn(`agent "${configId}": config-driven ${action} of "${sessionId}" failed: ${renderThrown(error)}`)
const args: unknown[] = ['agent-loop/config-start-failed', sessionId, error]
for (const callback of this.ctx.events.dispatch('emit', args)) {

5
packages/core/agent-loop/tests/config-session-id.spec.ts
View File

@@ -172,6 +172,8 @@ describe('config-driven session id', () => {
const listing = Promise.withResolvers<Awaited<ReturnType<typeof ctx.sessionPersistence.list>>>()
vi.spyOn(ctx.sessionPersistence, 'list').mockReturnValue(listing.promise)
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
const failures: unknown[] = []
ctx.on('agent-loop/config-start-failed', (_sessionId, error) => { failures.push(error) })
const loop = await ctx.plugin(AgentLoop, {
agents: [{ id: 'main', sessionId: SessionId('stdio-exact-dispose'), model: 'mock' }],
@@ -181,9 +183,10 @@ describe('config-driven session id', () => {
await Promise.resolve()
expect(disposed).toBe(false)
listing.resolve([])
listing.reject(new Error('startup cancelled by teardown'))
await disposal
expect(ctx.agents.get(SessionId('stdio-exact-dispose'))).toBeUndefined()
expect(failures).toEqual([])
expect(warn).not.toHaveBeenCalled()
warn.mockRestore()
await ctx.fiber.dispose()

2
packages/core/session/README.md
View File

@@ -53,7 +53,7 @@ Durable values need one accepted representation, not a check followed by a secon
### Request-header reconstruction (`request-header.ts`)
The `request/header` event records a full canonical `EpochHeader` snapshot with reason `initial`, `resume`, or `change`, making the request envelope logged session state and every conversation request a pure function of the log. `foldRequestHeader(events)` selects the latest snapshot from a log or prefix; `canonicalHeader` pins the one representation of absence (empty system/tools/messagePrefix ≡ absent fields), and `headerEquals` compares canonical headers. `EpochHeader.messagePrefix` is the durable record of the `agent/session-prefix` waterfall's product — composed once per loop instance, the request is `messagePrefix + derived history`, and `deriveMessages()` never returns it. Legacy v0 seeds containing the removed `request/header-delta` format are rejected rather than partially replayed.
The `request/header` event records a full canonical `EpochHeader` snapshot with reason `initial`, `resume`, or `change`, making the request envelope logged session state and every conversation request a pure function of the log. `foldRequestHeader(events)` selects the latest snapshot from a log or prefix; `canonicalHeader` pins the one representation of absence (empty system/tools/messagePrefix ≡ absent fields), and `headerEquals` compares canonical headers. `EpochHeader.messagePrefix` is the durable record of the `agent/session-prefix` waterfall's product — composed once per loop instance, the request is `messagePrefix + derived history`, and `deriveMessages()` never returns it. Legacy v0 seeds containing the removed `request/header-delta` event or its full-snapshot `fallback` reason are rejected rather than partially replayed.
### Session event vocabulary (`types.ts`)

11
packages/core/session/src/index.ts
View File

@@ -212,6 +212,15 @@ function assertSessionEventEnvelope(value: Record<string, unknown>, index: numbe
}
}
/** Reject request-header vocabulary removed with the legacy delta codec. */
function assertSupportedRequestHeader(type: string, data: unknown, location: string): void {
if (type === 'request/header'
&& data !== null && typeof data === 'object' && !Array.isArray(data)
&& (data as Record<string, unknown>)['reason'] === 'fallback') {
throw new Error(`${location} uses unsupported legacy request/header reason "fallback"`)
}
}
type SessionCallback = (...args: unknown[]) => unknown
/** Resolve one listener snapshot, including Cordis's internal dispatch checks. */
@@ -311,6 +320,7 @@ export class Session {
throw new Error(`seed event at index ${index} is not losslessly JSON-serializable`)
}
assertSessionEventEnvelope(snapshot, index)
assertSupportedRequestHeader(snapshot.type, snapshot.data, `seed event at index ${index}`)
if (snapshot.seq !== index) {
throw new Error(`seed event at index ${index} has seq ${snapshot.seq} (expected ${index}); seed must be contiguous from 0`)
}
@@ -396,6 +406,7 @@ export class Session {
if (dataSnapshot === undefined) {
throw new Error(`session event "${type}" carries non-JSON-serializable data`)
}
assertSupportedRequestHeader(type, dataSnapshot, `session event "${type}"`)
const surfaceMetadataSnapshot = snapshotJsonValue(surfaceMetadata)
if (surfaceMetadataSnapshot === undefined) {
throw new Error(`session event "${type}" carries non-JSON-serializable surface metadata`)

14
packages/core/session/tests/request-header.spec.ts
View File

@@ -68,4 +68,18 @@ describe('legacy request-header format', () => {
}] as unknown as SessionEvent[]
expect(() => new Session(SessionId('legacy'), legacy)).toThrow(/unsupported legacy request\/header-delta/)
})
it('rejects the removed fallback reason in seeds and untyped appends', () => {
const legacy = [{
type: 'request/header', seq: 0, time: 1, data: { header: { config: CONFIG }, reason: 'fallback' },
}] as unknown as SessionEvent[]
expect(() => new Session(SessionId('legacy-seed-reason'), legacy))
.toThrow('unsupported legacy request/header reason "fallback"')
const session = new Session(SessionId('legacy-append-reason'))
const appendLegacy = session.append.bind(session) as (type: string, data: unknown) => SessionEvent
expect(() => appendLegacy('request/header', { header: { config: CONFIG }, reason: 'fallback' }))
.toThrow('unsupported legacy request/header reason "fallback"')
expect(session.events).toHaveLength(0)
})
})

19
packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts
View File

@@ -162,6 +162,25 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
await expect(ctx.sessionPersistence.load(m.id)).rejects.toThrow(/unsupported legacy request\/header-delta event at seq 1/)
})
it('rejects a stored v0 full header carrying the legacy fallback reason', async () => {
const m = meta('legacy-header-fallback', '/legacy')
const path = logPath(root, m.cwd, m.id)
await mkdir(sessionDir(root, m.cwd), { recursive: true })
await writeFile(path, [
JSON.stringify(toHeaderLine(m)),
JSON.stringify({
type: 'request/header',
seq: 0,
time: 1,
data: { header: { config: { model: 'legacy' } }, reason: 'fallback' },
}),
'',
].join('\n'))
await expect(ctx.sessionPersistence.load(m.id))
.rejects.toThrow(/unsupported legacy request\/header reason "fallback" at seq 0/)
})
it('persists a forked child seed through the existing session write path', async () => {
const source = ctx.sessions.create(SessionId('persist-parent'), { meta: { cwd: '/workspace' } })
appendClosedTurn(source)

19
packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts
View File

@@ -161,6 +161,25 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
await mounted.dispose()
})
it('rejects a stored v0 full header carrying the legacy fallback reason', async () => {
const path = await freshDbPath()
const m = meta('legacy-header-fallback', '/legacy')
const db = openDatabase(path, 'wal')
db.prepare('INSERT INTO sessions (id, version, created_at, cwd, parent_session, seed_length) VALUES (?, ?, ?, ?, NULL, NULL)')
.run(m.id, m.version, m.createdAt, m.cwd ?? null)
db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, ?, ?, ?, ?)')
.run(m.id, 0, 'request/header', 1, JSON.stringify({
header: { config: { model: 'legacy' } },
reason: 'fallback',
}))
db.close()
const mounted = await backend(path)
await expect(mounted.ctx.sessionPersistence.load(m.id))
.rejects.toThrow(/unsupported legacy request\/header reason "fallback" at seq 0/)
await mounted.dispose()
})
it('an interrupted turn (rows after the last turn/end) is PRESERVED and closed during load', async () => {
const path = await freshDbPath()
const m = meta('crash')

5
packages/session-persistence/session-persistence/src/coordinator.ts
View File

@@ -158,6 +158,11 @@ function assertSupportedEvents(events: readonly SessionEvent[], id: SessionId):
if (legacy !== undefined) {
throw new Error(`session "${id}" contains unsupported legacy request/header-delta event at seq ${legacy.seq}`)
}
const fallback = events.find(event => event.type === 'request/header'
&& (event.data as { reason?: string }).reason === 'fallback')
if (fallback !== undefined) {
throw new Error(`session "${id}" contains unsupported legacy request/header reason "fallback" at seq ${fallback.seq}`)
}
}
/**

36
packages/session-persistence/session-persistence/tests/persistence.spec.ts
View File

@@ -22,6 +22,16 @@ function legacyHeaderDelta(seq = 0): SessionEvent {
} as unknown as SessionEvent
}
/** An obsolete full-header reason fixture from the removed delta codec. */
function legacyFallbackHeader(seq = 0): SessionEvent {
return {
type: 'request/header',
seq,
time: 1,
data: { header: { config: { model: 'legacy' } }, reason: 'fallback' },
} as unknown as SessionEvent
}
/** Optional plugin config: an EXTERNAL store shared across backend instances. */
interface MemoryConfig { store?: MemoryStore }
@@ -190,6 +200,19 @@ describe('SessionPersistence service registration', () => {
await fiber.dispose()
})
it('rejects a legacy fallback header buffered by a pre-change live producer', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const fiber = await ctx.plugin(MemoryPersistence)
const session = ctx.sessions.create(SessionId('legacy-fallback-live'), { meta: { cwd: '/legacy' } })
const appendLegacy = session.append.bind(session) as (type: string, data: unknown) => SessionEvent
expect(() => appendLegacy('request/header', legacyFallbackHeader().data))
.toThrow('unsupported legacy request/header reason "fallback"')
expect(session.events).toHaveLength(0)
await fiber.dispose()
})
it('rejects a legacy stored prefix during live HMR adoption', async () => {
const id = SessionId('legacy-hmr')
const m = meta(id, '/legacy')
@@ -207,4 +230,17 @@ describe('SessionPersistence service registration', () => {
.rejects.toThrow(/unsupported legacy request\/header-delta event at seq 0/)
await Promise.allSettled([fiber.dispose()])
})
it('rejects a stored legacy fallback header during load', async () => {
const id = SessionId('legacy-fallback-load')
const m = meta(id, '/legacy')
const store: MemoryStore = new Map([[id, { meta: m, events: [legacyFallbackHeader()] }]])
const ctx = new Context()
await ctx.plugin(SessionStore)
const fiber = await ctx.plugin(MemoryPersistence, { store })
await expect(ctx.sessionPersistence.load(id))
.rejects.toThrow('unsupported legacy request/header reason "fallback" at seq 0')
await fiber.dispose()
})
})

4
packages/support/acp-snapshot/src/launcher.ts
View File

@@ -178,13 +178,14 @@ export function launchAcpTestAgent(options: AcpTestLaunchOptions): LaunchedAcpTe
// `closed` follows parser exhaustion. Capture both eagerly so a caller that
// invokes close after process exit still joins the complete drain boundary.
const stdioClosed = new Promise<void>(resolve => child.once('close', () => { resolve() }))
const drained = Promise.all([stdioClosed, client.closed]).then(async () => {
const drained = Promise.allSettled([stdioClosed, client.closed]).then(async ([, clientResult]) => {
// The ACP SDK's readable loop dispatches client callbacks without awaiting
// them. Once `closed` settles no new callbacks can start, but callbacks
// already in flight still belong to this launch's teardown boundary.
while (inFlightClientCallbacks.size > 0) {
await Promise.allSettled([...inFlightClientCallbacks])
}
if (clientResult.status === 'rejected') throw clientResult.reason
})
// A caller may await a pending update without calling close(). Make natural
// stream exhaustion terminal for those waiters too, but only after the
@@ -206,6 +207,7 @@ export function launchAcpTestAgent(options: AcpTestLaunchOptions): LaunchedAcpTe
try {
await spawned
} catch (error: unknown) {
await drained.catch(() => undefined)
closeUpdateStream()
throw error
}

6
packages/support/acp-snapshot/src/suite.ts
View File

@@ -486,8 +486,10 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
const entries = await readdir(dir, { withFileTypes: true })
await Promise.all(entries
.filter(entry => entry.isFile()
&& entry.name.startsWith('session.')
&& entry.name.endsWith('.jsonl')
// Only valid numbered children are record-owned stale output.
// Malformed session-like names stay for the inventory guard to
// reject instead of being silently deleted during mutation.
&& /^session\.[1-9]\d*\.jsonl$/.test(entry.name)
&& !outputNames.has(entry.name))
.map(entry => rm(join(dir, entry.name))))
fixtureFiles = outputFixtureFiles

9
packages/support/acp-snapshot/tests/harness.spec.ts
View File

@@ -62,8 +62,17 @@ describe('runScenario', () => {
it('surfaces an asynchronous child spawn failure through startup and close', async () => {
const { dir } = await scenario({})
const launched = launchAcpTestAgent({ agent: AGENT, cwd: join(dir, 'missing') })
let stdioClosed = false
let clientClosed = false
launched.child.once('close', () => { stdioClosed = true })
void launched.client.closed.then(
() => { clientClosed = true },
() => { clientClosed = true },
)
await expect(launched.spawned).rejects.toMatchObject({ code: 'ENOENT' })
await expect(launched.close()).rejects.toMatchObject({ code: 'ENOENT' })
expect(stdioClosed).toBe(true)
expect(clientClosed).toBe(true)
})
it('centralizes ACP boot, captures, updates, fail-closed permissions, and shutdown', { timeout: 20_000 }, async () => {