fix(agents): preserve recoverable workspace state

packages/core/src/agents/workspace-agents/store.test.ts

@@ -64,9 +64,11 @@ import {
   writeThread,
 } from './store.js';
 import { postMessage, postMessageInTransaction } from './thread-actions.js';
+import { resolveThreadStatus } from './thread-status.js';
 import {
   HUMAN_AUTHOR_ID,
   AGENTS_SCHEMA_VERSION,
+  MAX_THREAD_RUNS,
   type WorkspaceAgent,
   type Thread,
   type ThreadEvent,
@@ -626,6 +628,78 @@ describe('retiring an agent', () => {
     );
   });
 
+  it('finishes disabling after the roster committed but thread cleanup failed', async () => {
+    await seed([ALICE]);
+    await writeThread(
+      PROJECT_ROOT,
+      thread({ runs: [run(1, 0, { status: 'queued', endedAt: undefined })] }),
+    );
+    atomicWriteFault.filePath = getThreadPath(PROJECT_ROOT, 'th_root');
+    atomicWriteFault.mode = 'throwBefore';
+    await expect(
+      setWorkspaceAgentEnabled(PROJECT_ROOT, ALICE.id, false),
+    ).rejects.toThrow('injected crash');
+    expect((await readWorkspaceAgents(PROJECT_ROOT))[0].enabled).toBe(false);
+    expect((await readThread(PROJECT_ROOT, 'th_root'))?.runs[0].status).toBe(
+      'queued',
+    );
+
+    await setWorkspaceAgentEnabled(PROJECT_ROOT, ALICE.id, false);
+    const stored = await readThread(PROJECT_ROOT, 'th_root');
+    expect(stored?.runs[0].status).toBe('cancelled');
+    expect(stored?.status).toBe('blocked');
+    await setWorkspaceAgentEnabled(PROJECT_ROOT, ALICE.id, false);
+    expect(await readThread(PROJECT_ROOT, 'th_root')).toEqual(stored);
+  });
+
+  it('refuses disabling before writing the roster when a thread is unreadable', async () => {
+    await seed([ALICE]);
+    await writeRaw(getThreadPath(PROJECT_ROOT, 'th_broken'), {});
+    await expect(
+      setWorkspaceAgentEnabled(PROJECT_ROOT, ALICE.id, false),
+    ).rejects.toThrow('thread records are unreadable');
+    expect((await readWorkspaceAgents(PROJECT_ROOT))[0].enabled).not.toBe(
+      false,
+    );
+  });
+
+  it('retains an unresolved close through trimming and releases it after acknowledgement', async () => {
+    await seed([ALICE]);
+    const blocker = run(1, 0, { closeKind: 'blocked' });
+    await writeThread(
+      PROJECT_ROOT,
+      thread({
+        status: 'blocked',
+        runs: [
+          blocker,
+          ...Array.from({ length: MAX_THREAD_RUNS }, (_, index) =>
+            run(index + 2, 0),
+          ),
+        ],
+      }),
+    );
+    const stored = (await readThread(PROJECT_ROOT, 'th_root'))!;
+    expect(stored.runs.some((entry) => entry.id === blocker.id)).toBe(true);
+    expect(
+      resolveThreadStatus({ thread: stored, hasLiveChildDependency: false })
+        .status,
+    ).toBe('blocked');
+
+    await writeThread(PROJECT_ROOT, {
+      ...stored,
+      runs: stored.runs.map((entry) =>
+        entry.id === blocker.id
+          ? { ...entry, closeAcknowledgedAtSequence: 0 }
+          : entry,
+      ),
+    });
+    expect(
+      (await readThread(PROJECT_ROOT, 'th_root'))?.runs.some(
+        (entry) => entry.id === blocker.id,
+      ),
+    ).toBe(false);
+  });
+
   it('reports cancellation when disabling the last queued run of a child', async () => {
     await seed([ALICE]);
     await writeThread(

packages/core/src/agents/workspace-agents/store.ts

@@ -15,6 +15,7 @@ import { Storage } from '../../config/storage.js';
 import { atomicWriteJSON } from '../../utils/atomicFileWrite.js';
 import { isNodeError } from '../../utils/errors.js';
 import { getProjectHash } from '../../utils/paths.js';
+import { outstandingCloseObligations } from './thread-status.js';
 import {
   DEFAULT_QUEUE_LIMIT,
   HUMAN_AUTHOR_ID,
@@ -935,9 +936,13 @@ function trimThread(thread: Thread): Thread {
     thread.messages.length - MAX_THREAD_MESSAGES,
   );
   const firstRetainedRun = Math.max(0, thread.runs.length - MAX_THREAD_RUNS);
+  const outstandingRunIds = new Set(
+    outstandingCloseObligations(thread).map((obligation) => obligation.runId),
+  );
   const retainedRuns = thread.runs.filter(
     (run, index) =>
       index >= firstRetainedRun ||
+      outstandingRunIds.has(run.id) ||
       run.usageByRound.length > 0 ||
       run.status === 'queued' ||
       run.status === 'running' ||
@@ -983,6 +988,7 @@ async function writeThreadUnlocked(
   return next;
 }
 
+/** Serialized workspace access; each file commits independently, without rollback. */
 export interface AgentStoreTransaction {
   readonly projectRoot: string;
   readonly workspaceId: string;
@@ -1144,29 +1150,31 @@ export async function setWorkspaceAgentEnabled(
     // success and change nothing a caller can observe — `isAgentAddressable`
     // still refuses it — which is worse than saying no.
     if (agent.retiredAt !== undefined) return 'retired';
-    if ((agent.enabled !== false) === enabled) return 'updated';
-    await transaction.writeAgents(
-      agents.map((candidate) =>
-        candidate.id === agentId ? { ...candidate, enabled } : candidate,
-      ),
-    );
-    if (!enabled) {
+    if (enabled && agent.enabled !== false) return 'updated';
+    const pending = enabled ? undefined : await transaction.listThreads();
+    if (pending && pending.unreadable.length > 0) {
+      throw new Error(
+        `Cannot change the agent roster while thread records are unreadable: ${pending.unreadable.join(', ')}.`,
+      );
+    }
+    if ((agent.enabled !== false) !== enabled) {
+      await transaction.writeAgents(
+        agents.map((candidate) =>
+          candidate.id === agentId ? { ...candidate, enabled } : candidate,
+        ),
+      );
+    }
+    if (pending) {
       // A disabled agent's queued runs can never start — candidate selection
       // skips non-addressable agents — but `queued` counts as live for
       // thread status and for `agentHasLiveWork`, so leaving one behind
       // pins its thread in `in_progress` and blocks retirement with no
-      // in-band explanation. Settle those runs here, in the roster change's
-      // own transaction; running work is left to finish.
-      const { threads, unreadable } = await transaction.listThreads();
-      if (unreadable.length > 0) {
-        throw new Error(
-          `Cannot change the agent roster while thread records are unreadable: ${unreadable.join(', ')}.`,
-        );
-      }
+      // in-band explanation. Repeat cleanup even when already disabled:
+      // an earlier call may have stopped after writing the roster.
       const now = Date.now();
       // Import lazily because run lifecycle is built on this store module.
       const { finishRunInTransaction } = await import('./run-lifecycle.js');
-      for (const thread of threads) {
+      for (const thread of pending.threads) {
         if (isThreadTerminal(thread.status)) continue;
         for (const run of thread.runs) {
           if (run.agentId !== agentId || run.status !== 'queued') continue;

packages/core/src/agents/workspace-agents/types.ts

@@ -393,8 +393,8 @@ export const DEFAULT_THREAD_AUTO_TURN_BUDGET = 12;
  */
 export const DEFAULT_THREAD_TOKEN_BUDGET = 1_000_000;
 
-/** Bound on retained posts per thread. */
+/** Recent-post retention target; referenced and idempotency records are retained. */
 export const MAX_THREAD_MESSAGES = 500;
 
-/** Bound on retained run records per thread. */
+/** Recent-run retention target; live, accounting and unresolved records are retained. */
 export const MAX_THREAD_RUNS = 200;