Skip to content

Commit fe0da99

Browse files
Yash-Singh1Yash-Singh1
committed
fix(chat): retain final replies before a steer
- Preserve Codex assistant message phases through orchestration - Keep completed final answers visible when a later steer interrupts the run
1 parent 7e565c5 commit fe0da99

7 files changed

Lines changed: 225 additions & 74 deletions

File tree

‎apps/mobile/src/lib/threadActivity.test.ts‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1309,6 +1309,22 @@ describe("buildThreadFeed", () => {
13091309
expect(rows(turn, "zen")).toEqual(["user", "final"]);
13101310
const [steer, answer] = [say("steer", 2, true, true), say("answer", 3)];
13111311
expect(rows([user, steer, answer, final], "zen")).toEqual(["user", "steer", "answer", "final"]);
1312+
const commentary = { ...step, phase: "commentary" as const };
1313+
expect(rows([user, commentary, steer, final], "zen")).toEqual(["user", "steer", "final"]);
1314+
for (const phase of ["commentary", "partial_answer", "final_answer", undefined] as const) {
1315+
const partial = { ...answer, ...(phase === undefined ? {} : { phase }) };
1316+
const interrupted = buildThreadFeed(
1317+
[user, passed, partial, steer, final].map((item, index) => projected(item, index)),
1318+
);
1319+
expect(
1320+
deriveThreadFeedPresentation(interrupted, null, new Set()).some(
1321+
(row) => row.type === "message" && row.message.id === answer.messageId,
1322+
),
1323+
).toBe(phase === "final_answer");
1324+
expect(rows([user, partial, steer, final], "zen").includes("answer")).toBe(
1325+
phase === "final_answer",
1326+
);
1327+
}
13121328
expect(rows([user, steer, step, passed, final], "zen")).toEqual(["user", "steer", "final"]);
13131329
expect(rows([user, steer, answer, step], "zen", running)).toEqual([
13141330
"user",

‎apps/mobile/src/lib/threadActivity.ts‎

Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1073,13 +1073,33 @@ export function messageFoldRunId(
10731073
return null;
10741074
}
10751075

1076+
function answersBeforeSteers(feed: ReadonlyArray<ThreadFeedEntry>): ReadonlySet<MessageId> {
1077+
const ids = new Set<MessageId>();
1078+
let previous: ThreadFeedEntry | undefined;
1079+
for (const entry of feed) {
1080+
if (
1081+
entry.type === "message" &&
1082+
isSteerMessage(entry.message) &&
1083+
previous?.type === "message" &&
1084+
previous.message.role === "assistant" &&
1085+
previous.message.projectedItem?.item.type === "assistant_message" &&
1086+
previous.message.projectedItem.item.phase === "final_answer"
1087+
) {
1088+
ids.add(previous.message.id);
1089+
}
1090+
previous = entry;
1091+
}
1092+
return ids;
1093+
}
1094+
10761095
function deriveThreadFeedRunFolds(
10771096
feed: ReadonlyArray<ThreadFeedEntry>,
10781097
latestRun: ThreadFeedLatestRun | null,
10791098
runlessWorkActive: boolean,
10801099
feedFailedRunIds: ReadonlySet<RunId>,
10811100
): ReadonlyMap<string, ThreadFeedRunFold> {
10821101
const terminalAssistantMessageIdByRun = new Map<RunId, string>();
1102+
const retainedAnswerIds = answersBeforeSteers(feed);
10831103
const interruptedRunIds = new Set<RunId>();
10841104
const failedRunIds = new Set(feedFailedRunIds);
10851105
const promptIndexByRunId = new Map<RunId, number>();
@@ -1166,6 +1186,7 @@ function deriveThreadFeedRunFolds(
11661186
.filter(
11671187
(entry) =>
11681188
entry.id !== terminalAssistantId &&
1189+
!(entry.type === "message" && retainedAnswerIds.has(entry.message.id)) &&
11691190
entry.type !== "html-render" &&
11701191
entry.type !== "mcp-app" &&
11711192
!(entry.type === "activity-group" && isContextCompactionActivityGroup(entry)) &&
@@ -1428,7 +1449,7 @@ function zenAnswerMessageIds(
14281449
): ReadonlySet<MessageId> {
14291450
const replies: Array<{ readonly message: ThreadFeedMessage; readonly index: number }> = [];
14301451
const lastReplyByResponse = new Map<string, MessageId>();
1431-
const steerAnswers = new Set<MessageId>();
1452+
const steerAnswers = new Set(answersBeforeSteers(feed));
14321453
let userMessages = 0;
14331454
let awaitingReply = false;
14341455
let candidate: MessageId | null = null;

‎apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts‎

Lines changed: 74 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -4105,10 +4105,50 @@ describe("CodexAdapterV2 post-settle continuation", () => {
41054105
Effect.scoped(
41064106
Effect.gen(function* () {
41074107
const transcript = finalAnswerTranscript("codex-partial-then-final", [
4108-
{ id: "partial", text: "CODEX_RECOVERY_OK", phase: "partial_answer", streamed },
4109-
{ id: "final", text: "CODEX_RECOVERY_OK", phase: "final_answer", streamed },
4108+
{
4109+
id: "status",
4110+
text: "Checking the files",
4111+
phase: "commentary",
4112+
streamed,
4113+
},
4114+
{
4115+
id: "partial",
4116+
text: "CODEX_RECOVERY_OK",
4117+
phase: "partial_answer",
4118+
streamed,
4119+
},
4120+
{
4121+
id: "final",
4122+
text: "CODEX_RECOVERY_OK",
4123+
phase: "final_answer",
4124+
streamed,
4125+
},
41104126
]);
4111-
const harness = yield* makeCodexReplayHarness(transcript);
4127+
const gates = yield* Effect.forEach(["status", "partial", "final"], (id) =>
4128+
Effect.gen(function* () {
4129+
return {
4130+
id,
4131+
ready: yield* Deferred.make<void>(),
4132+
release: yield* Deferred.make<void>(),
4133+
};
4134+
}),
4135+
);
4136+
const harness = yield* makeCodexReplayHarness(
4137+
transcript,
4138+
undefined,
4139+
undefined,
4140+
undefined,
4141+
(entry) => {
4142+
const gate = streamed
4143+
? gates.find((gate) => entry.label === `item/completed/${gate.id}`)
4144+
: undefined;
4145+
return gate === undefined
4146+
? Effect.void
4147+
: Deferred.succeed(gate.ready, undefined).pipe(
4148+
Effect.andThen(Deferred.await(gate.release)),
4149+
);
4150+
},
4151+
);
41124152
const now = yield* DateTime.now;
41134153
yield* harness.runtime.startTurn(
41144154
makeCodexTestTurnInput({
@@ -4119,11 +4159,41 @@ describe("CodexAdapterV2 post-settle continuation", () => {
41194159
text: "Reply with the requested recovery marker.",
41204160
}),
41214161
);
4162+
if (streamed) {
4163+
for (const gate of gates) {
4164+
yield* Deferred.await(gate.ready);
4165+
yield* TestClock.adjust("50 millis");
4166+
yield* harness.until(() =>
4167+
harness.events.some(
4168+
(event) =>
4169+
event.type === "turn_item.updated" &&
4170+
event.turnItem.type === "assistant_message" &&
4171+
event.turnItem.nativeItemRef?.nativeId === gate.id &&
4172+
event.turnItem.streaming,
4173+
),
4174+
);
4175+
yield* Deferred.succeed(gate.release, undefined);
4176+
}
4177+
}
41224178
yield* harness.awaitTerminal(1);
41234179
assert.deepEqual(
41244180
assistantMessages(harness.events).map((event) => event.message.text),
4125-
["CODEX_RECOVERY_OK", "CODEX_RECOVERY_OK"],
4181+
["Checking the files", "CODEX_RECOVERY_OK", "CODEX_RECOVERY_OK"],
4182+
);
4183+
const updates = harness.events.flatMap((event) =>
4184+
event.type === "turn_item.updated" && event.turnItem.type === "assistant_message"
4185+
? [event.turnItem]
4186+
: [],
41264187
);
4188+
for (const [text, phase] of [
4189+
["Checking the files", "commentary"],
4190+
["CODEX_RECOVERY_OK", "partial_answer"],
4191+
["CODEX_RECOVERY_OK", "final_answer"],
4192+
]) {
4193+
const matching = updates.filter((item) => item.text === text && item.phase === phase);
4194+
assert.isTrue(matching.some((item) => !item.streaming));
4195+
if (streamed) assert.isTrue(matching.some((item) => item.streaming));
4196+
}
41274197
}).pipe(
41284198
Effect.provide(
41294199
Layer.mergeAll(

‎apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1210,6 +1210,7 @@ interface ActiveCodexTurnContext {
12101210
readonly runningCompactionItemIds: Set<string>;
12111211
// Modern approval requests omit paths; keep the kinds without retaining patch bodies.
12121212
readonly fileChangeKindsByItem: Map<string, CodexFileChangeKinds>;
1213+
readonly messagePhases: Map<string, "commentary" | "partial_answer" | "final_answer">;
12131214
readonly itemPositions: Map<
12141215
string,
12151216
{ readonly ordinal: number; readonly startedAt: DateTime.Utc }
@@ -2193,6 +2194,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
21932194
startedAt: input.startedAt,
21942195
itemPositions: new Map(),
21952196
fileChangeKindsByItem: new Map(),
2197+
messagePhases: new Map(),
21962198
runningCompactionItemIds: new Set(),
21972199
};
21982200
const nativeThreadId = input.turnInput.providerThread.nativeThreadRef?.nativeId;
@@ -2799,6 +2801,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
27992801
startedAt: turn.startedAt,
28002802
itemPositions: new Map(),
28012803
fileChangeKindsByItem: new Map(),
2804+
messagePhases: new Map(),
28022805
runningCompactionItemIds: new Set(),
28032806
};
28042807
beginTurnTokenUsage(activeContext);
@@ -3476,6 +3479,9 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
34763479
completedAt,
34773480
updatedAt,
34783481
type: "assistant_message",
3482+
...(context.messagePhases.has(item.id)
3483+
? { phase: context.messagePhases.get(item.id) }
3484+
: {}),
34793485
messageId,
34803486
text: item.text,
34813487
streaming: !completed,
@@ -5036,6 +5042,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
50365042
}
50375043

50385044
if (payload.item.type === "agentMessage") {
5045+
if (payload.item.phase != null)
5046+
context.messagePhases.set(payload.item.id, payload.item.phase);
50395047
if (payload.item.phase == null || payload.item.phase === "final_answer") {
50405048
yield* Ref.update(finalAnswerItemIdsByTurn, (current) => {
50415049
const updated = new Map(current);
@@ -5425,6 +5433,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
54255433
return;
54265434
}
54275435

5436+
if (payload.item.phase != null)
5437+
context.messagePhases.set(payload.item.id, payload.item.phase);
54285438
const finalAnswer = payload.item.phase == null || payload.item.phase === "final_answer";
54295439
if (finalAnswer) {
54305440
yield* Ref.update(finalAnswerItemIdsByTurn, (current) => {
@@ -6829,6 +6839,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
68296839
startedAt: now,
68306840
itemPositions: new Map(),
68316841
fileChangeKindsByItem: new Map(),
6842+
messagePhases: new Map(),
68326843
runningCompactionItemIds: new Set(),
68336844
};
68346845
const providerTurn = {

‎apps/web/src/components/chat/MessagesTimeline.logic.test.ts‎

Lines changed: 73 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -3290,6 +3290,7 @@ describe("v2 run and attempt history", () => {
32903290
id: string,
32913291
attempt: typeof supersededAttempt | typeof nextAttempt,
32923292
completedSecond: number,
3293+
phase?: "commentary" | "partial_answer" | "final_answer",
32933294
): TimelineEntry => ({
32943295
id,
32953296
kind: "message",
@@ -3307,6 +3308,7 @@ describe("v2 run and attempt history", () => {
33073308
projectedItem: {
33083309
item: {
33093310
type: "assistant_message",
3311+
...(phase === undefined ? {} : { phase }),
33103312
status: "completed",
33113313
completedAt: DateTime.makeUnsafe(at(completedSecond)),
33123314
},
@@ -3315,51 +3317,58 @@ describe("v2 run and attempt history", () => {
33153317
const rowsFor = (
33163318
replyCompletedSecond: number,
33173319
runStatus: "running" | "completed" = "completed",
3320+
toolCallVisibility: "showAll" | "zen" = "showAll",
3321+
phase?: "commentary" | "partial_answer" | "final_answer",
3322+
nativeSteer = false,
33183323
) =>
33193324
deriveMessagesTimelineRows({
3320-
timelineEntries: [
3321-
{
3322-
id: "initial-user-entry",
3323-
kind: "message",
3324-
createdAt: at(0),
3325-
attempt: supersededAttempt,
3326-
message: {
3327-
id: MessageId.make("initial-user"),
3328-
role: "user",
3329-
text: "Build it",
3330-
runId,
3331-
inputIntent: "turn_start",
3325+
toolCallVisibility,
3326+
timelineEntries: (
3327+
[
3328+
{
3329+
id: "initial-user-entry",
3330+
kind: "message",
33323331
createdAt: at(0),
3333-
updatedAt: at(0),
3334-
streaming: false,
3332+
attempt: supersededAttempt,
3333+
message: {
3334+
id: MessageId.make("initial-user"),
3335+
role: "user",
3336+
text: "Build it",
3337+
runId,
3338+
inputIntent: "turn_start",
3339+
createdAt: at(0),
3340+
updatedAt: at(0),
3341+
streaming: false,
3342+
},
33353343
},
3336-
},
3337-
{
3338-
id: "old-work-entry",
3339-
kind: "work",
3340-
createdAt: at(1),
3341-
attempt: supersededAttempt,
3342-
entry: { id: "old-work", createdAt: at(1), runId, label: "Read", tone: "tool" },
3343-
},
3344-
assistantEntry("old-reply-entry", supersededAttempt, replyCompletedSecond),
3345-
{
3346-
id: "steer-user-entry",
3347-
kind: "message",
3348-
createdAt: at(10),
3349-
attempt: nextAttempt,
3350-
message: {
3351-
id: MessageId.make("steer-user"),
3352-
role: "user",
3353-
text: "What are you working on?",
3354-
runId,
3355-
inputIntent: "steer",
3344+
{
3345+
id: "old-work-entry",
3346+
kind: "work",
3347+
createdAt: at(1),
3348+
attempt: supersededAttempt,
3349+
entry: { id: "old-work", createdAt: at(1), runId, label: "Read", tone: "tool" },
3350+
},
3351+
assistantEntry("status-entry", supersededAttempt, 1, "commentary"),
3352+
assistantEntry("old-reply-entry", supersededAttempt, replyCompletedSecond, phase),
3353+
{
3354+
id: "steer-user-entry",
3355+
kind: "message",
33563356
createdAt: at(10),
3357-
updatedAt: at(10),
3358-
streaming: false,
3357+
attempt: nextAttempt,
3358+
message: {
3359+
id: MessageId.make("steer-user"),
3360+
role: "user",
3361+
text: "What are you working on?",
3362+
runId,
3363+
inputIntent: "steer",
3364+
createdAt: at(10),
3365+
updatedAt: at(10),
3366+
streaming: false,
3367+
},
33593368
},
3360-
},
3361-
assistantEntry("new-reply-entry", nextAttempt, 14),
3362-
],
3369+
assistantEntry("new-reply-entry", nextAttempt, 14),
3370+
] satisfies ReadonlyArray<TimelineEntry>
3371+
).map((entry) => (nativeSteer ? { ...entry, attempt: nextAttempt } : entry)),
33633372
latestRun: {
33643373
runId,
33653374
status: runStatus,
@@ -3371,23 +3380,33 @@ describe("v2 run and attempt history", () => {
33713380
supportsConversationRollback: false,
33723381
});
33733382

3374-
it.each(["running", "completed"] as const)(
3375-
"keeps a reply that finished before the steer visible while the run is %s",
3376-
(runStatus) => {
3377-
const rows = rowsFor(4, runStatus);
3378-
expect(rows.some((row) => row.kind === "attempt-fold")).toBe(false);
3379-
expect(rows.map((row) => row.id)).toContain("old-reply-entry");
3383+
it.each([
3384+
["final_answer", 4, true],
3385+
["final_answer", 11, true],
3386+
["commentary", 4, false],
3387+
["commentary", 11, false],
3388+
["partial_answer", 11, false],
3389+
[undefined, 4, true],
3390+
[undefined, 11, false],
3391+
] as const)(
3392+
"retains phase %s ending at %s only when it is a final reply: %s",
3393+
(phase, completedSecond, retained) => {
3394+
for (const runStatus of ["running", "completed"] as const) {
3395+
for (const visibility of ["showAll", "zen"] as const) {
3396+
const rows = rowsFor(completedSecond, runStatus, visibility, phase);
3397+
expect(rows.some((row) => row.id === "old-reply-entry")).toBe(retained);
3398+
if (phase !== undefined && runStatus === "completed") {
3399+
const nativeRows = rowsFor(completedSecond, runStatus, visibility, phase, true);
3400+
expect(nativeRows.some((row) => row.id === "old-reply-entry")).toBe(retained);
3401+
expect(nativeRows.some((row) => row.id === "status-entry")).toBe(false);
3402+
}
3403+
if (runStatus === "completed") {
3404+
expect(rows.some((row) => row.id === "status-entry")).toBe(false);
3405+
}
3406+
}
3407+
}
33803408
},
33813409
);
3382-
3383-
it("folds a reply the steer cut off as partial output", () => {
3384-
const rows = rowsFor(11, "running");
3385-
expect(rows.find((row) => row.kind === "attempt-fold")).toMatchObject({
3386-
attemptId: supersededAttempt.id,
3387-
label: "Superseded attempt",
3388-
});
3389-
expect(rows.map((row) => row.id)).not.toContain("old-reply-entry");
3390-
});
33913410
});
33923411

33933412
it("hides the interruption request while keeping intervening work and the result", () => {

0 commit comments

Comments
 (0)