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
1 change: 1 addition & 0 deletions apps/server/src/orchestration-v2/Orchestrator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3745,6 +3745,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
sourceProjection,
sourceRun,
sourceProviderThread,
sourceTokenUsage: providerTurnForRun(sourceProjection, sourceRun)?.tokenUsage,
canonicalSourcePoint: contextSourcePointForRun(sourceProjection, sourceRun),
transferId,
targetThreadId: command.targetThreadId,
Expand Down
8 changes: 8 additions & 0 deletions apps/server/src/orchestration-v2/ThreadForkService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import {
OrchestrationV2ContextTransfer,
OrchestrationV2CreationSource,
OrchestrationV2ProviderThread,
OrchestrationV2ProviderTurnTokenUsage,
OrchestrationV2Run,
OrchestrationV2ThreadProjection,
ThreadId,
Expand Down Expand Up @@ -54,6 +55,7 @@ export interface ThreadForkServiceV2Shape {
readonly sourceProjection: Pick<OrchestrationV2ThreadProjection, "thread">;
readonly sourceRun: OrchestrationV2Run;
readonly sourceProviderThread: OrchestrationV2ProviderThread | undefined;
readonly sourceTokenUsage?: OrchestrationV2ProviderTurnTokenUsage | undefined;
readonly canonicalSourcePoint: OrchestrationV2ContextSourcePoint;
readonly transferId: ContextTransferId;
readonly targetThreadId: ThreadId;
Expand Down Expand Up @@ -119,6 +121,12 @@ export const layer: Layer.Layer<ThreadForkServiceV2> = Layer.succeed(
sourcePoint: input.canonicalSourcePoint,
basePoint: null,
sourceProviderInstanceId: input.sourceRun.providerInstanceId,
// Only native forks retain this context; portable handoffs rebuild it.
sourceTokenUsage:
(input.sourceRun.status === "completed" || input.sourceRun.status === "waiting") &&
input.sourceProviderThread?.nativeThreadRef?.strength === "strong"
? input.sourceTokenUsage
: undefined,
targetProviderInstanceId: null,
targetRunId: null,
status: "pending",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -284,6 +284,8 @@ describe("orchestration V2 thread fork", () => {
assert.lengthOf(transfers, 1);
assert.equal(transfers[0]?.status, "consumed");
assert.equal(transfers[0]?.resolution?.strategy, "native_fork");
assert.isUndefined(sourceProjection.providerTurns[0]?.tokenUsage);
assert.isUndefined(transfers[0]?.sourceTokenUsage);
assert.equal(transfers[0]?.targetRunId, targetProjection.runs[0]?.id);

const transferCreatedIndex = result.domainEvents.findIndex(
Expand Down Expand Up @@ -452,6 +454,9 @@ describe("orchestration V2 thread fork", () => {
assert.lengthOf(transfers, 1);
assert.equal(transfers[0]?.status, "consumed");
assert.equal(transfers[0]?.resolution?.strategy, "native_fork");
const sourceUsage = sourceProjection.providerTurns[0]?.tokenUsage;
assert.isDefined(sourceUsage);
assert.deepEqual(transfers[0]?.sourceTokenUsage, sourceUsage);
}).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
);

Expand Down Expand Up @@ -780,6 +785,11 @@ describe("orchestration V2 thread fork", () => {
"forking from the first source run must not preserve later source turns in native Claude context",
);
assert.equal(targetProjection.contextTransfers[0]?.resolution?.strategy, "native_fork");
const sourceProjection = result.projections.get(materialized.sourceThreadId);
const sourceUsage = sourceProjection?.providerTurns[0]?.tokenUsage;
assert.isDefined(sourceUsage);
assert.notDeepEqual(sourceUsage, sourceProjection?.providerTurns[1]?.tokenUsage);
assert.deepEqual(targetProjection.contextTransfers[0]?.sourceTokenUsage, sourceUsage);
}).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
);

Expand Down
9 changes: 6 additions & 3 deletions apps/web/src/components/ChatView.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -1151,8 +1151,13 @@ export default function ChatView(props: ChatViewProps) {
status: threadStatus,
});
const threadDetailLoading = threadSyncPhase === "loading";
const activeProviderThread = serverProjection?.providerThreads.find(
(thread) => thread.id === serverProjection.thread.activeProviderThreadId,
);
const activeThreadLiveTokenUsage = latestProviderTurnTokenUsage(
serverProjection?.providerTurns ?? [],
serverProjection,
activeProviderThread,
);
const serverVisibleTurnItems = useThreadVisibleTurnItems(routeThreadDetailRef);
const serverThreadHistory = useThreadHistory(routeThreadDetailRef);
Expand Down Expand Up @@ -2532,9 +2537,7 @@ export default function ChatView(props: ChatViewProps) {
const activeContextWindow = deriveLatestContextWindowSnapshot(
serverVisibleTurnItems ?? [],
activeThreadLiveTokenUsage,
serverProjection?.providerThreads.find(
(thread) => thread.id === serverProjection.thread.activeProviderThreadId,
),
activeProviderThread,
);
const pendingBackgroundTasks = (() => {
if (serverProjection === null || serverProjection === undefined) {
Expand Down
9 changes: 6 additions & 3 deletions apps/web/src/components/chat/ComposerPrimaryActions.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -387,9 +387,12 @@ export const ComposerPrimaryActions = memo(function ComposerPrimaryActions({
<span>{compactTokens}</span>
</TooltipTrigger>
<TooltipPopup>
{keepFullHistory
? `Next send keeps all ${compactTokens} tokens. Click to compact first`
: `Next send compacts ${compactTokens} tokens first. Click to keep full history`}
<div>
{keepFullHistory
? `Next send keeps all ${compactTokens} tokens.`
: `Next send compacts ${compactTokens} tokens first.`}
</div>
<div>{keepFullHistory ? "Click to compact first" : "Click to keep full history"}</div>
</TooltipPopup>
</Tooltip>
{submit}
Expand Down
130 changes: 130 additions & 0 deletions apps/web/src/lib/contextWindow.test.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,14 @@
import { describe, expect, it } from "vite-plus/test";
import * as DateTime from "effect/DateTime";
import {
ContextTransferId,
ProviderDriverKind,
ProviderInstanceId,
RunId,
ThreadId,
} from "@t3tools/contracts";
import { makeThreadProjectionFixture } from "../test-fixtures";
import { shouldOfferResumeCompaction } from "../components/chat/ContextWindowMeter.logic";
import {
deriveLatestContextWindowSnapshot,
formatContextWindowTokens,
Expand Down Expand Up @@ -108,4 +117,125 @@ describe("latestProviderTurnTokenUsage", () => {
]);
expect(latest?.usedTokens).toBe(300_000);
});

const sourceUsage = {
usedTokens: 513_000,
maxTokens: 1_000_000,
promptCacheTtlMs: 5 * 60_000,
updatedAt: "2026-10-19T12:00:00.000Z",
};
function forkProjection() {
const projection = makeThreadProjectionFixture();
const sourceThreadId = ThreadId.make("parent");
const runId = RunId.make("fork-point");
return {
thread: {
...projection.thread,
forkedFrom: { type: "run" as const, threadId: sourceThreadId, runId },
},
contextTransfers: [
{
id: ContextTransferId.make("fork-transfer"),
type: "fork" as const,
sourceThreadId,
targetThreadId: projection.thread.id,
sourcePoint: { threadId: sourceThreadId, runId },
basePoint: null,
sourceProviderInstanceId: projection.thread.modelSelection.instanceId,
sourceTokenUsage: sourceUsage,
targetProviderInstanceId: null,
targetRunId: null,
status: "pending" as const,
resolution: null,
createdBy: "user" as const,
error: null,
createdAt: projection.updatedAt,
updatedAt: projection.updatedAt,
consumedAt: null,
},
],
};
}

it("keeps fork-point context and its cache expiry before the fork reports usage", () => {
const usage = latestProviderTurnTokenUsage([], forkProjection());
expect(usage).toEqual(sourceUsage);
const snapshot = deriveLatestContextWindowSnapshot([], usage);
expect(snapshot?.usedTokens).toBe(513_000);
expect(
shouldOfferResumeCompaction({
provider: "claudeAgent",
usedTokens: snapshot?.usedTokens,
updatedAt: snapshot?.updatedAt,
promptCacheTtlMs: usage?.promptCacheTtlMs,
now: "2026-10-19T12:06:00.000Z",
}),
).toBe(true);
});

it("replaces inherited usage with the fork's own report", () => {
const tokenUsage = { usedTokens: 42_000, updatedAt: "2026-10-19T12:10:00.000Z" };
expect(latestProviderTurnTokenUsage([{ tokenUsage }], forkProjection())).toEqual(tokenUsage);
});

it.each([0, 42_000])(
"prefers target provider-thread usage of %d tokens over inherited usage",
(usedTokens) => {
const providerThread = {
contextUsage: {
usedTokens,
maxTokens: 200_000,
cost: { amount: 0.42, currency: "USD" },
},
updatedAt: DateTime.makeUnsafe("2026-10-19T12:10:00.000Z"),
};
const usage = latestProviderTurnTokenUsage([], forkProjection(), providerThread);
const snapshot = deriveLatestContextWindowSnapshot([], usage, providerThread);

expect(snapshot).toMatchObject({
usedTokens,
maxTokens: 200_000,
cost: { amount: 0.42, currency: "USD" },
updatedAt: "2026-10-19T12:10:00.000Z",
});
},
);

it("keeps the fallback after a native fork is consumed", () => {
const projection = forkProjection();
const transfer = projection.contextTransfers[0]!;
expect(
latestProviderTurnTokenUsage([], {
...projection,
contextTransfers: [
{
...transfer,
status: "consumed",
resolution: {
strategy: "native_fork",
providerThreadRef: {
driver: ProviderDriverKind.make("claudeAgent"),
nativeId: "fork",
strength: "strong",
},
},
},
],
}),
).toEqual(sourceUsage);
});

it("does not reuse usage for portable handoffs, different providers, or missing history", () => {
const projection = forkProjection();
const transfer = projection.contextTransfers[0]!;
for (const contextTransfers of [
[],
[{ ...transfer, sourceTokenUsage: undefined }],
[{ ...transfer, status: "resolved_portable" as const }],
[{ ...transfer, sourceProviderInstanceId: ProviderInstanceId.make("other") }],
[{ ...transfer, sourcePoint: { ...transfer.sourcePoint, runId: RunId.make("later") } }],
]) {
expect(latestProviderTurnTokenUsage([], { ...projection, contextTransfers })).toBeNull();
}
});
});
18 changes: 17 additions & 1 deletion apps/web/src/lib/contextWindow.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ import type {
OrchestrationV2ProviderTurn,
OrchestrationV2ProviderTurnTokenUsage,
OrchestrationV2ProviderThread,
OrchestrationV2ThreadProjection,
OrchestrationV2TurnItem,
ThreadTokenUsageSnapshot,
} from "@t3tools/contracts";
Expand All @@ -28,6 +29,8 @@ export type ContextWindowSnapshot = NullableContextWindowUsage & {
by provider thread, not time, so pick by timestamp. */
export function latestProviderTurnTokenUsage(
turns: ReadonlyArray<Pick<OrchestrationV2ProviderTurn, "tokenUsage">>,
projection?: Pick<OrchestrationV2ThreadProjection, "thread" | "contextTransfers"> | null,
providerThread?: Pick<OrchestrationV2ProviderThread, "contextUsage"> | null,
): OrchestrationV2ProviderTurnTokenUsage | null {
let latest: OrchestrationV2ProviderTurnTokenUsage | null = null;
for (const { tokenUsage } of turns) {
Expand All @@ -37,7 +40,20 @@ export function latestProviderTurnTokenUsage(
latest = tokenUsage;
}
}
return latest;
if (latest !== null || projection == null || providerThread?.contextUsage != null) return latest;
const forkedFrom = projection.thread.forkedFrom;
if (forkedFrom?.type !== "run") return null;
return (
projection.contextTransfers.find(
(transfer) =>
transfer.type === "fork" &&
transfer.targetThreadId === projection.thread.id &&
transfer.sourceThreadId === forkedFrom.threadId &&
transfer.sourcePoint.runId === forkedFrom.runId &&
transfer.sourceProviderInstanceId === projection.thread.modelSelection.instanceId &&
(transfer.status === "pending" || transfer.resolution?.strategy === "native_fork"),
)?.sourceTokenUsage ?? null
);
}

export function deriveLatestContextWindowSnapshot(
Expand Down
27 changes: 14 additions & 13 deletions packages/contracts/src/orchestrationV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -163,6 +163,19 @@ export const OrchestrationV2ContextTransferResolution = Schema.Union([
export type OrchestrationV2ContextTransferResolution =
typeof OrchestrationV2ContextTransferResolution.Type;

export const OrchestrationV2ProviderTurnTokenUsage = Schema.Struct({
usedTokens: NonNegativeInt,
maxTokens: Schema.optional(Schema.NullOr(NonNegativeInt)),
inputTokens: Schema.optional(NonNegativeInt),
cachedInputTokens: Schema.optional(NonNegativeInt),
outputTokens: Schema.optional(NonNegativeInt),
reasoningOutputTokens: Schema.optional(NonNegativeInt),
promptCacheTtlMs: Schema.optional(PositiveInt),
updatedAt: Schema.String,
});
export type OrchestrationV2ProviderTurnTokenUsage =
typeof OrchestrationV2ProviderTurnTokenUsage.Type;

export const OrchestrationV2ContextTransfer = Schema.Struct({
id: ContextTransferId,
type: OrchestrationV2ContextTransferType,
Expand All @@ -171,6 +184,7 @@ export const OrchestrationV2ContextTransfer = Schema.Struct({
sourcePoint: OrchestrationV2ContextSourcePoint,
basePoint: Schema.NullOr(OrchestrationV2ContextSourcePoint),
sourceProviderInstanceId: Schema.NullOr(ProviderInstanceId),
sourceTokenUsage: Schema.optional(OrchestrationV2ProviderTurnTokenUsage),
targetProviderInstanceId: Schema.NullOr(ProviderInstanceId),
targetRunId: Schema.NullOr(RunId),
status: Schema.Literals([
Expand Down Expand Up @@ -950,19 +964,6 @@ export const OrchestrationV2ContextHandoff = Schema.Struct({
});
export type OrchestrationV2ContextHandoff = typeof OrchestrationV2ContextHandoff.Type;

export const OrchestrationV2ProviderTurnTokenUsage = Schema.Struct({
usedTokens: NonNegativeInt,
maxTokens: Schema.optional(Schema.NullOr(NonNegativeInt)),
inputTokens: Schema.optional(NonNegativeInt),
cachedInputTokens: Schema.optional(NonNegativeInt),
outputTokens: Schema.optional(NonNegativeInt),
reasoningOutputTokens: Schema.optional(NonNegativeInt),
promptCacheTtlMs: Schema.optional(PositiveInt),
updatedAt: Schema.String,
});
export type OrchestrationV2ProviderTurnTokenUsage =
typeof OrchestrationV2ProviderTurnTokenUsage.Type;

export const OrchestrationV2ProviderTurn = Schema.Struct({
id: ProviderTurnId,
providerThreadId: ProviderThreadId,
Expand Down
Loading