diff --git a/src/routes/chatSessionRoutes.ts b/src/routes/chatSessionRoutes.ts index d7ba002..5d6b2cc 100644 --- a/src/routes/chatSessionRoutes.ts +++ b/src/routes/chatSessionRoutes.ts @@ -54,6 +54,16 @@ const taskExecutionPayloadSchema = z.object({ workflow: z.string().trim().min(1).max(120).optional(), }); +const taskExecutionHistoryQuerySchema = z.object({ + run_id: z.string().uuid(), + session_id: z.string().trim().min(1).max(128), +}); + +const taskExecutionDebugPayloadSchema = z.object({ + run_id: z.string().uuid(), + session_id: z.string().trim().min(1).max(128), +}).strict(); + const sessionListQuerySchema = z.object({ purpose: z.enum(["ordinary_chat", "task_execution", "task_collaboration"]).optional(), task_id: z.string().uuid().optional(), @@ -240,6 +250,35 @@ export const registerChatSessionRoutes = ( } }; + const resolveTaskExecutionHistory = async ( + projectId: string, + credential: ProductCredential, + taskId: string, + runId: string, + sessionId: string, + ) => { + const validationSessionId = `task-history-access-${crypto.randomUUID()}`; + let taskTitle: string; + bindOperationalTaskCredential(validationSessionId, credential); + try { + taskTitle = (await operationalTaskClient.getAgentContext(validationSessionId, taskId)).task.title; + } finally { + releaseOperationalTaskCredential(validationSessionId); + clearOperationalTaskLease(validationSessionId); + } + + const record = await sessionMetadataStore.getTaskExecutionBinding( + projectId, + taskId, + runId, + sessionId, + ); + if (!record) return null; + const state = await sessionUiStateStore.read(toSessionUiStateContext(record)); + if (!state) return null; + return { record, state, taskTitle }; + }; + chatRouter.post("/session", async (req, res) => { const parsed = createSessionPayloadSchema.safeParse(req.body ?? {}); if (!parsed.success) { @@ -413,6 +452,152 @@ export const registerChatSessionRoutes = ( }); }); + chatRouter.get("/task/:task_id/execution-session", async (req, res) => { + const taskId = z.string().uuid().safeParse(req.params.task_id); + const query = taskExecutionHistoryQuerySchema.safeParse(req.query); + if (!taskId.success || !query.success) { + res.status(400).json({ message: "valid task_id, run_id and session_id are required" }); + return; + } + const authContext = getLocalAgentContext(req); + const credential = getTrustedProductCredential(req); + if (!authContext.projectId) { + res.status(403).json({ message: "task execution history requires project context" }); + return; + } + if (!credential) { + res.status(401).json({ message: "task execution history requires product credentials" }); + return; + } + + let history: Awaited>; + try { + history = await resolveTaskExecutionHistory( + authContext.projectId, + credential, + taskId.data, + query.data.run_id, + query.data.session_id, + ); + } catch (error) { + res.status(error && typeof error === "object" && "status" in error && typeof error.status === "number" ? error.status : 502).json({ + message: error instanceof Error ? error.message : "task access validation failed", + }); + return; + } + if (!history) { + res.status(404).json({ message: "task execution session not found" }); + return; + } + const { record, state, taskTitle } = history; + const runStatus = getSessionRunStatus(record.sessionId); + res.setHeader("Cache-Control", "no-store"); + res.json({ + id: record.sessionId, + session_id: record.sessionId, + title: `${taskTitle} · 执行记录`, + created_at: record.createdAt, + updated_at: record.updatedAt, + status: record.status, + messages: await withInteractionReplies(record, state.messages ?? []), + execution_plans: state.executionPlans ?? [], + is_streaming: runStatus === "running", + run_status: runStatus, + purpose: "task_execution", + task_id: record.taskId, + read_only: true, + }); + }); + + chatRouter.post("/task/:task_id/execution-session/debug", async (req, res) => { + const taskId = z.string().uuid().safeParse(req.params.task_id); + const payload = taskExecutionDebugPayloadSchema.safeParse(req.body); + if (!taskId.success || !payload.success) { + res.status(400).json({ message: "valid task_id, run_id and session_id are required" }); + return; + } + const authContext = getLocalAgentContext(req); + const credential = getTrustedProductCredential(req); + if (!authContext.projectId) { + res.status(403).json({ message: "task execution debugging requires project context" }); + return; + } + if (!credential) { + res.status(401).json({ message: "task execution debugging requires product credentials" }); + return; + } + + let history: Awaited>; + try { + history = await resolveTaskExecutionHistory( + authContext.projectId, + credential, + taskId.data, + payload.data.run_id, + payload.data.session_id, + ); + } catch (error) { + res.status(error && typeof error === "object" && "status" in error && typeof error.status === "number" ? error.status : 502).json({ + message: error instanceof Error ? error.message : "task access validation failed", + }); + return; + } + if (!history) { + res.status(404).json({ message: "task execution session not found" }); + return; + } + + const metadataContext = { + actorKey: authContext.actorKey, + projectId: authContext.projectId, + projectKey: authContext.projectKey, + userId: authContext.userId, + }; + const existing = (await sessionMetadataStore.list(metadataContext, { + purpose: "task_collaboration", + taskId: taskId.data, + })).find((record) => record.parentSessionId === history.record.sessionId); + if (existing) { + res.json({ + session_id: existing.sessionId, + title: existing.title, + purpose: readPurpose(existing), + task_id: existing.taskId, + parent_session_id: existing.parentSessionId, + created: false, + status: existing.status, + }); + return; + } + + const runtimeSession = await runtime.createSession(undefined, metadataContext); + const { record } = await sessionMetadataStore.ensure({ + ...metadataContext, + sessionId: runtimeSession.id, + parentSessionId: history.record.sessionId, + purpose: "task_collaboration", + taskId: taskId.data, + }); + const titled = await sessionMetadataStore.touch(record, { title: `${history.taskTitle} · 调试` }); + const messages = await withInteractionReplies(history.record, history.state.messages ?? []); + await sessionUiStateStore.write( + toSessionUiStateContext(titled), + buildForkedSessionUiState({ messages }, { + keepMessageCount: messages.length, + targetSessionId: titled.sessionId, + }), + ); + res.status(201).json({ + session_id: titled.sessionId, + title: titled.title, + purpose: readPurpose(titled), + task_id: titled.taskId, + parent_session_id: titled.parentSessionId, + created: true, + status: titled.status, + }); + }); + chatRouter.post("/task-executions", async (req, res) => { const parsed = taskExecutionPayloadSchema.safeParse(req.body); if (!parsed.success) { diff --git a/tests/routes/task-collaboration-sessions.test.ts b/tests/routes/task-collaboration-sessions.test.ts index ede857f..f280ba6 100644 --- a/tests/routes/task-collaboration-sessions.test.ts +++ b/tests/routes/task-collaboration-sessions.test.ts @@ -14,6 +14,7 @@ const http = globalThis.fetch; const projectId = "10000000-0000-4000-8000-000000000001"; const firstTask = "50000000-0000-4000-8000-000000000001"; const secondTask = "50000000-0000-4000-8000-000000000002"; +const firstRun = "70000000-0000-4000-8000-000000000001"; let root: string | undefined; let server: Server | undefined; @@ -75,9 +76,15 @@ async function fixture() { await new Promise((resolve) => server!.once("listening", resolve)); const address = server.address(); if (!address || typeof address === "string") throw new Error("missing server address"); - const request = async (path: string, method = "GET", user = "owner") => { + const request = async (path: string, method = "GET", user = "owner", body?: unknown) => { const response = await http(`http://127.0.0.1:${address.port}/api/v1${path}`, { - method, headers: { Authorization: `Bearer ${user}`, "X-Project-Id": projectId }, + method, + headers: { + Authorization: `Bearer ${user}`, + "X-Project-Id": projectId, + ...(body === undefined ? {} : { "Content-Type": "application/json" }), + }, + body: body === undefined ? undefined : JSON.stringify(body), }); return { status: response.status, body: await response.json() as Record }; }; @@ -87,7 +94,20 @@ async function fixture() { await states.write({ ...context, sessionId }, { sessionId, messages: [], isTitleManuallyEdited: manuallyEdited }); return titled; }; - return { request, seed, sessions, context, taskReads, setAvailable: (value: boolean) => { available = value; }, + const seedExecution = async (sessionId: string, taskId: string, runId: string) => { + const workerContext = { + actorKey: toActorKey("service:operational-worker"), + projectKey: context.projectKey, + projectId, + userId: "operational-worker", + }; + await sessions.ensure({ ...workerContext, sessionId, taskId, runId, purpose: "task_execution" }); + await states.write({ ...workerContext, sessionId }, { sessionId, messages: [ + { id: "execution-user", role: "user", parts: [{ type: "text", text: "执行后台调查任务" }] }, + { id: "execution-assistant", role: "assistant", parts: [{ type: "text", text: "已完成压力与流量证据核对" }] }, + ] }); + }; + return { request, seed, seedExecution, sessions, context, taskReads, setAvailable: (value: boolean) => { available = value; }, setTaskReadHook: (hook: () => Promise) => { onTaskRead = hook; }, waitForTitleAbort: () => { waitForAbort = true; }, didAbort: () => observedAbort }; } @@ -106,6 +126,72 @@ it("names new collaboration sessions after their verified tasks", async () => { expect(taskReads).toHaveLength(2); }); +it("opens an exact task execution binding as a read-only cross-actor history", async () => { + const { request, seedExecution } = await fixture(); + await seedExecution("worker-execution", firstTask, firstRun); + + expect((await request("/session/worker-execution")).status).toBe(404); + const result = await request(`/task/${firstTask}/execution-session?run_id=${firstRun}&session_id=worker-execution`); + + expect(result.status).toBe(200); + expect(result.body).toMatchObject({ + session_id: "worker-execution", + title: "南桥管道爆管调查 · 执行记录", + purpose: "task_execution", + task_id: firstTask, + read_only: true, + is_streaming: false, + }); + expect(result.body.messages.map((message: { id: string }) => message.id)).toEqual([ + "execution-user", + "execution-assistant", + ]); + expect((await request(`/task/${firstTask}/execution-session?run_id=${secondTask}&session_id=worker-execution`)).status).toBe(404); + expect((await request(`/task/${secondTask}/execution-session?run_id=${firstRun}&session_id=worker-execution`)).status).toBe(404); +}); + +it("creates one user-owned debug branch from an exact task execution history", async () => { + const { request, seedExecution } = await fixture(); + await seedExecution("worker-execution", firstTask, firstRun); + + const first = await request(`/task/${firstTask}/execution-session/debug`, "POST", "owner", { + run_id: firstRun, + session_id: "worker-execution", + }); + + expect(first.status).toBe(201); + expect(first.body).toMatchObject({ + title: "南桥管道爆管调查 · 调试", + purpose: "task_collaboration", + task_id: firstTask, + parent_session_id: "worker-execution", + created: true, + }); + const debug = await request(`/session/${first.body.session_id}`); + expect(debug.body.messages.map((message: { id: string }) => message.id)).toEqual([ + "execution-user", + "execution-assistant", + ]); + + const reused = await request(`/task/${firstTask}/execution-session/debug`, "POST", "owner", { + run_id: firstRun, + session_id: "worker-execution", + }); + expect(reused.status).toBe(200); + expect(reused.body).toMatchObject({ session_id: first.body.session_id, created: false }); +}); + +it("validates current task access before revealing an execution history", async () => { + const { request, seedExecution, setAvailable } = await fixture(); + await seedExecution("protected-execution", firstTask, firstRun); + setAvailable(false); + + const result = await request(`/task/${firstTask}/execution-session?run_id=${firstRun}&session_id=protected-execution`); + + expect(result.status).toBe(503); + expect(result.body.message).toBe("task context unavailable"); +}); + it("repairs returned legacy history without changing timestamps or custom names, then searches the repaired titles", async () => { const { request, seed, sessions, context, taskReads } = await fixture(); const first = await seed("legacy-first", firstTask, "任务协作");