Skip to content
Open
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
4 changes: 3 additions & 1 deletion apps/server/src/orchestration-v2/CommandPolicy.ts
Original file line number Diff line number Diff line change
Expand Up @@ -312,9 +312,11 @@ const ensureNativeFork: CommandPolicyV2Shape["ensureNativeFork"] = (input) => {
};

const ensureRollback: CommandPolicyV2Shape["ensureRollback"] = (input) => {
// Without native truncation, the next turn replays the kept conversation as a handoff.
if (
!input.capabilities.threads.canRollbackThread ||
!input.capabilities.checkpointing.providerCanRollbackConversation
(!input.capabilities.checkpointing.providerCanRollbackConversation &&
!input.capabilities.context.canConsumeHandoffSummaries)
) {
return Effect.fail(
unsupported(input, "rollback", "providerInstanceId conversation rollback is unavailable"),
Expand Down
89 changes: 77 additions & 12 deletions apps/server/src/orchestration-v2/ProviderTurnStartService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import {
type OrchestrationV2ProviderThread,
type OrchestrationV2Run,
type OrchestrationV2RunAttempt,
type OrchestrationV2ThreadProjection,
type OrchestrationV2TurnItem,
isCompactionCommandMessage,
parseGoalCommand,
Expand Down Expand Up @@ -115,6 +116,53 @@ const refusedBeforePrompt = (error: unknown): boolean =>
PROMPT_REFUSAL_TAGS.some((tag) => Predicate.isTagged(error, tag)) ||
(Predicate.hasProperty(error, "cause") && refusedBeforePrompt(error.cause));

/**
* Whether a non-truncating provider's native session lost the conversation kept before `run`.
* Rollback resets it: the newest accepted turn was rolled back (works without legacy-missing
* native ids) or ran under another native id (rollback of runs the provider never accepted).
*/
const nativeSessionMissesKeptTurns = (
projection: Pick<
OrchestrationV2ThreadProjection,
"thread" | "runs" | "attempts" | "providerTurns" | "contextHandoffs"
>,
providerThread: OrchestrationV2ProviderThread,
run: OrchestrationV2Run,
): boolean => {
const nativeId = providerThread.nativeThreadRef?.nativeId;
if (nativeId === undefined) return false;
const acceptedAttemptIds = new Set(projection.providerTurns.map((turn) => turn.runAttemptId));
const accepted = projection.attempts.filter((attempt) => acceptedAttemptIds.has(attempt.id));
const runsById = new Map(projection.runs.map((source) => [source.id, source]));
const ordinalOf = (attempt: OrchestrationV2RunAttempt) =>
runsById.get(attempt.runId)?.ordinal ?? 0;
const newest = accepted
.filter((attempt) => attempt.providerThreadId === providerThread.id)
.toSorted((a, b) => ordinalOf(a) - ordinalOf(b) || a.attemptOrdinal - b.attemptOrdinal)
.at(-1);
if (newest === undefined) return false;
// Inherited fork history is kept even when every local run was rolled back.
const keepsHistory =
(projection.thread.forkedFrom?.type === "run" && projection.thread.lineage.sideChat !== true) ||
accepted.some((attempt) => {
const source = runsById.get(attempt.runId);
return (
source !== undefined && source.ordinal < run.ordinal && source.status !== "rolled_back"
);
});
return (
keepsHistory &&
(runsById.get(newest.runId)?.status === "rolled_back" ||
(newest.nativeThreadId !== undefined && newest.nativeThreadId !== nativeId)) &&
!projection.contextHandoffs.some(
(handoff) =>
handoff.toProviderThreadId === providerThread.id &&
handoff.delivery?.nativeThreadId === nativeId &&
handoff.delivery.status !== "pending",
)
);
};

export interface ProviderTurnStartServiceV2Shape {
readonly start: (input: {
readonly threadId: ThreadId;
Expand Down Expand Up @@ -1092,13 +1140,22 @@ export const layer: Layer.Layer<
coveredRunOrdinals: { from: 1, to: Math.max(1, run.ordinal - 1) },
strategy: "full_thread_summary",
runs: projection.runs,
items: (yield* projectionStore.getTurnStartHistory(input.threadId)).filter(
(item) =>
item.runId === null ||
projection.runs.some(
(source) => source.id === item.runId && source.ordinal < run.ordinal,
),
),
items: [
...(yield* ContextHandoffService.readInheritedHandoffItems(
projectionStore,
projection.thread,
)),
...(yield* projectionStore.getTurnStartHistory(input.threadId)).filter(
(item) =>
item.runId === null ||
projection.runs.some(
(source) =>
source.id === item.runId &&
source.ordinal < run.ordinal &&
source.status !== "rolled_back",
),
),
],
createdAt,
});
return { handoff, transferId, createdAt };
Expand Down Expand Up @@ -1249,15 +1306,21 @@ export const layer: Layer.Layer<
handoff.delivery?.nativeThreadId === providerThread.nativeThreadRef?.nativeId &&
handoff.delivery?.status === "pending",
);
// A provider that cannot truncate rolls back into a fresh native session holding none of the kept turns.
const nativeSessionLacksKeptTurns =
!session.providerSession.capabilities.checkpointing.providerCanRollbackConversation &&
nativeSessionMissesKeptTurns(projection, providerThread, run);
const resumed = yield* Effect.result(
uncertainDelivery
uncertainDelivery || nativeSessionLacksKeptTurns
? Effect.fail(
new ProviderAdapter.ProviderAdapterTurnStartError({
driver: session.driver,
threadId: projection.thread.id,
providerThreadId: providerThread.id,
runId,
cause: "Uncertain native history injection",
cause: uncertainDelivery
? "Uncertain native history injection"
: "Native session holds none of the kept conversation",
}),
)
: session.resumeThread({
Expand Down Expand Up @@ -1329,9 +1392,11 @@ export const layer: Layer.Layer<
runId,
reason: uncertainDelivery
? "uncertain_history_delivery"
: resumed._tag === "Success"
? "handoff_budget_exhausted"
: "resume_failed",
: nativeSessionLacksKeptTurns
? "native_session_lacks_kept_turns"
: resumed._tag === "Success"
? "handoff_budget_exhausted"
: "resume_failed",
errorTag: resumed._tag === "Failure" ? resumed.failure._tag : "ContextHandoffBudgetError",
});
const replacement = yield* loadFromProvider(
Expand Down
Loading
Loading