You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by zh...@apache.org on 2026/09/22 13:03:21 UTC
(maka) branch main updated: fix(ui): prevent stale context refresh overwrite (#5569) (#5570)
This is an automated email from the ASF dual-hosted git repository.
me2seeks pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/main by this push:
new c5eacae370 fix(ui): prevent stale context refresh overwrite (#5569) (#5570)
c5eacae370 is described below
commit c5eacae37049c42699d818c4c426f78eeae5699e
Author: Zhang <96...@users.noreply.github.com>
AuthorDate: Tue Sep 22 21:03:10 2026 +0800
fix(ui): prevent stale context refresh overwrite (#5569) (#5570)
---
.../src/main/__tests__/live-context-usage.test.ts | 27 ++++++
.../session-inspector/live-context-usage.ts | 60 +++++-------
.../cli/src/__tests__/tui-context-refresh.test.ts | 28 ++++++
packages/cli/src/tui-context-refresh.ts | 43 +++------
packages/core/package.json | 1 +
.../src/__tests__/refresh-read-coordinator.test.ts | 104 +++++++++++++++++++++
packages/core/src/refresh-read-coordinator.ts | 84 +++++++++++++++++
7 files changed, 283 insertions(+), 64 deletions(-)
diff --git a/apps/desktop/src/main/__tests__/live-context-usage.test.ts b/apps/desktop/src/main/__tests__/live-context-usage.test.ts
index 0a7f0abd3d..774ffb2f7a 100644
--- a/apps/desktop/src/main/__tests__/live-context-usage.test.ts
+++ b/apps/desktop/src/main/__tests__/live-context-usage.test.ts
@@ -235,6 +235,33 @@ describe('createLiveContextUsageTracker', () => {
tracker.dispose();
});
+ it('discards an in-flight read invalidated by a newer event', async () => {
+ const timer = fakeTimer();
+ const query = scriptedQuery();
+ const seen: unknown[] = [];
+ const tracker = createLiveContextUsageTracker({
+ query: query.query,
+ delayMs: 400,
+ schedule: timer.schedule,
+ cancel: timer.cancel,
+ onChange: (usage) => seen.push(usage),
+ });
+ tracker.setTarget({ sessionId: 's1', route: ROUTE });
+ tracker.observe(event('token_usage'));
+
+ // The old read resolves before the debounced replacement is even issued.
+ // It must not overwrite the state that the newer event invalidated.
+ query.pending[0]!.resolve(available({ inputTokens: 10_000 }));
+ await Promise.resolve();
+ assert.deepEqual(seen, [undefined]);
+
+ timer.fire();
+ query.pending[1]!.resolve(available({ inputTokens: 60_000 }));
+ await Promise.resolve();
+ assert.deepEqual(seen, [undefined, { usageTokens: 60_000, contextWindow: 128_000 }]);
+ tracker.dispose();
+ });
+
it('keeps the last value when a read fails', async () => {
const timer = fakeTimer();
const query = scriptedQuery();
diff --git a/apps/desktop/src/renderer/application/contracts/session-inspector/live-context-usage.ts b/apps/desktop/src/renderer/application/contracts/session-inspector/live-context-usage.ts
index e6852c37e2..e94a6f08a2 100644
--- a/apps/desktop/src/renderer/application/contracts/session-inspector/live-context-usage.ts
+++ b/apps/desktop/src/renderer/application/contracts/session-inspector/live-context-usage.ts
@@ -18,8 +18,9 @@
*/
import type { SessionEvent } from '@maka/core/events';
+import { createRefreshReadCoordinator } from '@maka/core/refresh-read-coordinator';
import type { ContextDiagnosticsResult } from '@maka/runtime-host/protocol';
-import { createTraceRefreshCoalescer, type TraceRefreshCoalescer } from './session-trace-refresh.js';
+import { isTraceRelevantEvent } from './session-trace-refresh.js';
/**
* What the composer is about to send on, and therefore the only route a
@@ -117,13 +118,13 @@ export interface LiveContextUsageTracker {
* completion (#4545 spells out why the alternatives were rejected). This
* tracker pulls that snapshot on the same signal the inspector uses — the
* trace-relevant live events, coalesced — with the three protections the
- * inspector proved out: a revision counter drops reads that answer an older
- * question, a failed read keeps the last value standing, and a target change
- * clears that value first and discards whatever is still in flight. The last
- * two compose rather than collide: the value kept standing is only ever the
- * CURRENT target's, because switching targets clears the previous target's
- * reading before the first read on the new one — a rejected first read must
- * not pin the old target's number in place.
+ * inspector proved out: the refresh coordinator drops reads that answer an
+ * older question, a failed read keeps the last value standing, and a target
+ * change clears that value first and discards whatever is still in flight.
+ * The last two compose rather than collide: the value kept standing is only
+ * ever the CURRENT target's, because switching targets clears the previous
+ * target's reading before the first read on the new one — a rejected first
+ * read must not pin the old target's number in place.
*
* Framework-free on purpose: the timer and the query are injected, so the
* policy is testable without a DOM, and the hook in
@@ -137,31 +138,20 @@ export function createLiveContextUsageTracker(input: {
onChange: (usage: LiveContextUsage | undefined) => void;
}): LiveContextUsageTracker {
let target: LiveContextUsageTarget | undefined;
- let revision = 0;
- const coalescer: TraceRefreshCoalescer = createTraceRefreshCoalescer({
- refresh: () => refresh(),
+ const coordinator = createRefreshReadCoordinator({
+ read: () => target ? input.query(target.sessionId) : Promise.resolve(undefined),
+ apply: (diagnostics) => {
+ // Every target change invalidates in-flight reads before they can apply.
+ if (!diagnostics || !target) return;
+ input.onChange(liveContextUsageFromDiagnostics(diagnostics, target.route));
+ },
delayMs: input.delayMs,
- schedule: input.schedule,
- cancel: input.cancel,
+ schedule: (callback, delayMs) => {
+ const handle = input.schedule(callback, delayMs);
+ return () => input.cancel(handle);
+ },
});
- function refresh(): void {
- const current = target;
- if (!current) return;
- const readRevision = ++revision;
- void input.query(current.sessionId).then(
- (diagnostics) => {
- if (readRevision !== revision) return;
- input.onChange(liveContextUsageFromDiagnostics(diagnostics, current.route));
- },
- () => {
- // A failed read leaves the last value standing: it is still the newest
- // answer anyone has, and blanking it would report "no usage" for a
- // read that simply failed.
- },
- );
- }
-
return {
setTarget(next) {
// Any target change — another session, another route, or none — makes
@@ -172,24 +162,22 @@ export function createLiveContextUsageTracker(input: {
// would otherwise pin it there indefinitely. Re-aiming at the SAME
// target does not clear — the standing value still answers it, and
// blanking it would flicker.
- revision += 1;
const changed = !sameLiveContextUsageTarget(target, next);
target = next;
- coalescer.cancel();
+ coordinator.cancel();
if (!next) {
input.onChange(undefined);
return;
}
if (changed) input.onChange(undefined);
- refresh();
+ coordinator.refresh();
},
observe(event) {
- coalescer.observe(event);
+ if (isTraceRelevantEvent(event)) coordinator.observe();
},
dispose() {
- revision += 1;
target = undefined;
- coalescer.cancel();
+ coordinator.cancel();
},
};
}
diff --git a/packages/cli/src/__tests__/tui-context-refresh.test.ts b/packages/cli/src/__tests__/tui-context-refresh.test.ts
index 2bca471f4c..0446ff8b23 100644
--- a/packages/cli/src/__tests__/tui-context-refresh.test.ts
+++ b/packages/cli/src/__tests__/tui-context-refresh.test.ts
@@ -177,6 +177,34 @@ describe('createCtxRefresher', () => {
assert.deepEqual(applied, ['newer']);
});
+ test('a newer relevant event retires an in-flight query before the replacement runs', async () => {
+ const scheduler = createFakeScheduler();
+ const queries: Array<ReturnType<typeof deferred<string>>> = [];
+ const applied: string[] = [];
+ const refresher = createCtxRefresher({
+ query: () => {
+ const query = deferred<string>();
+ queries.push(query);
+ return query.promise;
+ },
+ apply: (result) => applied.push(result),
+ delayMs: 400,
+ schedule: scheduler.schedule,
+ });
+
+ refresher.observe(toolStartEvent());
+ scheduler.flush();
+ refresher.observe(toolStartEvent());
+ queries[0]!.resolve('stale');
+ await Promise.resolve();
+ assert.deepEqual(applied, []);
+
+ scheduler.flush();
+ queries[1]!.resolve('fresh');
+ await Promise.resolve();
+ assert.deepEqual(applied, ['fresh']);
+ });
+
test('a failed query leaves the last value standing and later events still refresh', async () => {
const scheduler = createFakeScheduler();
const queries: Array<ReturnType<typeof deferred<string>>> = [];
diff --git a/packages/cli/src/tui-context-refresh.ts b/packages/cli/src/tui-context-refresh.ts
index 98be3a6534..9eb3433f55 100644
--- a/packages/cli/src/tui-context-refresh.ts
+++ b/packages/cli/src/tui-context-refresh.ts
@@ -18,6 +18,7 @@
*/
import type { SessionEvent } from '@maka/core/events';
+import { createRefreshReadCoordinator } from '@maka/core/refresh-read-coordinator';
/**
* How full the context is right now, re-asked mid-turn (#4545).
@@ -71,11 +72,12 @@ export interface CtxRefresher {
/**
* Coalesces a burst of refresh-worthy events into one query, and lets only
- * the latest issued query apply. A query that resolves after a newer one was
- * issued — or after `cancel` retired it — is dropped: the answer a slow read
- * brings back describes an older snapshot than the one already shown, and a
- * torn-down session is not owed an update at all. A failed query leaves the
- * last value standing: it is still the newest answer anyone has.
+ * the latest valid query apply. A query that resolves after a newer one was
+ * issued, after a relevant event invalidated it, or after `cancel` retired it
+ * is dropped: the answer a slow read brings back describes an older snapshot
+ * than the one already shown, and a torn-down session is not owed an update
+ * at all. A failed query leaves the last value standing: it is still the
+ * newest answer anyone has.
*
* The scheduler is injected so the policy is testable without a wall clock,
* and follows the CLI's ticker convention: schedule returns the cancel.
@@ -86,35 +88,20 @@ export function createCtxRefresher<T>(input: {
delayMs: number;
schedule: (callback: () => void, delayMs: number) => CancelScheduled;
}): CtxRefresher {
- let cancelScheduled: CancelScheduled | undefined;
- let revision = 0;
- const dropScheduled = (): void => {
- cancelScheduled?.();
- cancelScheduled = undefined;
- };
- const run = (): void => {
- cancelScheduled = undefined;
- const requestRevision = ++revision;
- void input.query().then(
- (result) => {
- if (requestRevision !== revision) return;
- input.apply(result);
- },
- () => {},
- );
- };
+ const coordinator = createRefreshReadCoordinator({
+ read: input.query,
+ apply: input.apply,
+ delayMs: input.delayMs,
+ schedule: input.schedule,
+ });
return {
observe(event) {
if (!isCtxRefreshRelevantEvent(event)) return;
// Restart rather than stack: the last event of a burst is the one whose
// snapshot the reader wants, and an earlier timer would read before it.
- dropScheduled();
- cancelScheduled = input.schedule(run, input.delayMs);
- },
- cancel() {
- dropScheduled();
- revision += 1;
+ coordinator.observe();
},
+ cancel: coordinator.cancel,
};
}
diff --git a/packages/core/package.json b/packages/core/package.json
index da92a0bb04..cc4a3210f6 100644
--- a/packages/core/package.json
+++ b/packages/core/package.json
@@ -134,6 +134,7 @@
"./computer-use": "./dist/computer-use.js",
"./connection-error-copy": "./dist/connection-error-copy.js",
"./connection-usage": "./dist/connection-usage.js",
+ "./refresh-read-coordinator": "./dist/refresh-read-coordinator.js",
"./git-review": "./dist/git-review.js",
"./graph-command": "./dist/graph-command.js",
"./project": "./dist/project.js",
diff --git a/packages/core/src/__tests__/refresh-read-coordinator.test.ts b/packages/core/src/__tests__/refresh-read-coordinator.test.ts
new file mode 100644
index 0000000000..9514fdbae9
--- /dev/null
+++ b/packages/core/src/__tests__/refresh-read-coordinator.test.ts
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { strict as assert } from 'node:assert';
+import { describe, it } from 'node:test';
+import { createRefreshReadCoordinator } from '../refresh-read-coordinator.js';
+
+function fakeScheduler() {
+ let pending: (() => void) | undefined;
+ return {
+ schedule(callback: () => void) {
+ pending = callback;
+ return () => {
+ if (pending === callback) pending = undefined;
+ };
+ },
+ flush() {
+ const callback = pending;
+ pending = undefined;
+ callback?.();
+ },
+ };
+}
+
+function deferred<T>() {
+ let resolve!: (value: T) => void;
+ const promise = new Promise<T>((resolvePromise) => {
+ resolve = resolvePromise;
+ });
+ return { promise, resolve };
+}
+
+describe('createRefreshReadCoordinator', () => {
+ it('retires an in-flight read as soon as a newer authority event arrives', async () => {
+ const scheduler = fakeScheduler();
+ const reads: Array<ReturnType<typeof deferred<string>>> = [];
+ const applied: string[] = [];
+ const coordinator = createRefreshReadCoordinator({
+ read: () => {
+ const read = deferred<string>();
+ reads.push(read);
+ return read.promise;
+ },
+ apply: (result) => applied.push(result),
+ delayMs: 400,
+ schedule: scheduler.schedule,
+ });
+
+ coordinator.refresh();
+ assert.equal(reads.length, 1);
+ coordinator.observe();
+ reads[0]!.resolve('stale');
+ await Promise.resolve();
+ assert.deepEqual(applied, []);
+
+ scheduler.flush();
+ assert.equal(reads.length, 2);
+ reads[1]!.resolve('fresh');
+ await Promise.resolve();
+ assert.deepEqual(applied, ['fresh']);
+ });
+
+ it('retires an in-flight read when an immediate refresh replaces it', async () => {
+ const scheduler = fakeScheduler();
+ const reads: Array<ReturnType<typeof deferred<string>>> = [];
+ const applied: string[] = [];
+ const coordinator = createRefreshReadCoordinator({
+ read: () => {
+ const read = deferred<string>();
+ reads.push(read);
+ return read.promise;
+ },
+ apply: (result) => applied.push(result),
+ delayMs: 400,
+ schedule: scheduler.schedule,
+ });
+
+ coordinator.refresh();
+ coordinator.refresh();
+ assert.equal(reads.length, 2);
+ reads[0]!.resolve('old');
+ await Promise.resolve();
+ assert.deepEqual(applied, []);
+ reads[1]!.resolve('new');
+ await Promise.resolve();
+ assert.deepEqual(applied, ['new']);
+ });
+});
diff --git a/packages/core/src/refresh-read-coordinator.ts b/packages/core/src/refresh-read-coordinator.ts
new file mode 100644
index 0000000000..3103d1f587
--- /dev/null
+++ b/packages/core/src/refresh-read-coordinator.ts
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+export type CancelScheduledRefresh = () => void;
+
+export interface RefreshReadCoordinator {
+ /** An authority event arrived; invalidate older reads and debounce a refresh. */
+ observe(): void;
+ /** Read immediately, retiring any scheduled or in-flight older read. */
+ refresh(): void;
+ /** Cancel scheduled work and retire every in-flight read. */
+ cancel(): void;
+}
+
+/**
+ * Coordinates a debounced read of an authority-owned projection.
+ *
+ * The freshness invariant is intentionally stronger than "only the latest
+ * query may apply": an authority event also invalidates a read that is
+ * already in flight. That read may have captured the projection before the
+ * event and can otherwise resolve after the newer state has reached the
+ * caller. Callers decide which events are relevant; this module owns the
+ * ordering and cancellation rule behind that small interface.
+ */
+export function createRefreshReadCoordinator<T>(input: {
+ read: () => Promise<T>;
+ apply: (result: T) => void;
+ delayMs: number;
+ schedule: (callback: () => void, delayMs: number) => CancelScheduledRefresh;
+}): RefreshReadCoordinator {
+ let cancelScheduled: CancelScheduledRefresh | undefined;
+ let revision = 0;
+
+ const dropScheduled = (): void => {
+ cancelScheduled?.();
+ cancelScheduled = undefined;
+ };
+
+ const run = (): void => {
+ cancelScheduled = undefined;
+ const readRevision = ++revision;
+ void input.read().then(
+ (result) => {
+ if (readRevision !== revision) return;
+ input.apply(result);
+ },
+ () => {},
+ );
+ };
+
+ return {
+ observe() {
+ // Invalidate before scheduling. The previous read can resolve during the
+ // debounce window, before the replacement read is issued.
+ revision += 1;
+ dropScheduled();
+ cancelScheduled = input.schedule(run, input.delayMs);
+ },
+ refresh() {
+ dropScheduled();
+ run();
+ },
+ cancel() {
+ dropScheduled();
+ revision += 1;
+ },
+ };
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org