diff --git a/apps/server/tests/full-flow.integration.test.ts b/apps/server/tests/full-flow.integration.test.ts new file mode 100644 index 0000000..6be0612 --- /dev/null +++ b/apps/server/tests/full-flow.integration.test.ts @@ -0,0 +1,159 @@ +import { + FileAttachmentStore, + HarnessSession, + InMemoryApprovalBroker, + InMemoryEventStore, + InMemoryModelConnectionStore, + InMemorySessionRegistry, + SafeWorkspace, + SkillEventInference, + SkillIndex, + executeMockAdsMetrics, + FakeAgentSession, +} from "@agent-studio/harness"; +import { access, mkdir, mkdtemp, readFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { dirname, join, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; +import { Readable } from "node:stream"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { buildApp } from "../src/app.js"; + +const projectRoot = resolve(dirname(fileURLToPath(import.meta.url)), "../../.."); +const png = Buffer.from("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=", "base64"); +const createdRoots: string[] = []; + +afterEach(async () => { + await Promise.all(createdRoots.splice(0).map((root) => rm(root, { recursive: true, force: true }))); +}); + +describe("full fake Agent Studio flow", () => { + it.each(["approve", "deny", "timeout"] as const)("handles the %s report decision and cleans up", async (decision) => { + const root = await mkdtemp(join(tmpdir(), "agent-studio-full-flow-")); + createdRoots.push(root); + const sessionsRoot = join(root, "sessions"); + await mkdir(sessionsRoot); + const events = new InMemoryEventStore(); + const approvals = new InMemoryApprovalBroker({ eventStore: events, timeoutMs: decision === "timeout" ? 30 : 5_000 }); + const models = new InMemoryModelConnectionStore({ + validateUrl: (url) => Promise.resolve({ normalizedUrl: url, addresses: ["93.184.216.34"], redirectPolicy: "same-origin" }), + }); + const attachments = new FileAttachmentStore({ sessionsRoot, eventStore: events }); + const skillPath = join(projectRoot, ".agents/skills/ads-analysis/SKILL.md"); + const fixturePath = join(projectRoot, "fixtures/ads-account.json"); + const skillIndex = new SkillIndex([{ name: "ads-analysis", source: "project", canonicalFilePath: skillPath }]); + let fake: FakeAgentSession | undefined; + let reportPath = ""; + + const registry = new InMemorySessionRegistry({ + sessionsRoot, + createSession: ({ sessionId, workspaceRoot }) => { + events.append(sessionId, { type: "session.started", payload: { cwd: "." } }); + reportPath = join(workspaceRoot, "report.md"); + return new HarnessSession({ + sessionId, + workspaceRoot, + eventStore: events, + modelStore: models, + attachmentStore: attachments, + approvalBroker: approvals, + createSkillInference: (runId) => new SkillEventInference({ sessionId, runId, eventStore: events, skillIndex }), + createAgentPort: async ({ getActiveRunId }) => { + const workspace = await SafeWorkspace.create(workspaceRoot); + fake = new FakeAgentSession(async (port, prompt) => { + const runId = getActiveRunId(); + if (runId === undefined) throw new Error("Run was not active"); + expect(prompt.images).toHaveLength(1); + expect(prompt.images[0]).toMatchObject({ type: "image", mimeType: "image/png" }); + + port.emit({ type: "tool_execution_start", toolCallId: "tool_skill", toolName: "read", args: { path: skillPath } }); + await port.flush(); + port.emit({ type: "tool_execution_end", toolCallId: "tool_skill", toolName: "read", result: "skill metadata", isError: false }); + port.emit({ type: "message_update", assistantMessageEvent: { type: "text_delta", delta: "已读取分析方法。" } }); + + const metrics = await executeMockAdsMetrics(fixturePath, { accountId: "demo-account", days: 7 }); + port.emit({ type: "tool_execution_start", toolCallId: "tool_metrics", toolName: "mock_ads_metrics", args: { accountId: "demo-account", days: 7 } }); + port.emit({ type: "tool_execution_end", toolCallId: "tool_metrics", toolName: "mock_ads_metrics", result: metrics, isError: false }); + + const report = { title: "账户总结", body: `账户总结\nROAS ${String(metrics.roas)}` }; + port.emit({ type: "tool_execution_start", toolCallId: "tool_report", toolName: "save_report_draft", args: report }); + const resolution = await approvals.request({ + approvalId: `ap_${runId}`, + sessionId, + runId, + toolCallId: "tool_report", + toolName: "save_report_draft", + arguments: report, + riskLevel: "medium", + }); + if (resolution.status === "approved") { + const result = await workspace.writeReportDraft(report); + port.emit({ type: "tool_execution_end", toolCallId: "tool_report", toolName: "save_report_draft", result, isError: false }); + } else { + port.emit({ type: "tool_execution_end", toolCallId: "tool_report", toolName: "save_report_draft", result: `Approval ${resolution.status}`, isError: true }); + } + port.emit({ type: "message_update", assistantMessageEvent: { type: "text_delta", delta: resolution.status === "approved" ? "报告已保存。" : "报告未写入。" } }); + port.emit({ type: "message_end", message: { role: "assistant" } }); + }); + return fake; + }, + }); + }, + }); + const app = buildApp({ + sessionRegistry: registry, + skillCatalog: { list: () => [{ name: "ads-analysis", description: "分析广告指标", source: "project", filePath: ".agents/skills/ads-analysis/SKILL.md", diagnostics: [] }] }, + modelStore: models, + attachmentStore: attachments, + approvalBroker: approvals, + eventStore: events, + }); + + const created = await app.inject({ method: "POST", url: "/api/sessions" }); + const { sessionId } = created.json<{ sessionId: string }>(); + const configured = await app.inject({ + method: "PUT", + url: `/api/sessions/${sessionId}/model-config`, + payload: { providerId: "training", api: "openai-responses", apiKey: "sk-test-secret", modelId: "training-model", supportsImages: true }, + }); + expect(configured.body).not.toContain("sk-test-secret"); + const image = await attachments.add(sessionId, { name: "creative.png", mediaType: "image/png", stream: Readable.from(png) }); + const sent = await app.inject({ + method: "POST", + url: `/api/sessions/${sessionId}/messages`, + payload: { text: "分析 demo-account 并保存报告", attachmentIds: [image.attachmentId] }, + }); + expect(sent.statusCode).toBe(202); + + await vi.waitFor(() => { + expect(events.listAfter(sessionId).events.some((event) => event.type === "approval.required")).toBe(true); + }); + const approval = events.listAfter(sessionId).events.find((event) => event.type === "approval.required"); + if (approval?.type !== "approval.required") throw new Error("Approval event missing"); + if (decision !== "timeout") { + const response = await app.inject({ method: "POST", url: `/api/approvals/${approval.payload.approvalId}/${decision}` }); + expect(response.statusCode).toBe(200); + } + await vi.waitFor(() => { + expect(events.listAfter(sessionId).events.at(-1)?.type).toBe("run.completed"); + }); + + const allEvents = events.listAfter(sessionId).events; + expect(allEvents.map((event) => event.type)).toEqual(expect.arrayContaining([ + "session.started", "attachment.uploaded", "user.message", "run.started", "skill.requested", "skill.loaded", + "tool.requested", "tool.completed", "approval.required", "approval.resolved", "assistant.delta", "assistant.completed", "run.completed", + ])); + expect(allEvents.find((event) => event.type === "approval.resolved")).toMatchObject({ payload: { status: decision === "approve" ? "approved" : decision === "deny" ? "denied" : "expired" } }); + expect(JSON.stringify(allEvents)).not.toContain("base64"); + expect(JSON.stringify(allEvents)).not.toContain("sk-test-secret"); + expect(fake?.prompts[0]?.images[0]?.data).toBeTruthy(); + if (decision === "approve") expect(await readFile(reportPath, "utf8")).toContain("账户总结"); + else await expect(access(reportPath)).rejects.toMatchObject({ code: "ENOENT" }); + + const deleted = await app.inject({ method: "DELETE", url: `/api/sessions/${sessionId}` }); + expect(deleted.statusCode).toBe(200); + await expect(access(join(sessionsRoot, sessionId))).rejects.toMatchObject({ code: "ENOENT" }); + expect(fake?.disposed).toBe(true); + await app.close(); + }); +}); diff --git a/apps/web/tests/full-flow.test.tsx b/apps/web/tests/full-flow.test.tsx new file mode 100644 index 0000000..380ba27 --- /dev/null +++ b/apps/web/tests/full-flow.test.tsx @@ -0,0 +1,145 @@ +import { render, screen, within } from "@testing-library/react"; +import userEvent from "@testing-library/user-event"; +import type { HarnessEvent } from "@agent-studio/shared"; +import { describe, expect, it, vi } from "vitest"; +import { App } from "../src/App.js"; +import { + createInitialHarnessState, + reduceHarnessEvent, + type HarnessClientState, +} from "../src/store/reducer.js"; + +const base = { + sessionId: "ses_full_flow", + timestamp: "2026-09-08T08:00:00.000Z", +}; + +function event(value: Omit): T { + return { + ...base, + ...value, + eventId: `evt_${value.sequence}`, + } as T; +} + +function fullFlowSnapshot(): HarnessClientState { + const events: HarnessEvent[] = [ + event({ + type: "user.message", + sequence: 1, + payload: { messageId: "msg_user", text: "分析最近 7 天素材并生成报告", attachmentIds: ["att_1"] }, + }), + event({ type: "run.started", sequence: 2, runId: "run_1", payload: { modelId: "gpt-test" } }), + event({ + type: "skill.requested", + sequence: 3, + runId: "run_1", + payload: { name: "ads-analysis", source: "project" }, + }), + event({ + type: "skill.loaded", + sequence: 4, + runId: "run_1", + payload: { name: "ads-analysis", source: "project" }, + }), + event({ + type: "assistant.delta", + sequence: 5, + runId: "run_1", + payload: { messageId: "msg_assistant", delta: "已读取分析方法。" }, + }), + event({ + type: "tool.requested", + sequence: 6, + runId: "run_1", + payload: { + toolCallId: "tool_metrics", + toolName: "mock_ads_metrics", + arguments: { accountId: "demo-account", days: 7 }, + }, + }), + event({ + type: "tool.completed", + sequence: 7, + runId: "run_1", + payload: { + toolCallId: "tool_metrics", + toolName: "mock_ads_metrics", + result: { spend: 1024.5, revenue: 3088.2, roas: 3.01, conversions: 44 }, + }, + }), + event({ + type: "tool.requested", + sequence: 8, + runId: "run_1", + payload: { + toolCallId: "tool_report", + toolName: "save_report_draft", + arguments: { title: "广告素材分析报告", body: "近 7 天投放表现与下一步建议。" }, + }, + }), + event({ + type: "approval.required", + sequence: 9, + runId: "run_1", + payload: { + approvalId: "ap_full_flow", + toolCallId: "tool_report", + toolName: "save_report_draft", + arguments: { title: "广告素材分析报告", body: "近 7 天投放表现与下一步建议。" }, + riskLevel: "medium", + expiresAt: "2099-09-08T08:05:00.000Z", + }, + }), + ]; + + const state = events.reduce(reduceHarnessEvent, createInitialHarnessState({ + sessionId: base.sessionId, + skills: [{ + name: "ads-analysis", + description: "分析广告素材表现", + source: "project", + filePath: ".agents/skills/ads-analysis/SKILL.md", + diagnostics: [], + }], + })); + return { ...state, connection: "open" }; +} + +describe("complete training flow", () => { + it("renders the message, skill, tools, timeline, and server-owned approval as one flow", async () => { + const user = userEvent.setup(); + const approve = vi.fn().mockResolvedValue({ approvalId: "ap_full_flow", status: "approved" }); + const snapshot = fullFlowSnapshot(); + const store = { + getSnapshot: () => snapshot, + subscribe: () => () => undefined, + sendMessage: vi.fn(), + cancel: vi.fn(), + start: vi.fn().mockResolvedValue(undefined), + getModelConfig: vi.fn().mockResolvedValue({ + modelConfig: { configured: true, source: "session", modelId: "gpt-test", supportsImages: true }, + }), + approve, + deny: vi.fn(), + }; + + render(); + + expect(screen.getByText("分析最近 7 天素材并生成报告")).toBeVisible(); + expect(screen.getByText("已读取分析方法。")).toBeVisible(); + expect(screen.getByText("分析广告素材表现")).toBeVisible(); + + const metrics = screen.getByRole("region", { name: "广告指标工具" }); + expect(within(metrics).getByText("3.01")).toBeVisible(); + const report = screen.getByRole("region", { name: "保存报告草稿工具" }); + expect(within(report).getByText("等待审批")).toBeVisible(); + + expect(screen.getByRole("dialog", { name: "允许写入报告草稿吗？" })).toBeVisible(); + expect(screen.getByText("approval.required")).toBeVisible(); + + await user.click(screen.getByRole("button", { name: "批准并保存" })); + expect(approve).toHaveBeenCalledTimes(1); + expect(approve).toHaveBeenCalledWith("ap_full_flow"); + }); +}); diff --git a/apps/web/tests/setup.ts b/apps/web/tests/setup.ts index 1f07705..491ce19 100644 --- a/apps/web/tests/setup.ts +++ b/apps/web/tests/setup.ts @@ -13,3 +13,4 @@ class TestResizeObserver implements ResizeObserver { globalThis.ResizeObserver = TestResizeObserver; URL.createObjectURL = () => "blob:test"; URL.revokeObjectURL = () => undefined; +HTMLElement.prototype.scrollTo = () => undefined; diff --git a/packages/harness/src/index.ts b/packages/harness/src/index.ts index ccf4ca6..eca5b06 100644 --- a/packages/harness/src/index.ts +++ b/packages/harness/src/index.ts @@ -19,6 +19,7 @@ export * from "./sessions/in-memory-session-registry.js"; export * from "./sessions/session-registry.js"; export * from "./skills/skill-catalog.js"; export * from "./skills/skill-event-inference.js"; +export * from "./testing/fake-agent-session.js"; export * from "./tools/mock-ads-metrics.js"; export * from "./tools/save-report-draft.js"; export * from "./tools/tool-policy.js"; diff --git a/packages/harness/src/sessions/harness-session.ts b/packages/harness/src/sessions/harness-session.ts index e3418dc..890b2b2 100644 --- a/packages/harness/src/sessions/harness-session.ts +++ b/packages/harness/src/sessions/harness-session.ts @@ -82,8 +82,17 @@ export class HarnessSession { eventStore: this.#eventStore, ...(skillInference === undefined ? {} : { skillInference }), }); + let eventQueue = Promise.resolve(); + let eventError: unknown; const unsubscribe = port.subscribe((event) => { - void adapter.observe(event); + eventQueue = eventQueue.then(async () => { + if (eventError !== undefined) return; + try { + await adapter.observe(event); + } catch (error) { + eventError = error; + } + }); }); const active: ActiveRun = { runId, @@ -108,7 +117,11 @@ export class HarnessSession { void Promise.resolve() .then(() => port.prompt(message.text, { images })) - .then(() => this.#complete(active)) + .then(async () => { + await eventQueue; + if (eventError !== undefined) throw eventError; + this.#complete(active); + }) .catch((error: unknown) => this.#fail(active, error)); return { runId }; } diff --git a/packages/harness/src/testing/fake-agent-session.ts b/packages/harness/src/testing/fake-agent-session.ts new file mode 100644 index 0000000..2e009f0 --- /dev/null +++ b/packages/harness/src/testing/fake-agent-session.ts @@ -0,0 +1,46 @@ +import type { ImageContent } from "@earendil-works/pi-ai"; +import type { AgentSessionPort } from "../agent/agent-session-port.js"; + +type Listener = (event: unknown) => void; +type Driver = ( + session: FakeAgentSession, + prompt: { text: string; images: ImageContent[] }, +) => Promise; + +export class FakeAgentSession implements AgentSessionPort { + readonly #listeners = new Set(); + readonly prompts: Array<{ text: string; images: ImageContent[] }> = []; + public aborted = false; + public disposed = false; + + public constructor(private readonly driver: Driver) {} + + public subscribe(listener: Listener): () => void { + this.#listeners.add(listener); + return () => this.#listeners.delete(listener); + } + + public async prompt(text: string, options: { images: ImageContent[] }): Promise { + const prompt = { text, images: options.images.map((image) => ({ ...image })) }; + this.prompts.push(prompt); + await this.driver(this, prompt); + } + + public emit(event: unknown): void { + for (const listener of [...this.#listeners]) listener(event); + } + + public async flush(): Promise { + await new Promise((resolve) => setTimeout(resolve, 0)); + } + + public abort(): Promise { + this.aborted = true; + return Promise.resolve(); + } + + public dispose(): void { + this.disposed = true; + this.#listeners.clear(); + } +} diff --git a/packages/harness/tests/fakes/fake-agent-session.ts b/packages/harness/tests/fakes/fake-agent-session.ts new file mode 100644 index 0000000..4562753 --- /dev/null +++ b/packages/harness/tests/fakes/fake-agent-session.ts @@ -0,0 +1 @@ +export { FakeAgentSession } from "../../src/testing/fake-agent-session.js"; diff --git a/packages/harness/tests/session-registry.test.ts b/packages/harness/tests/session-registry.test.ts index f8371ca..abc725c 100644 --- a/packages/harness/tests/session-registry.test.ts +++ b/packages/harness/tests/session-registry.test.ts @@ -129,6 +129,44 @@ describe("HarnessSession", () => { await expect(session.startRun({ text: "look", attachmentIds: ["att_1"] })).rejects.toMatchObject({ code: "MODEL_DOES_NOT_SUPPORT_IMAGES" }); expect(createAgentPort).not.toHaveBeenCalled(); }); + + it("serializes async agent events before completing the run", async () => { + const port = new FakePort(); + const events = new harness.InMemoryEventStore(); + let finishSkillLookup: (() => void) | undefined; + const order: string[] = []; + const session = new harness.HarnessSession({ + sessionId: "ses_1", + workspaceRoot: root, + eventStore: events, + modelStore: await configuredModelStore(), + attachmentStore: { toPiImages: () => Promise.resolve([]), removeAll: () => Promise.resolve() }, + approvalBroker: { cancelForRun: () => 0, cancelForSession: () => 0 }, + createAgentPort: () => Promise.resolve(port), + createSkillInference: () => ({ + onReadRequested: () => new Promise((resolve) => { + finishSkillLookup = () => { + order.push("requested"); + resolve(); + }; + }), + onReadCompleted: () => { order.push("completed"); }, + }), + }); + + await session.startRun({ text: "load skill", attachmentIds: [] }); + for (const listener of port.listeners) { + listener({ type: "tool_execution_start", toolCallId: "call_1", toolName: "read", args: { path: "/skill/SKILL.md" } }); + listener({ type: "tool_execution_end", toolCallId: "call_1", toolName: "read", result: "ok", isError: false }); + } + port.resolvePrompt(); + await vi.waitFor(() => expect(finishSkillLookup).toBeDefined()); + expect(events.listAfter("ses_1").events.at(-1)?.type).not.toBe("run.completed"); + + finishSkillLookup?.(); + await vi.waitFor(() => expect(events.listAfter("ses_1").events.at(-1)?.type).toBe("run.completed")); + expect(order).toEqual(["requested", "completed"]); + }); }); describe("InMemorySessionRegistry", () => {