mirror of
https://github.com/deepseek-ai/deepseek-harness
synced 2026-08-15 21:04:50 +00:00
fix(web): fence plan mode projection races
This commit is contained in:
@@ -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 successful selection replaces the snapshot with the host-confirmed committed and pending state; failures retain the previous state. Logged live `plan/mode` events commit `active` and clear `pending`, while reconnect re-queries the full state. 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. 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
|
||||
|
||||
|
||||
@@ -59,6 +59,10 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
||||
private planCapabilityKnown = false
|
||||
/** Monotonic local fence for committed plan events observed on the mux stream. */
|
||||
private planEventVersion = 0
|
||||
/** Highest plan event seq observed through an append or replacement window. */
|
||||
private latestPlanEventSeq: number | null = null
|
||||
/** Monotonic fence shared by plan queries and selections; only the latest response may land. */
|
||||
private planRequestVersion = 0
|
||||
/** Latest valid commit, held until the initial capability query resolves. */
|
||||
private latestLivePlanMode: PlanModeState | null = null
|
||||
// Revision counters + caches backing the snapshot's reference-stability contract (§A.9.4/§C.2,
|
||||
@@ -145,6 +149,7 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
||||
*/
|
||||
async setPlanMode(active: boolean): Promise<RpcResult<PlanModeState | null>> {
|
||||
const planEventVersion = this.planEventVersion
|
||||
const planRequestVersion = ++this.planRequestVersion
|
||||
let result: RpcResult<PlanModeState | null>
|
||||
try {
|
||||
result = (await this.api.sessions.setPlanMode({ sessionId: this.sessionId, active })).result
|
||||
@@ -152,8 +157,9 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
||||
result = transportError(error)
|
||||
}
|
||||
if (result.ok) {
|
||||
this.applyPlanResponse(result.value, planEventVersion)
|
||||
this.notifier.notifyNow()
|
||||
if (this.applyPlanResponse(result.value, planEventVersion, planRequestVersion)) {
|
||||
this.notifier.notifyNow()
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
@@ -387,6 +393,7 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
||||
this.hasMore = hasMore
|
||||
this.foldAdapter.reset(this.events, this.baseSeq, this.views)
|
||||
this.rebuildDerivedFromWindow()
|
||||
this.applyLatestWindowPlanMode()
|
||||
const buffered = this.liveBuffer
|
||||
this.liveBuffer = []
|
||||
for (const item of buffered) this.appendLive(item.event, item.view)
|
||||
@@ -411,10 +418,11 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
||||
*/
|
||||
private async refreshPlanMode(generation: number): Promise<void> {
|
||||
const planEventVersion = this.planEventVersion
|
||||
const planRequestVersion = ++this.planRequestVersion
|
||||
try {
|
||||
const { result } = await this.api.sessions.planMode({ sessionId: this.sessionId })
|
||||
if (generation !== this.openGeneration) return
|
||||
if (result.ok) this.applyPlanResponse(result.value, planEventVersion)
|
||||
if (result.ok) this.applyPlanResponse(result.value, planEventVersion, planRequestVersion)
|
||||
else console.error('[web-runtime] plan-mode query failed:', result.error)
|
||||
} catch (error) {
|
||||
if (generation !== this.openGeneration) return
|
||||
@@ -429,6 +437,8 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
||||
if (typeof candidate.data !== 'object' || candidate.data === null) return
|
||||
const data = candidate.data as { active?: unknown }
|
||||
if (typeof data.active !== 'boolean') return
|
||||
if (this.latestPlanEventSeq !== null && event.seq <= this.latestPlanEventSeq) return
|
||||
this.latestPlanEventSeq = event.seq
|
||||
this.planEventVersion++
|
||||
this.latestLivePlanMode = { active: data.active }
|
||||
if (this.planCapabilityKnown && this.planMode !== null) {
|
||||
@@ -436,20 +446,38 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
||||
}
|
||||
}
|
||||
|
||||
/** Replacement windows bypass appendLive, so fold their newest valid plan commit explicitly. */
|
||||
private applyLatestWindowPlanMode(): void {
|
||||
for (let i = this.events.length - 1; i >= 0; i--) {
|
||||
const before = this.latestPlanEventSeq
|
||||
this.applyLivePlanMode(this.events[i] as SessionEvent)
|
||||
if (this.latestPlanEventSeq !== before) return
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply a unary plan snapshot unless a newer mux commit crossed the request.
|
||||
* Apply the latest unary plan snapshot unless a newer request or mux commit
|
||||
* crossed it.
|
||||
* A successful null response establishes absence and never promotes a raw
|
||||
* event into a capability.
|
||||
*
|
||||
* @returns Whether this response was current and applied.
|
||||
*/
|
||||
private applyPlanResponse(value: PlanModeState | null, requestVersion: number): void {
|
||||
private applyPlanResponse(
|
||||
value: PlanModeState | null,
|
||||
requestEventVersion: number,
|
||||
requestVersion: number,
|
||||
): boolean {
|
||||
if (requestVersion !== this.planRequestVersion) return false
|
||||
this.planCapabilityKnown = true
|
||||
if (value === null) {
|
||||
this.planMode = null
|
||||
return
|
||||
return true
|
||||
}
|
||||
this.planMode = requestVersion === this.planEventVersion
|
||||
this.planMode = requestEventVersion === this.planEventVersion
|
||||
? value
|
||||
: this.latestLivePlanMode ?? value
|
||||
return true
|
||||
}
|
||||
|
||||
/** Land a live session/event (open/repair in flight -> buffer; overlapping seq -> drop;
|
||||
|
||||
@@ -96,6 +96,29 @@ describe('plan mode projection', () => {
|
||||
expect(result).toEqual({ ok: true, value: { active: false, pending: true } })
|
||||
expect(api.callsOf('session.setPlanMode')).toEqual([{ sessionId: SID, active: true }])
|
||||
expect(session.getSnapshot().planMode).toEqual({ active: false, pending: true })
|
||||
|
||||
api.onSetPlanMode = () => Promise.resolve(ok({ active: false }))
|
||||
await session.setPlanMode(false)
|
||||
expect(session.getSnapshot().planMode).toEqual({ active: false })
|
||||
})
|
||||
|
||||
it('drops an older overlapping selection response after the newer selection lands', 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 selectDefault = session.setPlanMode(false)
|
||||
newer.resolve(ok({ active: false }))
|
||||
await selectDefault
|
||||
older.resolve(ok({ active: false, pending: true }))
|
||||
await selectPlan
|
||||
|
||||
expect(session.getSnapshot().planMode).toEqual({ active: false })
|
||||
})
|
||||
|
||||
it('retains the prior state when a selection fails at the business or transport layer', async () => {
|
||||
@@ -174,6 +197,26 @@ describe('plan mode projection', () => {
|
||||
expect(selection.session.getSnapshot().planMode).toEqual({ active: true })
|
||||
})
|
||||
|
||||
it('applies a plan commit recovered through a gap-repair replacement window', async () => {
|
||||
const { api, session } = makeSession()
|
||||
const initial = plainTurn(0, 0, 'a', 'b')
|
||||
api.onHistory = () => histResponse(initial)
|
||||
api.onPlanMode = () => Promise.resolve(ok({ active: false, pending: true }))
|
||||
await session.open()
|
||||
|
||||
const planCommit = at(6, { type: 'plan/mode', data: { active: true } })
|
||||
const later = ev.turnStart(7, 1)
|
||||
api.onHistory = () => histResponse([...initial, planCommit, later])
|
||||
session.handleMuxEnvelope('rp-gap' as never, {
|
||||
type: 'session/event', sessionId: SID, event: later,
|
||||
})
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(api.callsOf('session.history')).toHaveLength(2)
|
||||
expect(session.getSnapshot().planMode).toEqual({ active: true })
|
||||
})
|
||||
})
|
||||
|
||||
it('keeps history usable when the independent capability query fails', async () => {
|
||||
const business = makeSession()
|
||||
const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {})
|
||||
|
||||
@@ -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. Both 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.
|
||||
|
||||
## Carrier layer (`/client` + root)
|
||||
|
||||
|
||||
@@ -113,7 +113,10 @@ export const sessionCancelValueSchema = z.object({
|
||||
export const planModeStateSchema = z.object({
|
||||
active: z.boolean(),
|
||||
pending: z.boolean().optional(),
|
||||
}) satisfies z.ZodType<Wire<PlanModeState>>
|
||||
}).refine(
|
||||
state => state.pending === undefined || state.pending !== state.active,
|
||||
{ message: 'pending must differ from active when present', path: ['pending'] },
|
||||
) satisfies z.ZodType<Wire<PlanModeState>>
|
||||
|
||||
/** session.planMode request payload. */
|
||||
export const sessionPlanModeRequestSchema = z.object({
|
||||
|
||||
@@ -47,7 +47,7 @@ export interface SessionSummary {
|
||||
/**
|
||||
* Plan collaboration state exposed to clients. `active` is the logged state
|
||||
* shaping the current request; `pending`, when present, is the user's
|
||||
* next-boundary selection.
|
||||
* next-boundary selection and differs from `active`.
|
||||
*/
|
||||
export interface PlanModeState {
|
||||
active: boolean
|
||||
|
||||
@@ -114,6 +114,8 @@ describe('sessions domain schemas', () => {
|
||||
expect(sessionSetPlanModeValueSchema.parse({ active: true })).toEqual({ active: true })
|
||||
expect(() => sessionSetPlanModeRequestSchema.parse({ sessionId: 's1', active: 'yes' })).toThrow()
|
||||
expect(() => sessionPlanModeValueSchema.parse({ active: 'yes' })).toThrow()
|
||||
expect(() => sessionPlanModeValueSchema.parse({ active: false, pending: false })).toThrow()
|
||||
expect(() => sessionSetPlanModeValueSchema.parse({ active: true, pending: true })).toThrow()
|
||||
expect(contentBlockSchema.parse({ type: 'text', text: 'x', extra: 1 })).toMatchObject({ extra: 1 })
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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, and return `null` when it 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. 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
|
||||
|
||||
|
||||
@@ -12,7 +12,8 @@ import type { JsonValue, Session, SessionEvent, SessionHeader, SessionId } from
|
||||
import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
|
||||
import { foldSessionTitle } from '@deepseek-ai/dsh-session-title'
|
||||
import type {
|
||||
ApiProxy, HistoryEntry, HostFrame, MuxFrame, QuestionResponsePayload, SessionSummary, ToolEventView,
|
||||
ApiProxy, HistoryEntry, HostFrame, MuxFrame, PlanModeState, QuestionResponsePayload, SessionSummary,
|
||||
ToolEventView,
|
||||
} from '@deepseek-ai/dsh-host-apiproxy/api'
|
||||
import { questionResponsePayloadSchema } from '@deepseek-ai/dsh-host-apiproxy/api/questions.schema'
|
||||
import type { ClientResponse, RpcError, RpcReceipt, RpcRequest, RpcResponse } from '@deepseek-ai/dsh-host-apiproxy/api/rpc'
|
||||
@@ -65,6 +66,16 @@ function ok<T>(request: RpcRequest<unknown>, value: T): RpcResponse<T> {
|
||||
return { rpcId: request.rpcId, result: { ok: true, value } }
|
||||
}
|
||||
|
||||
/**
|
||||
* Project the service's boundary-cleanup intent into canonical wire state.
|
||||
* A target equal to the committed value has no user-visible pending effect.
|
||||
*/
|
||||
function projectPlanModeState(state: PlanModeState): PlanModeState {
|
||||
return state.pending !== undefined && state.pending === state.active
|
||||
? { active: state.active }
|
||||
: state
|
||||
}
|
||||
|
||||
/** Wrap an error result echoing the request's rpcId. */
|
||||
function err<T>(request: RpcRequest<unknown>, error: RpcError): RpcResponse<T> {
|
||||
return { rpcId: request.rpcId, result: { ok: false, error } }
|
||||
@@ -466,7 +477,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
const found = await agentFor(request.payload.sessionId)
|
||||
if ('error' in found) return err(request, found.error)
|
||||
const planMode = ctx.get('planMode')
|
||||
return ok(request, planMode?.get(found.agent) ?? null)
|
||||
return ok(request, planMode === undefined ? null : projectPlanModeState(planMode.get(found.agent)))
|
||||
},
|
||||
|
||||
async setPlanMode(request) {
|
||||
@@ -475,7 +486,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
const planMode = ctx.get('planMode')
|
||||
if (planMode === undefined) return ok(request, null)
|
||||
planMode.set(found.agent, request.payload.active)
|
||||
return ok(request, planMode.get(found.agent))
|
||||
return ok(request, projectPlanModeState(planMode.get(found.agent)))
|
||||
},
|
||||
},
|
||||
|
||||
|
||||
@@ -256,8 +256,8 @@ describe('sessions.planMode / setPlanMode', () => {
|
||||
})
|
||||
expect(expectOk(await running.api.sessions.setPlanMode(request({ sessionId, active: false })))).toEqual({
|
||||
active: false,
|
||||
pending: false,
|
||||
})
|
||||
expect(expectOk(await running.api.sessions.planMode(request({ sessionId })))).toEqual({ active: false })
|
||||
})
|
||||
|
||||
it('returns the normal session-not-found error for both methods', async () => {
|
||||
|
||||
Reference in New Issue
Block a user