Skip to content

fix(cli): resolve archive worktrees from machine Flock #382

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zxch3n merged 3 commits into from
Sep 12, 2026
Merged

fix(cli): resolve archive worktrees from machine Flock #382

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6f08cb1
fix(cli): resolve archive worktrees from machine Flock
UniversePeak Sep 4, 2026
c731b12
fix(cli): retry archive after Machine Flock read failures
UniversePeak Sep 7, 2026
03e2a9f
fix(cli): wait for authoritative archive metadata
UniversePeak Sep 8, 2026
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
45 changes: 39 additions & 6 deletions apps/cli/src/lib/message-handler.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3810,7 +3810,12 @@ export class MessageHandler {
this.getMachineFlockDocIdForMachine()
);
return readMachineFlockRowsFromFlock(handle.flock, {
families: ['archiveSessionCommand', 'deleteSessionCommand', 'deleteLocalProjectCommand'],
families: [
'archiveSessionCommand',
'deleteSessionCommand',
'deleteLocalProjectCommand',
'localProject',
],
});
}

Expand All @@ -3828,11 +3833,16 @@ export class MessageHandler {
}
}

private async readMachineCommandSnapshot(): Promise<MachineCommandSnapshot> {
private async readMachineCommandSnapshot(options?: {
strictFlock?: boolean;
}): Promise<MachineCommandSnapshot> {
const machineRoomId = getMachineRoomId(this.machineId);
const flockRows = options?.strictFlock
? this.readMachineFlockCommandRows()
: this.tryReadMachineFlockCommandRows();
const [machineMetaDoc, machineFlockRows] = await Promise.all([
this.workspaceDocument.repo.getDocMeta(machineRoomId),
this.tryReadMachineFlockCommandRows(),
flockRows,
]);
return buildMachineCommandSnapshot(
machineMetaDoc?.meta as MachineLegacyMetaFields | undefined,
Expand Down Expand Up @@ -3915,7 +3925,23 @@ export class MessageHandler {
}

private async processArchiveRequests(): Promise<void> {
const snapshot = await this.readMachineCommandSnapshot();
// Archive cleanup may need a local-project root that exists only in Machine
// Flock. In cloud mode, wait until its command watcher has completed the
// initial remote sync; a successful read before then can still be stale.
if (this.cloudPort.kind !== 'local' && !this.machineFlockCommandWatcher.isReady) {
this.logger.debug('[archive] Waiting for authoritative Machine Flock before processing');
return;
}

let snapshot: MachineCommandSnapshot;
try {
snapshot = await this.readMachineCommandSnapshot({ strictFlock: true });
} catch (error) {
this.logger.error(
`[archive] Failed to read Machine Flock; keeping archive request queued: ${formatErrorMessage(error)}`
);
return;
}
const sessionIds = snapshot.archiveSessionIds;
if (sessionIds.length === 0) {
this.logger.debug('[archive] No pending archive requests');
Expand All @@ -3938,7 +3964,9 @@ export class MessageHandler {
this.archiveInFlight.add(sessionId);
try {
this.logger.debug(`[archive] Start archiving session ${sessionId}`);
await this.archiveSessionResources(sessionId);
await this.archiveSessionResources(sessionId, {
machineFlockRows: snapshot.machineFlockRows,
});
await this.removeArchiveRequest(sessionId);
this.logger.debug(`[archive] Finished archiving session ${sessionId}`);
} catch (error) {
Expand All @@ -3953,7 +3981,10 @@ export class MessageHandler {

private async archiveSessionResources(
sessionId: SessionId,
options?: { preserveWorktree?: boolean }
options?: {
preserveWorktree?: boolean;
machineFlockRows?: MachineFlockRowMap;
}
): Promise<void> {
this.logger.debug(`[${sessionId}] Archiving session resources`);

Expand Down Expand Up @@ -3986,6 +4017,8 @@ export class MessageHandler {
const cleanupTarget = this.resolveWorktreeCleanupTarget({
sessionMeta,
machineMeta: archiveMachineMeta,
machineFlockRows:
options?.machineFlockRows ?? (await this.tryReadMachineFlockCommandRows()),
});
if (cleanupTarget) {
const worktreeManager = getWorktreeManager(this.buildWorktreeManagerConfig(cleanupTarget));
Expand Down
189 changes: 177 additions & 12 deletions apps/cli/tests/message-handler-terminal-cleanup.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ import {
type SessionMeta,
type WorkspaceId,
} from '@lody/shared';
import type { CloudPort } from '@lody/platform';
import { deriveRepoIdFromLocalProjectPath } from '@lody/shared/node/worktree-paths';
import { MessageHandler } from '../src/lib/message-handler';
import type { LoroDocumentManager } from '../src/lib/loro/doc';
Expand All @@ -39,9 +40,10 @@ const createSilentLogger = (): Logger => ({
});

type MessageHandlerInternals = {
processArchiveRequests: () => Promise<void>;
archiveSessionResources: (
sessionId: SessionId,
options?: { preserveWorktree?: boolean }
options?: { preserveWorktree?: boolean; machineFlockRows?: Record<string, MachineFlockScanRow> }
) => Promise<void>;
deleteLocalProjectResources: (
localProjectId: LocalProjectId,
Expand All @@ -66,6 +68,7 @@ type MessageHandlerInternals = {
previewService: {
closeSessionPreviewForCleanup: (sessionId: SessionId, reason: string) => Promise<void>;
};
machineFlockCommandWatcher: { isReady: boolean };
};

function createHarness(options?: {
Expand All @@ -75,8 +78,11 @@ function createHarness(options?: {
machineFlockRows?: MachineFlockScanRow[];
sessionMetas?: SessionMeta[];
activeSessionIds?: SessionId[];
archiveSessionIds?: SessionId[];
includeLegacySessionDeleteRequest?: boolean;
localProjectRootPaths?: Record<LocalProjectId, string>;
machineFlockOpenError?: Error;
cloudPort?: CloudPort;
}) {
const sessionId = options?.sessionId ?? ('session-1' as SessionId);
const childSessionIds = options?.childSessionIds ?? [];
Expand Down Expand Up @@ -128,7 +134,9 @@ function createHarness(options?: {
if (roomId === machineRoomId) {
return {
meta: {
needToArchiveSessions: {},
needToArchiveSessions: Object.fromEntries(
(options?.archiveSessionIds ?? []).map((id) => [id, true])
),
needToDeleteSessions:
options?.includeLegacySessionDeleteRequest === false ? {} : { [sessionId]: true },
localProjects: Object.fromEntries(
Expand All @@ -149,15 +157,20 @@ function createHarness(options?: {
: []
),
})),
openFlockDoc: vi.fn(async () => ({
flock: {
scan: () => machineFlockRows,
set: flockSet,
delete: flockDelete,
commit: flockCommit,
},
syncOnce: vi.fn(async () => {}),
})),
openFlockDoc: vi.fn(async () => {
if (options?.machineFlockOpenError) {
throw options.machineFlockOpenError;
}
return {
flock: {
scan: () => machineFlockRows,
set: flockSet,
delete: flockDelete,
commit: flockCommit,
},
syncOnce: vi.fn(async () => {}),
};
}),
upsertDocMeta: vi.fn(async (roomId: string, patch: Partial<SessionMeta>) => {
events.push(`meta:${roomId}:${patch.isArchived === true ? 'archived' : 'other'}`);
const current = sessionMetas.get(roomId);
Expand Down Expand Up @@ -199,7 +212,7 @@ function createHarness(options?: {
machineName: 'machine',
cliVersion: '0.0.0',
closeSessionTerminals,
cloudPort: createTestCloudPort(),
cloudPort: options?.cloudPort ?? createTestCloudPort(),
}
);
const internal = handler as unknown as MessageHandlerInternals;
Expand All @@ -224,6 +237,63 @@ function createHarness(options?: {
}

describe('MessageHandler terminal cleanup', () => {
it('cleans a local worktree when its project metadata exists only in the machine Flock', async () => {
const localProjectId = 'local-project-archive' as LocalProjectId;
const testDir = fs.mkdtempSync(path.join(os.tmpdir(), 'lody-archive-project-'));
const originalDataDir = process.env.LODY_DATA_DIR;
const originalLocksDir = process.env.LODY_LOCKS_DIR;
process.env.LODY_DATA_DIR = path.join(testDir, 'data');
process.env.LODY_LOCKS_DIR = path.join(testDir, 'locks');
const rootPath = createLocalRepo(testDir);
const sessionId = 'session-archive-worktree' as SessionId;
const sessionMeta = {
id: sessionId,
machineId: 'machine-1',
createdAt: '2026-01-01T00:00:00.000Z',
userId: 'user-1',
cliType: 'codex',
agentType: 'codex',
status: SessionStatusFactory.idle(),
project: { kind: 'local', localProjectId },
isWorktree: true,
} as SessionMeta;
const machineFlockRows = [
{
key: machineFlockKeys.localProject(localProjectId),
value: {
id: localProjectId,
name: 'Project',
rootPath,
createdAtMs: 1,
},
},
];
try {
const manager = getWorktreeManager({
repoId: deriveRepoIdFromLocalProjectPath(rootPath),
source: { kind: 'local-shared', originalRootPath: rootPath },
logger: createSilentLogger(),
});
const worktree = await manager.createWorktree(sessionId);
const { handler, sessionManager } = createHarness({
sessionId,
sessionMetas: [sessionMeta],
machineFlockRows,
});

await handler.archiveSessionResources(sessionId);

expect(fs.existsSync(worktree.hostPath)).toBe(false);
expect(sessionManager.archiveSession).toHaveBeenCalledWith(sessionId);
} finally {
if (originalDataDir === undefined) delete process.env.LODY_DATA_DIR;
else process.env.LODY_DATA_DIR = originalDataDir;
if (originalLocksDir === undefined) delete process.env.LODY_LOCKS_DIR;
else process.env.LODY_LOCKS_DIR = originalLocksDir;
fs.rmSync(testDir, { recursive: true, force: true });
}
});

it('closes session terminals when archiving resources even without an active session', async () => {
const { handler, sessionId, closeSessionTerminals, sessionManager } = createHarness();

Expand All @@ -233,6 +303,101 @@ describe('MessageHandler terminal cleanup', () => {
expect(sessionManager.terminateSession).not.toHaveBeenCalled();
});

it('keeps a cloud archive request queued until the Machine Flock watcher is authoritative', async () => {
const localProjectId = 'local-project-archive-authority' as LocalProjectId;
const testDir = fs.mkdtempSync(path.join(os.tmpdir(), 'lody-archive-authority-'));
const originalDataDir = process.env.LODY_DATA_DIR;
const originalLocksDir = process.env.LODY_LOCKS_DIR;
process.env.LODY_DATA_DIR = path.join(testDir, 'data');
process.env.LODY_LOCKS_DIR = path.join(testDir, 'locks');
const rootPath = createLocalRepo(testDir);
const sessionId = 'session-archive-awaiting-authority' as SessionId;
const sessionMeta = {
id: sessionId,
machineId: 'machine-1',
createdAt: '2026-01-01T00:00:00.000Z',
userId: 'user-1',
cliType: 'codex',
agentType: 'codex',
status: SessionStatusFactory.idle(),
project: { kind: 'local', localProjectId },
isWorktree: true,
} as SessionMeta;
const machineFlockRows = [
{
key: machineFlockKeys.localProject(localProjectId),
value: { id: localProjectId, name: 'Project', rootPath, createdAtMs: 1 },
},
];
const cloudPort = { ...createTestCloudPort(), kind: 'cloud' } as CloudPort;
try {
const manager = getWorktreeManager({
repoId: deriveRepoIdFromLocalProjectPath(rootPath),
source: { kind: 'local-shared', originalRootPath: rootPath },
logger: createSilentLogger(),
});
const worktree = await manager.createWorktree(sessionId);
const { handler, repo, sessionManager } = createHarness({
sessionId,
sessionMetas: [sessionMeta],
archiveSessionIds: [sessionId],
includeLegacySessionDeleteRequest: false,
machineFlockRows,
cloudPort,
});
handler.machineFlockCommandWatcher = { isReady: false };

await handler.processArchiveRequests();

expect(fs.existsSync(worktree.hostPath)).toBe(true);
expect(sessionManager.archiveSession).not.toHaveBeenCalled();
expect(repo.upsertDocMeta).not.toHaveBeenCalledWith(
getSessionRoomId(sessionId),
expect.objectContaining({ isArchived: true })
);
expect(repo.upsertDocMeta).not.toHaveBeenCalledWith(
getMachineRoomId('machine-1'),
expect.objectContaining({ needToArchiveSessions: {} })
);

handler.machineFlockCommandWatcher.isReady = true;
await handler.processArchiveRequests();

expect(fs.existsSync(worktree.hostPath)).toBe(false);
expect(sessionManager.archiveSession).toHaveBeenCalledWith(sessionId);
expect(repo.upsertDocMeta).toHaveBeenCalledWith(
getSessionRoomId(sessionId),
expect.objectContaining({ isArchived: true })
);
expect(repo.upsertDocMeta).toHaveBeenCalledWith(
getMachineRoomId('machine-1'),
expect.objectContaining({ needToArchiveSessions: {} })
);
} finally {
if (originalDataDir === undefined) delete process.env.LODY_DATA_DIR;
else process.env.LODY_DATA_DIR = originalDataDir;
if (originalLocksDir === undefined) delete process.env.LODY_LOCKS_DIR;
else process.env.LODY_LOCKS_DIR = originalLocksDir;
fs.rmSync(testDir, { recursive: true, force: true });
}
});

it('keeps archive requests queued when the Machine Flock read fails', async () => {
const sessionId = 'session-archive-flock-read-failure' as SessionId;
const { handler, sessionManager, repo } = createHarness({
sessionId,
archiveSessionIds: [sessionId],
machineFlockOpenError: new Error('temporary Machine Flock failure'),
});

await expect(handler.processArchiveRequests()).resolves.toBeUndefined();
expect(sessionManager.archiveSession).not.toHaveBeenCalled();
expect(repo.upsertDocMeta).not.toHaveBeenCalledWith(
getSessionRoomId(sessionId),
expect.objectContaining({ isArchived: true })
);
});

it('closes parent and active child terminals before permanent deletion cleanup', async () => {
const childSessionId = 'child-1' as SessionId;
const { handler, sessionId, closeSessionTerminals } = createHarness({
Expand Down
Loading