Skip to content

Commit 8024e50

Browse files
fix(server): a fork made after Edit from here no longer shows the rewound turns (pingdotgg#34823)
1 parent 0ac41ee commit 8024e50

2 files changed

Lines changed: 274 additions & 28 deletions

File tree

‎apps/server/src/orchestration-v2/ProjectionStore.test.ts‎

Lines changed: 187 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9006,6 +9006,193 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
90069006
}),
90079007
);
90089008

9009+
it.effect("hides source turns rewound before the fork from a fork's visible items", () =>
9010+
Effect.gen(function* () {
9011+
const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
9012+
const now = yield* DateTime.now;
9013+
const rewoundAt = DateTime.add(now, { minutes: 1 });
9014+
const forkedAt = DateTime.add(now, { minutes: 2 });
9015+
const prefix = "projection-fork-after-rewind";
9016+
const projectId = ProjectId.make(`project:${prefix}`);
9017+
const sourceThreadId = ThreadId.make(`thread:${prefix}:source`);
9018+
const targetThreadId = ThreadId.make(`thread:${prefix}:target`);
9019+
const sourceProviderThreadId = ProviderThreadId.make(`provider-thread:${prefix}:source`);
9020+
const runIds = [1, 2, 3].map((ordinal) => RunId.make(`run:${prefix}:${ordinal}`));
9021+
const nodeIds = [1, 2, 3].map((ordinal) => NodeId.make(`node:${prefix}:${ordinal}`));
9022+
const threadPayload = (input: {
9023+
readonly id: ThreadId;
9024+
readonly forkedFrom: { type: "run"; threadId: ThreadId; runId: RunId } | null;
9025+
readonly createdAt: DateTime.Utc;
9026+
}) => ({
9027+
createdBy: "user" as const,
9028+
creationSource: "web" as const,
9029+
id: input.id,
9030+
projectId,
9031+
title: input.id,
9032+
providerInstanceId,
9033+
modelSelection,
9034+
runtimeMode: "full-access" as const,
9035+
interactionMode: "default" as const,
9036+
branch: null,
9037+
worktreePath: null,
9038+
activeProviderThreadId: sourceProviderThreadId,
9039+
lineage:
9040+
input.forkedFrom === null
9041+
? { parentThreadId: null, relationshipToParent: null, rootThreadId: sourceThreadId }
9042+
: {
9043+
parentThreadId: sourceThreadId,
9044+
relationshipToParent: "fork" as const,
9045+
rootThreadId: sourceThreadId,
9046+
},
9047+
forkedFrom: input.forkedFrom,
9048+
createdAt: input.createdAt,
9049+
updatedAt: input.createdAt,
9050+
archivedAt: null,
9051+
settledOverride: null,
9052+
settledAt: null,
9053+
lastVisitedAt: null,
9054+
deletedAt: null,
9055+
});
9056+
const runPayload = (index: number, status: "completed" | "rolled_back") => ({
9057+
id: runIds[index]!,
9058+
threadId: sourceThreadId,
9059+
ordinal: index + 1,
9060+
providerInstanceId,
9061+
modelSelection,
9062+
providerThreadId: sourceProviderThreadId,
9063+
userMessageId: MessageId.make(`message:${prefix}:user:${index + 1}`),
9064+
rootNodeId: nodeIds[index]!,
9065+
activeAttemptId: null,
9066+
status,
9067+
requestedAt: now,
9068+
startedAt: now,
9069+
completedAt: status === "rolled_back" ? rewoundAt : now,
9070+
checkpointId: null,
9071+
contextHandoffId: null,
9072+
});
9073+
9074+
yield* projectionStore.apply({
9075+
id: EventId.make(`event:${prefix}:source-thread`),
9076+
type: "thread.created",
9077+
threadId: sourceThreadId,
9078+
occurredAt: now,
9079+
payload: threadPayload({ id: sourceThreadId, forkedFrom: null, createdAt: now }),
9080+
});
9081+
for (const [index, text] of ["one", "two", "three"].entries()) {
9082+
const runId = runIds[index]!;
9083+
const nodeId = nodeIds[index]!;
9084+
const ordinal = index + 1;
9085+
yield* projectionStore.apply({
9086+
id: EventId.make(`event:${prefix}:run-${ordinal}`),
9087+
type: "run.created",
9088+
threadId: sourceThreadId,
9089+
runId,
9090+
nodeId,
9091+
driver,
9092+
occurredAt: now,
9093+
payload: runPayload(index, "completed"),
9094+
});
9095+
for (const [offset, role] of (["user", "assistant"] as const).entries()) {
9096+
const itemBase = {
9097+
id: TurnItemId.make(`turn-item:${prefix}:${role}:${ordinal}`),
9098+
threadId: sourceThreadId,
9099+
runId,
9100+
nodeId,
9101+
providerThreadId: sourceProviderThreadId,
9102+
providerTurnId: null,
9103+
nativeItemRef: null,
9104+
parentItemId: null,
9105+
ordinal: ordinal * 100 + offset,
9106+
status: "completed" as const,
9107+
title: null,
9108+
startedAt: now,
9109+
completedAt: now,
9110+
updatedAt: now,
9111+
messageId: MessageId.make(`message:${prefix}:${role}:${ordinal}`),
9112+
};
9113+
yield* projectionStore.apply({
9114+
id: EventId.make(`event:${prefix}:${role}-item-${ordinal}`),
9115+
type: "turn-item.updated",
9116+
threadId: sourceThreadId,
9117+
runId,
9118+
nodeId,
9119+
driver,
9120+
occurredAt: now,
9121+
payload:
9122+
role === "user"
9123+
? {
9124+
...itemBase,
9125+
createdBy: "user",
9126+
creationSource: "web",
9127+
type: "user_message",
9128+
inputIntent: "turn_start",
9129+
text: `prompt ${text}`,
9130+
attachments: [],
9131+
}
9132+
: {
9133+
...itemBase,
9134+
type: "assistant_message",
9135+
text: `reply ${text}`,
9136+
streaming: false,
9137+
},
9138+
});
9139+
}
9140+
// Edit from here on prompt two rewinds run 2 before prompt three is sent.
9141+
if (ordinal === 2) {
9142+
yield* projectionStore.apply({
9143+
id: EventId.make(`event:${prefix}:run-2-rolled-back`),
9144+
type: "run.updated",
9145+
threadId: sourceThreadId,
9146+
runId,
9147+
nodeId,
9148+
driver,
9149+
occurredAt: rewoundAt,
9150+
payload: runPayload(index, "rolled_back"),
9151+
});
9152+
}
9153+
}
9154+
9155+
const forkedFrom = { type: "run" as const, threadId: sourceThreadId, runId: runIds[2]! };
9156+
yield* projectionStore.apply({
9157+
id: EventId.make(`event:${prefix}:target-thread`),
9158+
type: "thread.created",
9159+
threadId: targetThreadId,
9160+
occurredAt: forkedAt,
9161+
payload: threadPayload({ id: targetThreadId, forkedFrom, createdAt: forkedAt }),
9162+
});
9163+
9164+
const visibleTexts = (rows: ReadonlyArray<{ readonly item: OrchestrationV2TurnItem }>) =>
9165+
rows.map((row) =>
9166+
row.item.type === "user_message" || row.item.type === "assistant_message"
9167+
? row.item.text
9168+
: row.item.type,
9169+
);
9170+
const source = yield* projectionStore.getThreadProjection(sourceThreadId);
9171+
assert.deepEqual(visibleTexts(source.visibleTurnItems), [
9172+
"prompt one",
9173+
"reply one",
9174+
"prompt three",
9175+
"reply three",
9176+
]);
9177+
const expected = ["prompt one", "reply one", "prompt three", "reply three", "fork"];
9178+
const target = yield* projectionStore.getThreadProjection(targetThreadId);
9179+
assert.deepEqual(visibleTexts(target.visibleTurnItems), expected);
9180+
const page = yield* projectionStore.getTimelinePage(targetThreadId, {
9181+
view: "activity",
9182+
limit: 10,
9183+
});
9184+
assert.deepEqual(visibleTexts(page.items), expected);
9185+
const window = yield* projectionStore.getThreadSnapshotWindow(targetThreadId, {
9186+
rowLimit: 10,
9187+
});
9188+
assert.deepEqual(visibleTexts(window.projection.visibleTurnItems), expected);
9189+
assert.equal(
9190+
(yield* projectionStore.getThreadShell(targetThreadId))?.visibleItemCount,
9191+
expected.length,
9192+
);
9193+
}),
9194+
);
9195+
90099196
it.effect("a pull request watch keeps a finished thread working until it ends", () =>
90109197
Effect.gen(function* () {
90119198
const store = yield* ProjectionStore.ProjectionStoreV2;

0 commit comments

Comments
 (0)