mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-10 02:15:02 +00:00
refactor(server): share one timestamp across workspace archive
This commit is contained in:
@@ -14285,3 +14285,32 @@ test("turns keep distinct identities across an idle runtime collection and resum
|
||||
rmSync(workdir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("workspace archive preserves a loaded history agent stamp and stamps cascaded children", async () => {
|
||||
const workdir = mkdtempSync(join(tmpdir(), "workspace-stamp-"));
|
||||
const storage = new AgentStorage(join(workdir, "agents"), logger);
|
||||
const manager = new AgentManager({
|
||||
clients: { codex: new TestAgentClient() },
|
||||
registry: storage,
|
||||
logger,
|
||||
});
|
||||
try {
|
||||
const parent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
|
||||
workspaceId: undefined,
|
||||
});
|
||||
const child = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
|
||||
labels: { [PARENT_AGENT_ID_LABEL]: parent.id },
|
||||
workspaceId: undefined,
|
||||
});
|
||||
const stamp = "2026-11-09T00:00:00.000Z";
|
||||
await manager.archiveAgent(parent.id, stamp);
|
||||
expect((await storage.get(parent.id))?.archivedAt).toBe(stamp);
|
||||
expect((await storage.get(child.id))?.archivedAt).toBe(stamp);
|
||||
await ensureAgentLoaded(parent.id, { agentManager: manager, agentStorage: storage, logger });
|
||||
await manager.archiveAgent(parent.id, "2026-11-09T01:00:00.000Z");
|
||||
expect((await storage.get(parent.id))?.archivedAt).toBe(stamp);
|
||||
} finally {
|
||||
await storage.flush();
|
||||
rmSync(workdir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
@@ -2196,8 +2196,8 @@ export class AgentManager {
|
||||
}
|
||||
}
|
||||
|
||||
async archiveAgent(agentId: string): Promise<{ archivedAt: string }> {
|
||||
return this.runLifecycleMutation(agentId, () => this.archiveAgentUnlocked(agentId));
|
||||
async archiveAgent(agentId: string, archivedAt?: string): Promise<{ archivedAt: string }> {
|
||||
return this.runLifecycleMutation(agentId, () => this.archiveAgentUnlocked(agentId, archivedAt));
|
||||
}
|
||||
|
||||
private async archiveAgentUnlocked(
|
||||
@@ -2240,7 +2240,7 @@ export class AgentManager {
|
||||
await this.syncNativeArchiveState(stored.provider, stored.persistence, "archive");
|
||||
this.discardRetainedAgentState(agentId);
|
||||
|
||||
await this.cascadeArchiveChildren(stored);
|
||||
await this.cascadeArchiveChildren({ ...stored, archivedAt });
|
||||
|
||||
return { archivedAt };
|
||||
}
|
||||
@@ -2250,7 +2250,7 @@ export class AgentManager {
|
||||
// children so subagent fleets don't outlive their orchestrator. Detached
|
||||
// handoff agents omit this label, so they stand outside the cascade.
|
||||
private async cascadeArchiveChildren(
|
||||
parent: Pick<StoredAgentRecord, "id" | "workspaceId">,
|
||||
parent: Pick<StoredAgentRecord, "id" | "workspaceId" | "archivedAt">,
|
||||
): Promise<void> {
|
||||
const registry = this.registry;
|
||||
if (!registry) {
|
||||
@@ -2284,9 +2284,9 @@ export class AgentManager {
|
||||
this.dispatchStoredAgentState(detached.record);
|
||||
}
|
||||
} else if (this.agents.has(currentChild.id)) {
|
||||
await this.archiveAgentUnlocked(currentChild.id);
|
||||
await this.archiveAgentUnlocked(currentChild.id, parent.archivedAt!);
|
||||
} else {
|
||||
await this.archiveSnapshotUnlocked(currentChild.id, new Date().toISOString());
|
||||
await this.archiveSnapshotUnlocked(currentChild.id, parent.archivedAt!);
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -2302,7 +2302,7 @@ export class AgentManager {
|
||||
if (!current || getParentAgentIdFromLabels(current.labels) !== parentAgentId) {
|
||||
return;
|
||||
}
|
||||
await this.archiveAgentUnlocked(current.id);
|
||||
await this.archiveAgentUnlocked(current.id, parent.archivedAt!);
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -2312,7 +2312,7 @@ export class AgentManager {
|
||||
archivedAt = new Date().toISOString(),
|
||||
): Promise<ArchivedStoredAgentRecord> {
|
||||
const archivedRecord = await this.persistArchivedRecord(record, {
|
||||
archivedAt,
|
||||
archivedAt: record.archivedAt ?? archivedAt,
|
||||
updatedAt: archivedAt,
|
||||
});
|
||||
|
||||
@@ -2853,7 +2853,7 @@ export class AgentManager {
|
||||
if (!nextRecord.internal) this.dispatchStoredAgentState(nextRecord);
|
||||
|
||||
await this.fireAgentArchived(agentId);
|
||||
await this.cascadeArchiveChildren(record);
|
||||
await this.cascadeArchiveChildren(nextRecord);
|
||||
|
||||
return nextRecord;
|
||||
}
|
||||
|
||||
@@ -29,7 +29,11 @@ export interface AutoArchiveArchiveOptions {
|
||||
findWorkspaceIdForCwd: (cwd: string) => Promise<string | null>;
|
||||
listActiveWorkspaces: () => Promise<ActiveWorkspaceRef[]>;
|
||||
getAutoArchivedChangeRequestUrl: (workspaceId: string) => Promise<string | null>;
|
||||
archiveWorkspaceRecord: (workspaceId: string, context?: WorkspaceArchiveContext) => Promise<void>;
|
||||
archiveWorkspaceRecord: (
|
||||
workspaceId: string,
|
||||
context?: WorkspaceArchiveContext,
|
||||
archivedAt?: string,
|
||||
) => Promise<void>;
|
||||
markWorkspaceArchiving: (workspaceIds: Iterable<string>, archivingAt: string) => void;
|
||||
clearWorkspaceArchiving: (workspaceIds: Iterable<string>) => void;
|
||||
emitWorkspaceUpdatesForWorkspaceIds: (workspaceIds: Iterable<string>) => Promise<void>;
|
||||
@@ -101,10 +105,14 @@ export async function archiveIfSafe(input: {
|
||||
agentStorage: options.agentStorage,
|
||||
findWorkspaceIdForCwd: options.findWorkspaceIdForCwd,
|
||||
listActiveWorkspaces: options.listActiveWorkspaces,
|
||||
archiveWorkspaceRecord: (workspaceIdToArchive) =>
|
||||
options.archiveWorkspaceRecord(workspaceIdToArchive, {
|
||||
autoArchivedChangeRequestUrl: pullRequest.url,
|
||||
}),
|
||||
archiveWorkspaceRecord: (workspaceIdToArchive, archivedAt) =>
|
||||
options.archiveWorkspaceRecord(
|
||||
workspaceIdToArchive,
|
||||
{
|
||||
autoArchivedChangeRequestUrl: pullRequest.url,
|
||||
},
|
||||
archivedAt,
|
||||
),
|
||||
emitWorkspaceUpdatesForWorkspaceIds: options.emitWorkspaceUpdatesForWorkspaceIds,
|
||||
markWorkspaceArchiving: options.markWorkspaceArchiving,
|
||||
clearWorkspaceArchiving: options.clearWorkspaceArchiving,
|
||||
|
||||
@@ -1102,11 +1102,13 @@ export async function createPaseoDaemon(
|
||||
const archiveWorkspaceRecordExternal = async (
|
||||
workspaceId: string,
|
||||
context?: WorkspaceArchiveContext,
|
||||
archivedAt?: string,
|
||||
) => {
|
||||
const existingWorkspace = await archivePersistedWorkspaceRecord({
|
||||
workspaceId,
|
||||
workspaceRegistry,
|
||||
context,
|
||||
archivedAt,
|
||||
});
|
||||
if (!existingWorkspace || existingWorkspace.archivedAt) return;
|
||||
teardownArchivedWorkspaceRuntime(workspaceId);
|
||||
@@ -1301,7 +1303,8 @@ export async function createPaseoDaemon(
|
||||
findWorkspaceIdForCwd: findWorkspaceIdForCwdExternal,
|
||||
listActiveWorkspaces: listActiveWorkspacesExternal,
|
||||
getWorkspace: (workspaceIdToGet) => workspaceRegistry.get(workspaceIdToGet),
|
||||
archiveWorkspaceRecord: archiveWorkspaceRecordExternal,
|
||||
archiveWorkspaceRecord: (workspaceIdToArchive, archivedAt) =>
|
||||
archiveWorkspaceRecordExternal(workspaceIdToArchive, undefined, archivedAt),
|
||||
emitWorkspaceUpdatesForWorkspaceIds: emitWorkspaceUpdatesExternal,
|
||||
markWorkspaceArchiving: markWorkspaceArchivingExternal,
|
||||
clearWorkspaceArchiving: clearWorkspaceArchivingExternal,
|
||||
@@ -1326,7 +1329,7 @@ export async function createPaseoDaemon(
|
||||
archiveAgentCommand({ agentManager, agentStorage, logger }, agentId),
|
||||
findWorkspaceIdForCwd: findWorkspaceIdForCwdExternal,
|
||||
listActiveWorkspaces: listActiveWorkspacesExternal,
|
||||
archiveWorkspaceRecord: archiveWorkspaceRecordExternal,
|
||||
archiveWorkspaceRecord: (workspaceId) => archiveWorkspaceRecordExternal(workspaceId),
|
||||
emit: emitExternalSessionMessage,
|
||||
emitAgentRemove: async () => undefined,
|
||||
emitWorkspaceUpdatesForWorkspaceIds: emitWorkspaceUpdatesExternal,
|
||||
@@ -1422,7 +1425,8 @@ export async function createPaseoDaemon(
|
||||
findWorkspaceIdForCwd: findWorkspaceIdForCwdExternal,
|
||||
listActiveWorkspaces: listActiveWorkspacesExternal,
|
||||
getWorkspace: (workspaceIdToGet) => workspaceRegistry.get(workspaceIdToGet),
|
||||
archiveWorkspaceRecord: archiveWorkspaceRecordExternal,
|
||||
archiveWorkspaceRecord: (workspaceIdToArchive, archivedAt) =>
|
||||
archiveWorkspaceRecordExternal(workspaceIdToArchive, undefined, archivedAt),
|
||||
emitWorkspaceUpdatesForWorkspaceIds: emitWorkspaceUpdatesExternal,
|
||||
markWorkspaceArchiving: markWorkspaceArchivingExternal,
|
||||
clearWorkspaceArchiving: clearWorkspaceArchivingExternal,
|
||||
@@ -1544,7 +1548,8 @@ export async function createPaseoDaemon(
|
||||
workspaceGitService,
|
||||
findWorkspaceIdForCwd: findWorkspaceIdForCwdExternal,
|
||||
listActiveWorkspaces: listActiveWorkspacesExternal,
|
||||
archiveWorkspaceRecord: archiveWorkspaceRecordExternal,
|
||||
archiveWorkspaceRecord: (workspaceId, archivedAt) =>
|
||||
archiveWorkspaceRecordExternal(workspaceId, undefined, archivedAt),
|
||||
emitWorkspaceUpdatesForWorkspaceIds: emitWorkspaceUpdatesExternal,
|
||||
workspaceRegistry,
|
||||
projectRegistry,
|
||||
|
||||
@@ -4074,6 +4074,7 @@ export class Session {
|
||||
const removedWorkspaceIds: string[] = [];
|
||||
try {
|
||||
for (const workspaceId of activeWorkspaceIds) {
|
||||
const archivedAt = new Date().toISOString();
|
||||
await archiveWorkspaceContents(
|
||||
{
|
||||
agentManager: this.agentManager,
|
||||
@@ -4083,8 +4084,9 @@ export class Session {
|
||||
sessionLogger: this.sessionLogger,
|
||||
},
|
||||
workspaceId,
|
||||
archivedAt,
|
||||
);
|
||||
await this.archiveWorkspaceRecord(workspaceId);
|
||||
await this.archiveWorkspaceRecord(workspaceId, archivedAt);
|
||||
removedWorkspaceIds.push(workspaceId);
|
||||
}
|
||||
|
||||
@@ -5678,7 +5680,8 @@ export class Session {
|
||||
agentStorage: this.agentStorage,
|
||||
findWorkspaceIdForCwd: (cwd) => this.findWorkspaceIdForCwd(cwd),
|
||||
listActiveWorkspaces: () => this.listActiveWorkspaceRefs(),
|
||||
archiveWorkspaceRecord: (workspaceId) => this.archiveWorkspaceRecord(workspaceId),
|
||||
archiveWorkspaceRecord: (workspaceId, archivedAt) =>
|
||||
this.archiveWorkspaceRecord(workspaceId, archivedAt),
|
||||
emit: (message) => this.emit(message),
|
||||
emitWorkspaceUpdatesForWorkspaceIds: (workspaceIds) =>
|
||||
this.emitWorkspaceUpdatesForWorkspaceIds(workspaceIds),
|
||||
@@ -8019,7 +8022,8 @@ export class Session {
|
||||
findWorkspaceIdForCwd: (cwd) => this.findWorkspaceIdForCwd(cwd),
|
||||
getWorkspace: (workspaceId) => this.workspaceRegistry.get(workspaceId),
|
||||
listActiveWorkspaces: () => this.listActiveWorkspaceRefs(),
|
||||
archiveWorkspaceRecord: (workspaceId) => this.archiveWorkspaceRecord(workspaceId),
|
||||
archiveWorkspaceRecord: (workspaceId, archivedAt) =>
|
||||
this.archiveWorkspaceRecord(workspaceId, archivedAt),
|
||||
emitWorkspaceUpdatesForWorkspaceIds: (workspaceIds) =>
|
||||
this.emitWorkspaceUpdatesForWorkspaceIds(workspaceIds),
|
||||
markWorkspaceArchiving: (workspaceIds, archivingAt) =>
|
||||
|
||||
@@ -1255,3 +1255,38 @@ describe("resolveWorkspaceIdAtPath", () => {
|
||||
expect(result).toBe("ws-nested");
|
||||
});
|
||||
});
|
||||
|
||||
test("uses the operation timestamp for live agents, snapshots, and the workspace", async () => {
|
||||
const { repoDir } = createGitRepo();
|
||||
const deps = createArchiveDeps({
|
||||
paseoHome: repoDir,
|
||||
activeWorkspaces: [{ workspaceId: "stamped", cwd: repoDir, kind: "local_checkout" }],
|
||||
});
|
||||
const stamps: Array<string | undefined> = [];
|
||||
deps.agentManager.listAgents = () => [{ id: "live", workspaceId: "stamped" }] as ManagedAgent[];
|
||||
deps.agentManager.getAgent = () => ({ id: "live" }) as ManagedAgent;
|
||||
deps.agentManager.archiveAgent = async (_id, stamp) => {
|
||||
stamps.push(stamp);
|
||||
return { archivedAt: stamp! };
|
||||
};
|
||||
deps.agentStorage.listByWorkspace = async () =>
|
||||
[
|
||||
{ id: "stored", workspaceId: "stamped", archivedAt: null },
|
||||
{ id: "closed", workspaceId: "stamped", archivedAt: "2020-01-01T00:00:00.000Z" },
|
||||
] as StoredAgentRecord[];
|
||||
deps.agentManager.getAgent = (id) => (id === "live" ? ({ id } as ManagedAgent) : null);
|
||||
deps.agentManager.archiveSnapshot = async (_id, stamp) => {
|
||||
stamps.push(stamp);
|
||||
return {} as StoredAgentRecord;
|
||||
};
|
||||
deps.archiveWorkspaceRecord = async (_id, stamp) => {
|
||||
stamps.push(stamp);
|
||||
};
|
||||
await archiveByScope(deps, {
|
||||
scope: { kind: "workspace", workspaceId: "stamped" },
|
||||
requestId: "stamp",
|
||||
});
|
||||
expect(stamps).toHaveLength(3);
|
||||
expect(typeof stamps[0]).toBe("string");
|
||||
expect(new Set(stamps).size).toBe(1);
|
||||
});
|
||||
|
||||
@@ -45,7 +45,7 @@ export interface ArchiveDependencies {
|
||||
// break a same-cwd tie in favor of the worktree-kind record when archiving by
|
||||
// path (no explicit workspaceId).
|
||||
listActiveWorkspaces: () => Promise<ActiveWorkspaceRef[]>;
|
||||
archiveWorkspaceRecord: (workspaceId: string) => Promise<void>;
|
||||
archiveWorkspaceRecord: (workspaceId: string, archivedAt: string) => Promise<void>;
|
||||
emitWorkspaceUpdatesForWorkspaceIds: (workspaceIds: Iterable<string>) => Promise<void>;
|
||||
markWorkspaceArchiving: (workspaceIds: Iterable<string>, archivingAt: string) => void;
|
||||
clearWorkspaceArchiving: (workspaceIds: Iterable<string>) => void;
|
||||
@@ -165,6 +165,7 @@ async function archiveByScopeWithPriority(
|
||||
dependencies: ArchiveDependencies,
|
||||
request: ArchiveByScopeRequest,
|
||||
): Promise<ArchiveResult> {
|
||||
const archivedAt = new Date().toISOString();
|
||||
const progress: ArchiveProgress = { newlyArchivedAgentIds: new Set(), recordArchived: false };
|
||||
const checkpoint = () => {
|
||||
if (request.signal?.aborted) {
|
||||
@@ -198,7 +199,7 @@ async function archiveByScopeWithPriority(
|
||||
const { archivedAgents, archivedWorkspaceIds } = await archiveTargetRecords(
|
||||
dependencies,
|
||||
targetWorkspaceIds,
|
||||
{ requestId: request.requestId, checkpoint, progress },
|
||||
{ requestId: request.requestId, checkpoint, progress, archivedAt },
|
||||
);
|
||||
checkpoint();
|
||||
|
||||
@@ -393,21 +394,30 @@ async function resolveBackingDirectory(
|
||||
async function archiveTargetRecords(
|
||||
dependencies: ArchiveDependencies,
|
||||
targetWorkspaceIds: string[],
|
||||
options: { requestId: string; checkpoint: () => void; progress: ArchiveProgress },
|
||||
options: {
|
||||
requestId: string;
|
||||
checkpoint: () => void;
|
||||
progress: ArchiveProgress;
|
||||
archivedAt: string;
|
||||
},
|
||||
): Promise<{ archivedAgents: Set<string>; archivedWorkspaceIds: string[] }> {
|
||||
const archivedAgents = new Set<string>();
|
||||
const archivedWorkspaceIds: string[] = [];
|
||||
|
||||
const results = await Promise.allSettled(
|
||||
targetWorkspaceIds.map(async (workspaceId) => {
|
||||
const contents = await archiveWorkspaceContentsWithProgress(dependencies, workspaceId);
|
||||
const contents = await archiveWorkspaceContentsWithProgress(
|
||||
dependencies,
|
||||
workspaceId,
|
||||
options.archivedAt,
|
||||
);
|
||||
for (const agentId of contents.newlyArchivedAgentIds) {
|
||||
options.progress.newlyArchivedAgentIds.add(agentId);
|
||||
}
|
||||
// Agents and terminals are down, the record is not: an abort here leaves the workspace
|
||||
// listed, and the canceller restores the agents it reports.
|
||||
options.checkpoint();
|
||||
await dependencies.archiveWorkspaceRecord(workspaceId);
|
||||
await dependencies.archiveWorkspaceRecord(workspaceId, options.archivedAt);
|
||||
options.progress.recordArchived = true;
|
||||
return { workspaceId, agents: contents.agentIds };
|
||||
}),
|
||||
@@ -574,13 +584,16 @@ export type ArchiveWorkspaceContentsDependencies = Pick<
|
||||
export async function archiveWorkspaceContents(
|
||||
dependencies: ArchiveWorkspaceContentsDependencies,
|
||||
workspaceId: string,
|
||||
archivedAt = new Date().toISOString(),
|
||||
): Promise<Set<string>> {
|
||||
return (await archiveWorkspaceContentsWithProgress(dependencies, workspaceId)).agentIds;
|
||||
return (await archiveWorkspaceContentsWithProgress(dependencies, workspaceId, archivedAt))
|
||||
.agentIds;
|
||||
}
|
||||
|
||||
async function archiveWorkspaceContentsWithProgress(
|
||||
dependencies: ArchiveWorkspaceContentsDependencies,
|
||||
workspaceId: string,
|
||||
archivedAt: string,
|
||||
): Promise<{ agentIds: Set<string>; newlyArchivedAgentIds: string[] }> {
|
||||
const archivedAgents = new Set<string>();
|
||||
|
||||
@@ -605,7 +618,6 @@ async function archiveWorkspaceContentsWithProgress(
|
||||
archivedAgents.add(record.id);
|
||||
}
|
||||
|
||||
const archivedAt = new Date().toISOString();
|
||||
const agentIdsToArchive = [
|
||||
...new Set([
|
||||
...liveAgents.map((agent) => agent.id),
|
||||
@@ -625,7 +637,7 @@ async function archiveWorkspaceContentsWithProgress(
|
||||
Promise.allSettled(
|
||||
agentIdsToArchive.map((agentId) =>
|
||||
dependencies.agentManager.getAgent(agentId)
|
||||
? dependencies.agentManager.archiveAgent(agentId)
|
||||
? dependencies.agentManager.archiveAgent(agentId, archivedAt)
|
||||
: dependencies.agentManager.archiveSnapshot(agentId, archivedAt),
|
||||
),
|
||||
),
|
||||
|
||||
Reference in New Issue
Block a user