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

Merge branch 'codex/web-plan-wire' into codex/web-plan-mode

Browse Source
This commit is contained in:
fz
2026-07-24 16:34:56 +08:00
parent 4cd4c11871 76450b0c55
commit aed744dbbe
14 changed files with 252 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/client/runtime/README.md
View File

@@ -8,7 +8,7 @@ Client cordis boot + core services: SlotsService (Service wrapper over SlotCore
## Plan-mode projection
Each opened `Session` queries the optional plan capability independently of paginated history and exposes `planMode: null | { active, pending? }` in its `ConversationSnapshot`. `null` hides consumers that require the capability; a present `pending` differs from `active`. A successful selection replaces the snapshot with the host-confirmed committed and pending state; failures retain the previous state. A shared request fence drops stale plan query and selection responses, while an event-version fence preserves a commit that overtakes a current unary request. Logged live `plan/mode` events commit `active` and clear `pending`; replacement history windows also fold their latest plan event so gap repair cannot miss a recovered commit. Reconnect re-queries the full state, and a failed capability query never makes an otherwise usable conversation fail to open.
Each opened `Session` queries the optional plan capability independently of paginated history and exposes `planMode: null | { active, pending? }` in its `ConversationSnapshot`. `null` hides consumers that require the capability; a present `pending` differs from `active`. A successful selection replaces the snapshot with the host-confirmed committed and pending state; failures retain the previous state. Prompt admission waits for the latest selector request, follows a newer overlapping request when one supersedes it, stops on that latest selection's failure, and sends the resulting `pending ?? active` target with the prompt so the host cannot accept the message under another mode. A shared request fence drops stale plan query and selection responses, while an event-version fence preserves a commit that overtakes a current unary request. Logged live `plan/mode` events commit `active` and clear `pending`; replacement history windows also fold their latest plan event so gap repair cannot miss a recovered commit. Reconnect re-queries the full state, and a failed capability query never makes an otherwise usable conversation fail to open.
## Model Experience

37
packages/client/runtime/src/client/sessions/session.ts
View File

@@ -65,6 +65,11 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
private planRequestVersion = 0
/** Latest valid commit, held until the initial capability query resolves. */
private latestLivePlanMode: PlanModeState | null = null
/**
* Latest selector mutation, retained after settlement so prompt admission
* cannot miss a failure that completed before it began waiting.
*/
private planSelection: Promise<RpcResult<PlanModeState | null>> | null = null
// Revision counters + caches backing the snapshot's reference-stability contract (§A.9.4/§C.2,
// audit S5): buildSnapshot reuses the previous array when the revision is unchanged, so
// React.memo children survive unrelated snapshot swaps (chunk storms must not re-render every
@@ -108,9 +113,30 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
this.promptError = null
this.lastAgentError = null
this.notifier.markDirty()
while (this.planSelection !== null) {
const pending = this.planSelection
const selection = await pending
// A newer selection owns the target and its outcome, even when this
// superseded request fails after the replacement has already settled.
if (this.planSelection !== pending) continue
if (!selection.ok) {
this.promptError = { op: 'send', error: selection.error }
this.notifier.markDirty()
return selection
}
break
}
const planTarget = this.planMode === null
? undefined
: this.planMode.pending ?? this.planMode.active
let result: RpcResult<{ accepted: true }>
try {
result = (await this.api.sessions.prompt({ sessionId: this.sessionId, mode, content })).result
result = (await this.api.sessions.prompt({
sessionId: this.sessionId,
mode,
content,
...(planTarget === undefined ? {} : { planMode: planTarget }),
})).result
} catch (error) {
result = transportError(error)
}
@@ -147,7 +173,14 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
* @param active Whether plan mode should be selected.
* @returns The host-confirmed state, or null when plan mode is unavailable.
*/
async setPlanMode(active: boolean): Promise<RpcResult<PlanModeState | null>> {
setPlanMode(active: boolean): Promise<RpcResult<PlanModeState | null>> {
const selection = this.selectPlanMode(active)
this.planSelection = selection
return selection
}
/** Run one selector mutation; {@link setPlanMode} retains its latest outcome for prompt admission. */
private async selectPlanMode(active: boolean): Promise<RpcResult<PlanModeState | null>> {
const planEventVersion = this.planEventVersion
const planRequestVersion = ++this.planRequestVersion
let result: RpcResult<PlanModeState | null>

94
packages/client/runtime/tests/session.spec.ts
View File

@@ -384,6 +384,100 @@ describe('prompt and cancel errors', () => {
expect(session.getSnapshot().promptError).toMatchObject({ op: 'send', error: { code: 'agent-busy' } })
})
it('waits for the current selector target and admits it with the prompt', async () => {
const { api, session } = makeSession()
api.onPlanMode = () => Promise.resolve(ok({ active: false }))
await session.open()
const selected = deferred<Awaited<ReturnType<FakeApiClient['onSetPlanMode']>>>()
api.onSetPlanMode = () => selected.promise
const selecting = session.setPlanMode(true)
const prompting = session.prompt([{ type: 'text', text: 'plan this' }], 'queue')
await Promise.resolve()
expect(api.callsOf('session.prompt')).toEqual([])
selected.resolve(ok({ active: false, pending: true }))
await selecting
expect((await prompting).ok).toBe(true)
expect(api.callsOf('session.prompt')).toEqual([{
sessionId: SID,
mode: 'queue',
content: [{ type: 'text', text: 'plan this' }],
planMode: true,
}])
})
it('does not admit a prompt when the selector request fails', async () => {
const { api, session } = makeSession()
api.onPlanMode = () => Promise.resolve(ok({ active: false }))
await session.open()
const selected = deferred<Awaited<ReturnType<FakeApiClient['onSetPlanMode']>>>()
api.onSetPlanMode = () => selected.promise
const selecting = session.setPlanMode(true)
const prompting = session.prompt([{ type: 'text', text: 'do not send' }], 'queue')
selected.resolve(err({ code: 'internal', message: 'selection failed', details: {} }))
await selecting
expect((await prompting).ok).toBe(false)
expect(api.callsOf('session.prompt')).toEqual([])
expect(session.getSnapshot().promptError).toMatchObject({
op: 'send',
error: { code: 'internal', message: 'selection failed' },
})
})
it('ignores a superseded selector failure and admits the latest successful target', async () => {
const { api, session } = makeSession()
api.onPlanMode = () => Promise.resolve(ok({ active: false }))
await session.open()
const older = deferred<Awaited<ReturnType<FakeApiClient['onSetPlanMode']>>>()
const newer = deferred<Awaited<ReturnType<FakeApiClient['onSetPlanMode']>>>()
let call = 0
api.onSetPlanMode = () => ++call === 1 ? older.promise : newer.promise
const selectPlan = session.setPlanMode(true)
const prompting = session.prompt([{ type: 'text', text: 'use latest' }], 'queue')
const selectDefault = session.setPlanMode(false)
newer.resolve(ok({ active: false }))
await selectDefault
older.resolve(err({ code: 'internal', message: 'stale failure', details: {} }))
await selectPlan
expect((await prompting).ok).toBe(true)
expect(api.callsOf('session.prompt')).toEqual([{
sessionId: SID,
mode: 'queue',
content: [{ type: 'text', text: 'use latest' }],
planMode: false,
}])
})
it('retains the latest selector failure until prompt admission observes it', async () => {
const { api, session } = makeSession()
api.onPlanMode = () => Promise.resolve(ok({ active: false }))
await session.open()
const older = deferred<Awaited<ReturnType<FakeApiClient['onSetPlanMode']>>>()
const newer = deferred<Awaited<ReturnType<FakeApiClient['onSetPlanMode']>>>()
let call = 0
api.onSetPlanMode = () => ++call === 1 ? older.promise : newer.promise
const selectPlan = session.setPlanMode(true)
const prompting = session.prompt([{ type: 'text', text: 'must stay local' }], 'queue')
const selectDefault = session.setPlanMode(false)
newer.resolve(err({ code: 'internal', message: 'latest failure', details: {} }))
await selectDefault
older.resolve(ok({ active: false, pending: true }))
await selectPlan
expect((await prompting).ok).toBe(false)
expect(api.callsOf('session.prompt')).toEqual([])
expect(session.getSnapshot().promptError).toMatchObject({
op: 'send',
error: { code: 'internal', message: 'latest failure' },
})
})
it('lands cancel failures in promptError with op=stop', async () => {
const { api, session } = makeSession()
api.onCancel = () => Promise.reject(new Error('cancel transport down'))

2
packages/host/apiproxy/README.md
View File

@@ -10,7 +10,7 @@ The layering/protocol decisions are recorded in the [GUI layering and RPC protoc
The mux stream projects the latest log-backed title as a validated `session/title` control frame after each attached-session subscription baseline and immediately after the corresponding live raw title event. This projection does not add titles to `session.list`; cold sessions remain metadata-only there until opening or resuming attaches their logs.
Plan mode uses two unary methods instead of deriving current state from a history page: `session.planMode` returns the committed state plus any boundary-pending selection, and `session.setPlanMode` records a selection and returns the same authoritative shape. A present `pending` target must differ from `active`; a net-zero service cleanup intent projects as `{ active }`, and the wire schema rejects equal values. Both methods return `null` when the optional host service is absent; `null` is capability absence, while `{ active: false }` is a supported inactive session. Committed changes still arrive through the raw logged `plan/mode` session event.
Plan mode uses two unary methods instead of deriving current state from a history page: `session.planMode` returns the committed state plus any boundary-pending selection, and `session.setPlanMode` records a selection and returns the same authoritative shape. A present `pending` target must differ from `active`; a net-zero service cleanup intent projects as `{ active }`, and the wire schema rejects equal values. Both methods return `null` when the optional host service is absent; `null` is capability absence, while `{ active: false }` is a supported inactive session. Committed changes still arrive through the raw logged `plan/mode` session event. `session.prompt` may carry a `planMode` target so the host records that selection immediately before accepting the prompt; it fails when the capability is absent and restores the prior target after a synchronous prompt rejection.
## Carrier layer (`/client` + root)

1
packages/host/apiproxy/src/api/sessions.schema.ts
View File

@@ -92,6 +92,7 @@ export const sessionPromptRequestSchema = z.object({
sessionId: sessionIdSchema,
mode: z.union([z.literal('queue'), z.literal('steer')]),
content: z.array(contentBlockSchema),
planMode: z.boolean().optional(),
}) as unknown as z.ZodType<RequestPayload<'session.prompt'>>
/** session.prompt response value. */

13
packages/host/apiproxy/src/api/sessions.ts
View File

@@ -74,8 +74,17 @@ export interface SessionsApi {
history(request: RpcRequest<{ sessionId: SessionId; beforeSeq?: number; maxMessages?: number }>):
Promise<RpcResponse<{ events: HistoryEntry[]; hasMore: boolean }>>
/** Sends a message. content is core's ContentBlock[] verbatim; mode maps 1:1 — queue→send, steer→steer. */
prompt(request: RpcRequest<{ sessionId: SessionId; mode: 'queue' | 'steer'; content: ContentBlock[] }>):
/**
* Sends a message. `content` is core's ContentBlock[] verbatim and `mode`
* maps 1:1 — queue→send, steer→steer. An optional `planMode` target is
* admitted atomically with the prompt.
*/
prompt(request: RpcRequest<{
sessionId: SessionId
mode: 'queue' | 'steer'
content: ContentBlock[]
planMode?: boolean
}>):
Promise<RpcResponse<{ accepted: true }>>
/** Stops: clears both FIFOs + aborts the current step (1:1 with agent.cancel). */

7
packages/host/apiproxy/tests/fetch-carrier.spec.ts
View File

@@ -84,7 +84,12 @@ describe('unary round trip (handler ⇄ client, no network)', () => {
it('covers create/prompt/cancel/plan/describe passthrough', async () => {
const c = client()
expect((await c.sessions.create({})).result.ok).toBe(true)
expect((await c.sessions.prompt({ sessionId: 's' as never, mode: 'queue', content: [{ type: 'text', text: 'x' }] })).result.ok).toBe(true)
expect((await c.sessions.prompt({
sessionId: 's' as never,
mode: 'queue',
content: [{ type: 'text', text: 'x' }],
planMode: true,
})).result.ok).toBe(true)
expect((await c.sessions.cancel({ sessionId: 's' as never })).result.ok).toBe(true)
expect((await c.sessions.planMode({ sessionId: 's' as never })).result).toEqual({ ok: true, value: null })
expect((await c.sessions.setPlanMode({ sessionId: 's' as never, active: true })).result).toEqual({ ok: true, value: null })

9
packages/host/apiproxy/tests/rpc-schemas.spec.ts
View File

@@ -101,9 +101,14 @@ describe('sessions domain schemas', () => {
expect(sessionHistoryRequestSchema.parse({ sessionId: 's1', beforeSeq: 3, maxMessages: 5 }).beforeSeq).toBe(3)
expect(() => sessionHistoryRequestSchema.parse({ sessionId: 's1', maxMessages: 0 })).toThrow()
expect(sessionHistoryValueSchema.parse({ events: [], hasMore: false }).hasMore).toBe(false)
const prompt = sessionPromptRequestSchema.parse({ sessionId: 's1', mode: 'queue', content: [{ type: 'text', text: 'hi' }] })
expect(prompt.mode).toBe('queue')
const prompt = sessionPromptRequestSchema.parse({
sessionId: 's1', mode: 'queue', content: [{ type: 'text', text: 'hi' }], planMode: true,
})
expect(prompt).toMatchObject({ mode: 'queue', planMode: true })
expect(() => sessionPromptRequestSchema.parse({ sessionId: 's1', mode: 'inject', content: [] })).toThrow()
expect(() => sessionPromptRequestSchema.parse({
sessionId: 's1', mode: 'queue', content: [], planMode: 'plan',
})).toThrow()
expect(sessionPromptValueSchema.parse({ accepted: true }).accepted).toBe(true)
expect(sessionCancelRequestSchema.parse({ sessionId: 's1' }).sessionId).toBe('s1')
expect(sessionCancelValueSchema.parse({ accepted: true }).accepted).toBe(true)

2
packages/host/runtime/README.md
View File

@@ -18,7 +18,7 @@ Which plugins mount and with what defaults is decided only here — shells must
## ApiProxy implementation notes
Unary methods take the narrow `RpcRequest<P>` and echo `request.rpcId`; a prompt's rpcId rides `MessageSource` into the `user/message` event so clients can promote optimistic echoes. `history`/`prompt` on a cold session implicitly resume it, deduplicating concurrent calls through an in-flight table; `history` paginates backwards on message boundaries (never mid-message). `planMode` and `setPlanMode` use the same resume path, project the optional `ctx.planMode` service, canonicalize a net-zero cleanup intent by omitting `pending`, and return `null` when the service is not mounted. The mux stream replays a `session/subscribed` baseline per attached session and every still-pending question with its original rpcId. Question responses, including blank per-item answers, are validated against the owning session and exact request before an atomic first-wins claim; answer, whole-request cancellation, owner abort, and provider disposal broadcast `question/resolved`. The host stream carries session lifecycle, running flips, and `agent/error` as the only outlet for live failures with no turn position.
Unary methods take the narrow `RpcRequest<P>` and echo `request.rpcId`; a prompt's rpcId rides `MessageSource` into the `user/message` event so clients can promote optimistic echoes. `history`/`prompt` on a cold session implicitly resume it, deduplicating concurrent calls through an in-flight table; `history` paginates backwards on message boundaries (never mid-message). `planMode` and `setPlanMode` use the same resume path, project the optional `ctx.planMode` service, canonicalize a net-zero cleanup intent by omitting `pending`, and return `null` when the service is not mounted. A prompt carrying `planMode` sets that target and admits the message without an intervening await; missing capability fails closed, while synchronous admission failure restores the preceding target. The mux stream replays a `session/subscribed` baseline per attached session and every still-pending question with its original rpcId. Question responses, including blank per-item answers, are validated against the owning session and exact request before an atomic first-wins claim; answer, whole-request cancellation, owner abort, and provider disposal broadcast `question/resolved`. The host stream carries session lifecycle, running flips, and `agent/error` as the only outlet for live failures with no turn position.
## Model Experience

16
packages/host/runtime/src/api-proxy.ts
View File

@@ -443,16 +443,30 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
},
async prompt(request) {
const { sessionId, mode, content } = request.payload
const { sessionId, mode, content, planMode: planTarget } = request.payload
const found = await agentFor(sessionId)
if ('error' in found) return err(request, found.error)
const agent = found.agent
const planMode = ctx.get('planMode')
if (planTarget !== undefined && planMode === undefined) {
return err(request, {
code: 'internal',
message: 'prompt requested plan mode, but this host does not provide it',
details: {},
})
}
// No await separates selection from admission: another unary request
// cannot interleave a different target between these two operations.
const priorPlanState = planTarget === undefined ? undefined : planMode?.get(agent)
const priorPlanTarget = priorPlanState?.pending ?? priorPlanState?.active
if (planTarget !== undefined) planMode?.set(agent, planTarget)
// The rpcId rides MessageSource into user/message (merge declaration in api/sessions.ts; provisional correlation).
const source: MessageSource = { kind: 'user', rpcId: request.rpcId }
try {
if (mode === 'steer') agent.steer(content, { source })
else agent.send(content, { source })
} catch (error: unknown) {
if (priorPlanTarget !== undefined) planMode?.set(agent, priorPlanTarget)
// A synchronous throw from send/steer means disposed or invalid input; surface as agent-busy with the reason attached.
return err(request, { code: 'agent-busy', message: 'prompt rejected', details: { reason: String(error) } })
}

66
packages/host/runtime/tests/host-runtime.spec.ts
View File

@@ -273,6 +273,72 @@ describe('sessions.planMode / setPlanMode', () => {
})
describe('sessions.prompt / cancel', () => {
it('admits a prompt and its plan target through one host operation', async () => {
const running = await boot([textResponse('planned')])
await running.ctx.plugin(PlanModeService, { section: 'Plan before acting.' })
const { sessionId } = expectOk(await running.api.sessions.create(request({})))
const agent = running.ctx.agents.get(sessionId) as Agent
const idle = waitForIdle(running.ctx, agent)
expectOk(await running.api.sessions.prompt(request({
sessionId,
mode: 'queue' as const,
content: [{ type: 'text' as const, text: 'plan this' }],
planMode: true,
})))
await idle
const planEvent = agent.session.events.find(event => event.type === 'plan/mode')
const userEvent = agent.session.events.find(event => event.type === 'user/message')
const header = agent.session.events.find(event => event.type === 'request/header')
expect(planEvent?.type === 'plan/mode' && planEvent.data.active).toBe(true)
expect(planEvent?.seq).toBeLessThan(userEvent?.seq ?? Number.POSITIVE_INFINITY)
expect(header?.type === 'request/header' && header.data.header.system).toContain('Plan before acting.')
})
it('rolls back the plan target when prompt admission is rejected', async () => {
const running = await boot()
await running.ctx.plugin(PlanModeService, { section: 'Plan before acting.' })
const { sessionId } = expectOk(await running.api.sessions.create(request({})))
const agent = running.ctx.agents.get(sessionId) as Agent
const send = vi.spyOn(agent, 'send').mockImplementation(() => {
throw new Error('closed for admission')
})
try {
const response = await running.api.sessions.prompt(request({
sessionId,
mode: 'queue' as const,
content: [{ type: 'text' as const, text: 'plan this' }],
planMode: true,
}))
expect(response.result).toMatchObject({
ok: false,
error: { code: 'agent-busy', details: { reason: 'Error: closed for admission' } },
})
expect(expectOk(await running.api.sessions.planMode(request({ sessionId })))).toEqual({
active: false,
})
} finally {
send.mockRestore()
}
})
it('fails closed when a prompt targets unavailable plan mode', async () => {
const running = await boot()
const { sessionId } = expectOk(await running.api.sessions.create(request({})))
const response = await running.api.sessions.prompt(request({
sessionId,
mode: 'queue' as const,
content: [{ type: 'text' as const, text: 'must not run ambiguously' }],
planMode: true,
}))
expect(response.result).toMatchObject({
ok: false,
error: { code: 'internal', message: 'prompt requested plan mode, but this host does not provide it' },
})
expect((running.ctx.agents.get(sessionId) as Agent).session.events).toEqual([])
})
it.each([
{ name: 'host default', config: true, target: '5 words', maxTokens: 64 },
{