Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
137 changes: 135 additions & 2 deletions packages/core/src/agents/team/TeamManager.ghost-member.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,9 @@
*/

/**
* @fileoverview Regression test for #10208 — failed concurrent spawn
* can persist a ghost member in config.json.
* @fileoverview Regression tests for #10208 — failed concurrent spawn
* can persist a ghost member in config.json — and for the #10297
* commit-aware failed-spawn compensating-write gate.
*/

import { describe, it, expect, vi, afterEach } from 'vitest';
Expand Down Expand Up @@ -525,3 +526,135 @@ describe('TeamManager ghost member regression (#10208)', () => {
expect(inMemoryNames).not.toContain('beta');
});
});

describe('TeamManager commit-aware compensating-write gate (#10297)', () => {
let harness: TeamCoordinationHarness | undefined;

afterEach(async () => {
await harness?.cleanup();
harness = undefined;
vi.restoreAllMocks();
});

async function createHarness(): Promise<TeamCoordinationHarness> {
const h = await TeamCoordinationHarness.create();
setMockDir(h.tmpDir);
harness = h;
return h;
}

it('skips the compensating write when the only write in the window rejected (solo)', async () => {
const h = await createHarness();

const leaderSpy = vi.fn();
h.teamManager.setLeaderMessageCallback(leaderSpy);

// Disk full: the member's own roster write rejects, and a
// compensating write attempted while the disk is still full would
// reject too.
const writeSpy = vi
.spyOn(teamHelpers, 'writeTeamFile')
.mockRejectedValueOnce(new Error('ENOSPC: no space left on device'))
.mockRejectedValueOnce(new Error('ENOSPC: no space left on device'));

await expect(
h.teamManager.spawnTeammate({ name: 'alpha', cwd: h.tmpDir }),
).rejects.toThrow('ENOSPC');

// The only write that ran in the member's window rejected and
// persisted nothing, so there is nothing to compensate: exactly one
// write attempt...
expect(writeSpy).toHaveBeenCalledTimes(1);
// ...and no ghost-member notice reaches the leader on top of the
// spawn error that already carries the real cause (disk full).
expect(leaderSpy).not.toHaveBeenCalled();

// The member is rolled back in memory and nothing reached the disk.
const inMemory = (h.teamManager as unknown as { teamFile: TeamFile })
.teamFile;
expect(inMemory.members).toHaveLength(0);
const persisted = await readTeamFile(h.teamName);
expect(persisted).toBeDefined();
expect(persisted!.members).toHaveLength(0);
});

it('still repairs a member persisted by a committed window write (five-step interleaving)', async () => {
// The issue's counterexample to decrement-on-reject:
// 1. alpha's write starts (snapshot taken) and hangs in an fs await;
// 2. beta is pushed while it is in flight;
// 3. alpha's write rejects — a decrement would drop the counter
// back to beta's push watermark;
// 4. gamma's write starts, reusing that counter value, and commits
// a snapshot that still contains beta;
// 5. beta's spawn fails — the gate must still fire the repair,
// or gamma's committed write leaves ghost beta on disk (#10208).
const h = await createHarness();
const teamName = h.teamName;
const backend = h.backend;

const deferredA = createDeferred<void>();
const deferredB = createDeferred<void>();
gateSpawns(
backend,
new Map([
[formatAgentId('alpha', teamName), deferredA.promise],
[formatAgentId('beta', teamName), deferredB.promise],
]),
{ passthroughUnknown: true },
);

const realWriteTeamFile = teamHelpers.writeTeamFile;
let writeCalls = 0;
let releaseFirstWrite!: () => void;
const firstWriteGate = new Promise<void>((r) => {
releaseFirstWrite = r;
});
vi.spyOn(teamHelpers, 'writeTeamFile').mockImplementation(
async (name, tf) => {
writeCalls++;
if (writeCalls === 1) {
await firstWriteGate; // like the real await fs.mkdir
throw new Error('ENOSPC: no space left on device');
}
return realWriteTeamFile(name, tf);
},
);

const spawnA = h.teamManager.spawnTeammate({
name: 'alpha',
cwd: h.tmpDir,
});
await new Promise((r) => setTimeout(r, 50));

// Alpha succeeds; its write starts and hangs, then beta is pushed
// while it is in flight.
deferredA.resolve();
await vi.waitFor(() => expect(writeCalls).toBe(1));

const spawnB = h.teamManager.spawnTeammate({
name: 'beta',
cwd: h.tmpDir,
});
await new Promise((r) => setTimeout(r, 50));

// Alpha's write rejects; alpha rolls back. Nothing has committed.
releaseFirstWrite();
await expect(spawnA).rejects.toThrow('ENOSPC');

// Gamma succeeds; its write snapshots the live roster — beta is
// still in it — and commits beta to disk.
await h.teamManager.spawnTeammate({ name: 'gamma', cwd: h.tmpDir });

// Beta's spawn fails. A write inside beta's window committed a
// snapshot containing beta, so the compensating write must run.
deferredB.reject(new Error('spawn failed'));
await expect(spawnB).rejects.toThrow('spawn failed');

const persisted = await readTeamFile(teamName);
expect(persisted).toBeDefined();
const persistedNames = persisted!.members.map((m) => m.name);
expect(persistedNames).toContain('gamma');
expect(persistedNames).not.toContain('alpha');
expect(persistedNames).not.toContain('beta');
});
});
156 changes: 95 additions & 61 deletions packages/core/src/agents/team/TeamManager.ts
Original file line number Diff line number Diff line change
Expand Up @@ -221,19 +221,30 @@ export class TeamManager {
private teamFileWriteQueue: Promise<void> = Promise.resolve();

/**
* Count of roster writes that have started (reached their snapshot
* point) in `persistTeamFile`'s queue. Because each queued write
* snapshots the roster synchronously when it RUNS, a member pushed
* after the last started write cannot be on disk yet. The failed-spawn
* compensating write compares this counter against the value captured
* at push time and skips when no write could have persisted the
* failed member — writing anyway would serialize the live roster and
* persist sibling members whose own spawn is still pending, widening
* the #10208 ghost window in the "no write has landed yet"
* interleaving.
* Sequence number of roster writes in `persistTeamFile`'s queue:
* each queued write increments it synchronously at its snapshot point
* when it RUNS. A member's push captures the current value, and only
* writes with a higher sequence number snapshot the roster while the
* member is in it, so only those can persist the member. The
* failed-spawn compensating-write gate compares
* `teamFileWritesCommitted` against the captured value (see
* `persistTeamFile`). Deliberately monotonic — a rejected write keeps
* its number so a later write cannot reuse the value and hide at or
* below an earlier member's push watermark (#10297).
*/
private teamFileWritesStarted = 0;

/**
* Sequence number of the most recently committed roster write
* (`writeTeamFile` resolved). The queue is serial, so writes commit
* in sequence order and this watermark is monotonic. A member pushed
* when `teamFileWritesStarted` read N can be on disk only if a write
* with a sequence number above N committed, i.e. this watermark
* advanced past N — a write that started but rejected persisted
* nothing (#10297).
*/
private teamFileWritesCommitted = 0;

/**
* Cap on per-agent pending messages. Each message can be up to the
* `send_message` schema's `maxLength`, and a queue only drains when its
Expand Down Expand Up @@ -342,25 +353,46 @@ export class TeamManager {
* runs (see `teamFileWriteQueue`), so the last queued write always
* commits the newest in-memory state. A rejected write does not
* poison the queue — the chain survives for subsequent writes.
*
* With `onlyIfCommittedAfter`, the queued task writes only if a
* roster write with a sequence number above that value committed.
* The queue is serial, so by the time the task runs every earlier
* write — including any still in flight when the task was queued —
* has settled and `teamFileWritesCommitted` is final for the window.
* The failed-spawn compensating write uses this: a write that
* started inside the failed member's window but rejected persisted
* nothing, so there is nothing to repair and the redundant write
* (with its misleading ghost-member notice if it failed too) is
* skipped (#10297).
*/
private persistTeamFile(): Promise<void> {
private persistTeamFile(options?: {
onlyIfCommittedAfter?: number;
}): Promise<void> {
const write = this.teamFileWriteQueue.then(() => {
// Snapshot point for `teamFileWritesStarted`: the roster snapshot
// below is taken synchronously here, so any member already pushed
// could land on disk through this write. Counted before the
// write runs (not after it resolves) so an in-flight write that
// started inside a failed member's window still counts — the
// compensating write queues behind it and repairs whatever it
// persisted.
this.teamFileWritesStarted++;
// Snapshot synchronously at the counted point, before any await:
// `writeTeamFile` awaits `fs.mkdir` before stringifying its
// argument, so handing it the live roster would let a member
// pushed during that fs hop land on disk through a write this
// gate counts as pre-push — the failed member's compensating
// write would then be skipped and the ghost persisted (#10208).
if (
options?.onlyIfCommittedAfter !== undefined &&
this.teamFileWritesCommitted <= options.onlyIfCommittedAfter
) {
// No write that started inside the failed member's window
// committed, so nothing on disk can contain the rolled-back
// member — skip the compensating write.
return;
}
// Snapshot point: assign the sequence number and snapshot the
// roster synchronously, before any await. `writeTeamFile` awaits
// `fs.mkdir` before stringifying its argument, so handing it the
// live roster would let a member pushed during that fs hop land
// on disk through a write the gate counts as pre-push — the
// failed member's compensating write would then be skipped and
// the ghost persisted (#10208).
const writeSeq = ++this.teamFileWritesStarted;
const snapshot = structuredClone(this.teamFile);
return writeTeamFile(this.teamFile.name, snapshot);
return writeTeamFile(this.teamFile.name, snapshot).then(() => {
// Commit point: advance the watermark only after the write
// actually landed. Commits happen in sequence order (the queue
// is serial), so the watermark stays monotonic.
this.teamFileWritesCommitted = writeSeq;
});
});
this.teamFileWriteQueue = write.catch(() => {});
return write;
Expand Down Expand Up @@ -425,9 +457,10 @@ export class TeamManager {
this.lastActivityAt.set(agentId, Date.now());
this.agentIdentities.set(agentId, identity);

// Roster writes that started before this push cannot have persisted
// the member; the compensating write after a failed spawn compares
// against this to decide whether anything on disk needs repairing.
// Roster writes with a sequence number at or below this value
// snapshot the roster before this push and cannot persist the
// member. The compensating write after a failed spawn skips unless
// a write above this watermark committed (see `persistTeamFile`).
const writesStartedAtPush = this.teamFileWritesStarted;

let agentSpawned = false;
Expand Down Expand Up @@ -685,41 +718,42 @@ export class TeamManager {
// persisted this member in config.json, rewrite the file so
// persisted membership matches the post-rollback in-memory
// state. Best-effort — the original error is more important.
// Gated on `teamFileWritesStarted`: if no roster write started
// while the member was in the roster, nothing on disk can
// contain it, and writing anyway would serialize the live roster
// — persisting sibling members whose own spawn is still pending
// and widening the ghost window #10208 removes in the "failed
// spawn is the first write" interleaving.
if (this.teamFileWritesStarted > writesStartedAtPush) {
// Commit-aware gate (#10297): the queued task runs after every
// earlier write has settled and writes only if one with a
// sequence number above `writesStartedAtPush` actually committed.
// A write that started in the window but rejected persisted
// nothing, so compensating it would only add a redundant
// best-effort write — and if the disk is still full, a
// misleading ghost-member notice on top of the spawn error.
try {
await this.persistTeamFile({
onlyIfCommittedAfter: writesStartedAtPush,
});
} catch (writeErr) {
// Best-effort — the original error takes precedence, but
// leave a trail so a resurfaced ghost member can be told
// apart from a compensating write that itself failed.
debug.warn(
`Compensating team-file write after failed spawn of ` +
`${agentId} failed: ${getErrorMessage(writeErr)}`,
);
// Beyond the debug log (which is off in production), surface
// the failure to the leader as well, mirroring `fireAndForget`:
// the persisted roster may now keep a ghost member (#10208),
// and the leader is the only production-visible observer.
try {
await this.persistTeamFile();
} catch (writeErr) {
// Best-effort — the original error takes precedence, but
// leave a trail so a resurfaced ghost member can be told
// apart from a compensating write that itself failed.
this.leaderMessageCallback?.(
`<team_error>Compensating team-file write after failed ` +
`spawn of ${agentId} failed: ` +
`${getErrorMessage(writeErr)}</team_error>`,
`Team roster write after failed spawn of "${name}" failed`,
);
} catch (cbErr) {
const cbMsg = getErrorMessage(cbErr);
debug.warn(
`Compensating team-file write after failed spawn of ` +
`${agentId} failed: ${getErrorMessage(writeErr)}`,
`Compensating-write failure notice: leader message ` +
`callback threw: ${cbMsg}`,
);
// Beyond the debug log (which is off in production), surface
// the failure to the leader as well, mirroring `fireAndForget`:
// the persisted roster may now keep a ghost member (#10208),
// and the leader is the only production-visible observer.
try {
this.leaderMessageCallback?.(
`<team_error>Compensating team-file write after failed ` +
`spawn of ${agentId} failed: ` +
`${getErrorMessage(writeErr)}</team_error>`,
`Team roster write after failed spawn of "${name}" failed`,
);
} catch (cbErr) {
const cbMsg = getErrorMessage(cbErr);
debug.warn(
`Compensating-write failure notice: leader message ` +
`callback threw: ${cbMsg}`,
);
}
}
}
throw err;
Expand Down