diff --git a/.github/workflows/code-qa.yml b/.github/workflows/code-qa.yml index a936b0cd1b..7c71478266 100644 --- a/.github/workflows/code-qa.yml +++ b/.github/workflows/code-qa.yml @@ -90,6 +90,8 @@ jobs: run: pnpm lint - name: Check types run: pnpm check-types + - name: Model-check concurrent task lifecycle + run: pnpm lifecycle:model-check build-vsix: name: Build test VSIX diff --git a/AGENTS.md b/AGENTS.md index 9692463816..3b5be80ede 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -45,6 +45,13 @@ Prefer the narrowest test layer that proves the behavior. This follows standard - Keep e2e tests focused on high-value smoke coverage across boundaries. Avoid placing detailed protocol, parsing, storage, retry, or edge-case assertions in e2e when they can be covered reliably at a lower layer. - When fixing a regression, add the regression test at the lowest layer that would have failed for the bug. Add an e2e test only if lower-level tests cannot represent the failure mode. +## Task Lifecycle Changes + +- Read `docs/architecture/task-lifecycle-model.md` before changing task status, delegation, interruption, completion, abandonment, persistence ownership, or scheduler fan-out behavior. +- Keep lifecycle mutations in the shared reducers under `src/core/task-persistence/taskLifecycle.ts`. Update model actions, invariants, or named semantic landmarks for every new transition or concurrency bug class representable in the lifecycle model, then run `pnpm lifecycle:model-check`. +- Add extension-host E2E coverage only for a boundary the reducer model cannot prove, such as restart visibility, real persistence/rehydration, delayed provider streams, scheduler permits, or webview task scoping. Do not duplicate reducer interleavings in E2E. +- Run the focused lifecycle tests and `pnpm test` before completing a Zoo Code lifecycle change. + ## Shared Test Utilities - Use `src/test-utils/stream.ts` for mechanical async-stream setup and collection. diff --git a/docs/architecture/task-lifecycle-model.md b/docs/architecture/task-lifecycle-model.md new file mode 100644 index 0000000000..588ffd5204 --- /dev/null +++ b/docs/architecture/task-lifecycle-model.md @@ -0,0 +1,113 @@ +# Task lifecycle model check + +Zoo Code checks its persisted task delegation lifecycle with a bounded, exhaustive state explorer. Run it locally with: + +```sh +pnpm lifecycle:model-check +``` + +The check runs in the `compile` CI job after type checking. It fails if it finds an invariant violation, a modeled action becomes unreachable, or exploration exceeds its declared state budget. A violation includes the shortest breadth-first event trace, every intermediate state, and the active bounds so the sequence can be replayed as a focused regression test. + +## Why an executable TypeScript model + +The initial model uses a small explicit-state explorer rather than adding Quint, TLA+/TLC, or Alloy. This is deliberate: + +- Zoo's current risks are finite safety properties over a small persisted state machine, not yet temporal liveness or fairness properties. +- The explorer calls the production transition functions in `src/core/task-persistence/taskLifecycle.ts`. `ClineProvider` uses those same functions inside serialized and atomic store operations, reducing specification drift. +- Breadth-first exploration gives a deterministic, shortest-by-event counterexample with no Java or separate specification toolchain. +- Bounds and budget exhaustion are explicit. CI never reports a truncated exploration as a pass. + +This follows the same initial-state, next-state, reachable-state, invariant structure described by the [TLA+ high-level view](https://lamport.azurewebsites.net/tla/high-level-view.html) and [Quint's model-checker documentation](https://quint-lang.org/docs/model-checkers). The implementation connection is important: Quint's [model-based testing guidance](https://quint-lang.org/docs/model-based-testing) notes that checking a specification alone does not show that production code implements it. + +TLA+/PlusCal or Quint with TLC becomes a better fit when the lifecycle needs temporal properties, fairness assumptions, unbounded queues, or refinement between protocol layers. Alloy is better suited if relational ownership structure becomes harder than event ordering; Alloy analyses are explicitly bounded by scope, as described in the [Alloy tutorial](https://alloytools.org/tutorials/online/maintext-FS-1.html). Randomized model-based testing can complement, but not replace, the exhaustive bounded check when a production adapter is available; [fast-check documents command models](https://fast-check.dev/docs/advanced/model-based-testing/) and [controlled Promise scheduling](https://fast-check.dev/docs/advanced/race-conditions/). Jepsen-style history checking remains useful for distributed persistence behavior, but is heavier than this in-process lifecycle protocol; see Jepsen's [consistency model overview](https://jepsen.io/consistency). + +## Production mapping + +| Model concept | Production concept | +| ------------------------- | ------------------------------------------------------------------------------------ | +| Task record and status | `HistoryItem` persisted by `TaskHistoryStore` | +| `delegate(parent, child)` | `ClineProvider.delegateParentAndOpenChild` | +| `interrupt(child)` | cancellation or eviction through `markDelegatedChildInterrupted` | +| `complete(child)` | `ClineProvider.reopenParentFromDelegation` | +| `abandon(child)` | `ClineProvider.abandonSubtask` | +| Atomic event step | `atomicReadAndUpdate`, `atomicUpdatePair`, and per-parent delegation transition lock | +| Event interleaving | Competing completion, cancellation, abandonment, and new delegation calls | + +The model has three fixed task slots, enough to cover competing siblings and a nested parent-child-grandchild chain. It explores every reachable interleaving through depth 12, deduplicating canonical states. Representative checks also exercise rejected operations that do not create a new state: a second concurrent delegation while the first child is active, stale completion after re-delegation, late completion after abandonment, completion after interruption, and nested completion. Named semantic landmarks require the graph to retain interrupted-child re-delegation and nested delegation even when the raw state total changes. + +Production completion also accepts a recovery-compatible `active` parent that still awaits the returning child, then clears the stale pointers. Normal model transitions never create that intermediate state, so it is covered by a focused reducer test rather than admitted as a generally valid reachable state. + +## Shared-store concurrency model + +The same `pnpm lifecycle:model-check` command also runs a second bounded explorer over two `TaskHistoryStore` hosts. It imports the production `computeHistoryDelta` and `mergeHistoryDelta` functions, so its semantics match the store rather than assuming coherent caches or transactional pair writes: + +- each host has an independent cache and host-local mutex; +- store read/update operations hold the host mutex, while live-task snapshots used by completion and message saves may outlive it; +- a write delta is computed relative to that host's cache; +- revalidation under the per-file disk lock checks only status-transition legality; +- fields absent from the delta preserve the current disk value, `childIds` are unioned, and other same-field conflicts are last-writer-wins; +- `atomicUpdatePair` commits its files in order, with another host able to act between file commits; +- successful pair-operation cache entries publish together after both file writes; if the second write fails, the cache publishes only the first committed record; +- cache refresh is explicit and may occur after an external live-task snapshot was captured. + +There is no production record version or compare-and-swap token today. The model therefore does not invent one. It universally checks host-mutex and file-lock ownership, whole-file delta rejection, disk-field preservation, `childIds` union, and pair write order. Six scenarios, including distinct-task writes from #920 and a second-write pair failure, and all seven phases (`read`, `prepare`, `revalidate`, `commit`, `refresh`, `reject`, and `fail`) must remain reachable without exceeding the state/depth budgets. Positive semantic landmarks additionally require a stale cache beside newer disk state, the first pair write committed while the second is pending, and the same committed prefix retained after the second write fails. + +Two desired properties are currently false and remain issue-keyed shortest-witness ratchets rather than silently allowed assertion failures: + +- [#1469](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1469): an old completion can commit after a newer handoff and clear it because disk revalidation checks status legality, not exact-child ownership. +- [#1021](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1021): after abandonment and cache refresh, a stale live-task save can preserve the new interrupted status while restoring old lineage fields. + +CI fails if either exact causal witness or violation class changes, a witness disappears without being promoted to a universal invariant, a named semantic landmark or modeled phase becomes unreachable, a new safety violation appears, or exploration truncates. Raw reachable-state totals are printed as diagnostics, not used as ratchets: harmless representation changes can alter them without weakening protocol coverage. + +The known-unsafe witnesses currently compare exact shortest action sequences. This is intentionally simple and reviewable, but brittle to harmless action renames or serialization refactors. A causal partial-order comparator would reduce that brittleness but would add a second trace-equivalence protocol to maintain. Until that complexity is justified, update an exact witness only after confirming the terminal violation class and required causal ordering are unchanged. + +`TaskHistoryStore.realConcurrency.spec.ts` complements the abstract interleavings with one synchronized integration smoke check through the real `proper-lockfile` and filesystem rename path; broader VS Code E2E remains reserved for restart and extension-host behavior. + +## Invariants + +The checker currently enforces: + +1. A delegated parent has exactly one `awaitingChildId`, and `delegatedToId` matches it. +2. The awaited child exists, links back to the parent, is not completed, and remains in `childIds`. A delegated child may itself await a nested child. +3. Non-delegated parents retain no active delegation pointer. +4. Every active or delegated linked child is the child its parent currently awaits. An interrupted prior child may retain lineage after re-delegation but cannot complete back into that parent. +5. Parent-child lineage is acyclic. +6. Completed task records cannot be changed by later lifecycle events. +7. Active-child re-delegation, stale completion after ownership moves to another child, duplicate/late completion, and abandonment of a live child are rejected by the shared production guards. + +These are safety claims within the documented bounds. The check does not claim liveness, fairness, crash consistency, filesystem-lock correctness, API history correctness, or exhaustive coverage of arbitrary task counts. It also does not distinguish a delayed pre-interruption completion from a legitimate post-resume completion for the same child ID; that requires a persisted attempt/generation token before it can become a sound invariant. + +## Open-issue traceability + +The following map separates issue observations from the architectural interpretation encoded here. Open issues can change after this document is written; follow each link for current status. + +| Issue and directly observed evidence | Derived protocol rule | Production transition and current check | +| ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| [#1469](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1469): the issue report states that a barrier-controlled two-host run reproduced an old child completion clearing a newer handoff 25/25 times. | Completion is conditional on the parent still awaiting that exact child; a live-linked child must remain owned by its parent. | `completeDelegatedChild` rejects stale authoritative input. The lifecycle explorer checks that reducer rule, while the shared-store explorer reproduces the cross-host stale-cache counterexample with an exact causal witness. | +| [#1021](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1021): an in-flight `saveClineMessages` can restore parent/root IDs after abandonment cleared them. | Detachment should be monotonic: later lifecycle work must not reattach an abandoned child. | `abandonDelegatedChild` clears both sides. The shared-store explorer proves the detach commit occurs, then reproduces a refreshed-cache delta that preserves interrupted status while restoring stale live-task lineage. | +| [#1453](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1453), under user report [#1279](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1279): CI observed `TaskCompleted` before restart-visible API history once; 120 local repetitions did not reproduce it, while an Alloy abstraction permits the ordering. | A completion/readiness contract must define whether completion implies restart visibility. This is a liveness/durability boundary, not only a `HistoryItem` safety transition. | Not claimed by this checker. Add a controlled persistence barrier test after the contract decision; move to temporal model checking if eventual readiness and failure handling become protocol guarantees. | +| [#921](https://github.com/Zoo-Code-Org/Zoo-Code/issues/921): delegation across parallel tabs lacks coverage for different view-local mode/profile state. | Delegation must bind an explicit immutable execution-context snapshot rather than read whichever view is focused later. | The persisted ownership transition is covered; mode/profile snapshot isolation is outside this state model and belongs in a production adapter/model-based test. | +| [#920](https://github.com/Zoo-Code-Org/Zoo-Code/issues/920): issue analysis identifies a missing cross-instance history-update test and potential lost writes. | Distinct task writes must not overwrite one another, and same-task conflicts need an explicit merge/ownership rule. | The shared-store explorer checks distinct-task writes and same-record independent deltas. Cross-instance store tests retain production API coverage, and the synchronized real-filesystem smoke test exercises the actual lock/write path without claiming exhaustive filesystem proof. | +| [#369](https://github.com/Zoo-Code-Org/Zoo-Code/issues/369) and [#372](https://github.com/Zoo-Code-Org/Zoo-Code/issues/372): planned fan-out keeps a parent live while a child runs and requires completion routing by explicit parent ID, single-writer result readiness, permit release, and orphan cleanup. | Persisted `delegated` status is ownership, not proof that the parent instance is suspended. Completion must route by IDs; scheduler resources and live-instance state need separate invariants. | Nested and sibling lifecycle ownership are covered. Scheduler permits, live/suspended parent selection, orphan cancellation, and single-writer message readiness must be added when fan-out lands; they should not be folded into `HistoryItem` fields prematurely. | +| [#1468](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1468): a late chunk from one request combined tool identity with arguments from another request; rerun passed. | Every stream accumulator needs a request/task generation key, and late events cannot mutate another scope. | Separate protocol. It warrants a parser-scope model or deterministic interleaving test, not an unrelated field in the delegation model. | +| [#612](https://github.com/Zoo-Code-Org/Zoo-Code/issues/612): the CLI copied a status union and omitted `interrupted`. | Lifecycle vocabulary should have one type owner. | `HistoryItemStatus` is derived from `HistoryItem`, and production/checker transitions share `taskLifecycle.ts`; consumers should import rather than copy the union. | + +The issue-derived cases intentionally map to bug classes rather than issue-specific flags. In particular, stale event ownership, monotonic terminal/detached state, explicit scope, and single-writer boundaries generalize to future concurrent task work. + +## Extending the model + +When production lifecycle behavior changes: + +1. Define or update the pure transition in `taskLifecycle.ts`, then call it from the production operation. +2. Model the corresponding enabled event in `scripts/check-task-lifecycle.ts`. +3. Encode an invariant for the bug class, or a representative rejected-event scenario when the event intentionally leaves state unchanged. +4. Increase depth or task slots only when the new scenario requires it. Keep the state budget explicit and ensure CI completes quickly. +5. Convert any discovered counterexample into a focused production regression test as well as retaining the architectural invariant. + +Do not weaken bounds or remove an invariant merely to make CI pass. If state growth becomes difficult to control, split independent protocols or move the model to TLC/Quint with an implementation trace adapter rather than silently sampling the state space. + +## Test layering + +Keep reducer permutations in this model and focused Vitest suites. The real VS Code extension-host suite using a mocked provider in `apps/vscode-e2e/src/suite/subtasks.test.ts` already covers the boundaries the pure explorer cannot: task creation and rehydration, persisted parent-child state, cancellation during a delayed provider stream, interrupted-child resume, abandonment followed by a real resume/save/completion cycle, pending approvals across leave/return, and scheduler-driven resume. `restart-persistence.test.ts` separately verifies completion history through a fresh extension host. + +Add E2E coverage only when a lifecycle change crosses one of those runtime boundaries or introduces a new one. For example, #1453 persistence-readiness semantics require a controlled fresh-host test, and #369/#372 fan-out requires scheduler permit, live-parent routing, orphan cleanup, and task-scoping E2E. Do not add E2E cases solely to replay reducer orderings already exhausted here; they increase fixture and timing cost without strengthening the proof claim. diff --git a/package.json b/package.json index 2b90356386..d27f53bf21 100644 --- a/package.json +++ b/package.json @@ -12,6 +12,7 @@ "lint": "turbo lint --log-order grouped --output-logs new-only", "check-types": "turbo check-types --log-order grouped --output-logs new-only", "test": "turbo test --log-order grouped --output-logs new-only", + "lifecycle:model-check": "tsx scripts/check-task-lifecycle.ts && tsx scripts/check-task-store-concurrency.ts", "test:coverage": "turbo test:coverage --log-order grouped --output-logs new-only", "format": "turbo format --log-order grouped --output-logs new-only", "build": "turbo build --log-order grouped --output-logs new-only", diff --git a/scripts/check-task-lifecycle.ts b/scripts/check-task-lifecycle.ts new file mode 100644 index 0000000000..73e9078366 --- /dev/null +++ b/scripts/check-task-lifecycle.ts @@ -0,0 +1,287 @@ +import assert from "node:assert/strict" + +import type { HistoryItem } from "../packages/types/src/history" + +import { + abandonDelegatedChild, + completeDelegatedChild, + delegateTaskToChild, + interruptDelegatedChild, +} from "../src/core/task-persistence/taskLifecycle" + +const taskIds = ["parent", "child-a", "child-b"] as const +type TaskId = (typeof taskIds)[number] +type ModelState = Record + +interface Transition { + name: string + next: ModelState +} + +interface TraceStep { + action: string + state: ModelState +} + +const MAX_DEPTH = 12 +const MAX_STATES = 10_000 +const expectedActions = ["delegate", "interrupt", "complete", "abandon"] as const +const semanticLandmarks = { + "interrupted-child-redelegation": (state: ModelState) => + state.parent?.status === "delegated" && + state.parent.awaitingChildId === "child-b" && + state["child-a"]?.status === "interrupted", + "nested-delegation": (state: ModelState) => + state.parent?.status === "delegated" && + state.parent.awaitingChildId === "child-a" && + state["child-a"]?.status === "delegated" && + state["child-a"].awaitingChildId === "child-b", +} satisfies Record boolean> + +function task(id: TaskId, parentTaskId?: TaskId): HistoryItem { + return { + id, + number: taskIds.indexOf(id), + ts: taskIds.indexOf(id), + task: id, + tokensIn: 0, + tokensOut: 0, + totalCost: 0, + status: "active", + parentTaskId, + rootTaskId: parentTaskId ? "parent" : undefined, + childIds: [], + } +} + +function initialState(): ModelState { + return { parent: task("parent"), "child-a": undefined, "child-b": undefined } +} + +function replace(state: ModelState, ...updates: HistoryItem[]): ModelState { + const next = { ...state } + for (const update of updates) next[update.id as TaskId] = update + return next +} + +function transitions(state: ModelState): Transition[] { + const result: Transition[] = [] + for (const parentId of taskIds) { + const parent = state[parentId] + if (!parent) continue + + for (const childId of taskIds) { + if (childId === parentId || state[childId]) continue + const awaitedStatus = parent.awaitingChildId ? state[parent.awaitingChildId as TaskId]?.status : undefined + if (parent.status !== "active" && !(parent.status === "delegated" && awaitedStatus === "interrupted")) { + continue + } + const delegated = delegateTaskToChild(parent, childId, awaitedStatus) + result.push({ + name: `delegate(${parentId}, ${childId})`, + next: replace(state, delegated, task(childId, parentId)), + }) + } + } + + for (const childId of taskIds) { + const child = state[childId] + if (!child?.parentTaskId) continue + const parent = state[child.parentTaskId as TaskId] + if (!parent) continue + + if (parent.status === "delegated" && parent.awaitingChildId === child.id && child.status === "active") { + const interrupted = interruptDelegatedChild(parent, child) + result.push({ name: `interrupt(${childId})`, next: replace(state, interrupted) }) + } + + if ( + (parent.status === "delegated" || parent.status === "active") && + parent.awaitingChildId === child.id && + (child.status === "active" || child.status === "interrupted") + ) { + const completed = completeDelegatedChild(parent, child, `${childId} result`) + result.push({ + name: `complete(${childId})`, + next: replace(state, completed.parent, completed.child), + }) + } + + if (parent.status === "delegated" && parent.awaitingChildId === child.id && child.status === "interrupted") { + const abandoned = abandonDelegatedChild(parent, child) + result.push({ + name: `abandon(${childId})`, + next: replace(state, abandoned.parent, abandoned.child), + }) + } + } + return result +} + +function invariantViolations(state: ModelState): string[] { + const violations: string[] = [] + for (const id of taskIds) { + const current = state[id] + if (!current) continue + + if (current.status === "delegated") { + if (!current.awaitingChildId || current.delegatedToId !== current.awaitingChildId) { + violations.push(`${id}: delegated task must point to exactly one awaited child`) + continue + } + const child = state[current.awaitingChildId as TaskId] + if (!child || child.parentTaskId !== id || child.status === "completed") { + violations.push(`${id}: awaited child must exist, link back, and not be completed`) + } + if (!current.childIds?.includes(current.awaitingChildId)) { + violations.push(`${id}: awaited child must be retained in childIds`) + } + } else if (current.awaitingChildId || current.delegatedToId) { + violations.push(`${id}: only delegated tasks may retain an awaited-child pointer`) + } + + if (current.parentTaskId && current.status !== "interrupted") { + const parent = state[current.parentTaskId as TaskId] + if (current.status !== "completed" && parent?.awaitingChildId !== id) { + violations.push(`${id}: active or delegated linked child must be the child its parent awaits`) + } + } + + const ancestors = new Set([id]) + let cursor = current.parentTaskId + while (cursor) { + if (ancestors.has(cursor)) { + violations.push(`${id}: parentTaskId lineage must be acyclic`) + break + } + ancestors.add(cursor) + cursor = state[cursor as TaskId]?.parentTaskId + } + } + return violations +} + +function canonical(state: ModelState): string { + return JSON.stringify(taskIds.map((id) => state[id] ?? null)) +} + +function formatCounterexample(message: string, trace: TraceStep[]): string { + const steps = trace.map( + (step, index) => + `${index}. ${step.action}\n${JSON.stringify(step.state, null, 2) + .split("\n") + .map((line) => ` ${line}`) + .join("\n")}`, + ) + return [ + `Task lifecycle invariant failed: ${message}`, + `Bounds: depth=${MAX_DEPTH}, states=${MAX_STATES}`, + ...steps, + ].join("\n") +} + +function checkTransitionInvariants(previous: ModelState, transition: Transition): string[] { + const violations: string[] = [] + for (const id of taskIds) { + const before = previous[id] + const after = transition.next[id] + if (before?.status === "completed" && canonicalTask(before) !== canonicalTask(after)) { + violations.push(`${id}: completed task changed after ${transition.name}`) + } + } + return violations +} + +function canonicalTask(value: HistoryItem | undefined): string { + return JSON.stringify(value ?? null) +} + +function runModelCheck(): number { + const start = initialState() + const queue: Array<{ state: ModelState; trace: TraceStep[] }> = [ + { state: start, trace: [{ action: "initial", state: start }] }, + ] + const visited = new Set([canonical(start)]) + const reachedActions = new Set() + const reachedLandmarks = new Set() + const frontier: ModelState[] = [] + + for (let index = 0; index < queue.length; index++) { + const node = queue[index]! + for (const [name, predicate] of Object.entries(semanticLandmarks)) { + if (predicate(node.state)) reachedLandmarks.add(name) + } + const violations = invariantViolations(node.state) + if (violations.length) throw new Error(formatCounterexample(violations.join("; "), node.trace)) + if (node.trace.length - 1 === MAX_DEPTH) { + frontier.push(node.state) + continue + } + + for (const transition of transitions(node.state)) { + reachedActions.add(transition.name.slice(0, transition.name.indexOf("("))) + const transitionViolations = checkTransitionInvariants(node.state, transition) + const trace = [...node.trace, { action: transition.name, state: transition.next }] + if (transitionViolations.length) { + throw new Error(formatCounterexample(transitionViolations.join("; "), trace)) + } + const key = canonical(transition.next) + if (visited.has(key)) continue + visited.add(key) + queue.push({ state: transition.next, trace }) + if (visited.size > MAX_STATES) { + throw new Error( + `Task lifecycle exploration exceeded its ${MAX_STATES}-state budget; increase or reduce bounds`, + ) + } + } + } + const unreachableActions = expectedActions.filter((action) => !reachedActions.has(action)) + if (unreachableActions.length) { + throw new Error(`Task lifecycle model has unreachable actions: ${unreachableActions.join(", ")}`) + } + const missingLandmarks = Object.keys(semanticLandmarks).filter((name) => !reachedLandmarks.has(name)) + if (missingLandmarks.length) { + throw new Error(`Task lifecycle model has unreachable semantic landmarks: ${missingLandmarks.join(", ")}`) + } + const unexploredSuccessor = frontier + .flatMap((state) => transitions(state)) + .find((transition) => !visited.has(canonical(transition.next))) + if (unexploredSuccessor) { + throw new Error( + `Task lifecycle exploration reached depth ${MAX_DEPTH} with an unseen successor (${unexploredSuccessor.name}); increase the depth bound`, + ) + } + return visited.size +} + +function runRepresentativeScenarios(): void { + const parent = task("parent") + const childA = task("child-a", "parent") + const delegated = delegateTaskToChild(parent, childA.id) + + assert.throws(() => delegateTaskToChild(delegated, "child-b", "active"), /not interrupted/) + + const interruptedA = interruptDelegatedChild(delegated, childA) + const redelegated = delegateTaskToChild(delegated, "child-b", interruptedA.status) + assert.throws(() => completeDelegatedChild(redelegated, interruptedA, "stale"), /not delegated to child/) + + const abandoned = abandonDelegatedChild(delegated, interruptedA) + assert.throws(() => completeDelegatedChild(abandoned.parent, abandoned.child, "late"), /not delegated to child/) + + const childB = task("child-b", "child-a") + const nestedParent = delegateTaskToChild(childA, childB.id) + const nestedCompletion = completeDelegatedChild(nestedParent, childB, "nested result") + assert.equal(nestedCompletion.parent.status, "active") + assert.equal(nestedCompletion.parent.completedByChildId, childB.id) + + const interruptedCompletion = completeDelegatedChild(delegated, interruptedA, "resumed result") + assert.equal(interruptedCompletion.child.status, "completed") + assert.equal(interruptedCompletion.parent.status, "active") +} + +runRepresentativeScenarios() +const checkedStates = runModelCheck() +console.log( + `Task lifecycle model check passed: ${checkedStates} reachable states, ${expectedActions.length}/${expectedActions.length} actions reachable, ${Object.keys(semanticLandmarks).length}/${Object.keys(semanticLandmarks).length} landmarks reached, depth <= ${MAX_DEPTH}, ${taskIds.length} task slots`, +) diff --git a/scripts/check-task-store-concurrency.ts b/scripts/check-task-store-concurrency.ts new file mode 100644 index 0000000000..cf6f5f7d64 --- /dev/null +++ b/scripts/check-task-store-concurrency.ts @@ -0,0 +1,723 @@ +import type { HistoryItem } from "../packages/types/src/history" + +import { + abandonDelegatedChild, + completeDelegatedChild, + delegateTaskToChild, + interruptDelegatedChild, +} from "../src/core/task-persistence/taskLifecycle" +import { + computeHistoryDelta, + DeltaRejectedError, + mergeHistoryDelta, +} from "../src/core/task-persistence/taskStoreConcurrency" + +const hosts = ["A", "B"] as const +type Host = (typeof hosts)[number] +type TaskId = "parent" | "child-a" | "child-b" +type OperationId = + | "metadata-a" + | "metadata-b" + | "distinct-a" + | "distinct-b" + | "complete-a" + | "redelegate-b" + | "stale-save-a" + | "abandon-b" + | "reject-a" +type RecordMap = Partial> + +interface PreparedWrite { + taskId: TaskId + incoming: HistoryItem + delta: Partial +} + +interface OperationState { + phase: "idle" | "read" | "prepared" | "revalidated" | "done" | "rejected" | "failed" + snapshot?: RecordMap + writes?: PreparedWrite[] + writeIndex: number + candidate?: HistoryItem +} + +interface CommitEntry { + operationId: OperationId + taskId: TaskId + previous?: HistoryItem + delta: Partial + next: HistoryItem +} + +interface ModelState { + disk: RecordMap + caches: Record + hostMutexes: Partial> + locks: Partial> + operations: Partial> + commits: CommitEntry[] +} + +interface OperationSpec { + id: OperationId + host: Host + externalSnapshot?: boolean + allowRefreshAfterRead?: boolean + publishCacheAtEnd?: boolean + isEnabled?(snapshot: RecordMap): boolean + buildWrites(snapshot: RecordMap): HistoryItem[] +} + +interface Scenario { + name: string + operations: OperationSpec[] + targetViolation?: { issue: "#1469" | "#1021"; message: string; expectedActions: string[] } + check(state: ModelState): string[] +} + +interface TraceStep { + action: string + state: ModelState +} + +const MAX_DEPTH = 32 +const MAX_STATES = 100_000 +const commonInvariantNames = [ + "host mutex ownership", + "file lock ownership", + "disk field preservation", + "childIds union", + "pair write order", + "whole-delta rejection", +] as const +const expectedPhases = ["read", "prepare", "revalidate", "commit", "refresh", "reject", "fail"] as const +const semanticLandmarks = { + "stale-cache-newer-disk": (state: ModelState) => + state.commits.length > 0 && + hosts.some((host) => + (Object.keys(state.disk) as TaskId[]).some( + (taskId) => canonical(state.caches[host][taskId]) !== canonical(state.disk[taskId]), + ), + ), + "pair-first-commit-second-pending": (state: ModelState) => + (["complete-a", "abandon-b"] as OperationId[]).some((operationId) => { + const operation = state.operations[operationId] + return ( + operation?.writeIndex === 1 && + (operation.phase === "prepared" || operation.phase === "revalidated") && + state.commits.filter((entry) => entry.operationId === operationId).length === 1 + ) + }), + "pair-first-commit-second-failed": (state: ModelState) => + state.operations["complete-a"]?.phase === "failed" && + state.commits.filter((entry) => entry.operationId === "complete-a").length === 1 && + state.caches.A["child-a"]?.status === "completed" && + state.caches.A.parent?.status === "delegated", +} satisfies Record boolean> + +function item(id: TaskId, overrides: Partial = {}): HistoryItem { + return { + id, + number: id === "parent" ? 0 : id === "child-a" ? 1 : 2, + ts: id === "parent" ? 0 : id === "child-a" ? 1 : 2, + task: id, + tokensIn: 0, + tokensOut: 0, + totalCost: 0, + status: "active", + childIds: [], + ...overrides, + } +} + +function baseRecords(): RecordMap { + return { + parent: item("parent", { + status: "delegated", + awaitingChildId: "child-a", + delegatedToId: "child-a", + childIds: ["child-a"], + }), + "child-a": item("child-a", { parentTaskId: "parent", rootTaskId: "parent" }), + } +} + +function clone(value: T): T { + return structuredClone(value) +} + +function initialState(operationIds: OperationId[], disk = baseRecords()): ModelState { + return { + disk: clone(disk), + caches: { A: clone(disk), B: clone(disk) }, + hostMutexes: {}, + locks: {}, + operations: Object.fromEntries( + operationIds.map((id) => [id, { phase: "idle", writeIndex: 0 } satisfies OperationState]), + ), + commits: [], + } +} + +function getRequired(records: RecordMap, taskId: TaskId): HistoryItem { + const record = records[taskId] + if (!record) throw new Error(`Model setup is missing ${taskId}`) + return record +} + +const operationSpecs: Record = { + "metadata-a": { + id: "metadata-a", + host: "A", + buildWrites: (snapshot) => [{ ...getRequired(snapshot, "parent"), mode: "architect" }], + }, + "metadata-b": { + id: "metadata-b", + host: "B", + buildWrites: (snapshot) => [{ ...getRequired(snapshot, "parent"), totalCost: 42 }], + }, + "distinct-a": { + id: "distinct-a", + host: "A", + buildWrites: (snapshot) => [{ ...getRequired(snapshot, "parent"), mode: "architect" }], + }, + "distinct-b": { + id: "distinct-b", + host: "B", + buildWrites: (snapshot) => [{ ...getRequired(snapshot, "child-a"), totalCost: 42 }], + }, + "complete-a": { + id: "complete-a", + host: "A", + externalSnapshot: true, + publishCacheAtEnd: true, + isEnabled: (snapshot) => { + const parent = snapshot.parent + const child = snapshot["child-a"] + return ( + (parent?.status === "delegated" || parent?.status === "active") && + parent.awaitingChildId === "child-a" && + (child?.status === "active" || child?.status === "interrupted") + ) + }, + buildWrites: (snapshot) => { + const completed = completeDelegatedChild( + getRequired(snapshot, "parent"), + getRequired(snapshot, "child-a"), + "child-a result", + ) + return [completed.child, completed.parent] + }, + }, + "redelegate-b": { + id: "redelegate-b", + host: "B", + isEnabled: (snapshot) => + snapshot.parent?.status === "delegated" && + snapshot.parent.awaitingChildId === "child-a" && + snapshot["child-a"]?.status === "active", + buildWrites: (snapshot) => { + const parent = getRequired(snapshot, "parent") + const interrupted = interruptDelegatedChild(parent, getRequired(snapshot, "child-a")) + const delegated = delegateTaskToChild(parent, "child-b", "interrupted") + return [interrupted, item("child-b", { parentTaskId: "parent", rootTaskId: "parent" }), delegated] + }, + }, + "stale-save-a": { + id: "stale-save-a", + host: "A", + externalSnapshot: true, + allowRefreshAfterRead: true, + buildWrites: (snapshot) => { + const stale = getRequired(snapshot, "child-a") + return [{ ...stale, tokensOut: stale.tokensOut + 1 }] + }, + }, + "abandon-b": { + id: "abandon-b", + host: "B", + publishCacheAtEnd: true, + isEnabled: (snapshot) => + snapshot.parent?.status === "delegated" && + snapshot.parent.awaitingChildId === "child-a" && + snapshot["child-a"]?.status === "active", + buildWrites: (snapshot) => { + const parent = getRequired(snapshot, "parent") + const interrupted = interruptDelegatedChild(parent, getRequired(snapshot, "child-a")) + const abandoned = abandonDelegatedChild(parent, interrupted) + return [abandoned.child, abandoned.parent] + }, + }, + "reject-a": { + id: "reject-a", + host: "A", + buildWrites: (snapshot) => [ + { ...getRequired(snapshot, "parent"), status: "interrupted", mode: "must-not-commit" }, + ], + }, +} + +function prepareWrites(state: ModelState, spec: OperationSpec, operation: OperationState): PreparedWrite[] { + return spec.buildWrites(operation.snapshot!).map((built) => { + const taskId = built.id as TaskId + const cached = state.caches[spec.host][taskId] + // Task.saveClineMessages rebuilds lineage from the live Task but preserves the + // store's current status before upsert, so stale lineage is not accompanied by + // a stale status transition. + const incoming = spec.id === "stale-save-a" && cached?.status ? { ...built, status: cached.status } : built + return { + taskId, + incoming, + delta: cached ? { id: taskId, ...computeHistoryDelta(cached, incoming) } : { ...incoming }, + } + }) +} + +function transition(state: ModelState, action: string, mutate: (next: ModelState) => void): TraceStep { + const next = clone(state) + mutate(next) + return { action, state: next } +} + +function nextSteps(state: ModelState, scenario: Scenario): TraceStep[] { + const result: TraceStep[] = [] + for (const spec of scenario.operations) { + const operation = state.operations[spec.id]! + if ( + operation.phase === "idle" && + !state.hostMutexes[spec.host] && + (spec.isEnabled?.(state.caches[spec.host]) ?? true) + ) { + result.push( + transition(state, `${spec.id}.read`, (next) => { + const target = next.operations[spec.id]! + if (!spec.externalSnapshot) next.hostMutexes[spec.host] = spec.id + target.phase = "read" + target.snapshot = clone(next.caches[spec.host]) + }), + ) + } else if ( + operation.phase === "read" && + (spec.externalSnapshot ? !state.hostMutexes[spec.host] : state.hostMutexes[spec.host] === spec.id) + ) { + result.push( + transition(state, `${spec.id}.prepare`, (next) => { + const target = next.operations[spec.id]! + if (spec.externalSnapshot) next.hostMutexes[spec.host] = spec.id + target.writes = prepareWrites(next, spec, target) + target.phase = "prepared" + }), + ) + } else if (operation.phase === "prepared") { + const write = operation.writes![operation.writeIndex]! + if (!state.locks[write.taskId]) { + result.push( + transition(state, `${spec.id}.revalidate(${write.taskId})`, (next) => { + const target = next.operations[spec.id]! + const targetWrite = target.writes![target.writeIndex]! + next.locks[targetWrite.taskId] = spec.id + try { + target.candidate = mergeHistoryDelta( + next.disk[targetWrite.taskId], + targetWrite.incoming, + targetWrite.delta, + ) + target.phase = "revalidated" + } catch (error) { + if (!(error instanceof DeltaRejectedError)) throw error + target.phase = "rejected" + delete next.locks[targetWrite.taskId] + delete next.hostMutexes[spec.host] + } + }), + ) + } + } else if (operation.phase === "revalidated") { + const write = operation.writes![operation.writeIndex]! + result.push( + transition(state, `${spec.id}.commit(${write.taskId})`, (next) => { + const target = next.operations[spec.id]! + const targetWrite = target.writes![target.writeIndex]! + if (next.locks[targetWrite.taskId] !== spec.id || !target.candidate) { + throw new Error(`${spec.id} committed without owning ${targetWrite.taskId}`) + } + const previous = next.disk[targetWrite.taskId] + next.disk[targetWrite.taskId] = target.candidate + if (!spec.publishCacheAtEnd) next.caches[spec.host][targetWrite.taskId] = target.candidate + next.commits.push({ + operationId: spec.id, + taskId: targetWrite.taskId, + previous, + delta: targetWrite.delta, + next: target.candidate, + }) + delete next.locks[targetWrite.taskId] + target.candidate = undefined + target.writeIndex++ + target.phase = target.writeIndex === target.writes!.length ? "done" : "prepared" + if (target.phase === "done") { + if (spec.publishCacheAtEnd) { + for (const commit of next.commits.filter((entry) => entry.operationId === spec.id)) { + next.caches[spec.host][commit.taskId] = commit.next + } + } + delete next.hostMutexes[spec.host] + } + }), + ) + if (spec.publishCacheAtEnd && operation.writeIndex > 0) { + result.push( + transition(state, `${spec.id}.fail(${write.taskId})`, (next) => { + const target = next.operations[spec.id]! + const targetWrite = target.writes![target.writeIndex]! + if (next.locks[targetWrite.taskId] !== spec.id) { + throw new Error(`${spec.id} failed without owning ${targetWrite.taskId}`) + } + for (const commit of next.commits.filter((entry) => entry.operationId === spec.id)) { + next.caches[spec.host][commit.taskId] = commit.next + } + target.candidate = undefined + target.phase = "failed" + delete next.locks[targetWrite.taskId] + delete next.hostMutexes[spec.host] + }), + ) + } + } + } + + for (const host of hosts) { + const hostHasPreparedWork = + Boolean(state.hostMutexes[host]) || + scenario.operations.some((spec) => { + const operation = state.operations[spec.id]! + return ( + spec.host === host && + (["prepared", "revalidated"].includes(operation.phase) || + (operation.phase === "read" && !spec.allowRefreshAfterRead)) + ) + }) + if (!hostHasPreparedWork && canonical(state.caches[host]) !== canonical(state.disk)) { + result.push( + transition(state, `${host}.refresh`, (next) => { + next.caches[host] = clone(next.disk) + }), + ) + } + } + return result +} + +function commonViolations(state: ModelState, scenario: Scenario): string[] { + const violations: string[] = [] + for (const [host, owner] of Object.entries(state.hostMutexes) as Array<[Host, OperationId]>) { + const operation = state.operations[owner] + if (!operation || !["read", "prepared", "revalidated"].includes(operation.phase)) { + violations.push(`${owner} holds host ${host} mutex outside its write phase`) + } + } + for (const [taskId, owner] of Object.entries(state.locks) as Array<[TaskId, OperationId]>) { + const operation = state.operations[owner] + if (operation?.phase !== "revalidated" || operation.writes?.[operation.writeIndex]?.taskId !== taskId) { + violations.push(`${owner} holds ${taskId} without a revalidated write`) + } + } + for (const commit of state.commits) { + if (commit.previous) { + for (const [key, value] of Object.entries(commit.previous)) { + if (!(key in commit.delta) && !deepEqual(value, commit.next[key as keyof HistoryItem])) { + violations.push(`${commit.operationId} lost disk field ${key} absent from its delta`) + } + } + if (commit.delta.childIds && commit.previous.childIds) { + const expected = new Set([...commit.previous.childIds, ...commit.delta.childIds]) + if ([...expected].some((id) => !commit.next.childIds?.includes(id))) { + violations.push(`${commit.operationId} lost a concurrent childIds entry`) + } + } + } + } + for (const spec of scenario.operations) { + const operation = state.operations[spec.id]! + const committed = state.commits.filter((entry) => entry.operationId === spec.id) + const expectedOrder = operation.writes?.slice(0, committed.length).map((write) => write.taskId) ?? [] + if (committed.some((entry, index) => entry.taskId !== expectedOrder[index])) { + violations.push(`${spec.id} committed pair records out of production order`) + } + if (operation.phase === "rejected" && committed.length > operation.writeIndex) { + violations.push(`${spec.id} committed a rejected file delta`) + } + } + return violations +} + +function deepEqual(left: unknown, right: unknown): boolean { + return canonical(left) === canonical(right) +} + +function canonical(value: unknown): string { + return JSON.stringify(value) +} + +function phaseName(action: string): string { + if (action.endsWith(".read")) return "read" + if (action.endsWith(".prepare")) return "prepare" + if (action.includes(".revalidate(")) return "revalidate" + if (action.includes(".commit(")) return "commit" + if (action.includes(".fail(")) return "fail" + if (action.endsWith(".refresh")) return "refresh" + return "reject" +} + +function formatTrace(scenario: Scenario, message: string, trace: TraceStep[]): string { + return [ + `Shared-store model violation in ${scenario.name}: ${message}`, + `Bounds: depth=${MAX_DEPTH}, states=${MAX_STATES}`, + ...trace.map((step, index) => `${index}. ${step.action}\n${JSON.stringify(step.state, null, 2)}`), + ].join("\n") +} + +function targetViolation(state: ModelState, scenario: Scenario): string | undefined { + if (scenario.targetViolation?.issue === "#1469") { + const completionDone = state.operations["complete-a"]?.phase === "done" + const redelegationDone = state.operations["redelegate-b"]?.phase === "done" + const redelegationParentCommit = state.commits.findIndex( + (entry) => entry.operationId === "redelegate-b" && entry.taskId === "parent", + ) + const completionParentCommit = state.commits.findIndex( + (entry) => entry.operationId === "complete-a" && entry.taskId === "parent", + ) + const parent = state.disk.parent + const child = state.disk["child-b"] + if ( + completionDone && + redelegationDone && + redelegationParentCommit >= 0 && + redelegationParentCommit < completionParentCommit && + child?.status === "active" && + child.parentTaskId === "parent" + ) { + if (parent?.status !== "delegated" || parent.awaitingChildId !== "child-b") { + return scenario.targetViolation.message + } + } + } + if (scenario.targetViolation?.issue === "#1021") { + const abandonDone = state.operations["abandon-b"]?.phase === "done" + const staleSaveDone = state.operations["stale-save-a"]?.phase === "done" + const detachCommit = state.commits.findIndex( + (entry) => + entry.operationId === "abandon-b" && + entry.taskId === "child-a" && + entry.next.parentTaskId === undefined && + entry.next.rootTaskId === undefined, + ) + const reattachCommit = state.commits.findIndex( + (entry) => + entry.operationId === "stale-save-a" && + entry.taskId === "child-a" && + entry.previous?.parentTaskId === undefined && + entry.next.parentTaskId === "parent", + ) + if (abandonDone && staleSaveDone && detachCommit >= 0 && detachCommit < reattachCommit) { + return scenario.targetViolation.message + } + } + return undefined +} + +function runScenario(scenario: Scenario): { + states: number + witness?: TraceStep[] + phases: Set + landmarks: Set +} { + const startDisk = + scenario.name === "status rejection" + ? { parent: item("parent", { status: "completed", mode: "stable" }) } + : baseRecords() + const start = initialState( + scenario.operations.map((operation) => operation.id), + startDisk, + ) + const queue: Array<{ state: ModelState; trace: TraceStep[] }> = [ + { state: start, trace: [{ action: "initial", state: start }] }, + ] + const visited = new Set([canonical(start)]) + const frontier: ModelState[] = [] + const phases = new Set() + const landmarks = new Set() + let witness: TraceStep[] | undefined + + for (let index = 0; index < queue.length; index++) { + const node = queue[index]! + for (const [name, predicate] of Object.entries(semanticLandmarks)) { + if (predicate(node.state)) landmarks.add(name) + } + const violations = [...commonViolations(node.state, scenario), ...scenario.check(node.state)] + if (violations.length) throw new Error(formatTrace(scenario, violations.join("; "), node.trace)) + const expectedViolation = targetViolation(node.state, scenario) + if (expectedViolation && !witness) witness = node.trace + if (node.trace.length - 1 === MAX_DEPTH) { + frontier.push(node.state) + continue + } + + for (const step of nextSteps(node.state, scenario)) { + phases.add(phaseName(step.action)) + if (step.state.operations["reject-a"]?.phase === "rejected") phases.add("reject") + const key = canonical(step.state) + if (visited.has(key)) continue + visited.add(key) + queue.push({ state: step.state, trace: [...node.trace, step] }) + if (visited.size > MAX_STATES) throw new Error(`${scenario.name} exceeded ${MAX_STATES} states`) + } + } + + if (scenario.targetViolation && !witness) { + throw new Error( + `${scenario.name} no longer reproduces ${scenario.targetViolation.issue}; promote it to an invariant`, + ) + } + const unseen = frontier + .flatMap((state) => nextSteps(state, scenario)) + .find((step) => !visited.has(canonical(step.state))) + if (unseen) throw new Error(`${scenario.name} truncated before unseen action ${unseen.action}`) + return { states: visited.size, witness, phases, landmarks } +} + +const scenarios: Scenario[] = [ + { + name: "peer field merge", + operations: [operationSpecs["metadata-a"], operationSpecs["metadata-b"]], + check: (state) => { + if (state.operations["metadata-a"]?.phase !== "done" || state.operations["metadata-b"]?.phase !== "done") { + return [] + } + return state.disk.parent?.mode === "architect" && state.disk.parent.totalCost === 42 + ? [] + : ["concurrent writes to different fields lost an update"] + }, + }, + { + name: "status rejection", + operations: [operationSpecs["reject-a"]], + check: (state) => { + if (state.operations["reject-a"]?.phase !== "rejected") return [] + return state.disk.parent?.status === "completed" && state.disk.parent.mode === "stable" + ? [] + : ["rejected status delta applied companion fields"] + }, + }, + { + name: "pair second-write failure", + operations: [operationSpecs["complete-a"]], + check: (state) => { + if (state.operations["complete-a"]?.phase !== "failed") return [] + return state.disk["child-a"]?.status === "completed" && + state.disk.parent?.status === "delegated" && + state.caches.A["child-a"]?.status === "completed" && + state.caches.A.parent?.status === "delegated" + ? [] + : ["pair failure cache did not reflect the committed first-record prefix"] + }, + }, + { + name: "distinct task writes (#920)", + operations: [operationSpecs["distinct-a"], operationSpecs["distinct-b"]], + check: (state) => { + if (state.operations["distinct-a"]?.phase !== "done" || state.operations["distinct-b"]?.phase !== "done") { + return [] + } + return state.disk.parent?.mode === "architect" && state.disk["child-a"]?.totalCost === 42 + ? [] + : ["#920 distinct task writes lost an entry"] + }, + }, + { + name: "stale completion ownership", + operations: [operationSpecs["complete-a"], operationSpecs["redelegate-b"]], + targetViolation: { + issue: "#1469", + message: "stale child completion cleared a newer parent handoff", + expectedActions: [ + "complete-a.read", + "complete-a.prepare", + "redelegate-b.read", + "redelegate-b.prepare", + "redelegate-b.revalidate(child-a)", + "redelegate-b.commit(child-a)", + "complete-a.revalidate(child-a)", + "complete-a.commit(child-a)", + "redelegate-b.revalidate(child-b)", + "redelegate-b.commit(child-b)", + "redelegate-b.revalidate(parent)", + "redelegate-b.commit(parent)", + "complete-a.revalidate(parent)", + "complete-a.commit(parent)", + ], + }, + check: () => [], + }, + { + name: "stale save detachment", + operations: [operationSpecs["stale-save-a"], operationSpecs["abandon-b"]], + targetViolation: { + issue: "#1021", + message: "stale live-task save reattached abandoned lineage", + expectedActions: [ + "stale-save-a.read", + "abandon-b.read", + "abandon-b.prepare", + "abandon-b.revalidate(child-a)", + "abandon-b.commit(child-a)", + "abandon-b.revalidate(parent)", + "abandon-b.commit(parent)", + "A.refresh", + "stale-save-a.prepare", + "stale-save-a.revalidate(child-a)", + "stale-save-a.commit(child-a)", + ], + }, + check: () => [], + }, +] + +let totalStates = 0 +const reachedPhases = new Set() +const reachedLandmarks = new Set() +for (const scenario of scenarios) { + const result = runScenario(scenario) + totalStates += result.states + for (const phase of result.phases) reachedPhases.add(phase) + for (const landmark of result.landmarks) reachedLandmarks.add(landmark) + if (scenario.targetViolation) { + const actions = result.witness!.slice(1).map((step) => step.action) + if (canonical(actions) !== canonical(scenario.targetViolation.expectedActions)) { + throw new Error( + formatTrace( + scenario, + `${scenario.targetViolation.issue} shortest causal witness changed`, + result.witness!, + ), + ) + } + console.log( + `Known unsafe ${scenario.targetViolation.issue}: ${scenario.targetViolation.message}\n ${result + .witness!.slice(1) + .map((step) => step.action) + .join(" -> ")}`, + ) + } +} + +const missingPhases = expectedPhases.filter((phase) => !reachedPhases.has(phase)) +if (missingPhases.length) throw new Error(`Shared-store model has unreachable phases: ${missingPhases.join(", ")}`) +const missingLandmarks = Object.keys(semanticLandmarks).filter((name) => !reachedLandmarks.has(name)) +if (missingLandmarks.length) { + throw new Error(`Shared-store model has unreachable semantic landmarks: ${missingLandmarks.join(", ")}`) +} + +console.log( + `Shared-store model check passed: ${totalStates} states, ${scenarios.length} scenarios, ${commonInvariantNames.length} invariants, ${expectedPhases.length}/${expectedPhases.length} phases reachable, ${Object.keys(semanticLandmarks).length}/${Object.keys(semanticLandmarks).length} landmarks reached`, +) diff --git a/src/__tests__/history-resume-delegation.spec.ts b/src/__tests__/history-resume-delegation.spec.ts index e830798b16..d3a24a3140 100644 --- a/src/__tests__/history-resume-delegation.spec.ts +++ b/src/__tests__/history-resume-delegation.spec.ts @@ -1370,6 +1370,7 @@ describe("History resume delegation - parent metadata transitions", () => { parentTaskId: "parent-566", historyItem: { parentTaskId: "parent-566" }, providerRef: { deref: () => provider }, + ask: vi.fn().mockResolvedValue({ response: "yesButtonClicked", text: "", images: [] }), say: vi.fn().mockResolvedValue(undefined), emit: vi.fn(), getTokenUsage: vi.fn(() => ({})), diff --git a/src/core/task-persistence/TaskHistoryStore.ts b/src/core/task-persistence/TaskHistoryStore.ts index 075d21474f..3d4cc47604 100644 --- a/src/core/task-persistence/TaskHistoryStore.ts +++ b/src/core/task-persistence/TaskHistoryStore.ts @@ -9,66 +9,18 @@ import type { HistoryItem } from "@roo-code/types" import { GlobalFileNames } from "../../shared/globalFileNames" import { LOCK_STALE_MS, safeWriteJson } from "../../utils/safeWriteJson" import { getStorageBasePath } from "../../utils/storage" +import { assertValidTransition, type HistoryItemStatus } from "./taskLifecycle" +import { computeHistoryDelta, DeltaRejectedError, mergeHistoryDelta } from "./taskStoreConcurrency" -/** Valid status values for a task's HistoryItem. */ -export type HistoryItemStatus = NonNullable - -export class DeltaRejectedError extends Error { - constructor( - public readonly taskId: string, - public readonly diskStatus: HistoryItemStatus, - public readonly attemptedStatus: HistoryItemStatus, - ) { - super(`Delta rejected for task ${taskId}: disk status ${diskStatus} rejects transition to ${attemptedStatus}`) - this.name = "DeltaRejectedError" - } -} - -const VALID_TRANSITIONS: Record = { - active: ["delegated", "completed", "interrupted"], - delegated: ["active"], - interrupted: ["completed"], - completed: [], -} - -/** - * Asserts that a task status transition is valid, throwing if not. - * - * @throws {Error} When the transition is not allowed by the state machine. - */ -export function assertValidTransition(from: HistoryItemStatus | undefined, to: HistoryItemStatus): void { - const fromStatus: HistoryItemStatus = from ?? "active" - const validTargets = VALID_TRANSITIONS[fromStatus] - if (!validTargets.includes(to)) { - throw new Error(`Invalid task status transition: ${fromStatus} → ${to}`) - } -} +export { assertValidTransition, type HistoryItemStatus } from "./taskLifecycle" +export { DeltaRejectedError } from "./taskStoreConcurrency" /** * Build a `safeWriteJson` merge callback that applies only `delta` to the * current disk state, preserving fields written by another process. */ function mergeWithDisk(delta: Partial): (existing: unknown, incoming: unknown) => unknown { - return (existing, incoming) => { - if (!existing || typeof existing !== "object" || !("id" in existing)) { - return incoming - } - const disk = existing as HistoryItem - if (delta.status !== undefined) { - const diskStatus: HistoryItemStatus = disk.status ?? "active" - if (delta.status !== diskStatus) { - const validTargets = VALID_TRANSITIONS[diskStatus] - if (!validTargets?.includes(delta.status as HistoryItemStatus)) { - throw new DeltaRejectedError(disk.id, diskStatus, delta.status as HistoryItemStatus) - } - } - } - const merged = { ...disk, ...delta } - if (delta.childIds && disk.childIds) { - merged.childIds = [...new Set([...disk.childIds, ...delta.childIds])] - } - return merged - } + return (existing, incoming) => mergeHistoryDelta(existing, incoming as HistoryItem, delta) } /** @@ -882,9 +834,7 @@ export class TaskHistoryStore { * Return only the fields in `incoming` that differ from `cached`. */ private computeDelta(cached: HistoryItem, incoming: Partial): Partial { - return Object.fromEntries( - Object.entries(incoming).filter(([k, v]) => !deepEqual(v, (cached as Record)[k])), - ) as Partial + return computeHistoryDelta(cached, incoming) } private buildDelta(id: string, cached: HistoryItem, incoming: Partial): Partial { diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.realConcurrency.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.realConcurrency.spec.ts new file mode 100644 index 0000000000..d94ca8f782 --- /dev/null +++ b/src/core/task-persistence/__tests__/TaskHistoryStore.realConcurrency.spec.ts @@ -0,0 +1,127 @@ +import * as fs from "fs/promises" +import * as os from "os" +import * as path from "path" + +import type { HistoryItem } from "@roo-code/types" + +import { TaskHistoryStore } from "../TaskHistoryStore" + +type WriteTaskFile = (item: HistoryItem, delta?: Partial) => Promise + +interface WriteBarrier { + arrivals(): number + dispose(): void +} + +function synchronizeNextWrites(stores: TaskHistoryStore[], timeoutMs = 2_000): WriteBarrier { + let arrivals = 0 + let release!: () => void + let rejectBarrier!: (error: Error) => void + let settled = false + let timer: ReturnType | undefined + const barrier = new Promise((resolve, reject) => { + rejectBarrier = reject + release = () => { + if (settled) return + settled = true + if (timer) clearTimeout(timer) + resolve() + } + timer = setTimeout(() => { + if (settled) return + settled = true + reject(new Error(`Only ${arrivals}/${stores.length} stores reached writeTaskFile within ${timeoutMs}ms`)) + }, timeoutMs) + }) + void barrier.catch(() => {}) + + for (const store of stores) { + const value: unknown = Reflect.get(store, "writeTaskFile") + if (typeof value !== "function") throw new Error("TaskHistoryStore.writeTaskFile is unavailable") + const original = value.bind(store) as WriteTaskFile + Reflect.set(store, "writeTaskFile", async (historyItem: HistoryItem, delta?: Partial) => { + arrivals++ + if (arrivals === stores.length) release() + await barrier + return original(historyItem, delta) + }) + } + + return { + arrivals: () => arrivals, + dispose: () => { + if (settled) return + settled = true + if (timer) clearTimeout(timer) + rejectBarrier(new Error("Write barrier disposed before all stores arrived")) + }, + } +} + +function item(id: string): HistoryItem { + return { + id, + number: 1, + ts: 1, + task: id, + tokensIn: 0, + tokensOut: 0, + totalCost: 0, + status: "active", + childIds: [], + } +} + +describe("TaskHistoryStore real cross-host locking", () => { + it("preserves independent stale-cache deltas through the real per-file lock", async () => { + const storagePath = await fs.mkdtemp(path.join(os.tmpdir(), "task-history-real-lock-")) + const storeA = new TaskHistoryStore(storagePath) + const storeB = new TaskHistoryStore(storagePath) + let writeBarrier: WriteBarrier | undefined + + try { + await storeA.initialize() + await storeA.upsert(item("shared-task")) + await storeB.initialize() + writeBarrier = synchronizeNextWrites([storeA, storeB]) + + await Promise.all([ + storeA.atomicReadAndUpdate("shared-task", (current) => ({ ...current, mode: "architect" })), + storeB.atomicReadAndUpdate("shared-task", (current) => ({ ...current, totalCost: 42 })), + ]) + + expect(writeBarrier.arrivals()).toBe(2) + await storeA.invalidate("shared-task") + expect(storeA.get("shared-task")).toMatchObject({ mode: "architect", totalCost: 42 }) + } finally { + writeBarrier?.dispose() + storeA.dispose() + storeB.dispose() + await fs.rm(storagePath, { recursive: true, force: true }) + } + }) + + it("reports a bounded error when one store never reaches the write barrier", async () => { + const storagePath = await fs.mkdtemp(path.join(os.tmpdir(), "task-history-missed-barrier-")) + const storeA = new TaskHistoryStore(storagePath) + const storeB = new TaskHistoryStore(storagePath) + let writeBarrier: WriteBarrier | undefined + + try { + await storeA.initialize() + await storeA.upsert(item("shared-task")) + await storeB.initialize() + writeBarrier = synchronizeNextWrites([storeA, storeB], 50) + + await expect( + storeA.atomicReadAndUpdate("shared-task", (current) => ({ ...current, mode: "architect" })), + ).rejects.toThrow("Only 1/2 stores reached writeTaskFile within 50ms") + expect(writeBarrier.arrivals()).toBe(1) + } finally { + writeBarrier?.dispose() + storeA.dispose() + storeB.dispose() + await fs.rm(storagePath, { recursive: true, force: true }) + } + }) +}) diff --git a/src/core/task-persistence/__tests__/taskLifecycle.spec.ts b/src/core/task-persistence/__tests__/taskLifecycle.spec.ts new file mode 100644 index 0000000000..fe415f09f8 --- /dev/null +++ b/src/core/task-persistence/__tests__/taskLifecycle.spec.ts @@ -0,0 +1,104 @@ +import type { HistoryItem } from "@roo-code/types" + +import { + abandonDelegatedChild, + completeDelegatedChild, + delegateTaskToChild, + interruptDelegatedChild, +} from "../taskLifecycle" + +function item(id: string, overrides: Partial = {}): HistoryItem { + return { + id, + number: 1, + ts: 1, + task: id, + tokensIn: 0, + tokensOut: 0, + totalCost: 0, + status: "active", + ...overrides, + } +} + +describe("task lifecycle transitions", () => { + it("delegates an active parent and retains child history", () => { + const parent = delegateTaskToChild(item("parent", { childIds: ["older"] }), "child") + + expect(parent).toMatchObject({ + status: "delegated", + awaitingChildId: "child", + delegatedToId: "child", + childIds: ["older", "child"], + }) + }) + + it("treats a legacy unset status as active when delegating", () => { + expect(delegateTaskToChild(item("parent", { status: undefined }), "child")).toMatchObject({ + status: "delegated", + awaitingChildId: "child", + delegatedToId: "child", + }) + }) + + it("allows re-delegation only after the previous child is interrupted", () => { + const parent = item("parent", { + status: "delegated", + awaitingChildId: "old-child", + delegatedToId: "old-child", + childIds: ["old-child"], + }) + + expect(() => delegateTaskToChild(parent, "new-child", "active")).toThrow(/not interrupted/) + expect(delegateTaskToChild(parent, "new-child", "interrupted")).toMatchObject({ + status: "delegated", + awaitingChildId: "new-child", + childIds: ["old-child", "new-child"], + }) + }) + + it("interrupts a child without clearing the parent's ownership", () => { + const parent = item("parent", { status: "delegated", awaitingChildId: "child", delegatedToId: "child" }) + const child = item("child", { parentTaskId: "parent" }) + + expect(interruptDelegatedChild(parent, child)).toMatchObject({ status: "interrupted", parentTaskId: "parent" }) + }) + + it("completes only the child the parent still awaits", () => { + const parent = item("parent", { status: "delegated", awaitingChildId: "new-child", delegatedToId: "new-child" }) + const staleChild = item("old-child", { status: "interrupted", parentTaskId: "parent" }) + + expect(() => completeDelegatedChild(parent, staleChild, "stale result")).toThrow(/not delegated to child/) + + const child = item("new-child", { parentTaskId: "parent" }) + const completed = completeDelegatedChild(parent, child, "result") + expect(completed.child.status).toBe("completed") + expect(completed.parent).toMatchObject({ + status: "active", + completedByChildId: "new-child", + awaitingChildId: undefined, + }) + }) + + it("repairs an active parent that still awaits the returning child", () => { + const parent = item("parent", { status: "active", awaitingChildId: "child", delegatedToId: "child" }) + const child = item("child", { status: "interrupted", parentTaskId: "parent" }) + + expect(completeDelegatedChild(parent, child, "result").parent).toMatchObject({ + status: "active", + completedByChildId: "child", + awaitingChildId: undefined, + }) + }) + + it("abandons only an interrupted child and clears both sides of the live link", () => { + const parent = item("parent", { status: "delegated", awaitingChildId: "child", delegatedToId: "child" }) + const activeChild = item("child", { parentTaskId: "parent", rootTaskId: "parent" }) + + expect(() => abandonDelegatedChild(parent, activeChild)).toThrow(/status active/) + + const abandoned = abandonDelegatedChild(parent, { ...activeChild, status: "interrupted" }) + expect(abandoned.parent).toMatchObject({ status: "active", awaitingChildId: undefined }) + expect(abandoned.child).toMatchObject({ parentTaskId: undefined, rootTaskId: undefined }) + }) +}) diff --git a/src/core/task-persistence/__tests__/taskStoreConcurrency.spec.ts b/src/core/task-persistence/__tests__/taskStoreConcurrency.spec.ts new file mode 100644 index 0000000000..419ba03cb4 --- /dev/null +++ b/src/core/task-persistence/__tests__/taskStoreConcurrency.spec.ts @@ -0,0 +1,36 @@ +import type { HistoryItem } from "@roo-code/types" + +import { DeltaRejectedError, mergeHistoryDelta } from "../taskStoreConcurrency" + +function item(status: HistoryItem["status"]): HistoryItem { + return { + id: "task", + number: 1, + ts: 1, + task: "task", + tokensIn: 0, + tokensOut: 0, + totalCost: 0, + status, + } +} + +describe("mergeHistoryDelta", () => { + it("rejects an explicit undefined status that would revive a completed task", () => { + const disk = item("completed") + const incoming = { ...disk, status: undefined, mode: "must-not-commit" } + + expect(() => + mergeHistoryDelta(disk, incoming, { id: disk.id, status: undefined, mode: "must-not-commit" }), + ).toThrow(DeltaRejectedError) + expect(disk.status).toBe("completed") + expect("mode" in disk).toBe(false) + }) + + it("normalizes a legacy undefined disk status to explicit active", () => { + const disk = item(undefined) + const incoming = { ...disk, status: undefined } + + expect(mergeHistoryDelta(disk, incoming, { id: disk.id, status: undefined }).status).toBe("active") + }) +}) diff --git a/src/core/task-persistence/index.ts b/src/core/task-persistence/index.ts index edc4d860b5..463df8a0bb 100644 --- a/src/core/task-persistence/index.ts +++ b/src/core/task-persistence/index.ts @@ -1,4 +1,14 @@ export { type ApiMessage, readApiMessages, saveApiMessages } from "./apiMessages" export { readTaskMessages, saveTaskMessages } from "./taskMessages" export { taskMetadata } from "./taskMetadata" -export { TaskHistoryStore, assertValidTransition } from "./TaskHistoryStore" +export { TaskHistoryStore } from "./TaskHistoryStore" +export { + abandonDelegatedChild, + assertValidTransition, + completeDelegatedChild, + delegateTaskToChild, + interruptDelegatedChild, + LifecycleTransitionError, + type HistoryItemStatus, + VALID_TASK_STATUS_TRANSITIONS, +} from "./taskLifecycle" diff --git a/src/core/task-persistence/taskLifecycle.ts b/src/core/task-persistence/taskLifecycle.ts new file mode 100644 index 0000000000..efd2e1148f --- /dev/null +++ b/src/core/task-persistence/taskLifecycle.ts @@ -0,0 +1,115 @@ +import type { HistoryItem } from "@roo-code/types" + +/** Valid status values for a task's HistoryItem. */ +export type HistoryItemStatus = NonNullable + +export const VALID_TASK_STATUS_TRANSITIONS: Readonly> = { + active: ["delegated", "completed", "interrupted"], + delegated: ["active"], + interrupted: ["completed"], + completed: [], +} + +export class LifecycleTransitionError extends Error { + constructor(message: string) { + super(message) + this.name = "LifecycleTransitionError" + } +} + +export function assertValidTransition(from: HistoryItemStatus | undefined, to: HistoryItemStatus): void { + const fromStatus: HistoryItemStatus = from ?? "active" + if (!VALID_TASK_STATUS_TRANSITIONS[fromStatus].includes(to)) { + throw new Error(`Invalid task status transition: ${fromStatus} → ${to}`) + } +} + +export function delegateTaskToChild( + parent: HistoryItem, + childId: string, + awaitedChildStatus?: HistoryItemStatus, +): HistoryItem { + let base = parent + if (parent.status === "delegated") { + if (awaitedChildStatus !== "interrupted") { + throw new LifecycleTransitionError( + `Cannot re-delegate task ${parent.id}: existing child ${parent.awaitingChildId} is ${awaitedChildStatus}, not interrupted`, + ) + } + base = { + ...parent, + status: "active", + awaitingChildId: undefined, + delegatedToId: undefined, + } + } + + assertValidTransition(base.status, "delegated") + return { + ...base, + status: "delegated", + delegatedToId: childId, + awaitingChildId: childId, + childIds: Array.from(new Set([...(base.childIds ?? []), childId])), + } +} + +export function interruptDelegatedChild(parent: HistoryItem, child: HistoryItem): HistoryItem { + if (parent.status !== "delegated" || parent.awaitingChildId !== child.id) { + throw new LifecycleTransitionError(`Task ${parent.id} is not delegated to child ${child.id}`) + } + assertValidTransition(child.status, "interrupted") + return { ...child, status: "interrupted" } +} + +export function completeDelegatedChild( + parent: HistoryItem, + child: HistoryItem, + completionResultSummary: string, +): { parent: HistoryItem; child: HistoryItem } { + if ((parent.status !== "delegated" && parent.status !== "active") || parent.awaitingChildId !== child.id) { + throw new LifecycleTransitionError(`Task ${parent.id} is not delegated to child ${child.id}`) + } + assertValidTransition(child.status, "completed") + if (parent.status !== "active") assertValidTransition(parent.status, "active") + + return { + child: { + ...child, + status: "completed", + completionResultSummary, + }, + parent: { + ...parent, + status: "active", + completedByChildId: child.id, + completionResultSummary, + awaitingChildId: undefined, + delegatedToId: undefined, + childIds: Array.from(new Set([...(parent.childIds ?? []), child.id])), + }, + } +} + +export function abandonDelegatedChild( + parent: HistoryItem, + child: HistoryItem, +): { parent: HistoryItem; child: HistoryItem } { + if (parent.status !== "delegated" || parent.awaitingChildId !== child.id) { + throw new LifecycleTransitionError(`Task ${parent.id} is not delegated to child ${child.id}`) + } + if (child.status !== "interrupted") { + throw new LifecycleTransitionError(`Cannot abandon child ${child.id} with status ${child.status}`) + } + assertValidTransition(parent.status, "active") + + return { + child: { ...child, parentTaskId: undefined, rootTaskId: undefined }, + parent: { + ...parent, + status: "active", + awaitingChildId: undefined, + delegatedToId: undefined, + }, + } +} diff --git a/src/core/task-persistence/taskStoreConcurrency.ts b/src/core/task-persistence/taskStoreConcurrency.ts new file mode 100644 index 0000000000..0f747c4cb2 --- /dev/null +++ b/src/core/task-persistence/taskStoreConcurrency.ts @@ -0,0 +1,45 @@ +import deepEqual from "fast-deep-equal" +import type { HistoryItem } from "@roo-code/types" + +import { type HistoryItemStatus, VALID_TASK_STATUS_TRANSITIONS } from "./taskLifecycle" + +export class DeltaRejectedError extends Error { + constructor( + public readonly taskId: string, + public readonly diskStatus: HistoryItemStatus, + public readonly attemptedStatus: HistoryItemStatus, + ) { + super(`Delta rejected for task ${taskId}: disk status ${diskStatus} rejects transition to ${attemptedStatus}`) + this.name = "DeltaRejectedError" + } +} + +export function computeHistoryDelta(cached: HistoryItem, incoming: Partial): Partial { + return Object.fromEntries( + Object.entries(incoming).filter(([key, value]) => !deepEqual(value, (cached as Record)[key])), + ) as Partial +} + +export function mergeHistoryDelta(existing: unknown, incoming: HistoryItem, delta: Partial): HistoryItem { + if (!existing || typeof existing !== "object" || !("id" in existing)) { + return incoming + } + const disk = existing as HistoryItem + const normalizedDelta = { ...delta } + if ("status" in delta) { + const diskStatus: HistoryItemStatus = disk.status ?? "active" + const attemptedStatus: HistoryItemStatus = delta.status ?? "active" + if (attemptedStatus !== diskStatus) { + const validTargets = VALID_TASK_STATUS_TRANSITIONS[diskStatus] + if (!validTargets.includes(attemptedStatus)) { + throw new DeltaRejectedError(disk.id, diskStatus, attemptedStatus) + } + } + normalizedDelta.status = attemptedStatus + } + const merged = { ...disk, ...normalizedDelta } + if (normalizedDelta.childIds && disk.childIds) { + merged.childIds = [...new Set([...disk.childIds, ...normalizedDelta.childIds])] + } + return merged +} diff --git a/src/core/webview/ClineProvider.ts b/src/core/webview/ClineProvider.ts index b0237c8aa5..0a251aba5f 100644 --- a/src/core/webview/ClineProvider.ts +++ b/src/core/webview/ClineProvider.ts @@ -114,7 +114,10 @@ import { saveApiMessages, saveTaskMessages, TaskHistoryStore, - assertValidTransition, + abandonDelegatedChild, + completeDelegatedChild, + delegateTaskToChild, + interruptDelegatedChild, } from "../task-persistence" import { readTaskMessages } from "../task-persistence/taskMessages" import { getNonce } from "./getNonce" @@ -705,7 +708,7 @@ export class ClineProvider return } - const interruptedChild = { ...childHistory, status: "interrupted" as const } + const interruptedChild = interruptDelegatedChild(parentHistory, childHistory) await this.updateTaskHistory(interruptedChild) await this.postMessageToWebview({ type: "taskHistoryItemUpdated", taskHistoryItem: interruptedChild }) await this.postMessageToWebview({ type: "taskHistoryItemUpdated", taskHistoryItem: parentHistory }) @@ -3596,7 +3599,7 @@ export class ClineProvider if (parentHistory?.status === "delegated" && parentHistory?.awaitingChildId === task.taskId) { // Mark the child interrupted and leave parent delegated with awaitingChildId // intact — the user can resume this child later and it will report back. - historyItem = { ...historyItem!, status: "interrupted" } + historyItem = interruptDelegatedChild(parentHistory, historyItem!) await this.updateTaskHistory(historyItem) // Clear any stale fail-closed entry from a prior failed cancel attempt so // reopenParentFromDelegation is not incorrectly blocked on resume. @@ -3923,43 +3926,19 @@ export class ClineProvider // silently detached. try { await this.taskHistoryStore.atomicReadAndUpdate(parentTaskId, (historyItem) => { - let base = historyItem - if (pendingActionId && base.pendingAction?.actionId !== pendingActionId) { + if (pendingActionId && historyItem.pendingAction?.actionId !== pendingActionId) { throw new Error( - `[delegateParentAndOpenChild] Pending action mismatch for parent ${parentTaskId}: expected ${pendingActionId}, found ${base.pendingAction?.actionId}`, + `[delegateParentAndOpenChild] Pending action mismatch for parent ${parentTaskId}: expected ${pendingActionId}, found ${historyItem.pendingAction?.actionId}`, ) } - if (historyItem.status === "delegated") { - // Re-read the awaited child's current status under the store lock. - const awaitedChildStatus = historyItem.awaitingChildId - ? this.taskHistoryStore.get(historyItem.awaitingChildId)?.status - : undefined - // Only sever the stale link when the old child is confirmed interrupted. - // If it is still active, throw so the rollback path cleans up the new child - // rather than silently detaching a live task. - if (awaitedChildStatus !== "interrupted") { - throw new Error( - `[delegateParentAndOpenChild] Cannot re-delegate: existing child ${historyItem.awaitingChildId} is ${awaitedChildStatus}, not interrupted`, - ) - } - // Implicit sever of the stale interrupted-child link. - // The old child keeps its interrupted status; we just clear the parent's pointer. - base = { - ...historyItem, - status: "active" as const, - awaitingChildId: undefined, - delegatedToId: undefined, - } - } - assertValidTransition(base.status, "delegated") - const childIds = Array.from(new Set([...(base.childIds ?? []), child.taskId])) + const awaitedChildStatus = historyItem.awaitingChildId + ? this.taskHistoryStore.get(historyItem.awaitingChildId)?.status + : undefined + const delegated = delegateTaskToChild(historyItem, child.taskId, awaitedChildStatus) return { - ...base, - status: "delegated" as const, - delegatedToId: child.taskId, - awaitingChildId: child.taskId, - childIds, - pendingAction: base.pendingAction?.actionId === pendingActionId ? undefined : base.pendingAction, + ...delegated, + pendingAction: + delegated.pendingAction?.actionId === pendingActionId ? undefined : delegated.pendingAction, } }) this.recentTasksCache = undefined @@ -4207,6 +4186,7 @@ export class ClineProvider // any concurrent write that landed between step 1 and the lock acquisition // is preserved rather than silently overwritten. let updatedHistory!: typeof historyItem + let completingChild!: HistoryItem await this.taskHistoryStore.atomicUpdatePair( childTaskId, parentTaskId, @@ -4214,29 +4194,17 @@ export class ClineProvider if (pendingActionId && child.pendingAction?.actionId !== pendingActionId) { throw new Error(`[reopenParentFromDelegation] Pending action mismatch for child ${childTaskId}`) } - assertValidTransition(child.status, "completed") + completingChild = { ...child } + const lifecycleUpdate = completeDelegatedChild(historyItem, child, completionResultSummary) return { - ...child, - status: "completed" as const, - completionResultSummary, + ...lifecycleUpdate.child, pendingAction: child.pendingAction?.actionId === pendingActionId ? undefined : child.pendingAction, } }, (parent) => { - if (parent.status !== "active") { - assertValidTransition(parent.status, "active") - } - const childIds = Array.from(new Set([...(parent.childIds ?? []), childTaskId])) - updatedHistory = { - ...parent, - status: "active" as const, - completedByChildId: childTaskId, - completionResultSummary, - awaitingChildId: undefined, - delegatedToId: undefined, - childIds, - } + const lifecycleUpdate = completeDelegatedChild(parent, completingChild, completionResultSummary) + updatedHistory = lifecycleUpdate.parent return updatedHistory }, ) @@ -4346,8 +4314,6 @@ export class ClineProvider return false } - assertValidTransition(parentHistory.status, "active") - // Close the live child instance (if it's still the open task — the common case, // since an interrupted child is rehydrated onto the stack after cancelTask) BEFORE // clearing its persisted links. Task#saveClineMessages() rebuilds parentTaskId/ @@ -4362,13 +4328,8 @@ export class ClineProvider await this.taskHistoryStore.atomicUpdatePair( childTaskId, parentTaskId, - (child) => ({ ...child, parentTaskId: undefined, rootTaskId: undefined }), - (parent) => ({ - ...parent, - status: "active" as const, - awaitingChildId: undefined, - delegatedToId: undefined, - }), + (child) => abandonDelegatedChild(parentHistory, child).child, + (parent) => abandonDelegatedChild(parent, freshChild).parent, ) this.recentTasksCache = undefined