diff --git a/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.test.ts b/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.test.ts index 6cee039fe6..a1e3de78e2 100644 --- a/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.test.ts +++ b/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.test.ts @@ -57,6 +57,7 @@ import type { import { PRIME_AGENT_DAEMON_TRANSCRIPT_MAX_MESSAGES, primeAgentDaemonImageDigest, + decodePrimeAgentDaemonEvent, type PrimeDaemonEvent, type PrimeDaemonMessage, type PrimeDaemonPromptLifecycleCancellationResult, @@ -3057,6 +3058,127 @@ describe("PrimeAgentDaemonAdapter", () => { ); } + for (const changed of [false, true]) { + it.effect( + `verifies a background refinement before a later tool snapshot: changed=${changed}`, + () => + Effect.scoped( + Effect.gen(function* () { + const captures = makeCaptures(); + captures.correlatedPromptLifecycleAvailable = true; + captures.correlatedRecoveryProofEpoch = 0; + captures.correlatedPromptObserved = yield* Queue.unbounded(); + const resolutions = + yield* Queue.unbounded(); + captures.reconnectSnapshotResolutionObserved = (resolution) => { + Queue.offerUnsafe(resolutions, resolution); + }; + const adapter = yield* makePrimeAgentDaemonAdapter(decodeSettings({}), manager, { + instanceId, + runtimeFactory: fakeRuntimeFactory(captures), + }); + const subscription = yield* subscribe(adapter); + yield* adapter.startSession({ + threadId, + cwd: process.cwd(), + runtimeMode: "full-access", + }); + const refinement = decodePrimeAgentDaemonEvent({ + type: "session_event", + event: { + type: "message_end", + message: { + role: "custom", + customType: "refinement_outcome", + display: true, + content: "private memory update", + details: { refinementId: "refine-1", edits: [] }, + timestamp: 1, + }, + }, + }); + if (refinement?._tag !== "MessageCompleted") throw new Error("missing refinement"); + yield* offer(captures, { ...refinement, attribution: { scope: "session" } }); + // Drain startup notifications, then wait for an ordered event behind the refinement. + yield* Queue.takeAll(subscription.observed); + yield* offer(captures, { _tag: "ConnectionStatus", status: "connected" }); + yield* awaitObservedType(subscription.observed, "session.state.changed"); + const turnFiber = yield* adapter + .sendTurn({ threadId, input: "Read fixture" }) + .pipe(Effect.forkChild); + const correlationId = yield* Queue.take(captures.correlatedPromptObserved); + const delivered = lifecycleSnapshot(correlationId, "delivered", 2); + yield* offer(captures, { _tag: "PromptLifecycleUpdated", lifecycle: delivered }); + const prompt = { + role: "user", + timestamp: 2, + text: "Read fixture", + imageMimeTypes: [], + imageDigests: [], + } satisfies PrimeDaemonMessage; + const call = { + ...assistantMessage("", "toolUse"), + timestamp: 3, + toolCalls: [{ id: "read-1", name: "read" }], + }; + for (const message of [prompt, call]) { + yield* offer(captures, { + _tag: "MessageCompleted", + message, + attribution: { scope: "prompt", correlationId }, + }); + } + const result = { + role: "toolResult", + timestamp: 4, + toolCallId: "read-1", + toolName: "read", + text: "fixture", + imageMimeTypes: [], + isError: false, + } satisfies PrimeDaemonMessage; + yield* offer(captures, { + ...initialSnapshot(), + state: { ...initialSnapshot().state, isStreaming: true, messageCount: 4 }, + messages: [ + changed ? { ...refinement.message, timestamp: 999 } : refinement.message, + prompt, + call, + result, + ], + orderedSnapshot: true, + replayContinuity: "unknown", + connectionGeneration: 0, + correlatedProofEpoch: 0, + promptLifecycles: { records: [delivered], expired: [] }, + }); + expect((yield* Queue.take(resolutions)).reconciled).toBe(!changed); + if (!changed) { + yield* offer(captures, { + _tag: "MessageCompleted", + message: { ...assistantMessage("Read complete"), timestamp: 5 }, + attribution: { scope: "prompt", correlationId }, + }); + yield* offer(captures, { + _tag: "PromptLifecycleUpdated", + lifecycle: lifecycleSnapshot(correlationId, "completed", 3, { usage }), + }); + } + const settled = yield* Fiber.join(turnFiber); + const events = subscription.events.filter((event) => event.turnId === settled.turnId); + expect(events.findLast((event) => event.type === "turn.completed")).toMatchObject({ + payload: { state: changed ? "failed" : "completed" }, + }); + expect(events.filter((event) => event.type === "runtime.error")).toHaveLength( + changed ? 1 : 0, + ); + expect(encodeUnknownJson(subscription.events)).not.toContain("private memory"); + expect(encodeUnknownJson(subscription.events)).not.toContain("refinementOutcome"); + }), + ).pipe(Effect.provide(testLayer)), + ); + } + it.effect("accepts exact complete capable recovery with already observed output", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/provider/prime/PrimeAgentDaemonEvents.test.ts b/apps/server/src/provider/prime/PrimeAgentDaemonEvents.test.ts index 978b78ad53..822f5983d9 100644 --- a/apps/server/src/provider/prime/PrimeAgentDaemonEvents.test.ts +++ b/apps/server/src/provider/prime/PrimeAgentDaemonEvents.test.ts @@ -142,6 +142,50 @@ describe("PrimeAgentDaemonEvents", () => { } }); + it("retains refinement outcome identity in live events and snapshots without private content", () => { + const refinement = { + role: "custom", + customType: "refinement_outcome", + display: true, + content: "private memory update", + details: { refinementId: "refine-1", edits: [{ content: "private memory" }] }, + timestamp: 10, + }; + const completed = decodePrimeAgentDaemonEvent( + sessionEvent({ type: "message_end", message: refinement }), + ); + expect(completed).toMatchObject({ + _tag: "MessageCompleted", + message: { + role: "refinementOutcome", + timestamp: 10, + contentDigest: expect.stringMatching(/^[a-f0-9]{64}$/), + }, + }); + const snapshot = decodePrimeAgentDaemonEvent({ + type: "session_resynced", + snapshot: { state: { ...state, messageCount: 1 }, messages: [refinement] }, + }); + if (completed?._tag !== "MessageCompleted") throw new Error("missing refinement"); + expect(snapshot).toMatchObject({ + _tag: "SessionResynced", + state: { messageCount: 1 }, + messages: [completed.message], + }); + expect( + decodePrimeAgentDaemonEvent(sessionEvent({ type: "message_start", message: refinement })), + ).toEqual({ _tag: "MessageStarted", message: completed.message }); + expect(JSON.stringify(snapshot)).not.toContain("private"); + for (const changed of [ + { ...refinement, content: "changed" }, + { ...refinement, details: { refinementId: "refine-2" } }, + ]) { + expect( + decodePrimeAgentDaemonEvent(sessionEvent({ type: "message_end", message: changed })), + ).not.toEqual(completed); + } + }); + it("does not recognize unrelated or visible custom messages as hidden harness digests", () => { for (const hidden of [ { customType: "other", display: false, details: { digest: "x" } }, diff --git a/apps/server/src/provider/prime/PrimeAgentDaemonEvents.ts b/apps/server/src/provider/prime/PrimeAgentDaemonEvents.ts index 26465170c1..e904cdcd88 100644 --- a/apps/server/src/provider/prime/PrimeAgentDaemonEvents.ts +++ b/apps/server/src/provider/prime/PrimeAgentDaemonEvents.ts @@ -159,11 +159,23 @@ const PrimeAgentDaemonHarnessDigestMessage = Schema.Struct({ timestamp: Schema.Number, }); +// Refinement writes a native custom message even when no user turn is active. +// Keep its identity in the transcript; the separate refinement event owns UI projection. +const PrimeAgentDaemonRefinementOutcomeMessage = Schema.Struct({ + role: Schema.Literal("custom"), + customType: Schema.Literal("refinement_outcome"), + display: Schema.Boolean, + content: Schema.Union([Schema.String, Schema.Array(Schema.Union([textContent, imageContent]))]), + details: Schema.Unknown, + timestamp: Schema.Number, +}); + export const PrimeAgentDaemonMessage = Schema.Union([ PrimeAgentDaemonUserMessage, PrimeAgentDaemonAssistantMessage, PrimeAgentDaemonToolResultMessage, PrimeAgentDaemonHarnessDigestMessage, + PrimeAgentDaemonRefinementOutcomeMessage, ]); export type PrimeAgentDaemonMessage = typeof PrimeAgentDaemonMessage.Type; @@ -763,6 +775,12 @@ export interface PrimeDaemonPlanUpdate { } export type PrimeDaemonMessage = + | { + readonly role: "refinementOutcome"; + readonly timestamp: number; + /** Private refinement content participates in continuity, never client projection. */ + readonly contentDigest: string; + } | { readonly role: "harnessDigest"; readonly timestamp: number; @@ -1401,6 +1419,15 @@ function mapMessage(value: PrimeAgentDaemonMessage): PrimeDaemonMessage; function mapMessage(value: PrimeAgentDaemonMessage): PrimeDaemonMessage { switch (value.role) { case "custom": + if (value.customType === "refinement_outcome") { + return { + role: "refinementOutcome", + timestamp: value.timestamp, + contentDigest: NodeCrypto.createHash("sha256") + .update(JSON.stringify([value.display, value.content, value.details]), "utf8") + .digest("hex"), + }; + } return { role: "harnessDigest", timestamp: value.timestamp,