Skip to content

Stabilize provisional thought labels during streaming #665

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
marksftw merged 1 commit into from
Jul 24, 2026
Merged

Stabilize provisional thought labels during streaming #665

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
2 changes: 2 additions & 0 deletions frontend/src/components/AgentMode.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -214,6 +214,8 @@ describe("AgentMode runFinished thought labels", () => {
});
provisionalScheduler.observe(tracker.activePhase(SESSION_ID)!);
timers.run(AGENT_THOUGHT_LABEL_PROVISIONAL_MILESTONES_MS[1]);
expect(provisionalSignals).toHaveLength(1);
timers.run(AGENT_THOUGHT_LABEL_PROVISIONAL_MILESTONES_MS[2]);
expect(provisionalSignals).toHaveLength(2);

const finished = handleAgentModeThoughtRunFinished({
Expand Down
141 changes: 123 additions & 18 deletions frontend/src/services/agentThoughtLabels.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -450,15 +450,16 @@ describe("AgentThoughtLabelProvisionalScheduler", () => {
await Promise.resolve();
});

test("refreshes from the latest reasoning at one, five, and fifteen seconds", async () => {
test("refreshes from the latest reasoning until a provisional succeeds", async () => {
const timers = manualTimers();
const requestedReasoning: string[] = [];
const commits: string[] = [];
const results = [null, null, "Reviewing latest login evidence"];
const scheduler = new AgentThoughtLabelProvisionalScheduler({
schedule: timers.schedule,
request: async (source) => {
requestedReasoning.push(source.reasoningText);
return `Label ${requestedReasoning.length}`;
return results.shift() ?? null;
},
commit: (_source, label) => commits.push(label)
});
Expand All @@ -476,7 +477,7 @@ describe("AgentThoughtLabelProvisionalScheduler", () => {
await Promise.resolve();

expect(requestedReasoning.map((reasoning) => reasoning.length)).toEqual([100, 200, 300]);
expect(commits).toEqual(["Label 1", "Label 2", "Label 3"]);
expect(commits).toEqual(["Reviewing latest login evidence"]);
});

test("makes one request when enough reasoning arrives after multiple milestones", async () => {
Expand Down Expand Up @@ -526,15 +527,18 @@ describe("AgentThoughtLabelProvisionalScheduler", () => {
expect(commits).toEqual(["Comparing final options"]);
});

test("skips unchanged snapshots and exact duplicate labels", async () => {
test("keeps an early provisional stable until the fifteen-second refresh", async () => {
const timers = manualTimers();
const commits: string[] = [];
const requestedLengths: number[] = [];
const labels = ["Comparing authentication options", "Reviewing authentication decision"];
let requestCount = 0;
const scheduler = new AgentThoughtLabelProvisionalScheduler({
schedule: timers.schedule,
request: async () => {
request: async (requestedPhase) => {
requestCount += 1;
return "Comparing authentication options";
requestedLengths.push(requestedPhase.reasoningText.length);
return labels.shift() ?? null;
},
commit: (_source, label) => commits.push(label)
});
Expand All @@ -545,16 +549,62 @@ describe("AgentThoughtLabelProvisionalScheduler", () => {
scheduler.observe(source);
timers.run(firstMilestone);
await Promise.resolve();
scheduler.observe({ ...source, reasoningText: `${source.reasoningText}${"r".repeat(100)}` });
timers.run(secondMilestone);
scheduler.observe({ ...source, reasoningText: `${source.reasoningText} changed` });
await Promise.resolve();
expect(requestCount).toBe(1);

scheduler.observe({ ...source, reasoningText: `${source.reasoningText}${"r".repeat(200)}` });
timers.run(thirdMilestone);
await Promise.resolve();

expect(requestCount).toBe(2);
expect(commits).toEqual(["Comparing authentication options"]);
expect(requestedLengths).toEqual([100, 300]);
expect(commits).toEqual([
"Comparing authentication options",
"Reviewing authentication decision"
]);
});

test("aborts an obsolete snapshot and commits only the next milestone snapshot", async () => {
test("keeps a sampled request alive while reasoning continues streaming", async () => {
const timers = manualTimers();
const response = deferred<string | null>();
const requests: Array<{ reasoningText: string; signal: AbortSignal }> = [];
const commits: Array<{ reasoningText: string; label: string }> = [];
const scheduler = new AgentThoughtLabelProvisionalScheduler({
schedule: timers.schedule,
request: (source, signal) => {
requests.push({ reasoningText: source.reasoningText, signal });
return response.promise;
},
commit: (source, label) => commits.push({ reasoningText: source.reasoningText, label })
});
const source = phase("assistant:thought-0", 100);

scheduler.observe(source);
timers.run(AGENT_THOUGHT_LABEL_PROVISIONAL_MILESTONES_MS[0]);
scheduler.observe(phase("assistant:thought-0", 150));
scheduler.observe(phase("assistant:thought-0", 200));
scheduler.observe(phase("assistant:thought-0", 250));

expect(requests).toHaveLength(1);
expect(requests[0].signal.aborted).toBe(false);
response.resolve("Reviewing sampled login evidence");
await response.promise;
await Promise.resolve();

expect(commits).toEqual([
{
reasoningText: source.reasoningText,
label: "Reviewing sampled login evidence"
}
]);
expect(scheduler.complete(source.sessionId, source.phaseId)).toBe(
"Reviewing sampled login evidence"
);
});

test("aborts an obsolete sample and commits only the next milestone sample", async () => {
const timers = manualTimers();
const responses = [deferred<string | null>(), deferred<string | null>()];
const requests: Array<{ reasoningText: string; signal: AbortSignal }> = [];
Expand All @@ -573,9 +623,10 @@ describe("AgentThoughtLabelProvisionalScheduler", () => {
scheduler.observe(phaseA);
timers.run(AGENT_THOUGHT_LABEL_PROVISIONAL_MILESTONES_MS[0]);
scheduler.observe(phaseB);
expect(requests[0].signal.aborted).toBe(true);
expect(requests[0].signal.aborted).toBe(false);

timers.run(AGENT_THOUGHT_LABEL_PROVISIONAL_MILESTONES_MS[1]);
expect(requests[0].signal.aborted).toBe(true);
expect(requests.map(({ reasoningText }) => reasoningText)).toEqual([
phaseA.reasoningText,
phaseB.reasoningText
Expand All @@ -593,7 +644,7 @@ describe("AgentThoughtLabelProvisionalScheduler", () => {
expect(scheduler.complete(phaseB.sessionId, phaseB.phaseId)).toBe("Reviewing current snapshot");
});

test("never commits a stale result that resolves before the replacement starts", async () => {
test("invalidates an in-flight sample when the overall task context changes", async () => {
const timers = manualTimers();
const responses = [deferred<string | null>(), deferred<string | null>()];
const commits: string[] = [];
Expand Down Expand Up @@ -626,14 +677,22 @@ describe("AgentThoughtLabelProvisionalScheduler", () => {
expect(commits).toEqual(["Reviewing current logout request"]);
});

test("uses snapshot generations when reasoning changes from A to B and back to A", async () => {
test("uses sampled generations when reasoning changes from A to B and back to A", async () => {
const timers = manualTimers();
const responses = [deferred<string | null>(), deferred<string | null>()];
const responses = [
deferred<string | null>(),
deferred<string | null>(),
deferred<string | null>()
];
const commits: string[] = [];
const signals: AbortSignal[] = [];
let requestCount = 0;
const scheduler = new AgentThoughtLabelProvisionalScheduler({
schedule: timers.schedule,
request: () => responses[requestCount++].promise,
request: (_source, signal) => {
signals.push(signal);
return responses[requestCount++].promise;
},
commit: (_source, label) => commits.push(label)
});
const phaseA = phase("assistant:thought-0", 100);
Expand All @@ -642,16 +701,22 @@ describe("AgentThoughtLabelProvisionalScheduler", () => {
scheduler.observe(phaseA);
timers.run(AGENT_THOUGHT_LABEL_PROVISIONAL_MILESTONES_MS[0]);
scheduler.observe(phaseB);
timers.run(AGENT_THOUGHT_LABEL_PROVISIONAL_MILESTONES_MS[1]);
expect(signals[0].aborted).toBe(true);
scheduler.observe(phaseA);
timers.run(AGENT_THOUGHT_LABEL_PROVISIONAL_MILESTONES_MS[2]);
expect(signals[1].aborted).toBe(true);

responses[0].resolve("Reviewing old A snapshot");
await responses[0].promise;
responses[1].resolve("Reviewing old B snapshot");
await responses[1].promise;
await Promise.resolve();
expect(commits).toEqual([]);

timers.run(AGENT_THOUGHT_LABEL_PROVISIONAL_MILESTONES_MS[1]);
expect(requestCount).toBe(2);
responses[1].resolve("Reviewing current A snapshot");
await responses[1].promise;
expect(requestCount).toBe(3);
responses[2].resolve("Reviewing current A snapshot");
await responses[2].promise;
await Promise.resolve();
expect(commits).toEqual(["Reviewing current A snapshot"]);
});
Expand Down Expand Up @@ -764,6 +829,46 @@ describe("AgentThoughtLabelProvisionalScheduler", () => {
expect(commits).toEqual([]);
});

test("retains the early label when completion cancels the fifteen-second refresh", async () => {
const timers = manualTimers();
const responses = [deferred<string | null>(), deferred<string | null>()];
const signals: AbortSignal[] = [];
const commits: string[] = [];
let requestCount = 0;
const scheduler = new AgentThoughtLabelProvisionalScheduler({
schedule: timers.schedule,
request: (_source, signal) => {
signals.push(signal);
return responses[requestCount++].promise;
},
commit: (_source, label) => commits.push(label)
});
const earlySample = phase("assistant:thought-0", AGENT_THOUGHT_LABEL_PROVISIONAL_MIN_LENGTH);
const refreshSample = {
...earlySample,
reasoningText: `${earlySample.reasoningText} more evidence`
};

scheduler.observe(earlySample);
timers.run(AGENT_THOUGHT_LABEL_PROVISIONAL_MILESTONES_MS[0]);
responses[0].resolve("Reviewing initial login evidence");
await responses[0].promise;
await Promise.resolve();

scheduler.observe(refreshSample);
timers.run(AGENT_THOUGHT_LABEL_PROVISIONAL_MILESTONES_MS[2]);
expect(signals[1].aborted).toBe(false);
expect(scheduler.complete(refreshSample.sessionId, refreshSample.phaseId)).toBe(
"Reviewing initial login evidence"
);
expect(signals[1].aborted).toBe(true);

responses[1].resolve("Reviewing later login evidence");
await responses[1].promise;
await Promise.resolve();
expect(commits).toEqual(["Reviewing initial login evidence"]);
});

test("cancels matching pending provisionals during turn invalidation", async () => {
const timers = manualTimers();
const response = deferred<string | null>();
Expand Down
37 changes: 25 additions & 12 deletions frontend/src/services/agentThoughtLabels.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,10 +7,11 @@ export const AGENT_THOUGHT_LABEL_PENDING_TEXT = "Thinking";
export const AGENT_THOUGHT_LABEL_FALLBACK_TEXT = "Thought";
export const AGENT_THOUGHT_LABEL_FALLBACK_DELAY_MS = 5_000;
export const AGENT_THOUGHT_LABEL_PROVISIONAL_DELAY_MS = 1_000;
export const AGENT_THOUGHT_LABEL_PROVISIONAL_REFRESH_MS = 15_000;
export const AGENT_THOUGHT_LABEL_PROVISIONAL_MILESTONES_MS = [
AGENT_THOUGHT_LABEL_PROVISIONAL_DELAY_MS,
5_000,
15_000
AGENT_THOUGHT_LABEL_PROVISIONAL_REFRESH_MS
] as const;
export const AGENT_THOUGHT_LABEL_PROVISIONAL_MIN_LENGTH = 100;
export const AGENT_THOUGHT_LABEL_PROVISIONAL_DEADLINE_MS = 3_000;
Expand Down Expand Up @@ -127,6 +128,7 @@ type AgentThoughtLabelSchedule = (callback: () => void, delayMs: number) => () =

interface ProvisionalEntry {
phase: AgentThoughtLabelPhase;
sampledPhase: AgentThoughtLabelPhase | null;
snapshotGeneration: number;
waitingForMinimumLength: boolean;
visibleLabel: string | null;
Expand Down Expand Up @@ -214,11 +216,10 @@ export class AgentThoughtLabelProvisionalScheduler {
const key = thoughtLabelPhaseKey(phase.sessionId, phase.phaseId);
const existing = this.entries.get(key);
if (existing) {
const snapshotChanged =
existing.phase.userRequest !== phase.userRequest ||
existing.phase.reasoningText !== phase.reasoningText;
const contextChanged = existing.phase.userRequest !== phase.userRequest;
existing.phase = phase;
if (snapshotChanged) {
if (contextChanged) {
existing.sampledPhase = null;
existing.snapshotGeneration += 1;
const abortController = existing.abortController;
abortController?.abort();
Expand All @@ -231,6 +232,7 @@ export class AgentThoughtLabelProvisionalScheduler {
const schedule = this.options.schedule ?? scheduleAgentThoughtLabelTimer;
const entry: ProvisionalEntry = {
phase,
sampledPhase: null,
snapshotGeneration: 0,
waitingForMinimumLength: false,
visibleLabel: null,
Expand All @@ -241,7 +243,10 @@ export class AgentThoughtLabelProvisionalScheduler {
};
this.entries.set(key, entry);
entry.cancelMilestones = AGENT_THOUGHT_LABEL_PROVISIONAL_MILESTONES_MS.map((delayMs) =>
schedule(() => this.tryStart(key, entry), delayMs)
schedule(
() => this.tryStart(key, entry, delayMs === AGENT_THOUGHT_LABEL_PROVISIONAL_REFRESH_MS),
delayMs
)
);
}

Expand All @@ -260,20 +265,28 @@ export class AgentThoughtLabelProvisionalScheduler {
}
}

private tryStart(key: string, entry: ProvisionalEntry): void {
private tryStart(key: string, entry: ProvisionalEntry, refreshVisibleLabel = false): void {
if (this.entries.get(key) !== entry) return;
if (entry.abortController) return;
if (entry.visibleLabel && !refreshVisibleLabel) return;
const reasoningCharacters = reasoningCharacterLength(entry.phase.reasoningText);
if (reasoningCharacters < AGENT_THOUGHT_LABEL_PROVISIONAL_MIN_LENGTH) {
entry.waitingForMinimumLength = true;
return;
}
entry.waitingForMinimumLength = false;
if (this.activeRequests >= AGENT_THOUGHT_LABEL_MAX_CONCURRENT_PROVISIONAL_REQUESTS) return;
// Live reasoning stays buffered in phase until a milestone promotes the next request sample.
if (!entry.sampledPhase || !thoughtLabelSnapshotsMatch(entry.sampledPhase, entry.phase)) {
entry.sampledPhase = { ...entry.phase };
entry.snapshotGeneration += 1;
const abortController = entry.abortController;
abortController?.abort();
if (abortController) this.finishRequest(entry, abortController);
}
if (entry.snapshotGeneration === entry.lastRequestedGeneration) return;
if (this.activeRequests >= AGENT_THOUGHT_LABEL_MAX_CONCURRENT_PROVISIONAL_REQUESTS) return;

this.activeRequests += 1;
const requestPhase = { ...entry.phase };
const requestPhase = { ...entry.sampledPhase };
const requestSnapshotGeneration = entry.snapshotGeneration;
entry.lastRequestedGeneration = requestSnapshotGeneration;
const abortController = new AbortController();
Expand Down Expand Up @@ -312,8 +325,8 @@ export class AgentThoughtLabelProvisionalScheduler {
if (this.entries.get(key) !== entry || !this.finishRequest(entry, abortController)) return;
if (
entry.snapshotGeneration !== requestSnapshotGeneration ||
entry.phase.userRequest !== requestPhase.userRequest ||
entry.phase.reasoningText !== requestPhase.reasoningText
!entry.sampledPhase ||
!thoughtLabelSnapshotsMatch(entry.sampledPhase, requestPhase)
) {
return;
}
Expand Down
Loading