Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
122 changes: 122 additions & 0 deletions apps/server/src/provider/prime/PrimeAgentDaemonAdapter.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@ import type {
import {
PRIME_AGENT_DAEMON_TRANSCRIPT_MAX_MESSAGES,
primeAgentDaemonImageDigest,
decodePrimeAgentDaemonEvent,
type PrimeDaemonEvent,
type PrimeDaemonMessage,
type PrimeDaemonPromptLifecycleCancellationResult,
Expand Down Expand Up @@ -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<string>();
const resolutions =
yield* Queue.unbounded<FakeCaptures["reconnectResolutions"][number]>();
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* () {
Expand Down
44 changes: 44 additions & 0 deletions apps/server/src/provider/prime/PrimeAgentDaemonEvents.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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" } },
Expand Down
27 changes: 27 additions & 0 deletions apps/server/src/provider/prime/PrimeAgentDaemonEvents.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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,
Expand Down
Loading