Skip to content
Closed
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
59 changes: 37 additions & 22 deletions apps/mobile/src/lib/threadActivity.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -143,28 +143,43 @@ function assistantMessage(updatedAt = "2026-06-20T00:00:03.000Z") {
}

describe("buildThreadFeed", () => {
it("keeps async answers in question history instead of user bubbles", () => {
const requestId = RuntimeRequestId.make("question");
const question: OrchestrationV2TurnItem = {
...base("question", "2026-06-20T00:00:01.000Z", 0),
type: "user_input_request",
requestId,
questions: [],
questionAnswer: { requestId, answers: { color: "Blue" }, attachmentsByQuestionId: {} },
};
const reply: OrchestrationV2TurnItem = {
...userMessage(),
id: TurnItemId.make("answer"),
messageId: MessageId.make(`async-answer:${requestId}`),
};
const feed = buildThreadFeed([projected(question, 0), projected(reply, 1)]);
expect(feed).toHaveLength(1);
expect(feed[0]).toMatchObject({
type: "activity-group",
activities: [{ workEntry: { questionAnswer: question.questionAnswer } }],
});
expect(buildThreadFeed([projected(reply, 0)])[0]?.type).toBe("message");
});
it.each([runId, RunId.make("later-answer-run")])(
"folds async answers only into questions from the same run (%s)",
(answerRunId) => {
const requestId = RuntimeRequestId.make("question");
const question: OrchestrationV2TurnItem = {
...base("question", "2026-06-20T00:00:01.000Z", 0),
type: "user_input_request",
requestId,
questions: [],
questionAnswer: { requestId, answers: { color: "Blue" }, attachmentsByQuestionId: {} },
};
const reply: OrchestrationV2TurnItem = {
...userMessage(),
id: TurnItemId.make("answer"),
runId: answerRunId,
messageId: MessageId.make(`async-answer:${requestId}`),
};
const cache = createThreadFeedCache();
const rows = [projected(question, 0)];
buildThreadFeed(rows, { cache });
const nextRows = [...rows, projected(reply, 1)];
const feed = buildThreadFeed(nextRows, { cache });
expect(feed).toEqual(buildThreadFeed(nextRows));
expect(feed).toHaveLength(answerRunId === runId ? 1 : 2);
if (answerRunId !== runId) {
expect(feed[1]).toMatchObject({
type: "message",
message: { id: reply.messageId, runId: answerRunId, text: reply.text },
});
}
expect(feed[0]).toMatchObject({
type: "activity-group",
activities: [{ workEntry: { questionAnswer: question.questionAnswer } }],
});
expect(buildThreadFeed([projected(reply, 0)])[0]?.type).toBe("message");
},
);

it.each([false, true])(
"keeps answered questions outside settled work (attachments only: %s)",
Expand Down
13 changes: 11 additions & 2 deletions apps/mobile/src/lib/threadActivity.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2044,18 +2044,27 @@ export function buildThreadFeed(
};
// Thread-wide facts that change rows anywhere in the feed. A cached prefix
// is reused only while these match the build that produced it.
const foldedAnswerMessageIds = new Set<string>();
const answeredQuestionRuns = new Map<string, RunId | null>();
const delegatedChildrenByRun = new Map<RunId, Set<string>>();
for (const { item } of visibleTurnItems) {
if (item.type === "user_input_request" && item.questionAnswer) {
foldedAnswerMessageIds.add(`async-answer:${item.questionAnswer.requestId}`);
answeredQuestionRuns.set(`async-answer:${item.questionAnswer.requestId}`, item.runId);
} else if (item.type === "subagent" && item.origin === "app_owned" && item.runId !== null) {
const children = delegatedChildrenByRun.get(item.runId) ?? new Set<string>();
children.add(item.subagentId);
delegatedChildrenByRun.set(item.runId, children);
}
}

const foldedAnswerMessageIds = new Set(
visibleTurnItems.flatMap(({ item }) =>
item.type === "user_message" &&
item.runId !== null &&
answeredQuestionRuns.get(item.messageId) === item.runId
? [item.messageId]
: [],
),
);
const cache = options?.cache;
const previous = cache?.feedBuild ?? null;
let restartIndex = -1;
Expand Down
45 changes: 45 additions & 0 deletions apps/server/src/orchestration-v2/ThreadFind.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ import * as ProjectStore from "./ProjectStore.ts";
import * as EventStore from "./EventStore.ts";
import { encodeThreadHistoryCursor, selectHistoryPageFromCursor } from "./threadHistoryPaging.ts";
import { projectTurnItemForWire } from "./WireProjection.ts";
import { findProjectedThreadItems } from "./ThreadFind.ts";

const layerTest = Layer.mergeAll(EventStore.layer, ProjectionStore.layer, ProjectStore.layer).pipe(
Layer.provideMerge(SqlitePersistence.layerMemory),
Expand Down Expand Up @@ -471,6 +472,50 @@ describe("V2 thread find", () => {
}).pipe(Effect.provide(layerTest)),
);

it.effect.each([runId, RunId.make("later-answer-run")])(
"searches an async answer only when it belongs to a later run (%s)",
(answerRunId) =>
Effect.gen(function* () {
const projection = yield* setup;
if (answerRunId !== runId) yield* commit([run(threadId, answerRunId, 2)]);
const question: OrchestrationV2TurnItem = {
...item("question", 1, ""),
type: "user_input_request",
requestId: RuntimeRequestId.make("question"),
questions: [],
questionAnswer: { requestId: "question", answers: {}, attachmentsByQuestionId: {} },
};
const answer = {
...item("async-answer:question", 2, "needle", "user_message"),
runId: answerRunId,
};
const items = [question, answer];
yield* putItems(items);
const input = { threadId, query: "needle" };
const persisted = yield* projection.searchThread(input);
const memory = findProjectedThreadItems(
items.map((item, position) => ({
position,
visibility: "local",
sourceThreadId: item.threadId,
sourceItemId: item.id,
item,
})),
input,
0,
);
for (const result of [persisted, memory]) {
assert.equal(result.totalMatches, answerRunId === runId ? 0 : 1);
assert.deepEqual(
result.match,
answerRunId === runId
? null
: { entryId: "async-answer:question", runId: answerRunId, occurrence: 0 },
);
}
}).pipe(Effect.provide(layerTest)),
);

it.effect(
"does not search undispatched inputs, cancelled queue rows, or folded question answers",
() =>
Expand Down
25 changes: 18 additions & 7 deletions apps/server/src/orchestration-v2/ThreadFind.ts
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ interface FindPayload extends FindRow {
readonly payload: string;
readonly entryId: string;
readonly type: OrchestrationV2TurnItem["type"];
readonly runId: OrchestrationV2TurnItem["runId"];
}
interface FindDocument extends FindRow {
readonly entryId: string;
Expand Down Expand Up @@ -83,17 +84,21 @@ function parseText(key: TextKey): readonly string[] {
}

function searchableRows(items: readonly OrchestrationV2ProjectedTurnItem[]) {
const foldedAnswers = new Set(
const answeredQuestionRuns = new Map<string, OrchestrationV2TurnItem["runId"]>(
items.flatMap(({ item }) =>
item.type === "user_input_request" && item.questionAnswer
? [`async-answer:${item.questionAnswer.requestId}`]
? [[`async-answer:${item.questionAnswer.requestId}`, item.runId] as const]
: [],
),
);
return items.filter(
({ item }) =>
["user_message", "assistant_message", "proposed_plan"].includes(item.type) &&
!(item.type === "user_message" && foldedAnswers.has(item.messageId)),
!(
item.type === "user_message" &&
item.runId !== null &&
answeredQuestionRuns.get(item.messageId) === item.runId
),
);
}

Expand Down Expand Up @@ -236,8 +241,9 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* (
payload_json: string;
entry_id: string;
item_type: OrchestrationV2TurnItem["type"];
run_id: OrchestrationV2TurnItem["runId"];
}>`
SELECT item.thread_id, item.turn_item_id, item.payload_json,
SELECT item.thread_id, item.turn_item_id, item.payload_json, item.run_id,
COALESCE(json_extract(item.payload_json, '$.messageId'), item.turn_item_id) AS entry_id,
json_extract(item.payload_json, '$.type') AS item_type
FROM orchestration_v2_projection_turn_items AS item
Expand All @@ -260,6 +266,7 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* (
payload: payload.payload_json,
entryId: payload.entry_id,
type: payload.item_type,
runId: payload.run_id,
});
});
});
Expand Down Expand Up @@ -370,17 +377,21 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* (
),
);
// Answers rendered inside question cards must not appear as separate matches.
const foldedAnswers = new Set<string>();
const answeredQuestionRuns = new Map<string, OrchestrationV2TurnItem["runId"]>();
for (const payload of payloads) {
if (payload.type !== "user_input_request") continue;
const { item } = yield* decode(payload);
if (item.type === "user_input_request" && item.questionAnswer)
foldedAnswers.add(`async-answer:${item.questionAnswer.requestId}`);
answeredQuestionRuns.set(`async-answer:${item.questionAnswer.requestId}`, item.runId);
}
const rows = payloads.filter(
(row) =>
row.type !== "user_input_request" &&
!(row.type === "user_message" && foldedAnswers.has(row.entryId)),
!(
row.type === "user_message" &&
row.runId !== null &&
answeredQuestionRuns.get(row.entryId) === row.runId
),
);
const documents = new Map<number, FindDocument>();
const count = Effect.fnUntraced(function* (row: FindPayload) {
Expand Down
102 changes: 57 additions & 45 deletions apps/web/src/session-logic.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1251,52 +1251,64 @@ describe("native provider presentation in the v2 timeline", () => {
item,
});

it("keeps async answers in the question row, including incrementally appended replies", () => {
const requestId = RuntimeRequestId.make("question");
const question: OrchestrationV2TurnItem = {
...base,
type: "user_input_request",
requestId,
questions: [],
questionAnswer: { requestId, answers: { color: "Blue" }, attachmentsByQuestionId: {} },
};
const reply: OrchestrationV2TurnItem = {
...base,
id: TurnItemId.make("answer"),
type: "user_message",
messageId: MessageId.make(`async-answer:${requestId}`),
inputIntent: "steer",
text: "Which color?\nBlue",
createdBy: "user",
creationSource: "server",
attachments: [],
};
const input = { visibleTurnItems: [visible(question)], optimisticMessages: [] };
const previous = deriveTimelineEntriesFromVisibleTurnItemsWithState(input);
const nextInput = { ...input, visibleTurnItems: [...input.visibleTurnItems, visible(reply)] };
const next = deriveTimelineEntriesFromVisibleTurnItemsWithState(nextInput, previous);
expect(next.entries).toEqual(deriveTimelineEntriesFromVisibleTurnItems(nextInput));
expect(next.entries).toHaveLength(1);
const replyFirst = { ...input, visibleTurnItems: [visible(reply)] };
const replyProjection = deriveTimelineEntriesFromVisibleTurnItemsWithState(replyFirst);
const questionAfterReply = {
...input,
visibleTurnItems: [...replyFirst.visibleTurnItems, visible(question)],
};
expect(
deriveTimelineEntriesFromVisibleTurnItemsWithState(questionAfterReply, replyProjection)
.entries,
).toEqual(deriveTimelineEntriesFromVisibleTurnItems(questionAfterReply));
it.each([base.runId, RunId.make("later-answer-run")])(
"folds async answers only into questions from the same run (%s)",
(answerRunId) => {
const requestId = RuntimeRequestId.make("question");
const question: OrchestrationV2TurnItem = {
...base,
type: "user_input_request",
requestId,
questions: [],
questionAnswer: { requestId, answers: { color: "Blue" }, attachmentsByQuestionId: {} },
};
const reply: OrchestrationV2TurnItem = {
...base,
id: TurnItemId.make("answer"),
runId: answerRunId,
type: "user_message",
messageId: MessageId.make(`async-answer:${requestId}`),
inputIntent: "steer",
text: "Which color?\nBlue",
createdBy: "user",
creationSource: "server",
attachments: [],
};
const input = { visibleTurnItems: [visible(question)], optimisticMessages: [] };
const previous = deriveTimelineEntriesFromVisibleTurnItemsWithState(input);
const nextInput = { ...input, visibleTurnItems: [...input.visibleTurnItems, visible(reply)] };
const next = deriveTimelineEntriesFromVisibleTurnItemsWithState(nextInput, previous);
expect(next.entries).toEqual(deriveTimelineEntriesFromVisibleTurnItems(nextInput));
expect(next.entries).toHaveLength(answerRunId === base.runId ? 1 : 2);
if (answerRunId !== base.runId) {
expect(next.entries[1]).toMatchObject({
kind: "message",
message: { id: reply.messageId, runId: answerRunId, text: reply.text },
});
}
const replyFirst = { ...input, visibleTurnItems: [visible(reply)] };
const replyProjection = deriveTimelineEntriesFromVisibleTurnItemsWithState(replyFirst);
const questionAfterReply = {
...input,
visibleTurnItems: [...replyFirst.visibleTurnItems, visible(question)],
};
expect(
deriveTimelineEntriesFromVisibleTurnItemsWithState(questionAfterReply, replyProjection)
.entries,
).toEqual(deriveTimelineEntriesFromVisibleTurnItems(questionAfterReply));

expect(next.entries[0]).toMatchObject({
kind: "work",
entry: { questionAnswer: question.questionAnswer },
});
expect(
deriveTimelineEntriesFromVisibleTurnItems({ ...input, visibleTurnItems: [visible(reply)] })[0]
?.kind,
).toBe("message");
});
expect(next.entries[0]).toMatchObject({
kind: "work",
entry: { questionAnswer: question.questionAnswer },
});
expect(
deriveTimelineEntriesFromVisibleTurnItems({
...input,
visibleTurnItems: [visible(reply)],
})[0]?.kind,
).toBe("message");
},
);

it("excludes checkpoint-only work from the timeline", () => {
const message: OrchestrationV2TurnItem = {
Expand Down
13 changes: 11 additions & 2 deletions apps/web/src/session-logic.ts
Original file line number Diff line number Diff line change
Expand Up @@ -648,10 +648,19 @@ export function deriveTimelineEntriesFromVisibleTurnItems(
return undefined;
};

const foldedAnswerMessageIds = new Set(
const answeredQuestionRuns = new Map<string, RunId | null>(
input.visibleTurnItems.flatMap(({ item }) =>
item.type === "user_input_request" && item.questionAnswer
? [`async-answer:${item.questionAnswer.requestId}`]
? [[`async-answer:${item.questionAnswer.requestId}`, item.runId] as const]
: [],
),
);
const foldedAnswerMessageIds = new Set(
input.visibleTurnItems.flatMap(({ item }) =>
item.type === "user_message" &&
item.runId !== null &&
answeredQuestionRuns.get(item.messageId) === item.runId
? [item.messageId]
: [],
),
);
Expand Down
Loading