Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
324 changes: 169 additions & 155 deletions packages/core/src/agents/team/TeamManager.ts

Large diffs are not rendered by default.

60 changes: 60 additions & 0 deletions packages/core/src/agents/team/mailbox.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ import {
clearInbox,
clearAllInboxes,
sendStructuredMessage,
disposeInboxLocks,
} from './mailbox.js';

vi.mock('../../config/storage.js', async (importOriginal) => {
Expand Down Expand Up @@ -105,6 +106,39 @@ describe('mailbox', () => {
expect(messages[1]!.text).toBe('second');
});

it('compacts aged read messages but keeps recent and unread ones', async () => {
// Bounds the leader inbox once consumption marks messages read:
// read entries past the retention window are dropped on the next
// write, while recent read entries and all unread entries survive.
const old = new Date(Date.now() - 10 * 60 * 1000).toISOString();
const recent = new Date(Date.now() - 60 * 1000).toISOString();

await writeMessage(
'team',
'leader',
makeMessage({ text: 'aged-read', read: true, timestamp: old }),
);
await writeMessage(
'team',
'leader',
makeMessage({ text: 'recent-read', read: true, timestamp: recent }),
);
await writeMessage(
'team',
'leader',
makeMessage({ text: 'aged-unread', read: false, timestamp: old }),
);

// A subsequent write runs the retention compaction.
await writeMessage('team', 'leader', makeMessage({ text: 'fresh' }));

const texts = (await readInbox('team', 'leader')).map((m) => m.text);
expect(texts).not.toContain('aged-read'); // read + aged → dropped
expect(texts).toContain('recent-read'); // read but within window → kept
expect(texts).toContain('aged-unread'); // unread → never dropped
expect(texts).toContain('fresh');
});

// ─── consumeUnread ─────────────────────────────────────────

it('returns unread messages and marks them read', async () => {
Expand Down Expand Up @@ -187,6 +221,32 @@ describe('mailbox', () => {
expect(await readInbox('team', 'w2')).toEqual([]);
});

// ─── disposeInboxLocks ─────────────────────────────────────

it('evicts inbox locks for one team and reports the count', async () => {
// writeMessage creates a per-inbox lock under the team's inboxes dir.
await writeMessage('alpha', 'w1', makeMessage());
await writeMessage('alpha', 'w2', makeMessage());
await writeMessage('beta', 'w1', makeMessage());

// Evicts only alpha's two locks.
expect(disposeInboxLocks('alpha')).toBe(2);
// Idempotent: nothing left for alpha.
expect(disposeInboxLocks('alpha')).toBe(0);
// Team isolation: beta's lock survived alpha's eviction.
expect(disposeInboxLocks('beta')).toBe(1);
});

it('clearAllInboxes evicts the team inbox locks too', async () => {
await writeMessage('gamma', 'w1', makeMessage());
await writeMessage('gamma', 'w2', makeMessage());

await clearAllInboxes('gamma');

// Already evicted by clearAllInboxes.
expect(disposeInboxLocks('gamma')).toBe(0);
});

// ─── sendStructuredMessage ─────────────────────────────────

it('sends a structured message with type', async () => {
Expand Down
31 changes: 29 additions & 2 deletions packages/core/src/agents/team/mailbox.ts
Original file line number Diff line number Diff line change
Expand Up @@ -92,8 +92,9 @@ const LOCK_OPTIONS: lockfile.LockOptions = {
// One `Mutex` per inbox path, keyed by absolute path. Distinct
// inboxes never block each other; concurrent operations on the same
// inbox queue in memory so only one of them ever reaches for the
// `proper-lockfile` file lock at a time. Entries are never evicted,
// but the key space is bounded by team size (one per agent inbox).
// `proper-lockfile` file lock at a time. Entries are evicted on team
// teardown (`disposeInboxLocks`) so a long-running process doesn't
// accumulate a dead `Mutex` per inbox across team create/delete cycles.

const inboxLocks = new Map<string, Mutex>();

Expand All @@ -106,6 +107,31 @@ function getInboxLock(inboxPath: string): Mutex {
return lock;
}

/**
* Evict the in-process inbox locks for a team. The lock map keys on
* absolute inbox path and would otherwise retain a `Mutex` for every
* inbox the process ever touched — a slow leak across many team
* create/delete cycles in a long-lived daemon. All of a team's inbox
* paths sit directly under its inboxes dir, so match on the parent.
*
* Best-effort: a teammate's late `writeMessage` racing team teardown
* can re-create an entry afterwards, but the next same-name
* `team_create` resets inboxes and evicts it again.
*
* Returns the number of locks evicted.
*/
export function disposeInboxLocks(teamName: string): number {
const dir = getInboxesDir(teamName);
let evicted = 0;
for (const key of inboxLocks.keys()) {
if (path.dirname(key) === dir) {
inboxLocks.delete(key);
evicted++;
}
}
return evicted;
}

/**
* Run `fn` while holding both the in-process inbox mutex and the
* cross-process file lock for `inboxPath`.
Expand Down Expand Up @@ -252,6 +278,7 @@ export async function clearInbox(
export async function clearAllInboxes(teamName: string): Promise<void> {
const dir = getInboxesDir(teamName);
await fs.rm(dir, { recursive: true, force: true });
disposeInboxLocks(teamName);
}

// ─── Convenience: send structured message ───────────────────
Expand Down
40 changes: 40 additions & 0 deletions packages/core/src/agents/team/teamHelpers.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import {
setMemberActive,
findMemberById,
findMemberByName,
classifyShutdownResponse,
readTeamFile,
writeTeamFile,
deleteTeamDirs,
Expand Down Expand Up @@ -252,6 +253,45 @@ describe('findMemberByName', () => {
});
});

describe('classifyShutdownResponse', () => {
it('classifies a reply that leads with the approve token', () => {
expect(classifyShutdownResponse('shutdown_approved')).toBe(
'shutdown_approved',
);
// Verbose lead-in form an exact-string match would miss.
expect(classifyShutdownResponse('shutdown_approved, work finished')).toBe(
'shutdown_approved',
);
expect(classifyShutdownResponse(' shutdown_approved\nbye')).toBe(
'shutdown_approved',
);
});

it('classifies a reply that leads with the reject token', () => {
expect(classifyShutdownResponse('shutdown_rejected: still mid-task')).toBe(
'shutdown_rejected',
);
});

it('does not classify a mid-prose mention of the token', () => {
// The false-abort bug: a token mentioned mid-report is not a
// response.
expect(
classifyShutdownResponse(
'I reviewed the shutdown_approved handler and it looks correct.',
),
).toBeUndefined();
expect(
classifyShutdownResponse('I will send shutdown_approved when done.'),
).toBeUndefined();
});

it('returns undefined for an unrelated message', () => {
expect(classifyShutdownResponse('task complete')).toBeUndefined();
expect(classifyShutdownResponse('')).toBeUndefined();
});
});

// ─── File I/O tests ─────────────────────────────────────────

describe('file I/O', () => {
Expand Down
24 changes: 24 additions & 0 deletions packages/core/src/agents/team/teamHelpers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -214,6 +214,30 @@ export function findMemberByName(
return members.find((m) => m.name === sanitized);
}

/**
* Classify a teammate's free-text reply to a shutdown request.
*
* The leader asks the teammate to reply with `shutdown_approved` or
* `shutdown_rejected: <reason>`. A compliant reply leads with the
* token, so match only at the start (after leading whitespace) — never
* anywhere in the body. That anchoring is what stops a teammate that
* merely *mentions* the token mid-report (e.g. while reviewing
* shutdown-related code) from being read as an approval and aborted,
* while still accepting the verbose `shutdown_approved, work finished`
* form that an exact-string match would miss.
*
* Returns the structured response type, or undefined when the reply is
* not a shutdown response.
*/
export function classifyShutdownResponse(
message: string,
): 'shutdown_approved' | 'shutdown_rejected' | undefined {
const trimmed = message.trimStart();
if (/^shutdown_approved\b/i.test(trimmed)) return 'shutdown_approved';
if (/^shutdown_rejected\b/i.test(trimmed)) return 'shutdown_rejected';
return undefined;
}

// ─── File I/O ───────────────────────────────────────────────

/**
Expand Down
Loading
Loading