jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(app): introduce attachment storage layer and draft lifecycle management

Browse Source 
Replace inline data-URL image handling with a persistent attachment storage service, add draft lifecycle states (active/abandoned/sent) with generation tracking, and update fetchAgent to return project placement alongside agent snapshots.
This commit is contained in:
Mohamed Boudra
2026-02-22 16:22:21 +07:00
parent 2652bdda2e
commit 0c710b09f5
55 changed files with 2381 additions and 350 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/client/daemon-client.ts
+18 -7
View File
@@ -12,6 +12,7 @@ import {
import type {
AgentStreamEventPayload,
AgentSnapshotPayload,
ProjectPlacementPayload,
AgentPermissionResolvedMessage,
CreateAgentRequestMessage,
FileDownloadTokenResponse,
@@ -202,6 +203,7 @@ export type CreateAgentRequestOptions = {
provider?: AgentProvider
cwd?: string
initialPrompt?: string
clientMessageId?: string
outputSchema?: Record<string, unknown>
images?: CreateAgentRequestMessage['images']
git?: GitSetupOptions
@@ -284,6 +286,11 @@ export type FetchAgentsOptions = Omit<FetchAgentsRequest, 'type' | 'requestId'>
export type FetchAgentsEntry = FetchAgentsPayload['entries'][number]
export type FetchAgentsPageInfo = FetchAgentsPayload['pageInfo']
export type FetchAgentResult = {
agent: AgentSnapshotPayload
project: ProjectPlacementPayload | null
}
export type WaitForFinishResult = {
status: 'idle' | 'error' | 'permission' | 'timeout'
final: AgentSnapshotPayload | null
@@ -1089,7 +1096,7 @@ export class DaemonClient {
})
}
async fetchAgent(agentId: string, requestId?: string): Promise<AgentSnapshotPayload | null> {
async fetchAgent(agentId: string, requestId?: string): Promise<FetchAgentResult | null> {
const resolvedRequestId = this.createRequestId(requestId)
const message = SessionInboundMessageSchema.parse({
type: 'fetch_agent_request',
@@ -1114,7 +1121,10 @@ export class DaemonClient {
if (payload.error) {
throw new Error(payload.error)
}
return payload.agent
if (!payload.agent) {
return null
}
return { agent: payload.agent, project: payload.project ?? null }
}
private resubscribeCheckoutDiffSubscriptions(): void {
@@ -1158,6 +1168,7 @@ export class DaemonClient {
requestId,
config,
...(options.initialPrompt ? { initialPrompt: options.initialPrompt } : {}),
...(options.clientMessageId ? { clientMessageId: options.clientMessageId } : {}),
...(options.outputSchema ? { outputSchema: options.outputSchema } : {}),
...(options.images && options.images.length > 0 ? { images: options.images } : {}),
...(options.git ? { git: options.git } : {}),
@@ -2316,9 +2327,9 @@ export class DaemonClient {
predicate: (snapshot: AgentSnapshotPayload) => boolean,
timeout = 60000
): Promise<AgentSnapshotPayload> {
const initial = await this.fetchAgent(agentId).catch(() => null)
if (initial && predicate(initial)) {
return initial
const initialResult = await this.fetchAgent(agentId).catch(() => null)
if (initialResult && predicate(initialResult.agent)) {
return initialResult.agent
}
const deadline = Date.now() + timeout
@@ -2374,8 +2385,8 @@ export class DaemonClient {
}
pollInFlight = true
try {
const snapshot = await this.fetchAgent(agentId).catch(() => null)
maybeResolve(snapshot)
const result = await this.fetchAgent(agentId).catch(() => null)
maybeResolve(result?.agent ?? null)
} finally {
pollInFlight = false
}
packages/server/src/server/daemon-client.e2e.test.ts
+13 -9
View File
@@ -140,7 +140,7 @@ describe("daemon client E2E", () => {
expect(ctx.client.isConnected).toBe(true);
const agents = await ctx.client.fetchAgents();
expect(Array.isArray(agents)).toBe(true);
expect(Array.isArray(agents.entries)).toBe(true);
const cwd = tmpCwd();
const created = await ctx.client.createAgent({
@@ -218,10 +218,12 @@ describe("daemon client E2E", () => {
const result = await ctx.client.archiveAgent(created.id);
expect(result.archivedAt).toBeTruthy();
const archived = await ctx.client.fetchAgent(created.id);
expect(archived).not.toBeNull();
expect(archived?.archivedAt).toBeTruthy();
expect(archived?.status).not.toBe("running");
const archivedResult = await ctx.client.fetchAgent(created.id);
expect(archivedResult).not.toBeNull();
expect(archivedResult?.agent.archivedAt).toBeTruthy();
expect(archivedResult?.agent.status).not.toBe("running");
expect(archivedResult?.project).not.toBeNull();
expect(archivedResult?.project?.checkout.cwd).toBe(cwd);
const runningAgents = await ctx.client.fetchAgents({
filter: { includeArchived: true, statuses: ["running"] },
@@ -342,8 +344,10 @@ describe("daemon client E2E", () => {
}),
]);
expect(Array.isArray(first)).toBe(true);
expect(Array.isArray(second)).toBe(true);
expect(first.requestId).toBe(firstRequestId);
expect(second.requestId).toBe(secondRequestId);
expect(Array.isArray(first.entries)).toBe(true);
expect(Array.isArray(second.entries)).toBe(true);
}, 15000);
test(
@@ -399,8 +403,8 @@ describe("daemon client E2E", () => {
expect(agent.id).toBeTruthy();
expect(agent.status).toBe("idle");
const fetched = await ctx.client.fetchAgent(agent.id);
expect(fetched?.id).toBe(agent.id);
const fetchedResult = await ctx.client.fetchAgent(agent.id);
expect(fetchedResult?.agent.id).toBe(agent.id);
const agentUpdate = await agentUpdatePromise;
expect(agentUpdate.payload.agent.id).toBe(agent.id);
packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts
+17 -17
View File
@@ -68,14 +68,14 @@ describe("daemon E2E", () => {
limit: 200,
projection: "projected",
});
const refreshedState = await ctx.client.fetchAgent(agent.id);
const refreshedResult = await ctx.client.fetchAgent(agent.id);
// Verify agent is still idle
expect(refreshedState?.status).toBe("idle");
expect(refreshedResult?.agent.status).toBe("idle");
// CRITICAL: The timestamp should NOT have changed
// Just opening/clicking an agent should not update its updatedAt
expect(refreshedState?.updatedAt).toBe(initialUpdatedAt);
expect(refreshedResult?.agent.updatedAt).toBe(initialUpdatedAt);
// Also clear attention (what happens when opening an agent with notification)
await ctx.client.clearAgentAttention(agent.id);
@@ -86,10 +86,10 @@ describe("daemon E2E", () => {
limit: 200,
projection: "projected",
});
const stateAfterClear = await ctx.client.fetchAgent(agent.id);
const clearResult = await ctx.client.fetchAgent(agent.id);
// Timestamp should STILL not have changed
expect(stateAfterClear?.updatedAt).toBe(initialUpdatedAt);
expect(clearResult?.agent.updatedAt).toBe(initialUpdatedAt);
// Cleanup
rmSync(cwd, { recursive: true, force: true });
@@ -341,7 +341,7 @@ describe("daemon E2E", () => {
// Initially, there should be no agents (fresh session)
const initialAgents = await ctx.client.fetchAgents();
expect(initialAgents).toHaveLength(0);
expect(initialAgents.entries).toHaveLength(0);
// Create first agent
const agent1 = await ctx.client.createAgent({
@@ -355,10 +355,10 @@ describe("daemon E2E", () => {
// fetchAgents should now return 1 agent
const afterFirst = await ctx.client.fetchAgents();
expect(afterFirst).toHaveLength(1);
expect(afterFirst[0].id).toBe(agent1.id);
expect(afterFirst.entries).toHaveLength(1);
expect(afterFirst.entries[0]?.agent.id).toBe(agent1.id);
// Title may or may not be set depending on timing
expect(afterFirst[0].cwd).toBe(cwd1);
expect(afterFirst.entries[0]?.agent.cwd).toBe(cwd1);
// Create second agent
const agent2 = await ctx.client.createAgent({
@@ -372,15 +372,15 @@ describe("daemon E2E", () => {
// fetchAgents should now return 2 agents
const afterSecond = await ctx.client.fetchAgents();
expect(afterSecond).toHaveLength(2);
expect(afterSecond.entries).toHaveLength(2);
// Verify both agents are present with correct IDs and states
const ids = afterSecond.map((a) => a.id);
const ids = afterSecond.entries.map((entry) => entry.agent.id);
expect(ids).toContain(agent1.id);
expect(ids).toContain(agent2.id);
const agent1State = afterSecond.find((a) => a.id === agent1.id);
const agent2State = afterSecond.find((a) => a.id === agent2.id);
const agent1State = afterSecond.entries.find((entry) => entry.agent.id === agent1.id)?.agent;
const agent2State = afterSecond.entries.find((entry) => entry.agent.id === agent2.id)?.agent;
// Title may or may not be set depending on timing
expect(agent1State?.cwd).toBe(cwd1);
@@ -395,12 +395,12 @@ describe("daemon E2E", () => {
// fetchAgents should now return only 1 agent
const afterDelete = await ctx.client.fetchAgents();
expect(afterDelete).toHaveLength(1);
expect(afterDelete[0].id).toBe(agent2.id);
expect(afterDelete[0].cwd).toBe(cwd2);
expect(afterDelete.entries).toHaveLength(1);
expect(afterDelete.entries[0]?.agent.id).toBe(agent2.id);
expect(afterDelete.entries[0]?.agent.cwd).toBe(cwd2);
// Verify agent1 is no longer in the list
const deletedAgent = afterDelete.find((a) => a.id === agent1.id);
const deletedAgent = afterDelete.entries.find((entry) => entry.agent.id === agent1.id);
expect(deletedAgent).toBeUndefined();
// Cleanup
packages/server/src/server/daemon-e2e/claude-autonomous-wake.real.e2e.test.ts
+10 -9
View File
@@ -545,8 +545,8 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () =
const settled = await client.waitForFinish(agent.id, 20_000);
expect(settled.status).toBe("idle");
const finalSnapshot = await client.fetchAgent(agent.id);
expect(finalSnapshot?.status).toBe("idle");
const finalResult = await client.fetchAgent(agent.id);
expect(finalResult?.agent.status).toBe("idle");
} finally {
await client.close();
await daemon.close();
@@ -750,13 +750,13 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () =
try {
secondCompletion = await client.waitForFinish(agent.id, 20_000);
} catch {
const atTimeout = await client.fetchAgent(agent.id);
const atTimeoutResult = await client.fetchAgent(agent.id);
// eslint-disable-next-line no-console
console.log(
JSON.stringify({
cycle,
phase: "second_completion_timeout",
statusAtTimeout: atTimeout?.status ?? "unknown",
statusAtTimeout: atTimeoutResult?.agent.status ?? "unknown",
})
);
secondCompletion = await client.waitForFinish(agent.id, 30_000);
@@ -830,12 +830,12 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () =
try {
secondCompletion = await client.waitForFinish(agent.id, 20_000);
} catch {
const atTimeout = await client.fetchAgent(agent.id);
const atTimeoutResult = await client.fetchAgent(agent.id);
// eslint-disable-next-line no-console
console.log(
JSON.stringify({
phase: "second_background_completion_timeout",
statusAtTimeout: atTimeout?.status ?? "unknown",
statusAtTimeout: atTimeoutResult?.agent.status ?? "unknown",
})
);
secondCompletion = await client.waitForFinish(agent.id, 30_000);
@@ -995,9 +995,9 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () =
error instanceof Error ? error : new Error(String(error ?? "wait_for_finish failed"));
}
let afterWait = await client.fetchAgent(agent.id);
let afterWaitResult = await client.fetchAgent(agent.id);
let cancelRecovered = true;
if (afterWait?.status === "running") {
if (afterWaitResult?.agent.status === "running") {
await client.cancelAgent(agent.id);
try {
const afterCancel = await client.waitForFinish(agent.id, 15_000);
@@ -1005,12 +1005,13 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () =
} catch {
cancelRecovered = false;
}
afterWait = await client.fetchAgent(agent.id);
afterWaitResult = await client.fetchAgent(agent.id);
}
expect(waitError).toBeNull();
expect(cancelRecovered).toBe(true);
const afterWait = afterWaitResult?.agent ?? null;
const sessionId =
afterWait?.persistence?.sessionId ??
afterWait?.runtimeInfo?.sessionId ??
packages/server/src/server/daemon-e2e/send-while-running-stuck-claude.real.e2e.test.ts
+3 -3
View File
@@ -102,9 +102,9 @@ describe("daemon E2E (real claude) - send while running recovery", () => {
}
await secondary.waitForFinish(agent.id, 180_000);
const finalSnapshot = await secondary.fetchAgent(agent.id);
if (finalSnapshot) {
applySnapshot(finalSnapshot);
const finalResult = await secondary.fetchAgent(agent.id);
if (finalResult) {
applySnapshot(finalResult.agent);
}
// Sending while running should clear processing even if reconnect misses the
packages/server/src/server/daemon-e2e/send-while-running-stuck.real.e2e.test.ts
+3 -3
View File
@@ -99,9 +99,9 @@ describe("daemon E2E (real codex) - send while running recovery", () => {
}
await secondary.waitForFinish(agent.id, 120_000);
const finalSnapshot = await secondary.fetchAgent(agent.id);
if (finalSnapshot) {
applySnapshot(finalSnapshot);
const finalResult = await secondary.fetchAgent(agent.id);
if (finalResult) {
applySnapshot(finalResult.agent);
}
// Sending while running should clear processing even if reconnect misses the
packages/server/src/server/session.ts
+69 -33
View File
@@ -4926,6 +4926,22 @@ export class Session {
}
}
private getFetchAgentsSortValueFromAgent(
agent: AgentSnapshotPayload,
key: FetchAgentsRequestSort['key']
): string | number | null {
switch (key) {
case 'status_priority':
return this.getStatusPriority(agent)
case 'created_at':
return Date.parse(agent.createdAt)
case 'updated_at':
return Date.parse(agent.updatedAt)
case 'title':
return agent.title?.toLocaleLowerCase() ?? ''
}
}
private compareSortValues(left: string | number | null, right: string | number | null): number {
if (left === right) {
return 0
@@ -4942,21 +4958,21 @@ export class Session {
return String(left).localeCompare(String(right))
}
private compareFetchAgentsEntries(
left: FetchAgentsResponseEntry,
right: FetchAgentsResponseEntry,
private compareFetchAgentsAgents(
left: AgentSnapshotPayload,
right: AgentSnapshotPayload,
sort: FetchAgentsRequestSort[]
): number {
for (const spec of sort) {
const leftValue = this.getFetchAgentsSortValue(left, spec.key)
const rightValue = this.getFetchAgentsSortValue(right, spec.key)
const leftValue = this.getFetchAgentsSortValueFromAgent(left, spec.key)
const rightValue = this.getFetchAgentsSortValueFromAgent(right, spec.key)
const base = this.compareSortValues(leftValue, rightValue)
if (base === 0) {
continue
}
return spec.direction === 'asc' ? base : -base
}
return left.agent.id.localeCompare(right.agent.id)
return left.id.localeCompare(right.id)
}
private encodeFetchAgentsCursor(
@@ -5050,13 +5066,13 @@ export class Session {
}
}
private compareEntryWithCursor(
entry: FetchAgentsResponseEntry,
private compareAgentWithCursor(
agent: AgentSnapshotPayload,
cursor: FetchAgentsCursor,
sort: FetchAgentsRequestSort[]
): number {
for (const spec of sort) {
const leftValue = this.getFetchAgentsSortValue(entry, spec.key)
const leftValue = this.getFetchAgentsSortValueFromAgent(agent, spec.key)
const rightValue =
cursor.values[spec.key] !== undefined ? (cursor.values[spec.key] ?? null) : null
const base = this.compareSortValues(leftValue, rightValue)
@@ -5065,7 +5081,7 @@ export class Session {
}
return spec.direction === 'asc' ? base : -base
}
return entry.agent.id.localeCompare(cursor.id)
return agent.id.localeCompare(cursor.id)
}
private async listFetchAgentsEntries(
@@ -5092,31 +5108,45 @@ export class Session {
return placementPromise
}
let entries = await Promise.all(
agents.map(async (agent) => ({
agent,
project: await getPlacement(agent.cwd),
}))
)
entries = entries.filter((entry) =>
this.matchesAgentFilter({
agent: entry.agent,
project: entry.project,
filter,
})
)
entries.sort((left, right) => this.compareFetchAgentsEntries(left, right, sort))
let candidates = [...agents]
candidates.sort((left, right) => this.compareFetchAgentsAgents(left, right, sort))
const cursorToken = request.page?.cursor
if (cursorToken) {
const cursor = this.decodeFetchAgentsCursor(cursorToken, sort)
entries = entries.filter((entry) => this.compareEntryWithCursor(entry, cursor, sort) > 0)
candidates = candidates.filter((agent) => this.compareAgentWithCursor(agent, cursor, sort) > 0)
}
const limit = request.page?.limit ?? entries.length
const pagedEntries = entries.slice(0, limit)
const hasMore = entries.length > limit
const limit = request.page?.limit ?? 200
const matchedEntries: FetchAgentsResponseEntry[] = []
const batchSize = 25
for (let start = 0; start < candidates.length && matchedEntries.length <= limit; start += batchSize) {
const batch = candidates.slice(start, start + batchSize)
const batchEntries = await Promise.all(
batch.map(async (agent) => ({
agent,
project: await getPlacement(agent.cwd),
}))
)
for (const entry of batchEntries) {
if (
!this.matchesAgentFilter({
agent: entry.agent,
project: entry.project,
filter,
})
) {
continue
}
matchedEntries.push(entry)
if (matchedEntries.length > limit) {
break
}
}
}
const pagedEntries = matchedEntries.slice(0, limit)
const hasMore = matchedEntries.length > limit
const nextCursor =
hasMore && pagedEntries.length > 0
? this.encodeFetchAgentsCursor(pagedEntries[pagedEntries.length - 1], sort)
@@ -5197,7 +5227,7 @@ export class Session {
if (!resolved.ok) {
this.emit({
type: 'fetch_agent_response',
payload: { requestId, agent: null, error: resolved.error },
payload: { requestId, agent: null, project: null, error: resolved.error },
})
return
}
@@ -5206,14 +5236,20 @@ export class Session {
if (!agent) {
this.emit({
type: 'fetch_agent_response',
payload: { requestId, agent: null, error: `Agent not found: ${resolved.agentId}` },
payload: {
requestId,
agent: null,
project: null,
error: `Agent not found: ${resolved.agentId}`,
},
})
return
}
const project = await this.buildProjectPlacement(agent.cwd)
this.emit({
type: 'fetch_agent_response',
payload: { requestId, agent, error: null },
payload: { requestId, agent, project, error: null },
})
}
packages/server/src/shared/messages.create-agent-client-message-id.test.ts
+23
View File
@@ -0,0 +1,23 @@
import { describe, expect, it } from "vitest";
import { SessionInboundMessageSchema } from "./messages";
describe("create_agent_request clientMessageId", () => {
it("accepts clientMessageId for stable initial prompt transfer", () => {
const parsed = SessionInboundMessageSchema.parse({
type: "create_agent_request",
requestId: "req-1",
clientMessageId: "client-msg-1",
config: {
provider: "claude",
cwd: "/tmp/project",
},
initialPrompt: "hello",
});
expect(parsed.type).toBe("create_agent_request");
if (parsed.type !== "create_agent_request") {
throw new Error("Expected create_agent_request");
}
expect(parsed.clientMessageId).toBe("client-msg-1");
});
});
packages/server/src/shared/messages.ts
+3 -1
View File
@@ -491,7 +491,7 @@ export const FetchAgentsRequestMessageSchema = z.object({
.optional(),
page: z
.object({
limit: z.number().int().positive().max(1000),
limit: z.number().int().positive().max(200),
cursor: z.string().min(1).optional(),
})
.optional(),
@@ -578,6 +578,7 @@ export const CreateAgentRequestMessageSchema = z.object({
config: AgentSessionConfigSchema,
worktreeName: z.string().optional(),
initialPrompt: z.string().optional(),
clientMessageId: z.string().optional(),
outputSchema: z.record(z.unknown()).optional(),
images: z
.array(
@@ -1469,6 +1470,7 @@ export const FetchAgentResponseMessageSchema = z.object({
payload: z.object({
requestId: z.string(),
agent: AgentSnapshotPayloadSchema.nullable(),
project: ProjectPlacementPayloadSchema.nullable().optional(),
error: z.string().nullable(),
}),
})