3dtours/penguin-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(web): take the Session's elapsed time from Trace timestamps, so a reload stops restarting it (#124)

Browse Source 
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Yaowei Zheng
2026-07-29 23:30:13 +08:00
committed by GitHub
parent ed1ea509d3
commit b9de3da879
7 changed files with 289 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/web/src/api/client.ts
+27 -3
View File
@@ -42,8 +42,29 @@ export interface ApiFetchOptions {
query?: Record<string, string | number | undefined>;
}
/** Response metadata a caller may need alongside the parsed body. */
export interface ApiFetchMeta {
/**
* The server's own clock at the moment it produced the response, read from the HTTP `Date`
* header; null when absent or unparseable. Lets a caller measure a server-side interval
* entirely in server time — differencing it against a server-supplied timestamp cancels any
* client/server clock offset, which a local `Date.now()` cannot do. Whole-second precision
* (RFC 9110 fixes the header's format), so treat it as ±1s. `Date` is CORS-safelisted, so it
* is readable cross-origin too.
*/
serverNowMs: number | null;
}
/** Makes an API request; non-2xx responses uniformly throw ApiError; 204/empty body returns undefined. */
export async function apiFetch<T>(path: string, options: ApiFetchOptions = {}): Promise<T> {
return (await apiFetchWithMeta<T>(path, options)).data;
}
/** {@link apiFetch} plus the response metadata in {@link ApiFetchMeta}; identical in every other respect. */
export async function apiFetchWithMeta<T>(
path: string,
options: ApiFetchOptions = {},
): Promise<{ data: T } & ApiFetchMeta> {
let url = path;
if (options.query) {
const params = new URLSearchParams();
@@ -84,8 +105,11 @@ export async function apiFetch<T>(path: string, options: ApiFetchOptions = {}):
throw new ApiError(response.status, code, message);
}
if (response.status === 204) return undefined as T;
const headerDate = Date.parse(response.headers.get("date") ?? "");
const serverNowMs = Number.isFinite(headerDate) ? headerDate : null;
if (response.status === 204) return { data: undefined as T, serverNowMs };
const text = await response.text();
if (!text) return undefined as T;
return JSON.parse(text) as T;
if (!text) return { data: undefined as T, serverNowMs };
return { data: JSON.parse(text) as T, serverNowMs };
}
packages/web/src/api/endpoints.ts
+11 -2
View File
@@ -71,7 +71,7 @@ import type {
VersionResponse,
WorkspaceFilesResponse,
} from "@prismshadow/penguin-server/api";
import { apiFetch } from "./client";
import { apiFetch, apiFetchWithMeta } from "./client";
// Auth & user -----------------------------------------------------------------
@@ -246,8 +246,17 @@ export const patchSession = (sessionId: string, body: SessionPatchRequest) =>
export const deleteSession = (sessionId: string) =>
apiFetch<void>(`/api/sessions/${encodeURIComponent(sessionId)}`, { method: "DELETE" });
/**
* History rebuild. Carries the server's clock at read time (see ApiFetchMeta.serverNowMs)
* alongside the messages: a Task still running has no Trace entry for the event currently in
* flight, so its elapsed can only be measured by differencing this against the Task's first
* message timestamp — both server-side values, so no client clock offset enters the result
* (see pushMessages).
*/
export const getMessages = (sessionId: string) =>
apiFetch<MessagesResponse>(`/api/sessions/${encodeURIComponent(sessionId)}/messages`);
apiFetchWithMeta<MessagesResponse>(
`/api/sessions/${encodeURIComponent(sessionId)}/messages`,
).then(({ data, serverNowMs }) => ({ ...data, serverNowMs }));
// Task execution, approval, abort, compaction ------------------------------------------------------
packages/web/src/lib/omni/stream-controller.ts
+14 -5
View File
@@ -73,8 +73,17 @@ function parseEventId(id: string): { epoch: string; seq: number } | null {
}
export interface StreamControllerDeps {
/** Fetch history messages (GET /api/sessions/:id/messages), including the live tail while running. */
loadMessages: () => Promise<{ messages: OmniMessage[]; live?: MessagesLiveTail }>;
/**
* Fetch history messages (GET /api/sessions/:id/messages), including the live tail while
* running. `serverNowMs` is the server's clock at read time (the response's `Date` header);
* omitted/null just costs a running Task's header the time its in-flight event has taken so
* far, which falls back to the Trace's own span (see pushMessages).
*/
loadMessages: () => Promise<{
messages: OmniMessage[];
live?: MessagesLiveTail;
serverNowMs?: number | null;
}>;
/** Authoritative running state from the stream (covers both the subscription snapshot and transition events). */
onTaskState: (state: SessionStatus) => void;
/** Queued follow-up count carried on task_state events (absent on old servers -> 0). */
@@ -183,7 +192,7 @@ export function createStreamController(deps: StreamControllerDeps): StreamContro
deps.onQueuedFollowUps?.(ev.queued ?? 0);
if (ev.state === "idle") {
// Task ended (or the snapshot confirms idle): finalize the current Task's stats; pending approvals have already converged server-side.
notifyTaskIdle(model, now());
notifyTaskIdle(model);
clearPending();
deps.onModelChange();
}
@@ -263,14 +272,14 @@ export function createStreamController(deps: StreamControllerDeps): StreamContro
const load = async (currentEpoch: number, freshModel?: StreamModel): Promise<void> => {
try {
const { messages, live } = await deps.loadMessages();
const { messages, live, serverNowMs } = await deps.loadMessages();
if (disposed || currentEpoch !== epoch) return;
// Rebuild path: make the freshly-built model visible only now, atomically — the old model
// stayed on screen throughout the refetch above (see rebuild). Initial load / retry pass no
// freshModel and keep operating on the current model.
if (freshModel) model = freshModel;
const target = model;
pushMessages(target, messages, now());
pushMessages(target, messages, now(), serverNowMs ?? null);
const dedup = buildDedupIndex(messages, 100);
// Replay the buffer (events that arrived while fetching history), with dedup; while a
// Task runs, the live tail is woven in so the in-progress message is seeded too.
packages/web/src/lib/omni/stream-model.ts
+83 -35
View File
@@ -39,9 +39,9 @@
* session's user side starts a new Task; a Task ends when the live
* stream receives task_state:idle (notifyTaskIdle), or — during history
* rebuild — when the next Task starts / the stream ends
* (finalizeHistory). Live finalization takes the duration as the larger
* of the local-clock delta and the message-timestamp span (the local
* clock only covers the time since a mid-stream join). A stats row is only added if token_usage occurred during the Task.
* (finalizeHistory). Either way the duration comes from Trace timestamps
* alone and never the local clock, so a round settles to the same figure
* watched live and replayed after a reload. A stats row is only added if token_usage occurred during the Task.
* - Overlap-dedup helpers: buildDedupIndex/isDuplicate
* judge duplicates by exact match of the envelope JSON; when a complete
* message hits the dedup check, discardFragmentFor also discards the corresponding in-flight fragment.
@@ -378,13 +378,30 @@ export interface StreamModel {
lastAuthFailureMs: number | null;
/** Task segmentation state. */
taskOpen: boolean;
/**
* Local-clock instant the running Task's header elapsed ticks from — display
* only; no settled duration is ever derived from it (see finalizeOpenTask,
* which reads Trace timestamps alone). A live stream sets it to the real
* start; a history rebuild would otherwise stamp the page-load instant and
* restart the ticking value from zero on every reload, so pushMessages
* back-dates it by the elapsed already behind the Task — measured entirely
* in server time, so no client/server clock offset reaches it, and taken
* from the server's own clock rather than the Trace's tail so that an event
* still in flight is counted too.
*/
taskStartLocalMs: number;
/** The Task's first message timestamp, in SERVER time: both the settled duration and the back-dated live anchor measure from it. */
taskFirstTsMs: number;
/**
* The latest timestamp seen among this round's messages, used only as a
* **fallback for the round's end**: only used for a degenerate round that
* has no request_end at all (interrupted before its first Request even
* ran). The normal round-end is taken from taskLastReqEndMs.
* The latest timestamp seen among this round's messages. Two readers:
* - the **fallback for the round's end**, used only for a degenerate
* round that has no request_end at all (interrupted before its first
* Request even ran) — the normal round-end is taken from taskLastReqEndMs;
* - the floor under the anchor pushMessages back-dates taskStartLocalMs
* to, for a round still open when a history rebuild ends. That reader
* fires for every such round, degenerate or not, but only decides the
* anchor when the server's own clock did not come back with the
* response — it cannot see an event still in flight.
*/
taskLastTsMs: number;
/**
@@ -566,22 +583,55 @@ function advanceLastTs(model: StreamModel, timestamp: string): void {
if (Number.isFinite(ms)) model.lastTsMs = ms;
}
/**
* Replay a history rebuild. `serverNowMs` is the server's clock when it produced the
* response (see StreamControllerDeps.loadMessages); null when unavailable.
*/
export function pushMessages(
model: StreamModel,
messages: OmniMessage[],
nowMs: number = Date.now(),
serverNowMs: number | null = null,
): void {
for (const msg of messages) pushMessage(model, msg, nowMs);
// Re-anchor a Task still open at the end of the replay. Every message in a
// rebuild is fed the same `nowMs`, so startTask stamped taskStartLocalMs
// with the instant the page loaded — and the header's live elapsed, which
// ticks over `now − taskStartLocalMs`, would restart from zero on every
// reload of a running Session. Back-date the anchor by the elapsed already
// behind this Task, so the ticking value resumes where it left off:
//
// now − anchor == (now − loadInstant) + elapsedSoFar
//
// elapsedSoFar is measured in SERVER time and applied to the local clock, so
// a client/server clock offset cancels out and never enters the result. Two
// readings of it, the larger winning:
// - serverNowMs − taskFirstTsMs: the true elapsed, and the only one that
// covers an event still in flight — a tool executing, a Request
// streaming, a compaction running — where nothing has been appended to
// the Trace since it began. Whole-second precision (the `Date` header's
// format), which a chip ticking in whole seconds cannot show.
// - taskLastTsMs − taskFirstTsMs: the span the Trace itself proves. The
// fallback when no `Date` header came back, and a floor under a stale
// one: a cached or intermediary-rewritten reading can only be older
// than the true now, so it can under-report but never overshoot.
// A live stream pushes one message at a time with the real current clock,
// where both readings are still zero at startTask and this is a no-op.
if (model.taskOpen) {
const tracedSpan = model.taskLastTsMs - model.taskFirstTsMs;
const serverSpan = serverNowMs === null ? 0 : serverNowMs - model.taskFirstTsMs;
model.taskStartLocalMs = nowMs - Math.max(0, tracedSpan, serverSpan);
}
}
/** The live stream received task_state:idle: finalize the current Task using the local clock. */
export function notifyTaskIdle(model: StreamModel, nowMs: number = Date.now()): void {
finalizeOpenTask(model, "live", nowMs);
/** The live stream received task_state:idle: finalize the current Task from its Trace timestamps. */
export function notifyTaskIdle(model: StreamModel): void {
finalizeOpenTask(model);
}
/** History rebuild is complete (end of stream): finalize the last Task using message timestamps. */
export function finalizeHistory(model: StreamModel): void {
finalizeOpenTask(model, "history");
finalizeOpenTask(model);
}
/** Register an approval clicked on this end (so the subsequent approval_decision event is labeled "manual"). */
@@ -620,12 +670,15 @@ export function findToolCard(
// ---------------------------------------------------------------------------
/**
* Advance this round's "latest timestamp seen among its messages" — used
* only as a **fallback for the round's end** (see taskLastReqEndMs). The
* normal round-end is set by request_end; this only guarantees a usable
* upper bound for a degenerate round with no request_end at all
* (interrupted before its first Request even ran). Compaction forms its
* own round, and messages within its range don't belong to this round, so this isn't advanced for them.
* Advance this round's "latest timestamp seen among its messages" — the
* **fallback for the round's end** (see taskLastReqEndMs: the normal
* round-end is set by request_end, and this only guarantees a usable
* upper bound for a degenerate round with no request_end at all,
* interrupted before its first Request even ran), and the floor under the
* live anchor back-dated on a history rebuild when the server's own clock
* did not come back with the response (see taskLastTsMs, pushMessages).
* Compaction forms its own round, and messages within its range don't
* belong to this round, so this isn't advanced for them.
*/
function touchTask(model: StreamModel, timestamp: string): void {
if (!model.taskOpen) return;
@@ -637,7 +690,7 @@ function touchTask(model: StreamModel, timestamp: string): void {
function startTask(model: StreamModel, timestamp: string, nowMs: number): void {
// The previous Task is finalized by "the next Task starting" (the history-rebuild convention).
finalizeOpenTask(model, "history");
finalizeOpenTask(model);
// Finalize any retry state left over from the previous Task: when the
// server dies during a backoff window, the Trace's tail is
// request_end(timeout) with no abort, and history rebuild would leave a
@@ -659,7 +712,7 @@ function startTask(model: StreamModel, timestamp: string, nowMs: number): void {
resetTaskCounters(model.stats);
}
function finalizeOpenTask(model: StreamModel, mode: "history" | "live", nowMs?: number): void {
function finalizeOpenTask(model: StreamModel): void {
if (!model.taskOpen) return;
model.taskOpen = false;
// No more tool output will arrive once a Task is finalized: close cards still "executing" and stop their LiveDuration.
@@ -672,23 +725,18 @@ function finalizeOpenTask(model: StreamModel, mode: "history" | "live", nowMs?:
// and is naturally excluded — no compaction wall-clock addition/subtraction
// is needed at all, and history rebuild and live share the same
// convention, consistent before and after a refresh (see taskLastReqEndMs).
//
// Trace timestamps are the ONLY source here — the local clock is never
// consulted, so a round settles to the same number whether it was watched
// live or replayed from the Trace after a reload. A degenerate round (no
// request_end at all, e.g. interrupted before its first Request even ran)
// falls back to its message span, which the abort event's own timestamp
// still bounds; that span is what a later reload would compute, so taking
// the local clock instead — as this did before — only bought a number that
// silently changed on refresh, along with idle-detection and mid-join
// latency folded into it.
const endMs = model.taskLastReqEndMs ?? model.taskLastTsMs;
const tsElapsed = Math.max(0, endMs - model.taskFirstTsMs);
// Only a degenerate round (no request_end at all for the whole round,
// e.g. interrupted before its first Request even ran) falls back to the
// local clock: during a mid-stream join / resync rebuild, the local clock
// only covers the time since joining, so it's compared against the
// message span and the larger is taken to avoid underestimating. When
// there is a request_end, it's the accurate round-end and is used
// directly (not compared against the local clock, to avoid folding in noise like idle-detection latency).
let elapsed: number;
if (model.taskLastReqEndMs !== null) {
elapsed = tsElapsed;
} else if (mode === "live") {
elapsed = Math.max((nowMs ?? Date.now()) - model.taskStartLocalMs, tsElapsed);
} else {
elapsed = tsElapsed;
}
const elapsed = Math.max(0, endMs - model.taskFirstTsMs);
const stats = endTask(model.stats, elapsed);
if (model.nested) return;
const reply = collectTaskAssistant(model);
packages/web/src/lib/omni/task-stats.ts
+13 -5
View File
@@ -316,11 +316,19 @@ export function endTask(t: TaskStatsTracker, elapsedMs: number): TaskStats | nul
/**
* Session elapsed time for live display (the header chip): the settled cross-Task cumulative
* plus the currently running Task's wall clock so far. While no Task is open this is exactly
* `sessionElapsedMs`, so the idle rendering equals the settled value; the running addition
* counts from the model's task-start local clock (`taskStartLocalMs`), clamped so a clock
* anomaly never makes the sum go backwards. At task end {@link endTask} folds the Task's
* elapsed into `sessionElapsedMs` in the same model update that flips `taskOpen` off, so the
* live addition never double-counts across the boundary.
* `sessionElapsedMs`, so the idle rendering equals the settled value. At task end {@link endTask}
* folds the Task's elapsed into `sessionElapsedMs` in the same model update that flips `taskOpen`
* off, so the live addition never double-counts across the boundary.
*
* The running addition counts from the model's task-start local clock (`taskStartLocalMs`),
* clamped so a clock anomaly never makes the sum go backwards.
*
* This is the only place a local clock reaches the elapsed figure at all, and only to animate the
* Task in flight. `taskStartLocalMs` is not the instant this client loaded the page: on a history
* rebuild it is back-dated by the elapsed already behind the Task, measured entirely in server
* time (see pushMessages), so reloading mid-run resumes the ticking value instead of restarting
* it, covers an event still in flight, and carries no client/server clock offset. The moment the
* Task settles the number is replaced by one computed from Trace timestamps alone.
*/
export function liveSessionElapsedMs(
t: TaskStatsTracker,