diff --git a/tests/runner-e2e/PRODUCT-JOURNEYS.md b/tests/runner-e2e/PRODUCT-JOURNEYS.md index bcf42b014dc..8f810b78567 100644 --- a/tests/runner-e2e/PRODUCT-JOURNEYS.md +++ b/tests/runner-e2e/PRODUCT-JOURNEYS.md @@ -93,6 +93,17 @@ bot is not a general-work fixture. The private provider supplies its exact execution identity through the `agent` capability and collects normalized run usage. The existing Cloud provider still owns authentication, conditional leases, fixture ownership, and cleanup. + +The Plan and Stop final-answer checks reload the owned task and bind its exact, +unique visible reply to the final successful native run. A settled native turn +can render its saved reply before the comments projection exists; that path also +requires exactly one matching final assistant message in the complete durable +event stream, with the same company, agent, run and validated PRP source. If the run +has a projected selected completion comment, only that attributed comment can +satisfy the check. A pending or suppressed comment can use the same proven native +reply path. A finish summary by itself, an earlier turn, a progress +bubble, or a cancelled run cannot satisfy it. Private receipts record the run, +thread anchor and observed text; classic protocol comment matchers stay strict. A model allowance is an accounting threshold; reused direct-provider routes do not acquire the onboarding gateway's request-level cash guarantee. The question observer allows one run and waits at most four minutes. Cleanup stops only diff --git a/tests/runner-e2e/native-task-reply.test.ts b/tests/runner-e2e/native-task-reply.test.ts new file mode 100644 index 00000000000..06243c6c562 --- /dev/null +++ b/tests/runner-e2e/native-task-reply.test.ts @@ -0,0 +1,75 @@ +import { expect, it } from "vitest"; +import { taskReplyBelongsToRun, uniqueBoundTaskReply, type TaskReplyObservation } from "./native-task-reply.js"; + +const run = { id: "final-run", companyId: "owned-company", agentId: "owned-agent", status: "succeeded", runtimeMode: "native", resultJson: {} }; +const observed: TaskReplyObservation = { body: "The exact final reply", anchor: "final-run:turn", settled: true, nativeFinal: true, visible: true }; +function events() { + return [{ seq: 1, runId: run.id, companyId: run.companyId, agentId: run.agentId, eventType: "item.completed", protocolSchemaVersion: 1, + sourceEventId: "event", sourceInstanceId: "instance", sourceSeq: 1, + payload: { prpEvent: { schema: "paperclip.prp.event.v1", schemaVersion: 1, runId: run.id, sourceKind: "runner", eventType: "item.completed", + sourceEventId: "event", sourceInstanceId: "instance", sourceSeq: 1, payload: { kind: "agentMessage", channel: "final", text: observed.body } } } }]; +} + +it("binds a settled native final reply to its actual run without inventing a comment", () => { + expect(taskReplyBelongsToRun(observed, run, "owned-agent", [], events())).toBe(true); + for (const change of [{ anchor: "earlier-run:turn" }, { anchor: "final-run:turnish" }, { settled: false }, { nativeFinal: false }, { visible: false }, { body: "" }]) + expect(taskReplyBelongsToRun({ ...observed, ...change }, run, "owned-agent", [], events())).toBe(false); + for (const change of [{ agentId: "other-agent" }, { status: "cancelled" }, { runtimeMode: "legacy" }]) + expect(taskReplyBelongsToRun(observed, { ...run, ...change }, "owned-agent", [], events())).toBe(false); +}); + +it("requires one exact visible answer and rejects duplicates, stale answers and extra text", () => { + expect(uniqueBoundTaskReply([observed], run, "owned-agent", [], observed.body, events())).toEqual(observed); + for (const all of [[], [observed, observed], [{ ...observed, body: observed.body + " extra" }], + [{ ...observed, anchor: "old:turn" }], [observed, { ...observed, anchor: "old:turn" }]]) + expect(uniqueBoundTaskReply(all, run, "owned-agent", [], observed.body, events())).toBeNull(); +}); + +it("accepts item-normalized final messages but rejects contradictory payload fields", () => { + const copy: any[] = events(); + const payload = copy[0].payload.prpEvent.payload; + payload.item = { type: payload.kind, channel: payload.channel, text: payload.text }; + delete payload.kind; delete payload.channel; delete payload.text; + expect(taskReplyBelongsToRun(observed, run, "owned-agent", [], copy)).toBe(true); + for (const change of [{ kind: "tool_call" }, { channel: "commentary" }, { text: "other reply" }]) { + expect(taskReplyBelongsToRun(observed, run, "owned-agent", [], [{ ...copy[0], payload: { prpEvent: { + ...copy[0].payload.prpEvent, payload: { ...payload, ...change }, + } } }])).toBe(false); + } +}); + +it("rejects a rendered finish summary with no durable final assistant message", () => { + const summaryOnly = { ...run, resultJson: { nativeResult: { summary: observed.body } } }; + expect(taskReplyBelongsToRun(observed, summaryOnly, "owned-agent", [])).toBe(false); + const changes = [ + (e: ReturnType[number]) => { e.runId = "other"; }, + (e: ReturnType[number]) => { e.companyId = "other"; }, + (e: ReturnType[number]) => { e.agentId = "other"; }, + (e: ReturnType[number]) => { e.protocolSchemaVersion = 99; }, + (e: ReturnType[number]) => { e.payload.prpEvent.sourceKind = "control_plane"; }, + (e: ReturnType[number]) => { e.payload.prpEvent.payload.channel = "commentary"; }, + (e: ReturnType[number]) => { e.payload.prpEvent.payload.text = "different reply"; }, + (e: ReturnType[number]) => { e.sourceSeq = 2; }, + ]; + for (const change of changes) { const copy = events(); change(copy[0]!); expect(taskReplyBelongsToRun(observed, run, "owned-agent", [], copy)).toBe(false); } + expect(taskReplyBelongsToRun(observed, run, "owned-agent", [], [...events(), ...events()])).toBe(false); +}); + +it("requires the selected, attributed completion comment when a presentation decision exists", () => { + const published = { ...run, resultJson: { presentationDecision: { commentId: "selected" } } }; + const comment = { id: "selected", authorAgentId: "owned-agent", createdByRunId: "final-run", body: observed.body }; + const bubble = { ...observed, anchor: "selected", nativeFinal: false, settled: false }; + expect(taskReplyBelongsToRun(bubble, published, "owned-agent", [comment])).toBe(true); + expect(taskReplyBelongsToRun(observed, published, "owned-agent", [comment])).toBe(false); + for (const change of [{ id: "other" }, { authorAgentId: "other-agent" }, { createdByRunId: "earlier-run" }, { body: "wrong reply" }]) + expect(taskReplyBelongsToRun(bubble, published, "owned-agent", [{ ...comment, ...change }])).toBe(false); + expect(taskReplyBelongsToRun(bubble, published, "owned-agent", [])).toBe(false); +}); + +it("accepts a proven native final before the selected comment is projected, and when comment creation is suppressed", () => { + for (const presentationDecision of [{ commentId: "not-yet-projected", commentAction: "create" }, { commentId: null, commentAction: "none" }]) { + const pendingProjection = { ...run, resultJson: { presentationDecision } }; + expect(taskReplyBelongsToRun(observed, pendingProjection, "owned-agent", [], events())).toBe(true); + expect(taskReplyBelongsToRun(observed, pendingProjection, "owned-agent", [])).toBe(false); + } +}); diff --git a/tests/runner-e2e/native-task-reply.ts b/tests/runner-e2e/native-task-reply.ts new file mode 100644 index 00000000000..f67e347c449 --- /dev/null +++ b/tests/runner-e2e/native-task-reply.ts @@ -0,0 +1,82 @@ +import { expect, type Page } from "@playwright/test"; +import { isValidNativePrpEnvelope } from "./native-event-envelope.js"; + +type Row = Record; +export interface TaskReplyObservation { + body: string; anchor: string | null; settled: boolean; nativeFinal: boolean; visible: boolean; +} + +/** Native task replies can render from a saved turn before a completion comment + * is projected. Respect an explicit presentation decision when one exists. */ +export function nativeFinalReplyEvents(events: Row[], run: Row) { + return events.flatMap(event => { + const envelope = event.payload?.prpEvent, payload = envelope?.payload, item = payload?.item; + if (!envelope || !Number.isSafeInteger(event.seq) || event.seq < 1 || !isValidNativePrpEnvelope(envelope, event.protocolSchemaVersion) + || event.runId !== run.id || event.companyId !== run.companyId || event.agentId !== run.agentId + || envelope.runId !== run.id || envelope.sourceKind !== "runner" || event.eventType !== "item.completed" || envelope.eventType !== event.eventType + || typeof envelope.sourceEventId !== "string" || !envelope.sourceEventId || event.sourceEventId !== envelope.sourceEventId + || typeof envelope.sourceInstanceId !== "string" || !envelope.sourceInstanceId || event.sourceInstanceId !== envelope.sourceInstanceId + || !Number.isSafeInteger(envelope.sourceSeq) || envelope.sourceSeq < 1 || event.sourceSeq !== envelope.sourceSeq + || !["agentMessage", "assistant_message"].includes(payload?.kind ?? item?.type) || (payload.channel ?? item?.channel) !== "final" + || payload.kind != null && item?.type != null && payload.kind !== item.type + || payload.channel != null && item?.channel != null && payload.channel !== item.channel) return []; + const text = typeof payload.text === "string" ? payload.text : item?.text; + if (typeof text !== "string" || !text.trim() || typeof payload.text === "string" && typeof item?.text === "string" && payload.text !== item.text) return []; + return [{ seq: event.seq, sourceEventId: envelope.sourceEventId, sourceInstanceId: envelope.sourceInstanceId, + sourceSeq: envelope.sourceSeq, protocolSchemaVersion: event.protocolSchemaVersion, body: text.trim() }]; + }); +} + +export function taskReplyBelongsToRun(observation: TaskReplyObservation, run: Row, agentId: string, comments: Row[], events: Row[] = []) { + if (typeof run.id !== "string" || !run.id || !agentId || run.agentId !== agentId || run.status !== "succeeded" || run.runtimeMode !== "native" || !observation.body || !observation.visible) return false; + const decision = run.resultJson?.presentationDecision; + if (decision != null && (typeof decision !== "object" || Array.isArray(decision))) return false; + if (decision?.commentId != null && typeof decision.commentId !== "string") return false; + const projected = typeof decision?.commentId === "string" ? comments.find(comment => comment.id === decision.commentId) : null; + if (projected) { + return observation.anchor === projected.id && projected.authorAgentId === agentId + && projected.createdByRunId === run.id && typeof projected.body === "string" && projected.body.trim() === observation.body; + } + const finals = nativeFinalReplyEvents(events, run); + return finals.length === 1 && finals[0]!.body === observation.body && observation.settled && observation.nativeFinal + && (observation.anchor === `${run.id}:turn` || observation.anchor?.startsWith(`${run.id}:turn:`) === true); +} + +export function uniqueBoundTaskReply(all: TaskReplyObservation[], run: Row, agentId: string, comments: Row[], expectedText?: string, events: Row[] = []) { + const matches = all.filter(row => taskReplyBelongsToRun(row, run, agentId, comments, events) + && (expectedText === undefined || row.body === expectedText)); + return matches.length === 1 && (expectedText === undefined || all.filter(row => row.visible && row.body === expectedText).length === 1) + ? matches[0]! : null; +} + +/** Observe the real task thread after a fresh load, rather than a finish summary + * or a provider's claimed result. The caller separately validates task/run scope. */ +export async function readBoundTaskFinalReply(input: { + page: Page; runId: string; agentId: string; expectedText?: string; deadlineAt: number; + load(): Promise<{ run: Row; comments: Row[]; events: Row[] }>; +}) { + await input.page.reload(); + const replies = input.page.getByTestId("task-chat-thread").getByTestId("task-chat-agent-bubble"); + const observe = () => replies.evaluateAll(elements => elements.map(element => ({ + body: (element as HTMLElement).innerText.trim(), + anchor: element.closest("[data-thread-anchor]")?.getAttribute("data-thread-anchor") ?? null, + settled: element.closest('[data-testid="task-chat-turn"]')?.getAttribute("data-settled") === "true", + nativeFinal: element.closest('[data-testid="task-chat-final-response"]') !== null, + visible: element.getClientRects().length > 0 && getComputedStyle(element).visibility !== "hidden" && getComputedStyle(element).display !== "none", + }))); + const read = async () => { + const state = await input.load(), all = await observe(); + const match = state.run.id === input.runId ? uniqueBoundTaskReply(all, state.run, input.agentId, state.comments, input.expectedText, state.events) : null; + return { ...state, match }; + }; + const timeout = input.deadlineAt - Date.now(); + if (timeout <= 0) throw new Error("native_reply_deadline"); + await expect.poll(async () => { + return Boolean((await read()).match); + }, { timeout, intervals: [1000, 2000, 5000] }).toBe(true); + const { match, run, events } = await read(); + if (!match) throw new Error("native_reply_changed"); + const source = match.anchor === run.resultJson?.presentationDecision?.commentId ? "posted-comment" : "native-turn"; + return { runId: input.runId, agentId: input.agentId, observedAfterReload: true, + source, finalEvent: source === "native-turn" ? nativeFinalReplyEvents(events, run)[0] : null, ...match }; +} diff --git a/tests/runner-e2e/product-journeys/agent-flows.ts b/tests/runner-e2e/product-journeys/agent-flows.ts index 679e5555d11..0492a2a15ad 100644 --- a/tests/runner-e2e/product-journeys/agent-flows.ts +++ b/tests/runner-e2e/product-journeys/agent-flows.ts @@ -12,6 +12,7 @@ import { decideProtectedActionThroughUi, readServiceBriefing, serviceBriefingChe import { editCreatedSkillThroughStudio } from "../skill-studio-flow.js"; import { downloadStoryArtifact, runArtifactOracle } from "../story-artifact.js"; import { collectRunEvents } from "../run-observations.js"; +import { readBoundTaskFinalReply } from "../native-task-reply.js"; import { createTaskThroughUi, submitTaskReply } from "../user-actions.js"; import { agentJourneyRuns } from "./catalog.js"; import { JourneyError, type JourneyContext, type JourneyEvidence, type TargetSession } from "./types.js"; @@ -92,6 +93,21 @@ export async function runAgentJourney(id: AgentJourney, context: JourneyContext, const settled = (issue: Row, count: number) => waitFor(options.deadlineAt, () => read(issue), state => state.currentIssue.status === "done" && state.taskRuns.length === count && state.taskRuns.every(run => !active(run)) && state.interactions.every(row => row.status !== "pending"), "agent_task_not_completed"); + const finalReply = async (issue: Row, state: TaskState, marker: string) => { + const runs = state.taskRuns.filter(run => run.status === "succeeded"); + if (!runs.length || runs.some(run => !Number.isFinite(Date.parse(run.finishedAt)))) throw new JourneyError("failed", "final_run_time_missing"); + const run = [...runs].sort((a, b) => Date.parse(a.finishedAt) - Date.parse(b.finishedAt)).at(-1)!; + await open(issue); + const receipt = await readBoundTaskFinalReply({ page: context.page, runId: run.id, agentId: agent.id, + load: async () => { + const [latest, events] = await Promise.all([read(issue), collectRunEvents((afterSeq, limit) => + context.api.get(`/api/heartbeat-runs/${run.id}/events?afterSeq=${afterSeq}&limit=${limit}`))]); + return { run: latest.taskRuns.find(row => row.id === run.id)!, comments: latest.comments, events }; + }, + expectedText: marker, deadlineAt: options.deadlineAt }); + await retain(`${issue.id}-final-reply.json`, receipt); + return receipt; + }; const reply = async (issue: Row, message: string) => { const before = (await read(issue)).comments; await open(issue); @@ -124,9 +140,8 @@ export async function runAgentJourney(id: AgentJourney, context: JourneyContext, const final = await settled(issue, 3); const marker = fixture.buildVisibleMarker(nonce); check("plan-revision-and-approval", receipt.draftInteraction.id !== receipt.revisedInteraction.id); - check("plan-final-answer", final.comments.some(row => row.authorAgentId === agent.id && row.body?.trim() === marker)); - await open(issue); await context.page.reload(); - await expect(context.page.getByText(marker, { exact: true }).last()).toBeVisible(); + await finalReply(issue, final, marker); + check("plan-final-answer"); } else if (id === "build-revise") { await runArtifactOracle(null, "base", options.deadlineAt); const issue = await create("", SLUGIFY_REQUIREMENTS); @@ -240,8 +255,8 @@ export async function runAgentJourney(id: AgentJourney, context: JourneyContext, const marker = redirected.marker; const userReply = redirected.reply as Row; const final = await settled(issue, 3); - check("new-direction-completed", final.comments.some(row => row.authorAgentId === agent.id && row.body?.trim() === marker - && row.createdByRunId !== original.id)); + const completion = await finalReply(issue, final, marker); + check("new-direction-completed", completion.runId !== original.id); check("stopped-run-remained-terminal", final.taskRuns.find(run => run.id === original.id)?.status === "cancelled"); const afterEvents = await runActivity(); check("stopped-run-no-later-events", nativeCancellationHasNoLaterWork(afterEvents) diff --git a/tests/runner-e2e/product-journeys/launch.ts b/tests/runner-e2e/product-journeys/launch.ts index bb18c868ea4..601c1238832 100644 --- a/tests/runner-e2e/product-journeys/launch.ts +++ b/tests/runner-e2e/product-journeys/launch.ts @@ -103,7 +103,7 @@ export async function launchProductJourneys(args: string[], repositoryRoot: stri const source = resolveRunnerE2ESource({ sha, ref: null, workflowRunUrl: null }); const definitionHash = createHash("sha256").update(JSON.stringify(productJourneys)); for (const name of ["catalog.ts", "task-properties.ts", "basic-flows.ts", "basic-evidence.ts", "agent-flows.ts", "success-contract.ts", "product-screenshot.ts", "cloud-onboarding.ts", "onboarding-connect.ts", "onboarding-first-task.ts", "workspace-deferral.ts", "onboarding-pre-acceptance.ts", "artifact-evidence.ts", "fixtures.ts", "launch.ts", "types.ts", "collection.ts", "local-target.ts"]) definitionHash.update(await readFile(path.join(import.meta.dirname, name))); - for (const file of ["catalog.ts", "api.ts", "run-observations.ts", "user-actions.ts", "plan-revision-flow.ts", "skill-studio-flow.ts", "protected-action-flow.ts", "story-artifact.ts", "stop-redirect-flow.ts", "everyday-cases.ts", "everyday-delivery.ts", "everyday-decisions.ts", "connection-guidance-evidence.ts", "everyday-artifact.py"]) definitionHash.update(await readFile(path.join(import.meta.dirname, "..", file))); + for (const file of ["catalog.ts", "api.ts", "run-observations.ts", "native-task-reply.ts", "native-event-envelope.ts", "user-actions.ts", "plan-revision-flow.ts", "skill-studio-flow.ts", "protected-action-flow.ts", "story-artifact.ts", "stop-redirect-flow.ts", "everyday-cases.ts", "everyday-delivery.ts", "everyday-decisions.ts", "connection-guidance-evidence.ts", "everyday-artifact.py"]) definitionHash.update(await readFile(path.join(import.meta.dirname, "..", file))); definitionHash.update(runnerSuites.flatMap(suite => suite.tasks).find(task => task.id === "ask-question")!.buildPrompt("definition")); const suiteDefinitionHash = definitionHash.digest("hex"); const expected = config.journeys.map(id => `product-journeys.${config.id}.${id === "cloud-onboarding" ? config.modelProfileId : isAgentJourney(id) ? "configured-agent" : "browser"}.${config.executionEnvironment}.${id}`);