From 3129e8694bacf98e3f051d9cf1d118ea2d838921 Mon Sep 17 00:00:00 2001 From: marks Date: Tue, 21 Jul 2026 13:58:53 -0500 Subject: [PATCH] Stabilize provisional thought label sampling --- frontend/src/components/AgentMode.test.ts | 2 + .../src/services/agentThoughtLabels.test.ts | 141 +++++++++++++++--- frontend/src/services/agentThoughtLabels.ts | 37 +++-- 3 files changed, 150 insertions(+), 30 deletions(-) diff --git a/frontend/src/components/AgentMode.test.ts b/frontend/src/components/AgentMode.test.ts index 620e7bc93..58a2ede4f 100644 --- a/frontend/src/components/AgentMode.test.ts +++ b/frontend/src/components/AgentMode.test.ts @@ -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({ diff --git a/frontend/src/services/agentThoughtLabels.test.ts b/frontend/src/services/agentThoughtLabels.test.ts index 599344c52..afb21b025 100644 --- a/frontend/src/services/agentThoughtLabels.test.ts +++ b/frontend/src/services/agentThoughtLabels.test.ts @@ -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) }); @@ -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 () => { @@ -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) }); @@ -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(); + 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(), deferred()]; const requests: Array<{ reasoningText: string; signal: AbortSignal }> = []; @@ -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 @@ -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(), deferred()]; const commits: string[] = []; @@ -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(), deferred()]; + const responses = [ + deferred(), + deferred(), + deferred() + ]; 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); @@ -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"]); }); @@ -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(), deferred()]; + 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(); diff --git a/frontend/src/services/agentThoughtLabels.ts b/frontend/src/services/agentThoughtLabels.ts index 155d02ab1..b35ee4e19 100644 --- a/frontend/src/services/agentThoughtLabels.ts +++ b/frontend/src/services/agentThoughtLabels.ts @@ -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; @@ -127,6 +128,7 @@ type AgentThoughtLabelSchedule = (callback: () => void, delayMs: number) => () = interface ProvisionalEntry { phase: AgentThoughtLabelPhase; + sampledPhase: AgentThoughtLabelPhase | null; snapshotGeneration: number; waitingForMinimumLength: boolean; visibleLabel: string | null; @@ -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(); @@ -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, @@ -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 + ) ); } @@ -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(); @@ -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; }