feat(task): add runState field, write helpers, and projection (RSK-04)

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.

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<TaskEvents> 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<TaskEvents> 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<Ask, { tag: "none" }>): 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<void> {
 		if (isAbandoned) {
 			this.abandoned = true

src/core/task/__tests__/Task.run-shadow.spec.ts

@@ -0,0 +1,131 @@
+// 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("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)
+
+		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([])
+	})
+})

src/core/task/run-state/__tests__/runStateShadow.spec.ts

@@ -0,0 +1,104 @@
+import { initialRunState, nextRunState, type RunEvent, type RunState } from "../runState"
+import {
+	applyRunEvent,
+	MAX_RUN_REJECTIONS,
+	projectRunState,
+	readRunStateFields,
+	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)
+}
+
+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(readRunStateFields(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 })
+	})
+})

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 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 {
+	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
+}