From 5506980216722b642e782262ab47372b86dee2ac Mon Sep 17 00:00:00 2001 From: Elliott de Launay Date: Fri, 9 Oct 2026 21:38:56 -0400 Subject: [PATCH 1/2] feat(task): add runState field, write helpers, and projection --- ...task-lifecycle-run-state-kernel-tickets.md | 8 +- src/core/task/Task.ts | 64 ++++++++++ .../task/__tests__/Task.run-shadow.spec.ts | 120 ++++++++++++++++++ .../__tests__/runStateShadow.spec.ts | 115 +++++++++++++++++ src/core/task/run-state/runStateShadow.ts | 51 ++++++++ 5 files changed, 354 insertions(+), 4 deletions(-) create mode 100644 src/core/task/__tests__/Task.run-shadow.spec.ts create mode 100644 src/core/task/run-state/__tests__/runStateShadow.spec.ts create mode 100644 src/core/task/run-state/runStateShadow.ts diff --git a/docs/architecture/task-lifecycle-run-state-kernel-tickets.md b/docs/architecture/task-lifecycle-run-state-kernel-tickets.md index 039914acaa..46a6154bde 100644 --- a/docs/architecture/task-lifecycle-run-state-kernel-tickets.md +++ b/docs/architecture/task-lifecycle-run-state-kernel-tickets.md @@ -284,11 +284,11 @@ A direct swap is unsafe. The fields are read across `src`. The safe path runs th ### Developer Notes -- [ ] Add a `run: RunState` field to `Task`. It always runs. It is not gated. +- [ ] Add a `runState: RunState` field to `Task`. It always runs. It is not gated. - [ ] Add the write-helper pattern: one method per event that sets the old field and drives the event into `nextRunState`. Keep the field authoritative. - [ ] If the kernel rejects an event that the code applies, keep the field write. Record the rejection for RSK-05. -- [ ] Add the projection: a pure function from the current field values to the expected `RunState`. RSK-05 compares against it. -- [ ] Do not read `run` for control flow. +- [ ] Add the projection: a pure function from the current field values to the expected `RunStateFields`, the values the kernel getters report. RSK-05 compares against it. +- [ ] Do not read `runState` for control flow. Exit criteria: the helpers and the projection exist and are unit-tested. No write site uses them yet. Behavior is unchanged. @@ -309,7 +309,7 @@ Shadow mode must report disagreement. Before any read flips, the kernel must agr ### Developer Notes -- [ ] After each helper call, compare `run` with the projection from RSK-04, for the routed fields only. A field whose writes are not routed yet gets no events, so a comparison of it reports a false mismatch. Keep the routed fields in one list that each routing ticket extends: RSK-06 adds the latches and `phase`, and RSK-10 adds the stream, ask, and turn regions. +- [ ] After each helper call, compare `runState` with the projection from RSK-04, for the routed fields only. A field whose writes are not routed yet gets no events, so a comparison of it reports a false mismatch. Keep the routed fields in one list that each routing ticket extends: RSK-06 adds the latches and `isInitialized`, and RSK-10 adds the stream, ask, and turn regions. - [ ] Log each mismatch and each rejected event with the state, the event, the field values, and `taskId`. - [ ] Tag each mismatch with a race ID when it matches a known window. Count tagged and untagged mismatches as separate metrics. - [ ] Load the ignore list from the known-races table. An entry with a fix ticket stays in the list only until that ticket merges. R-1 stays until RSK-17 merges. R-3 stays until RSK-20 merges. R-4 stays until RSK-19 merges. diff --git a/src/core/task/Task.ts b/src/core/task/Task.ts index 5ef3e54b56..a549264782 100644 --- a/src/core/task/Task.ts +++ b/src/core/task/Task.ts @@ -147,6 +147,8 @@ import { mergeConsecutiveApiMessages } from "./mergeConsecutiveApiMessages" import { prepareApiConversationMessage } from "./apiConversationHistory" import { shouldAddUserMessageToHistory } from "./messageCounting" import { type TaskExecutionContext } from "./providerHandoff" +import { initialRunState, type Ask, type RunEvent, type RunState } from "./run-state/runState" +import { applyRunEvent, type RunRejection } from "./run-state/runStateShadow" const MAX_EXPONENTIAL_BACKOFF_SECONDS = 600 // 10 minutes const DEFAULT_USAGE_COLLECTION_TIMEOUT_MS = 5000 // 5 seconds @@ -403,6 +405,9 @@ export class Task extends EventEmitter implements TaskLike { abortReason?: ClineApiReqCancelReason isInitialized = false + runState: RunState = initialRunState + readonly runRejections: RunRejection[] = [] + // API apiConfiguration: ProviderSettings api: ApiHandler @@ -3365,6 +3370,65 @@ export class Task extends EventEmitter implements TaskLike { this.debouncedEmitTokenUsage.flush() } + markInitialized(): void { + this.isInitialized = true + this.driveRunEvent({ tag: "initialized" }) + } + + markStreamStarted(): void { + this.isStreaming = true + this.driveRunEvent({ tag: "streamStarted" }) + } + + markStreamCleanupFinished(): void { + this.didFinishAbortingStream = true + this.driveRunEvent({ tag: "streamCleanupFinished" }) + } + + markStreamEnded(): void { + this.isStreaming = false + this.driveRunEvent({ tag: "streamEnded" }) + } + + markAskStarted(ask: Exclude): void { + this.driveRunEvent({ tag: "askStarted", ask }) + } + + markAskSettled(askTs: number): void { + this.driveRunEvent({ tag: "askSettled", askTs }) + } + + markCompletionAccepted(askTs: number): void { + this.driveRunEvent({ tag: "completionAccepted", askTs }) + } + + setAbortReason(reason: ClineApiReqCancelReason): void { + this.abortReason ??= reason + this.driveRunEvent({ tag: "reasonSet", reason }) + } + + requestAbort(isAbandoned: boolean): void { + if (isAbandoned) { + this.abandoned = true + } + this.abort = true + this.driveRunEvent({ tag: "abortRequested", abandoned: isAbandoned }) + } + + requestAbandon(): void { + this.abandoned = true + this.driveRunEvent({ tag: "abandonRequested" }) + } + + requestDispose(): void { + this.abort = true + this.driveRunEvent({ tag: "disposeRequested" }) + } + + private driveRunEvent(event: RunEvent): void { + this.runState = applyRunEvent(this.runState, event, this.runRejections) + } + public abortTask(isAbandoned = false): Promise { if (isAbandoned) { this.abandoned = true diff --git a/src/core/task/__tests__/Task.run-shadow.spec.ts b/src/core/task/__tests__/Task.run-shadow.spec.ts new file mode 100644 index 0000000000..3c8e6a9ea9 --- /dev/null +++ b/src/core/task/__tests__/Task.run-shadow.spec.ts @@ -0,0 +1,120 @@ +// npx vitest run core/task/__tests__/Task.run-shadow.spec.ts + +import { providerIdentifiers } from "@roo-code/types/provider-identifiers" +import { TelemetryService } from "@roo-code/telemetry" + +import { makeExtensionContext, makeUri } from "../../../test-utils/vscode" +import { Task } from "../Task" +import { ClineProvider } from "../../webview/ClineProvider" +import { initialRunState } from "../run-state/runState" + +vi.mock("../../../i18n", () => ({ t: (key: string) => key })) +vi.mock("../../ignore/RooIgnoreController") + +function makeTask(): Task { + const provider = { + log: vi.fn(), + context: makeExtensionContext({ globalStorageUri: makeUri("/mock/global-storage") }), + } + // Double cast: the constructor reads only these provider members. + return new Task({ + provider: provider as unknown as ClineProvider, + apiConfiguration: { apiProvider: providerIdentifiers.anthropic, apiKey: "test-api-key" }, + task: "test task", + startTask: false, + }) +} + +describe("Task run-state write helpers", () => { + let task: Task + + beforeEach(() => { + if (!TelemetryService.hasInstance()) { + TelemetryService.createInstance([]) + } + task = makeTask() + }) + + it("starts with the initial run state and no rejections", () => { + expect(task.runState).toEqual(initialRunState) + expect(task.runRejections).toEqual([]) + }) + + it("sets the field and drives the event for each stream and phase helper", () => { + task.markInitialized() + task.markStreamStarted() + task.markStreamCleanupFinished() + + expect(task.isInitialized).toBe(true) + expect(task.isStreaming).toBe(true) + expect(task.didFinishAbortingStream).toBe(true) + expect(task.runState.phase).toBe("running") + expect(task.runState.stream).toEqual({ tag: "live", generation: 1, cleanupFinished: true }) + + task.markStreamEnded() + expect(task.isStreaming).toBe(false) + expect(task.runState.stream).toEqual({ tag: "none" }) + expect(task.runRejections).toEqual([]) + }) + + it("requestAbort(false) sets abort only", () => { + task.requestAbort(false) + + expect([task.abort, task.abandoned]).toEqual([true, false]) + expect(task.runState.latches).toMatchObject({ abort: true, abandoned: false }) + }) + + it("requestAbort(true) sets abort and abandoned", () => { + task.requestAbort(true) + + expect([task.abort, task.abandoned]).toEqual([true, true]) + expect(task.runState.latches).toMatchObject({ abort: true, abandoned: true }) + }) + + it("requestAbandon sets abandoned after an abort", () => { + task.requestAbort(false) + task.requestAbandon() + + expect(task.abandoned).toBe(true) + expect(task.runState.latches.abandoned).toBe(true) + expect(task.runRejections).toEqual([]) + }) + + it("requestDispose sets abort and the disposed latch", () => { + task.requestDispose() + + expect(task.abort).toBe(true) + expect(task.runState.latches).toMatchObject({ abort: true, disposed: true }) + }) + + it("drives the ask events", () => { + task.markInitialized() + task.markAskStarted({ tag: "approval", askTs: 10 }) + expect(task.runState.ask).toEqual({ tag: "approval", askTs: 10 }) + + task.markAskSettled(10) + expect(task.runState.ask).toEqual({ tag: "none" }) + + task.markAskStarted({ tag: "completion", askTs: 20 }) + task.markCompletionAccepted(20) + expect(task.runState.phase).toBe("completed") + expect(task.runRejections).toEqual([]) + }) + + it("keeps the field write and records the event when the kernel rejects it", () => { + task.requestAbandon() + + expect(task.abandoned).toBe(true) + expect(task.runState.latches.abandoned).toBe(false) + expect(task.runRejections).toEqual([{ event: { tag: "abandonRequested" }, state: initialRunState }]) + }) + + it("keeps the first abort reason in the field and in the kernel", () => { + task.setAbortReason("user_cancelled") + task.setAbortReason("streaming_failed") + + expect(task.abortReason).toBe("user_cancelled") + expect(task.runState.latches.reason).toBe("user_cancelled") + expect(task.runRejections).toEqual([]) + }) +}) diff --git a/src/core/task/run-state/__tests__/runStateShadow.spec.ts b/src/core/task/run-state/__tests__/runStateShadow.spec.ts new file mode 100644 index 0000000000..40822243d6 --- /dev/null +++ b/src/core/task/run-state/__tests__/runStateShadow.spec.ts @@ -0,0 +1,115 @@ +import { initialRunState, nextRunState, type RunEvent, type RunState } from "../runState" +import { + applyRunEvent, + MAX_RUN_REJECTIONS, + projectRunState, + runStateFieldGetters, + type RunRejection, + type RunStateFields, +} from "../runStateShadow" + +const idle: RunStateFields = { + abort: false, + abandoned: false, + isInitialized: false, + isStreaming: false, + didFinishAbortingStream: false, +} + +function drive(events: RunEvent[]): RunState { + return events.reduce((state, event) => { + const result = nextRunState(state, event) + if (!("ok" in result)) { + throw new Error(`expected ${event.tag} to be accepted`) + } + return result.ok + }, initialRunState as RunState) +} + +function kernelFields(state: RunState): RunStateFields { + return { + abort: runStateFieldGetters.abort(state), + abandoned: runStateFieldGetters.abandoned(state), + abortReason: runStateFieldGetters.abortReason(state), + isInitialized: runStateFieldGetters.isInitialized(state), + isStreaming: runStateFieldGetters.isStreaming(state), + didFinishAbortingStream: runStateFieldGetters.didFinishAbortingStream(state), + } +} + +describe("projectRunState", () => { + it("keeps each field value in its own slot", () => { + const fields: RunStateFields = { + abort: true, + abandoned: false, + abortReason: "user_cancelled", + isInitialized: true, + isStreaming: true, + didFinishAbortingStream: true, + } + expect(projectRunState(fields)).toEqual(fields) + expect(projectRunState({ ...fields, abort: false, abandoned: true })).toMatchObject({ + abort: false, + abandoned: true, + }) + }) + + it("reports didFinishAbortingStream false without a live stream, even when the field is stale", () => { + expect(projectRunState({ ...idle, didFinishAbortingStream: true }).didFinishAbortingStream).toBe(false) + }) + + it("agrees with the kernel getters after the same sequence of writes", () => { + const completed = drive([ + { tag: "initialized" }, + { tag: "askStarted", ask: { tag: "completion", askTs: 1 } }, + { tag: "completionAccepted", askTs: 1 }, + ]) + const cases: Array<[RunState, RunStateFields]> = [ + [initialRunState, idle], + [drive([{ tag: "initialized" }]), { ...idle, isInitialized: true }], + [ + drive([{ tag: "initialized" }, { tag: "streamStarted" }, { tag: "streamCleanupFinished" }]), + { ...idle, isInitialized: true, isStreaming: true, didFinishAbortingStream: true }, + ], + [ + drive([ + { tag: "initialized" }, + { tag: "reasonSet", reason: "user_cancelled" }, + { tag: "abortRequested", abandoned: false }, + ]), + { ...idle, isInitialized: true, abort: true, abortReason: "user_cancelled" }, + ], + [drive([{ tag: "abortRequested", abandoned: true }]), { ...idle, abort: true, abandoned: true }], + [completed, { ...idle, isInitialized: true }], + ] + for (const [state, fields] of cases) { + expect(projectRunState(fields)).toEqual(kernelFields(state)) + } + }) +}) + +describe("applyRunEvent", () => { + it("returns the next state and records nothing when the kernel accepts the event", () => { + const rejections: RunRejection[] = [] + const next = applyRunEvent(initialRunState, { tag: "initialized" }, rejections) + expect(next.phase).toBe("running") + expect(rejections).toEqual([]) + }) + + it("keeps the state and records the event and state when the kernel rejects the event", () => { + const rejections: RunRejection[] = [] + const next = applyRunEvent(initialRunState, { tag: "streamEnded" }, rejections) + expect(next).toBe(initialRunState) + expect(rejections).toEqual([{ event: { tag: "streamEnded" }, state: initialRunState }]) + }) + + it("drops the oldest rejection when the buffer is full", () => { + const rejections: RunRejection[] = [] + for (let askTs = 0; askTs < MAX_RUN_REJECTIONS + 5; askTs++) { + applyRunEvent(initialRunState, { tag: "completionAccepted", askTs }, rejections) + } + expect(rejections).toHaveLength(MAX_RUN_REJECTIONS) + expect(rejections[0].event).toEqual({ tag: "completionAccepted", askTs: 5 }) + expect(rejections.at(-1)?.event).toEqual({ tag: "completionAccepted", askTs: MAX_RUN_REJECTIONS + 4 }) + }) +}) diff --git a/src/core/task/run-state/runStateShadow.ts b/src/core/task/run-state/runStateShadow.ts new file mode 100644 index 0000000000..f78291b1d2 --- /dev/null +++ b/src/core/task/run-state/runStateShadow.ts @@ -0,0 +1,51 @@ +import { + abandoned, + abort, + abortReason, + didFinishAbortingStream, + isInitialized, + isStreaming, + nextRunState, + type AbortReason, + type RunEvent, + type RunState, +} from "./runState" + +export type RunStateFields = { + abort: boolean + abandoned: boolean + abortReason?: AbortReason + isInitialized: boolean + isStreaming: boolean + didFinishAbortingStream: boolean +} + +export type RunRejection = { event: RunEvent; state: RunState } + +export const MAX_RUN_REJECTIONS = 100 + +export const runStateFieldGetters: { [K in keyof RunStateFields]-?: (state: RunState) => RunStateFields[K] } = { + abort, + abandoned, + abortReason, + isInitialized, + isStreaming, + didFinishAbortingStream, +} + +// The kernel reports didFinishAbortingStream only for a live stream, so a leftover field (race R-1) projects false. +export function projectRunState(fields: RunStateFields): RunStateFields { + return { ...fields, didFinishAbortingStream: fields.isStreaming && fields.didFinishAbortingStream } +} + +export function applyRunEvent(state: RunState, event: RunEvent, rejections: RunRejection[]): RunState { + const result = nextRunState(state, event) + if ("ok" in result) { + return result.ok + } + rejections.push({ event, state }) + if (rejections.length > MAX_RUN_REJECTIONS) { + rejections.shift() + } + return state +} From 6eb65750de668f24599d009769d6e73364e91ff0 Mon Sep 17 00:00:00 2001 From: Elliott de Launay Date: Fri, 9 Oct 2026 22:01:30 -0400 Subject: [PATCH 2/2] refactor(task): read kernel fields through a function --- src/core/task/__tests__/Task.run-shadow.spec.ts | 11 +++++++++++ .../run-state/__tests__/runStateShadow.spec.ts | 15 ++------------- src/core/task/run-state/runStateShadow.ts | 16 ++++++++-------- 3 files changed, 21 insertions(+), 21 deletions(-) diff --git a/src/core/task/__tests__/Task.run-shadow.spec.ts b/src/core/task/__tests__/Task.run-shadow.spec.ts index 3c8e6a9ea9..4e0c2b169b 100644 --- a/src/core/task/__tests__/Task.run-shadow.spec.ts +++ b/src/core/task/__tests__/Task.run-shadow.spec.ts @@ -57,6 +57,17 @@ describe("Task run-state write helpers", () => { expect(task.runRejections).toEqual([]) }) + it("keeps a stale didFinishAbortingStream on the next stream so the kernel mismatch shows race R-1", () => { + task.markInitialized() + task.markStreamStarted() + task.markStreamCleanupFinished() + task.markStreamEnded() + task.markStreamStarted() + + expect(task.didFinishAbortingStream).toBe(true) + expect(task.runState.stream).toEqual({ tag: "live", generation: 2, cleanupFinished: false }) + }) + it("requestAbort(false) sets abort only", () => { task.requestAbort(false) diff --git a/src/core/task/run-state/__tests__/runStateShadow.spec.ts b/src/core/task/run-state/__tests__/runStateShadow.spec.ts index 40822243d6..fa7712ff77 100644 --- a/src/core/task/run-state/__tests__/runStateShadow.spec.ts +++ b/src/core/task/run-state/__tests__/runStateShadow.spec.ts @@ -3,7 +3,7 @@ import { applyRunEvent, MAX_RUN_REJECTIONS, projectRunState, - runStateFieldGetters, + readRunStateFields, type RunRejection, type RunStateFields, } from "../runStateShadow" @@ -26,17 +26,6 @@ function drive(events: RunEvent[]): RunState { }, initialRunState as RunState) } -function kernelFields(state: RunState): RunStateFields { - return { - abort: runStateFieldGetters.abort(state), - abandoned: runStateFieldGetters.abandoned(state), - abortReason: runStateFieldGetters.abortReason(state), - isInitialized: runStateFieldGetters.isInitialized(state), - isStreaming: runStateFieldGetters.isStreaming(state), - didFinishAbortingStream: runStateFieldGetters.didFinishAbortingStream(state), - } -} - describe("projectRunState", () => { it("keeps each field value in its own slot", () => { const fields: RunStateFields = { @@ -83,7 +72,7 @@ describe("projectRunState", () => { [completed, { ...idle, isInitialized: true }], ] for (const [state, fields] of cases) { - expect(projectRunState(fields)).toEqual(kernelFields(state)) + expect(projectRunState(fields)).toEqual(readRunStateFields(state)) } }) }) diff --git a/src/core/task/run-state/runStateShadow.ts b/src/core/task/run-state/runStateShadow.ts index f78291b1d2..1010108714 100644 --- a/src/core/task/run-state/runStateShadow.ts +++ b/src/core/task/run-state/runStateShadow.ts @@ -24,14 +24,14 @@ export type RunRejection = { event: RunEvent; state: RunState } export const MAX_RUN_REJECTIONS = 100 -export const runStateFieldGetters: { [K in keyof RunStateFields]-?: (state: RunState) => RunStateFields[K] } = { - abort, - abandoned, - abortReason, - isInitialized, - isStreaming, - didFinishAbortingStream, -} +export const readRunStateFields = (state: RunState): RunStateFields => ({ + abort: abort(state), + abandoned: abandoned(state), + abortReason: abortReason(state), + isInitialized: isInitialized(state), + isStreaming: isStreaming(state), + didFinishAbortingStream: didFinishAbortingStream(state), +}) // The kernel reports didFinishAbortingStream only for a live stream, so a leftover field (race R-1) projects false. export function projectRunState(fields: RunStateFields): RunStateFields {