From a38c400eb721c1d5309d5ca8c59119bd28e04d1a Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Fri, 2 Oct 2026 17:20:22 +0300 Subject: [PATCH] =?UTF-8?q?push=20architecture=20slice=204:=20job=20writes?= =?UTF-8?q?=20announce=20jobs-changed=20from=20the=20store;=20chat=20view?= =?UTF-8?q?=20fetches=20on=20announcement=20(debounced)=20=E2=80=94=20the?= =?UTF-8?q?=20jobs=20interval=20poll=20and=20pollMs=20machinery=20retire?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- components/chat.tsx | 201 ++++++++++++++++++++++---------------------- src/store/jobs.ts | 13 +++ tests/jobs.test.ts | 32 +++++++ 3 files changed, 146 insertions(+), 100 deletions(-) diff --git a/components/chat.tsx b/components/chat.tsx index b7e17d0..16541e0 100644 --- a/components/chat.tsx +++ b/components/chat.tsx @@ -26,7 +26,7 @@ import { DictationBar, joinDictation, useDictation } from "@/components/dictatio import { JobsContext } from "@/components/jobs-context"; import { StatsBar } from "@/components/stats-bar"; import { useActivity } from "@/components/activity-dock"; -import { useServerEvent } from "@/components/server-events"; +import { useServerEvent, useServerResync } from "@/components/server-events"; import { isProviderKeyRefusal, needsContinuation, type ChatInput, type ChatSettings, type LhMessage, type ModelSlotName } from "@/src/core"; import { PlanBlock, PlanPopup, latestTree } from "@/components/plan-block"; import { PlanMarginalia, planExists } from "@/components/plan-window"; @@ -688,50 +688,34 @@ export function Chat({ }, [status, chatId, syncFromServer]); // ── Central job observability (contracts.ts "THE ASYNC PROTOCOL" §4) ── - // ONE poll of /api/jobs drives every job surface (transcript dispatch + // ONE fetch of /api/jobs drives every job surface (transcript dispatch // cards, the sub-agents panel) and re-syncs the transcript exactly once - // when a job settles. A settle no longer implies the record is complete: - // CONTINUATION COALESCING flips the job done at claim time — while the - // learner's own answer turn is still streaming (and a settle can race a - // second job's wake turn). Syncing mid-turn would swap in a stale snapshot - // mid-stream, which breaks the SDK's in-place continuation merge (duplicate - // message ids). So: a settle during a busy turn defers the sync; the next - // idle poll tick adopts the finished record. Poll fast while jobs run, - // slow otherwise (new-dispatch discovery), and not at all when this - // session has never dispatched a job. + // when a job settles. PUSH (architecture S4, 2026-10-02): the store + // announces every job write on the events channel; the interval poll and + // its pollMs/jobPartCount gating are RETIRED. The view fetches once on + // mount, once per jobs-changed announcement (trailing-debounced — a + // sub-agent run writes activity-line bursts; 250ms coalesces them at or + // below the old 1s poll cadence), and once on channel resync. + // + // A settle no longer implies the record is complete: CONTINUATION + // COALESCING flips the job done at claim time — while the learner's own + // answer turn is still streaming (and a settle can race a second job's + // wake turn). Syncing mid-turn would swap in a stale snapshot + // mid-stream, which breaks the SDK's in-place continuation merge + // (duplicate message ids). So: a settle during a busy turn defers the + // sync; the next idle fetch adopts the finished record. const [jobs, setJobs] = useState([]); - const jobPartCount = useMemo( - () => - messages.reduce( - (n, m) => - n + - m.parts.filter( - // sources v0 — "tool-pdf" added to the job-part enumerations - // (mechanical, UI-stream review at merge): poll cadence + - // dispatch nudge + settle-sync all key on these parts. - (p) => - p.type === "tool-research" || - p.type === "tool-demo" || - p.type === "tool-visual" || - p.type === "tool-pdf", - ).length, - 0, - ), - [messages], - ); - const anyRunning = jobs.some((j) => j.status === "running"); - const pollMs = jobPartCount === 0 && jobs.length === 0 ? 0 : anyRunning ? 1000 : 4000; const localStreamRef = useRef(false); localStreamRef.current = localStream; const jobStatusRef = useRef>(new Map()); // A settle seen during a busy turn — the sync is owed, not dropped. const pendingSyncRef = useRef(false); - // Dispatch nudge (accuracy wave, 2026-09-19): a research/demo/visual part - // that cleared input streaming proves the server-side execute ran (the - // bridge emits input-available only at tool_execution_end — AFTER + // Dispatch nudge (accuracy wave, 2026-09-19): a research/demo/visual/pdf + // part that cleared input streaming proves the server-side execute ran + // (the bridge emits input-available only at tool_execution_end — AFTER // dispatch() created the job record on disk), so nudging the activity // feed here is race-free. The dock learns a dispatched job at the moment - // of cause, not at the next idle poll tick (up to 10s). Keyed by + // of cause (dies at S5 when the dock goes event-driven). Keyed by // toolCallId; the persistent set makes each dispatch nudge exactly once // (transcript parts stay forever — the diff only fires on new ids). const dispatchedRef = useRef>(new Set()); @@ -743,7 +727,7 @@ export function Chat({ (p.type === "tool-research" || p.type === "tool-demo" || p.type === "tool-visual" || - p.type === "tool-pdf") && // sources v0 — see the poll-cadence enumeration above + p.type === "tool-pdf") && p.toolCallId && p.state !== "input-streaming" && p.state !== "output-error" @@ -757,72 +741,89 @@ export function Chat({ } dispatchedRef.current = live; }, [messages, refreshActivity]); - useEffect(() => { - if (pollMs === 0) return; - let cancelled = false; - // First-tick guard: the poll's first read adopts existing history - // (jobs recorded before mount) without nudging — a nudge is only owed - // for a running job the feed hasn't seen THIS session. - let firstPoll = jobStatusRef.current.size === 0; - const poll = async () => { - if (cancelled) return; - try { - const res = await fetch(`/api/jobs?chatId=${encodeURIComponent(chatId)}`); - if (!res.ok) return; - const data = (await res.json()) as { jobs?: JobRecord[]; turnInFlight?: boolean }; - if (cancelled) return; - const list = data.jobs ?? []; - let nudgeActivity = false; // settle or new running job this tick - for (const j of list) { - const prev = jobStatusRef.current.get(j.id); - if (prev === "running" && j.status !== "running") { - nudgeActivity = true; // a settle — the blob should leave the feed NOW - // The record is only guaranteed complete when no turn is in - // flight — locally or server-side (a coalesced settle fires - // mid-turn by design), so the FULL sync waits for idle. - // Ground rule (2026-09-15): delivered content shows - // immediately anyway — adopt append-only NOW; the idle - // sync still settles the record properly. - if (localStreamRef.current || data.turnInFlight === true) { - pendingSyncRef.current = true; - void adoptAppendOnly(); - } else { - void syncFromServer(); - } + // First-fetch guard (the old first-tick guard): the initial read adopts + // existing history (jobs recorded before mount) without nudging — a + // nudge is only owed for a running job the feed hasn't seen THIS session. + const firstFetchRef = useRef(true); + // The trailing debounce for jobs-changed announcements (one timer), and + // the ref the timer fires through (a stale closure must never fetch an + // old chatId). + const fetchDebounceRef = useRef | undefined>(undefined); + const fetchJobsRef = useRef<() => void>(() => {}); + const fetchJobs = useCallback(async () => { + try { + const res = await fetch(`/api/jobs?chatId=${encodeURIComponent(chatId)}`); + if (!res.ok) return; + const data = (await res.json()) as { jobs?: JobRecord[]; turnInFlight?: boolean }; + const list = data.jobs ?? []; + let nudgeActivity = false; // settle or new running job this fetch + for (const j of list) { + const prev = jobStatusRef.current.get(j.id); + if (prev === "running" && j.status !== "running") { + nudgeActivity = true; // a settle — the blob should leave the feed NOW + // The record is only guaranteed complete when no turn is in + // flight — locally or server-side (a coalesced settle fires + // mid-turn by design), so the FULL sync waits for idle. + // Ground rule (2026-09-15): delivered content shows + // immediately anyway — adopt append-only NOW; the idle + // sync still settles the record properly. + if (localStreamRef.current || data.turnInFlight === true) { + pendingSyncRef.current = true; + void adoptAppendOnly(); + } else { + void syncFromServer(); } - // A running job this feed has never seen (wake/respawn started - // server-side, or a dispatch in ANOTHER tab) — nudge so the + } + // A running job this feed has never seen (wake/respawn started + // server-side, or a dispatch in ANOTHER tab) — nudge so the // dock adopts it at once. - if (prev === undefined && j.status === "running" && !firstPoll) nudgeActivity = true; - jobStatusRef.current.set(j.id, j.status); - } - firstPoll = false; - if ( - pendingSyncRef.current && - !localStreamRef.current && - data.turnInFlight !== true - ) { - pendingSyncRef.current = false; - void syncFromServer(); - } - setJobs(list); - if (nudgeActivity) refreshActivity(); - // Wake-turn adoption: a server-side teacher turn with no client request - // of ours (sub-agent delivery) — attach to it LIVE through the same - // reattach path (instant catch-up + deltas); while we hold a local - // stream the turn is already ours — nothing to attach to. - if (data.turnInFlight === true && !localStreamRef.current) attachToRunningTurn(); - } catch { - /* transient fetch failure — keep polling */ + if (prev === undefined && j.status === "running" && !firstFetchRef.current) nudgeActivity = true; + jobStatusRef.current.set(j.id, j.status); } - }; - void poll(); - const timer = setInterval(poll, pollMs); - return () => { - cancelled = true; - clearInterval(timer); - }; - }, [pollMs, jobPartCount, chatId, syncFromServer, adoptAppendOnly, refreshActivity, attachToRunningTurn]); + firstFetchRef.current = false; + if ( + pendingSyncRef.current && + !localStreamRef.current && + data.turnInFlight !== true + ) { + pendingSyncRef.current = false; + void syncFromServer(); + } + setJobs(list); + if (nudgeActivity) refreshActivity(); + // (Wake-turn adoption that lived here — the turnInFlight attach + // line — died at S3: turn-started on the events channel owns + // discovery now. turnInFlight is STILL READ above: the settle + // logic's idle check needs it.) + } catch { + /* transient fetch failure — the next jobs-changed announcement re-fetches */ + } + }, [chatId, syncFromServer, adoptAppendOnly, refreshActivity]); + fetchJobsRef.current = () => void fetchJobs(); + // Mount + chat switch: the IMMEDIATE (non-debounced) initial read. The + // firstFetch guard resets with the old rule — true only when this + // session has never seen a job (a chat switch keeps the map: job ids + // are unique per chat, and a newly-seen running job may still nudge). + useEffect(() => { + firstFetchRef.current = jobStatusRef.current.size === 0; + void fetchJobs(); + }, [fetchJobs]); // identity changes exactly on chatId + useServerEvent("jobs-changed", (p) => { + if (p.chatId != null && p.chatId !== chatId) return; // ours or unattributed + if (fetchDebounceRef.current) clearTimeout(fetchDebounceRef.current); + fetchDebounceRef.current = setTimeout(() => { + fetchDebounceRef.current = undefined; + fetchJobsRef.current(); + }, 250); + }); + // Reconnect catch-up: events missed while the channel was down. + useServerResync(() => fetchJobsRef.current()); + useEffect( + () => () => { + if (fetchDebounceRef.current) clearTimeout(fetchDebounceRef.current); + }, + [], + ); // ── Server-initiated turn discovery (push-architecture S3, 2026-10-02) ── // Every turn THIS chat runs is announced on the events channel — the diff --git a/src/store/jobs.ts b/src/store/jobs.ts index 32b55bc..e8df50c 100644 --- a/src/store/jobs.ts +++ b/src/store/jobs.ts @@ -11,6 +11,7 @@ import * as fs from "node:fs"; import * as path from "node:path"; import { PATHS } from "../../config.ts"; +import { emit } from "../core/events.ts"; import type { JobStore } from "../core/ports.ts"; import type { CreateJobInit, JobPatch } from "../core/ports.ts"; import type { JobRecord } from "../core/contracts.ts"; @@ -23,6 +24,10 @@ function jobFile(id: string): string { return path.join(JOBS_DIR, `${id}.json`); } +/** Writes announce on the events channel AFTER the write lands (the + * notify-after-write law, beginTurn precedent) — jobs-changed is what + * retires the view's interval poll (push S4, 2026-10-02): dispatch cards, + * progress lines, settles, and aborts all surface as one fetch. */ function createJob(init: CreateJobInit): JobRecord { const now = new Date().toISOString(); const record: JobRecord = { @@ -37,6 +42,7 @@ function createJob(init: CreateJobInit): JobRecord { }; fs.mkdirSync(JOBS_DIR, { recursive: true }); fs.writeFileSync(jobFile(record.id), JSON.stringify(record, null, "\t"), "utf-8"); + emit("jobs-changed", { chatId: init.chatId }); return record; } @@ -56,6 +62,10 @@ function updateJob(id: string, patch: JobPatch): JobRecord | null { if (!record) return null; const next: JobRecord = { ...record, ...patch, updatedAt: patch.updatedAt ?? new Date().toISOString() }; fs.writeFileSync(jobFile(id), JSON.stringify(next, null, "\t"), "utf-8"); + // EVERY write, not just status flips: progress-line updates must reach + // the dispatch cards (the view debounces the fetch bursts). A missing id + // wrote nothing and stays silent. + emit("jobs-changed", { chatId: next.chatId }); return next; } @@ -93,6 +103,9 @@ function deleteJobsForChat(chatId: string): number { /* already gone — the count is best-effort truth */ } } + // Only when something actually died — a zero-file delete wrote nothing + // (the endTurn present-entry law). + if (removed > 0) emit("jobs-changed", { chatId }); return removed; } diff --git a/tests/jobs.test.ts b/tests/jobs.test.ts index cc0c7f8..21b865e 100644 --- a/tests/jobs.test.ts +++ b/tests/jobs.test.ts @@ -69,6 +69,7 @@ import type { PiEngine } from "../src/agent/models.ts"; import type { SearchProvider } from "../src/core/search.ts"; import { PATHS } from "../config.ts"; import { projectView as projectNative } from "../src/core/view.ts"; +import { subscribe } from "../src/core/events.ts"; import type { LhMessage } from "../src/core/native-messages.ts"; import type { SteerInput, TurnInput } from "../src/core/input-events.ts"; @@ -224,6 +225,37 @@ describe("job store", () => { assert.equal(byTool?.id, job.id); assert.equal(jobStore.findJobByToolCallId("chat_storetest", "call-missing"), null); }); + + // ── push-architecture S4 (2026-10-02): every WRITE announces on the + // events channel (notify-after-write); reads stay silent ── + it("every mutation emits jobs-changed {chatId}; reads emit nothing", () => { + const got: { chatId?: string }[] = []; + const off = subscribe((e) => { + if (e.type === "jobs-changed") got.push({ chatId: e.chatId }); + }); + const job = jobStore.createJob({ kind: "research", chatId: "chat_storeemit", toolCallId: "call-emit" }); + trackJob(job.id); + assert.deepEqual(got, [{ chatId: "chat_storeemit" }], "createJob announces after the write"); + got.length = 0; + + jobStore.updateJob(job.id, { activity: "still working" }); + assert.deepEqual(got, [{ chatId: "chat_storeemit" }], "updateJob announces on EVERY write — progress lines included"); + got.length = 0; + + jobStore.updateJob("job_missing", { activity: "noop" }); + assert.deepEqual(got, [], "a missing id wrote nothing — silent"); + + jobStore.listJobs("chat_storeemit"); + jobStore.getJob(job.id); + assert.deepEqual(got, [], "reads are silent"); + + jobStore.deleteJobsForChat("chat_storeemit"); + assert.deepEqual(got, [{ chatId: "chat_storeemit" }], "a real delete announces"); + got.length = 0; + jobStore.deleteJobsForChat("chat_storeemit"); + assert.deepEqual(got, [], "a zero-file delete wrote nothing — silent"); + off(); + }); }); // ─── 2. steer helpers (pure) ─────────────────────────────────────────────────