fix(desktop): anchor active-turn badge to skew-corrected agent start (#1068)

Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Co-authored-by: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
Will Pfleger
2026-06-16 11:18:37 -04:00
committed by GitHub
co-authored by npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7
parent ba776b9959
commit 2d26db6d8a
5 changed files with 310 additions and 73 deletions
@@ -450,11 +450,14 @@ describe("activeAgentTurnsStore", () => {
assert.equal(ref1, ref2, "should return cached array reference");
});
it("preserves a desktop-clock observedAt per channel", () => {
it("anchors a turn to its skew-corrected start, not the local insert clock", () => {
// The badge anchor must reflect the agent's true start translated into
// desktop time (startedAt + clock offset), so a turn whose event arrives
// with a stale timestamp does NOT reset to ~Date.now(). With a single
// event the offset is exactly Date.now() - startedAt, so the anchor lands
// on Date.now() here — the regression coverage for skew lives below.
const before = Date.now();
syncAgentTurnsFromEvents(AGENT, [
// startedAt comes from the (stale) event timestamp; observedAt must
// instead anchor to the local clock at insert time.
makeEvent({
seq: 1,
turnId: "t1",
@@ -466,69 +469,114 @@ describe("activeAgentTurnsStore", () => {
const [summary] = getActiveTurnsForAgent(AGENT);
assert.equal(summary.channelId, "c1");
assert.ok(
summary.observedAt >= before && summary.observedAt <= after,
"observedAt must be the local clock at insert, not the event timestamp",
summary.anchorAt >= before && summary.anchorAt <= after,
"anchorAt must be the skew-corrected start, here equal to the local clock",
);
});
it("collapses two turns in one channel to the earliest observedAt", () => {
it("gives two turns with different startedAt different anchors (no lockstep)", () => {
// The lockstep bug: turns processed in the same JS tick were all anchored
// to one shared Date.now(), so their elapsed counters ticked in unison.
// Anchoring to startedAt + offset makes distinct agent-host starts produce
// distinct anchors. A single sampleClockOffset minimum is shared, so the
// anchor difference equals the startedAt difference.
syncAgentTurnsFromEvents(AGENT, [
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
makeEvent({
seq: 1,
turnId: "t1",
channelId: "c-early",
timestamp: "2024-01-01T00:00:00Z",
}),
makeEvent({
seq: 2,
turnId: "t2",
channelId: "c-late",
timestamp: "2024-01-01T00:05:00Z",
}),
]);
const firstObservedAt = getActiveTurnsForAgent(AGENT)[0].observedAt;
const byChannel = new Map(
getActiveTurnsForAgent(AGENT).map((s) => [s.channelId, s.anchorAt]),
);
assert.notEqual(
byChannel.get("c-early"),
byChannel.get("c-late"),
"distinct startedAt must yield distinct anchors",
);
assert.equal(
byChannel.get("c-late") - byChannel.get("c-early"),
5 * 60_000,
"anchor spacing must equal the agent-host start spacing",
);
});
// Second turn in the same channel — its observedAt is >= the first
// because the clock is monotonic, so the earliest must still win.
it("collapses two turns in one channel to the earliest anchor", () => {
// Same agent-host start timestamp, distinct turns (seq bumped so the
// second passes the watermark). Identical timestamps mean the offset does
// not move, so the surfaced anchor is stable and the earliest wins.
syncAgentTurnsFromEvents(AGENT, [
makeEvent({ seq: 2, turnId: "t2", channelId: "c1" }),
makeEvent({
seq: 1,
turnId: "t1",
channelId: "c1",
timestamp: "2024-01-01T00:00:00Z",
}),
]);
const firstAnchor = getActiveTurnsForAgent(AGENT)[0].anchorAt;
syncAgentTurnsFromEvents(AGENT, [
makeEvent({
seq: 2,
turnId: "t2",
channelId: "c1",
timestamp: "2024-01-01T00:00:00Z",
}),
]);
const summaries = getActiveTurnsForAgent(AGENT);
assert.equal(summaries.length, 1, "same channel collapses to one entry");
assert.equal(
summaries[0].observedAt,
firstObservedAt,
"earliest observedAt for the channel must be surfaced",
summaries[0].anchorAt,
firstAnchor,
"earliest start's anchor must be surfaced",
);
});
it("advances to the surviving turn's observedAt after the earliest ends", () => {
it("advances to the surviving turn's anchor after the earliest ends", () => {
// Two turns in one channel; the array must be rebuilt from the LIVE map
// on every mutation, so ending the earliest-observed turn must surface
// the survivor's observedAt — not a stale cached minimum.
// on every mutation, so ending the earliest-started turn must surface the
// survivor's (later) anchor — not a stale cached minimum.
syncAgentTurnsFromEvents(AGENT, [
makeEvent({ seq: 1, turnId: "t-early", channelId: "c1" }),
makeEvent({
seq: 1,
turnId: "t-early",
channelId: "c1",
timestamp: "2024-01-01T00:00:00Z",
}),
makeEvent({
seq: 2,
turnId: "t-later",
channelId: "c1",
timestamp: "2024-01-01T00:02:00Z",
}),
]);
const tEarly = getActiveTurnsForAgent(AGENT)[0].observedAt;
const tEarly = getActiveTurnsForAgent(AGENT)[0].anchorAt;
// Force the second turn's observedAt strictly past the first so the
// advance is observable even when Date.now() would otherwise collide.
const spinUntil = Date.now() + 2;
while (Date.now() < spinUntil) {
/* busy-wait one clock tick */
}
syncAgentTurnsFromEvents(AGENT, [
makeEvent({ seq: 2, turnId: "t-later", channelId: "c1" }),
]);
assert.equal(
getActiveTurnsForAgent(AGENT)[0].observedAt,
tEarly,
"earliest wins while both turns survive",
);
// End the earliest turn by its turnId.
// End the earliest turn by its turnId. Reuse t-later's timestamp (seq
// bumped to pass the watermark) so the offset does not tighten and the
// surviving anchor's advance is exactly the 2-minute start gap.
syncAgentTurnsFromEvents(AGENT, [
makeEvent({
seq: 3,
kind: "turn_completed",
turnId: "t-early",
channelId: "c1",
timestamp: "2024-01-01T00:02:00Z",
}),
]);
const [survivor] = getActiveTurnsForAgent(AGENT);
assert.equal(survivor.channelId, "c1");
assert.ok(
survivor.observedAt > tEarly,
"surfaced observedAt must advance to the surviving turn after eviction",
assert.equal(
survivor.anchorAt - tEarly,
2 * 60_000,
"surfaced anchor must advance to the surviving turn after eviction",
);
});
@@ -642,6 +690,148 @@ describe("activeAgentTurnsStore", () => {
);
});
});
describe("skew-corrected elapsed (real-time arrival)", () => {
// The clock offset estimate (running minimum of Date.now() - event time)
// is only meaningful when events arrive at distinct real times — exactly
// how the harness streams them. Faking Date lets us advance the desktop
// clock between events so an earlier event calibrates the offset before the
// measured turn starts. The fixed epoch is the desktop clock floor.
const EPOCH = Date.parse("2024-06-01T00:00:00Z");
beforeEach(() => {
mock.timers.enable({ apis: ["Date"], now: EPOCH });
});
afterEach(() => {
mock.timers.reset();
});
/** Agent-host clock = desktop clock + skew, as an ISO timestamp. */
const agentTs = (desktopMs, skew) =>
new Date(desktopMs + skew).toISOString();
it("shows a large elapsed for a turn that started well in the past", () => {
// Clocks synced (skew 0). An early event at the true present calibrates
// offset ≈ 0. Five true minutes pass. Then the desktop first observes a
// turn whose start timestamp is that 5-minutes-ago instant — the badge
// must read ~5 minutes, not reset to 0s on first sight.
syncAgentTurnsFromEvents(AGENT, [
makeEvent({
seq: 1,
kind: "turn_liveness",
turnId: "warm",
channelId: "c0",
timestamp: agentTs(EPOCH - 1_000, 0),
}),
]);
mock.timers.tick(5 * 60_000); // 5 true minutes elapse
syncAgentTurnsFromEvents(AGENT, [
makeEvent({
seq: 2,
turnId: "t1",
channelId: "c1",
timestamp: agentTs(EPOCH, 0), // started 5 minutes ago
}),
]);
const summary = getActiveTurnsForAgent(AGENT).find(
(s) => s.channelId === "c1",
);
assert.equal(
Date.now() - summary.anchorAt,
5 * 60_000 - 1_000,
"a 5-minute-old turn must show ~5 minutes elapsed, not 0s",
);
});
it("corrects for agent-host clock skew so elapsed tracks true duration", () => {
// Agent host is 1 hour AHEAD of the desktop. A liveness event received at
// the true present (desktop EPOCH) carries a timestamp an hour in the
// future, calibrating offset ≈ -1h. The turn then starts 30s later in
// true time; its future-stamped start, corrected by the offset, anchors
// to the true start — without correction elapsed would be deeply negative.
const SKEW = 60 * 60_000;
syncAgentTurnsFromEvents(AGENT, [
makeEvent({
seq: 1,
kind: "turn_liveness",
turnId: "warm",
channelId: "c0",
timestamp: agentTs(EPOCH, SKEW),
}),
]);
mock.timers.tick(30_000); // 30s of true time passes
syncAgentTurnsFromEvents(AGENT, [
makeEvent({
seq: 2,
turnId: "t1",
channelId: "c1",
timestamp: agentTs(EPOCH + 30_000, SKEW),
}),
]);
const summary = getActiveTurnsForAgent(AGENT).find(
(s) => s.channelId === "c1",
);
assert.equal(
Date.now() - summary.anchorAt,
0,
"a just-started turn under heavy skew must read ~0s, not a negative/huge value",
);
// Let the turn run 45s; elapsed must track that true duration exactly.
mock.timers.tick(45_000);
const stillRunning = getActiveTurnsForAgent(AGENT).find(
(s) => s.channelId === "c1",
);
assert.equal(
Date.now() - stillRunning.anchorAt,
45_000,
"skew-corrected elapsed must track true duration as the clock advances",
);
});
it("retroactively corrects a live turn's anchor when the offset tightens", () => {
// The design's load-bearing invariant: anchors are derived at READ time,
// so a later, tighter offset must shift an ALREADY-LIVE turn earlier.
// The turn first goes live under a loose offset (its start arrives with a
// +5s processing delay → offset +5000), then a delay-free liveness sample
// tightens the running minimum to 0. The live turn's surfaced anchor must
// move earlier by exactly that 5000ms delta. A regression that froze
// anchorAt at startTurn would leave the anchor at its loose value and
// fail this assertion.
syncAgentTurnsFromEvents(AGENT, [
makeEvent({
seq: 1,
turnId: "t1",
channelId: "c1",
timestamp: agentTs(EPOCH - 5_000, 0), // observed 5s after its start
}),
]);
const looseAnchor = getActiveTurnsForAgent(AGENT).find(
(s) => s.channelId === "c1",
).anchorAt;
mock.timers.tick(1_000); // 1s of true time so the liveness arrives later
syncAgentTurnsFromEvents(AGENT, [
makeEvent({
seq: 2,
kind: "turn_liveness",
turnId: "t1",
channelId: "c1",
timestamp: agentTs(EPOCH + 1_000, 0), // delay-free → offset tightens to 0
}),
]);
const tightAnchor = getActiveTurnsForAgent(AGENT).find(
(s) => s.channelId === "c1",
).anchorAt;
assert.equal(
tightAnchor - looseAnchor,
-5_000,
"a tighter offset must shift the live turn's read-time anchor earlier by the tightening delta",
);
});
});
});
describe("formatElapsed", () => {
@@ -25,20 +25,36 @@ type ActiveTurn = {
turnId: string;
channelId: string;
startedAt: number;
observedAt: number;
lastActivityAt: number;
};
/** One working channel surfaced to the UI, anchored to the desktop clock. */
export type ActiveTurnSummary = {
channelId: string;
observedAt: number;
anchorAt: number;
};
// Module-level state: agentPubkey → turnId → ActiveTurn
const activeTurnsByAgent = new Map<string, Map<string, ActiveTurn>>();
const listeners = new Set<() => void>();
// Per-agent clock offset: the desktop clock minus the agent-host clock, in
// milliseconds. Estimated as the running minimum of
// (Date.now() - Date.parse(event.timestamp)) across that agent's events. The
// minimum converges on true skew minus the smallest network/processing delay
// seen — a monotonically tightening estimate immune to per-event jitter. While
// true skew is constant or shrinking it is conservative: elapsed under-reports
// by the minimum delay and never inflates. The minimum never loosens, so under
// GROWING skew (an NTP step forward, or the host clock drifting further behind
// mid-session) the stored estimate goes stale-too-small and elapsed can over-
// report — bounded by how far the skew grows, sub-second over a session. A
// turn's badge anchor is startedAt + offset: the agent's own start, translated
// into desktop-clock terms. Anchors are derived at read time so a later, tighter
// offset retroactively corrects every live turn — distinct agent starts then
// yield distinct anchors (no lockstep) and a turn started long ago anchors into
// the past (large elapsed) instead of resetting to Date.now().
const clockOffsetByAgent = new Map<string, number>();
// Cached snapshots for useSyncExternalStore reference stability.
// Only regenerated when the underlying turn map for an agent actually changes.
const cachedTurnSummaries = new Map<string, ActiveTurnSummary[]>();
@@ -61,6 +77,23 @@ function notifyListeners() {
}
}
/**
* Refine this agent's clock-offset estimate from one observer event. Samples
* Date.now() - Date.parse(timestamp) and keeps the running minimum. When the
* minimum tightens, every live anchor for the agent shifts, so the cache is
* invalidated. Events with an unparseable timestamp contribute no sample.
* Returns true when the offset changed.
*/
function sampleClockOffset(agentKey: string, timestamp: string): boolean {
const sample = Date.now() - Date.parse(timestamp);
if (Number.isNaN(sample)) return false;
const prior = clockOffsetByAgent.get(agentKey);
if (prior !== undefined && sample >= prior) return false;
clockOffsetByAgent.set(agentKey, sample);
invalidateCache(agentKey);
return true;
}
function startTurn(
agentPubkey: string,
channelId: string,
@@ -89,15 +122,12 @@ function startTurn(
}
}
const now = Date.parse(timestamp) || Date.now();
const startedAt = Date.parse(timestamp) || Date.now();
agentTurns.set(turnId, {
turnId,
channelId,
startedAt: now,
// Desktop-clock anchor for the live elapsed counter. Must NOT use startedAt
// (agent-host clock) — ticking the desktop clock against it skews remote agents.
observedAt: Date.now(),
lastActivityAt: now,
startedAt,
lastActivityAt: Date.now(),
});
invalidateCache(key);
}
@@ -179,6 +209,11 @@ function processEvent(agentPubkey: string, event: ObserverEvent) {
}
lastProcessed.set(key, event);
// Refine the clock offset from every fresh event. A tighter offset shifts
// every live anchor for this agent, so a change must reach the UI even when
// the event itself surfaces no new turn.
const offsetChanged = sampleClockOffset(key, event.timestamp);
switch (event.kind) {
case "turn_started":
if (event.channelId) {
@@ -189,6 +224,7 @@ function processEvent(agentPubkey: string, event: ObserverEvent) {
event.timestamp,
);
notifyListeners();
return;
}
break;
case "turn_completed":
@@ -196,16 +232,20 @@ function processEvent(agentPubkey: string, event: ObserverEvent) {
case "agent_panic":
endTurn(agentPubkey, event.turnId ?? null, event.channelId ?? null);
notifyListeners();
break;
return;
case "acp_read":
case "acp_write":
// turn_liveness keeps a quiet-but-alive turn from being pruned; same
// refresh-only path as stream activity — no surfaced summary change, so
// no notifyListeners().
// refresh-only path as stream activity — no surfaced summary change on its
// own, so it only notifies when the offset above actually moved.
case "turn_liveness":
recordActivity(agentPubkey, event.turnId ?? null);
break;
}
if (offsetChanged) {
notifyListeners();
}
}
function ensurePruneInterval() {
@@ -237,7 +277,7 @@ export function subscribeActiveAgentTurns(listener: () => void) {
/**
* Returns the channels where the given agent has active turns, sorted by
* channelId, each anchored to the earliest `observedAt` for that channel.
* channelId, each anchored to the earliest `anchorAt` for that channel.
* The array reference is cached and stable until the turn map mutates a
* requirement for `useSyncExternalStore`.
*/
@@ -252,18 +292,24 @@ export function getActiveTurnsForAgent(
const cached = cachedTurnSummaries.get(key);
if (cached) return cached;
// Collapse multiple turns in one channel to the earliest observation —
// the badge should count from when the channel first went active.
const offset = clockOffsetByAgent.get(key) ?? 0;
// Collapse multiple turns in one channel to the earliest start — the badge
// should count from when the channel's oldest live turn began. Anchors are
// derived here (startedAt + offset) so the latest skew estimate applies.
const earliestByChannel = new Map<string, number>();
for (const turn of agentTurns.values()) {
const prior = earliestByChannel.get(turn.channelId);
if (prior === undefined || turn.observedAt < prior) {
earliestByChannel.set(turn.channelId, turn.observedAt);
if (prior === undefined || turn.startedAt < prior) {
earliestByChannel.set(turn.channelId, turn.startedAt);
}
}
const result = [...earliestByChannel.entries()]
.map(([channelId, observedAt]) => ({ channelId, observedAt }))
.map(([channelId, startedAt]) => ({
channelId,
anchorAt: startedAt + offset,
}))
.sort((a, b) => a.channelId.localeCompare(b.channelId));
cachedTurnSummaries.set(key, result);
return result;
@@ -286,7 +332,7 @@ export function syncAgentTurnsFromEvents(
/**
* Hook: returns the channels where the given agent is currently working, each
* with the desktop-clock `observedAt` to anchor a live elapsed counter.
* with the desktop-clock `anchorAt` to anchor a live elapsed counter.
* Re-renders when the set of channels changes not when the clock ticks.
*/
export function useActiveAgentTurns(
@@ -324,6 +370,7 @@ export function useActiveAgentTurnsBridge(
export function resetActiveAgentTurnsStore() {
activeTurnsByAgent.clear();
lastProcessed.clear();
clockOffsetByAgent.clear();
cachedTurnSummaries.clear();
notifyListeners();
}
@@ -90,10 +90,10 @@ export function ManagedAgentRow({
const activeWorkingChannels = React.useMemo(
() =>
activeTurns
.map(({ channelId, observedAt }) => ({
.map(({ channelId, anchorAt }) => ({
id: channelId,
name: channelIdToName[channelId] ?? channelId,
observedAt,
anchorAt,
}))
.slice(0, 3),
[activeTurns, channelIdToName],
@@ -222,7 +222,7 @@ function AgentSummary({
personaLabel,
presenceStatus,
}: {
activeWorkingChannels: { id: string; name: string; observedAt: number }[];
activeWorkingChannels: { id: string; name: string; anchorAt: number }[];
agent: ManagedAgent;
channelNames: { id: string; name: string }[];
isExpandable: boolean;
@@ -291,7 +291,7 @@ function AgentSummary({
key={`working-${channel.id}`}
channelId={channel.id}
name={channel.name}
observedAt={channel.observedAt}
anchorAt={channel.anchorAt}
onNavigate={goChannel}
/>
))}
@@ -306,12 +306,12 @@ function AgentSummary({
function WorkingBadge({
channelId,
name,
observedAt,
anchorAt,
onNavigate,
}: {
channelId: string;
name: string;
observedAt: number;
anchorAt: number;
onNavigate: (channelId: string) => void;
}) {
// The 1s tick lives here, at the leaf, so only visible working badges
@@ -327,7 +327,7 @@ function WorkingBadge({
onNavigate(channelId);
}}
>
Working in #{name} · {formatElapsed(now - observedAt)}
Working in #{name} · {formatElapsed(now - anchorAt)}
</Badge>
);
}
@@ -173,12 +173,12 @@ export function ProfileSummaryView({
{activeTurns.length > 0 ? (
<div className="flex flex-wrap justify-center gap-1.5">
{activeTurns.map(({ channelId, observedAt }) => (
{activeTurns.map(({ channelId, anchorAt }) => (
<ProfileWorkingBadge
key={channelId}
channelId={channelId}
name={channelIdToName[channelId] ?? channelId}
observedAt={observedAt}
anchorAt={anchorAt}
onNavigate={goChannel}
/>
))}
@@ -238,12 +238,12 @@ export function ProfileSummaryView({
function ProfileWorkingBadge({
channelId,
name,
observedAt,
anchorAt,
onNavigate,
}: {
channelId: string;
name: string;
observedAt: number;
anchorAt: number;
onNavigate: (channelId: string) => void;
}) {
const now = useNow(1000);
@@ -254,7 +254,7 @@ function ProfileWorkingBadge({
variant="default"
onClick={() => onNavigate(channelId)}
>
Working in #{name} · {formatElapsed(now - observedAt)}
Working in #{name} · {formatElapsed(now - anchorAt)}
</Badge>
);
}
@@ -264,11 +264,11 @@ export function UserProfilePopover({
{activeTurns.length > 0 ? (
<div className="flex flex-wrap gap-1.5">
{activeTurns.map(({ channelId, observedAt }) => (
{activeTurns.map(({ channelId, anchorAt }) => (
<PopoverWorkingBadge
key={channelId}
name={channelIdToName[channelId] ?? channelId}
observedAt={observedAt}
anchorAt={anchorAt}
/>
))}
</div>
@@ -302,16 +302,16 @@ export function UserProfilePopover({
function PopoverWorkingBadge({
name,
observedAt,
anchorAt,
}: {
name: string;
observedAt: number;
anchorAt: number;
}) {
const now = useNow(1000);
return (
<span className="inline-flex items-center rounded-full bg-primary/10 px-2 py-0.5 text-xs font-medium text-primary motion-safe:animate-pulse">
Working in #{name} · {formatElapsed(now - observedAt)}
Working in #{name} · {formatElapsed(now - anchorAt)}
</span>
);
}