From d6450e260ab1e2215a50fa27100f9bdd5826b5f1 Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Fri, 2 Oct 2026 17:31:35 +0300 Subject: [PATCH] =?UTF-8?q?push=20architecture=20slice=205:=20extraction/p?= =?UTF-8?q?ractice=20registries=20announce=20activity-changed;=20dock=20su?= =?UTF-8?q?bscribes=20(debounced)=20=E2=80=94=20the=20adaptive=20poll=20an?= =?UTF-8?q?d=20every=20chat-view=20dock=20nudge=20retire?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- components/activity-dock.tsx | 77 ++++++++++++++---------- components/chat.tsx | 77 +++++------------------- src/agent/subagents/extraction.ts | 18 +++++- src/agent/subagents/practice-pipeline.ts | 24 ++++++-- tests/extraction.test.ts | 31 ++++++++++ tests/practice-pipeline.test.ts | 39 ++++++++++++ 6 files changed, 164 insertions(+), 102 deletions(-) diff --git a/components/activity-dock.tsx b/components/activity-dock.tsx index 2e62e4c..287524a 100644 --- a/components/activity-dock.tsx +++ b/components/activity-dock.tsx @@ -6,6 +6,7 @@ import { cn } from "@/lib/utils"; import type { CallUsage, JobKind, JobRecord } from "@/src/core"; import { isModelsRefusal, isProviderKeyRefusal, jobCost } from "@/src/core"; import { formatTokens, formatUsd } from "@/components/usage-format"; +import { useServerEvent, useServerResync } from "@/components/server-events"; /** * THE BACKGROUND-AGENTS SECTION (owner, sidebar round — the successor of @@ -85,7 +86,7 @@ export interface ActivityFeed { const EMPTY_FEED: ActivityFeed = { jobs: [], extractions: [] }; -// ─── The one poll (composition root: the root layout) ──────────────────────── +// ─── The activity feed (composition root: the root layout) ─────────────────── const ActivityContext = createContext<{ feed: ActivityFeed; refresh: () => void }>({ feed: EMPTY_FEED, @@ -93,52 +94,59 @@ const ActivityContext = createContext<{ feed: ActivityFeed; refresh: () => void }); /** - * Polls /api/activity for the whole app shell. Adaptive interval: fast while - * anything runs, slow when idle, ~4× slower on hidden tabs. A failed fetch - * keeps the last feed (the dock never flickers to empty on a blip). - * - * ACCURACY (2026-09-19): polling alone can only be as fresh as its interval, - * so the start sites that the client can know about nudge refresh() - * directly (chat.tsx at turn start/end and dispatch; the extract/import - * buttons), and THIS provider adds the last client-known event: visibility — - * a tab becoming visible refetches at once, so hidden-tab staleness (×4) - * ends the moment the learner looks. The poll stays the safety net for - * everything with no client-visible cause (other tabs, wake turns, - * recovery). + * THE ACTIVITY FEED, PUSH-DRIVEN (architecture S5, 2026-10-02): the + * adaptive poll is RETIRED — the dock refetches exactly when truth moves. + * Every cause is announced on the events channel: extraction phase + * transitions and practice rows (activity-changed), job writes + * (jobs-changed — the running-jobs blobs), and channel resync (reconnect / + * dev-restart catch-up) — plus the one client-local cause that predates + * the channel: visibility, a tab becoming visible refetches at once. + * Announced fetches ride a 250ms trailing debounce (extraction phases and + * practice progress burst); resync and visibility are immediate. A failed + * fetch keeps the last feed (the dock never flickers to empty on a blip). */ export function ActivityProvider({ children }: { children: React.ReactNode }) { const [feed, setFeed] = useState(EMPTY_FEED); const [nonce, setNonce] = useState(0); const refresh = useCallback(() => setNonce((n) => n + 1), []); - const hasRunning = feed.jobs.length > 0 || feed.extractions.length > 0 || (feed.practice?.length ?? 0) > 0; + const refreshRef = useRef(refresh); + refreshRef.current = refresh; + const debounceRef = useRef | undefined>(undefined); useEffect(() => { let alive = true; - let timer: ReturnType | undefined; - const poll = async () => { + void (async () => { try { const res = await fetch("/api/activity"); if (res.ok) { const data = (await res.json()) as Partial; if (alive) - setFeed({ jobs: data.jobs ?? [], extractions: data.extractions ?? [], practice: data.practice ?? [] }); + setFeed({ jobs: data.jobs ?? [], extractions: data.extractions ?? [], practice: data.practice ?? [] }); } } catch { - // transport blip: keep the last feed, try again + // transport blip: keep the last feed, announce again next event } - if (!alive) return; - const base = hasRunning ? 2500 : 10000; - timer = setTimeout(poll, document.hidden ? base * 4 : base); - }; - void poll(); + })(); return () => { alive = false; - if (timer) clearTimeout(timer); }; - }, [hasRunning, nonce]); + }, [nonce]); - // The visibility nudge: returning to the tab refetches immediately (and - // the re-run effect below resets the poll timer — no double polling). + // Announcements → one trailing fetch per burst (the S4 debounce law). + const scheduleAnnouncedFetch = () => { + if (debounceRef.current) clearTimeout(debounceRef.current); + debounceRef.current = setTimeout(() => { + debounceRef.current = undefined; + refreshRef.current(); + }, 250); + }; + useServerEvent("activity-changed", scheduleAnnouncedFetch); + useServerEvent("jobs-changed", scheduleAnnouncedFetch); + // Reconnect catch-up: events missed while the channel was down. + useServerResync(() => refreshRef.current()); + + // The visibility nudge: returning to the tab refetches immediately — + // anything missed while hidden lands at once. useEffect(() => { const onVisibility = () => { if (document.visibilityState === "visible") setNonce((n) => n + 1); @@ -147,14 +155,21 @@ export function ActivityProvider({ children }: { children: React.ReactNode }) { return () => document.removeEventListener("visibilitychange", onVisibility); }, []); + // The debounce timer never outlives the provider. + useEffect( + () => () => { + if (debounceRef.current) clearTimeout(debounceRef.current); + }, + [], + ); + return {children}; } /** The activity feed seam for the rest of the shell: the dock's data + - * its refresh nudge. The chat page nudges on turn start (owner ground - * rules, 2026-09-15): background processes that start at send — - * extraction and sub-agent jobs — must be visible in the dock the - * moment they begin, without the dock polling faster while idle. */ + * its refresh nudge. The remaining nudgers are client-visible causes + * (the extract/import buttons — extraction-coverage, graph-view); every + * server-side cause announces itself on the events channel now. */ export function useActivity(): { feed: ActivityFeed; refresh: () => void } { return useContext(ActivityContext); } diff --git a/components/chat.tsx b/components/chat.tsx index 16541e0..f771b8a 100644 --- a/components/chat.tsx +++ b/components/chat.tsx @@ -25,7 +25,6 @@ import { Composer } from "@/components/composer"; import { DictationBar, joinDictation, useDictation } from "@/components/dictation"; import { JobsContext } from "@/components/jobs-context"; import { StatsBar } from "@/components/stats-bar"; -import { useActivity } from "@/components/activity-dock"; 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"; @@ -529,10 +528,9 @@ export function Chat({ // outgoing message, BEFORE the stream produces anything — so a bounded, // self-terminating window of three record fetches (400ms/1s/2s) catches // the insert and shows the marker during the turn, then stops. No - // standing poll — the window exists only for these seconds. The same - // instant nudges the activity feed: extraction and sub-agent jobs - // start at send (server-side), and the dock must say so immediately. - const { refresh: refreshActivity } = useActivity(); + // standing poll — the window exists only for these seconds. (The dock + // nudges that lived here died at S5: dispatch writes announce jobs-changed + // from the store, extraction announces activity-changed at setPhase.) const turnStartStatusRef = useRef(status); // The window must survive the submitted→streaming flip (the stream starts // in well under 400ms), so the timers are guarded by this ref — dropped at @@ -544,18 +542,12 @@ export function Chat({ turnStartStatusRef.current = status; if (was === "submitted" || status !== "submitted") return; // A turn just started (status flipped ready → submitted): open the - // bounded adoption window and nudge the dock's feed. - refreshActivity(); + // bounded adoption window. adoptionWindowOpenRef.current = true; const timers = [400, 1000, 2000].map((ms) => setTimeout(() => { if (adoptionWindowOpenRef.current) { void adoptAppendOnly(); - // Turn-start nudge re-fire (accuracy wave): extraction and - // sub-agent jobs start at send — the initial refresh above can - // race the server's job-record write; the window's ticks retry - // it so the dock always catches the job within the window (≤2s). - refreshActivity(); } }, ms), ); @@ -564,7 +556,7 @@ export function Chat({ timers.forEach((t) => { turnStartTimerRefs.current.push(t); }); - }, [status, adoptAppendOnly, refreshActivity]); + }, [status, adoptAppendOnly]); useEffect(() => { const timers = turnStartTimerRefs; return () => { @@ -647,17 +639,9 @@ export function Chat({ // The coverage ring re-reads at turn settle too (extraction-UX // redesign, 2026-09-24); the ring's own running-poll handles runs. setCoverageTick((t) => t + 1); - // Turn-end nudge (accuracy wave, 2026-09-19): the extraction - // pipeline fires at this turn's onFinish — the dock's poll would - // discover a started run up to 10s late. Refetch the activity feed - // now (stream settle) and again at +600/+1500ms (bounded window — - // covers onFinish→triggerExtraction→setPhase lag); the feed's own - // fast interval takes over once it sees the run. (Dies at S5 when - // the dock goes event-driven.) - refreshActivity(); - const nudgeTimers = [600, 1500].map((ms) => setTimeout(() => refreshActivity(), ms)); - return () => nudgeTimers.forEach(clearTimeout); - }, [status, syncTurns, refreshActivity]); + // (The extraction-discovery nudges that lived here died at S5: the + // pipeline announces every phase on the events channel.) + }, [status, syncTurns]); // A stream that died MID-FLIGHT (network): the runner decoupled the turn // from its view, so the server turn runs ON behind the error banner — @@ -710,37 +694,9 @@ export function Chat({ 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/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 (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()); - useEffect(() => { - const live = new Set(); - for (const m of messages) { - for (const p of m.parts) { - if ( - (p.type === "tool-research" || - p.type === "tool-demo" || - p.type === "tool-visual" || - p.type === "tool-pdf") && - p.toolCallId && - p.state !== "input-streaming" && - p.state !== "output-error" - ) { - live.add(p.toolCallId); - } - } - } - for (const id of live) { - if (!dispatchedRef.current.has(id)) refreshActivity(); - } - dispatchedRef.current = live; - }, [messages, refreshActivity]); + // (The dispatch nudge effect that lived here — refreshActivity per new + // dispatch tool part — died at S5: the store announces jobs-changed at + // the dispatch write; the dock subscribes.) // 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. @@ -756,11 +712,11 @@ export function Chat({ 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 + // A settle — the blob leaves the dock via the store's jobs-changed + // announce at this very write (S5); this handler owns the RECORD. // 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. @@ -774,10 +730,6 @@ export function Chat({ void syncFromServer(); } } - // 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" && !firstFetchRef.current) nudgeActivity = true; jobStatusRef.current.set(j.id, j.status); } firstFetchRef.current = false; @@ -790,7 +742,6 @@ export function Chat({ 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 @@ -798,7 +749,7 @@ export function Chat({ } catch { /* transient fetch failure — the next jobs-changed announcement re-fetches */ } - }, [chatId, syncFromServer, adoptAppendOnly, refreshActivity]); + }, [chatId, syncFromServer, adoptAppendOnly]); 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 diff --git a/src/agent/subagents/extraction.ts b/src/agent/subagents/extraction.ts index d730d4d..d0c3653 100644 --- a/src/agent/subagents/extraction.ts +++ b/src/agent/subagents/extraction.ts @@ -25,6 +25,7 @@ import type { } from "../../core/contracts.ts"; import { getGraph, mergeFindings, type DerivedGraph, type EmissionStats, type GraphDeps } from "../../core/graph.ts"; import type { GraphStore } from "../../core/ports.ts"; +import { emit } from "../../core/events.ts"; import { settleGraphLayout } from "../../core/layout-settle.ts"; import { batchRuns, coverageOf, messageViewLines } from "../../core/coverage.ts"; import { teacherProseBlocks } from "../../core/coverage.ts"; @@ -722,6 +723,17 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti usage: prev?.usage ?? {}, // phase changes never lose usage controller: prev?.controller ?? new AbortController(), }); + // Every phase transition announces the dock (push S5, 2026-10-02): + // notify-after-write — the registry is the feed's truth. + emit("activity-changed", {}); + } + + /** Run over (done, failed, aborted — every exit is a finally): the row + * leaves the feed. ONE seam for all delete sites so no exit path can + * miss the announce (push S5, 2026-10-02). */ + function finishExtraction(jobKey: string): void { + runningExtractions.delete(jobKey); + emit("activity-changed", {}); } /** Accumulate one billed step into the run's registry entry (live). */ @@ -899,7 +911,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti persistRunUsage(jobKey, Math.round(performance.now() - t0), undefined); throw err; } finally { - runningExtractions.delete(jobKey); + finishExtraction(jobKey); } } @@ -1062,7 +1074,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti throw err; } finally { inFlight.delete(chatId); - runningExtractions.delete(chatId); // done or failed — no longer running + finishExtraction(chatId); // done or failed — no longer running } } @@ -1217,7 +1229,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti throw err; } finally { inFlight.delete(id); - runningExtractions.delete(id); // done or failed — no longer running + finishExtraction(id); // done or failed — no longer running } } diff --git a/src/agent/subagents/practice-pipeline.ts b/src/agent/subagents/practice-pipeline.ts index 12d5db4..9317840 100644 --- a/src/agent/subagents/practice-pipeline.ts +++ b/src/agent/subagents/practice-pipeline.ts @@ -43,6 +43,7 @@ */ import type { GraphStore, PracticeStore } from "../../core/ports.ts"; +import { emit } from "../../core/events.ts"; import type { Problem, ProblemSet, ReasoningChain } from "../../core/contracts.ts"; import { chainsForSelection } from "../../core/practice.ts"; import { getGraph } from "../../core/graph.ts"; @@ -197,9 +198,20 @@ function registerRow( controller: new AbortController(), }; practiceRuns().set(key, row); + // The dock learns the row at the moment of cause (push S5, 2026-10-02): + // notify-after-write — the registry is the feed's truth. + emit("activity-changed", {}); return row; } +/** Row over (settled, failed, aborted — every exit): the row leaves the + * feed. ONE seam for all delete sites so no exit path can miss the + * announce (push S5, 2026-10-02). */ +function finishPractice(key: string): void { + practiceRuns().delete(key); + emit("activity-changed", {}); +} + /** DEDUP INPUT (computed ONCE at creation, passed to partitioner and * every generator): ready problems from the most recent ≤5 sets sharing * at least one selected node. */ @@ -318,7 +330,7 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { partitions = submitted.partitions; } catch (err) { await markSetError(set.id, isAbortLike(err) || controller.signal.aborted ? "Aborted" : messageOf(err)); - practiceRuns().delete(partitionRow.key); + finishPractice(partitionRow.key); return; } @@ -340,7 +352,7 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { deps.practiceStore.saveSet(cur); return cur; }); - practiceRuns().delete(partitionRow.key); + finishPractice(partitionRow.key); if (!saved) return; // deleted mid-run — stop quietly const genRows = stubs.map((stub, i) => @@ -367,6 +379,7 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { throw err; } row.phase = "generating"; + emit("activity-changed", {}); // in-place phase write — announce it const generated = await agents.runProblemGenerator({ target: stub.chains, fullChains: set.chains, @@ -404,7 +417,7 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { deps.practiceStore.saveSet(cur); }); } finally { - practiceRuns().delete(row.key); + finishPractice(row.key); } }), ); @@ -535,6 +548,7 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { await withBgLlmSlot(controller.signal, async () => { if (controller.signal.aborted) return; // dropped — no grade, no error state row.phase = "grading"; + emit("activity-changed", {}); // in-place phase write — announce it const set = deps.practiceStore.getSet(setId); if (!set) return; const problem = set.problems.find((p) => p.id === problemId); @@ -571,7 +585,7 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { // stands — one attempt; the UI shows "grade interrupted"). console.error(`[practice] grading failed for ${problemId}:`, isAbortLike(err) ? "aborted" : messageOf(err)); } finally { - practiceRuns().delete(row.key); + finishPractice(row.key); } } @@ -588,7 +602,7 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { const row = practiceRuns().get(key); if (!row) return; row.controller.abort(); - practiceRuns().delete(key); + finishPractice(key); } /** SET-level abort — composes from row aborts (every live row of the diff --git a/tests/extraction.test.ts b/tests/extraction.test.ts index c3af4ce..04aed81 100644 --- a/tests/extraction.test.ts +++ b/tests/extraction.test.ts @@ -57,6 +57,7 @@ import { simplifyChains } from "../src/core/graph.ts"; import { ModelsNotConfiguredError, ProviderKeyMissingError, providerKeyMissing, type ModelSlotName } from "../src/core/settings.ts"; import { coverageOf, messageViewLines } from "../src/core/coverage.ts"; import { chatStore } from "../src/store/chats.ts"; +import { subscribe } from "../src/core/events.ts"; import { PATHS } from "../config.ts"; // ─── fixtures ──────────────────────────────────────────────────────────────── @@ -372,6 +373,36 @@ describe("transcriptView (pure — teacher prose only, owner 2026-09-28)", () => // ─── the pipeline ─────────────────────────────────────────────────────────── describe("the extraction pipeline (extractor → auditor → merger)", () => { + it("push S5: every registry mutation announces activity-changed; reads stay silent", async () => { + const { chatId } = fixtureChat(); + const ledgers = memoryLedgers(); + const verdict: AuditVerdicts = { links: [] }; + const pipeline = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + extractor: submitModel("submit_findings", FINDINGS), + auditor: submitModel("submit_review", verdict), + }); + + let announces = 0; + const off = subscribe((e) => { + if (e.type === "activity-changed") announces++; + }); + // A read announces nothing (reads are silent — the bus hygiene law). + extractionStatuses(); + assert.equal(announces, 0); + + await pipeline.runExtraction(chatId); + // The run announced its lifecycle: setPhase at registration and every + // phase transition, finishExtraction at the exit — several events, + // and NONE after settle (the registry is empty; nothing left to say). + assert.ok(announces >= 2, `phases + finish announced (got ${announces})`); + const afterSettle = announces; + extractionStatuses(); + assert.equal(announces, afterSettle, "reads stay silent"); + off(); + }); + it("stringified-payload heal (owner, 2026-09-28): propositions arriving as ONE JSON string still validate and merge", async () => { const { chatId } = fixtureChat(); const ledgers = memoryLedgers(); diff --git a/tests/practice-pipeline.test.ts b/tests/practice-pipeline.test.ts index 7650109..2bba47c 100644 --- a/tests/practice-pipeline.test.ts +++ b/tests/practice-pipeline.test.ts @@ -26,6 +26,7 @@ import { BadAnswersError, type RunningPractice, } from "../src/agent/subagents/practice-pipeline.ts"; +import { subscribe } from "../src/core/events.ts"; import { chainCoverage } from "../src/core/practice.ts"; import type { GraphChain, GraphNode, ProblemSet } from "../src/core/contracts.ts"; import type { ChatStore, GraphStore, PracticeStore } from "../src/core/ports.ts"; @@ -403,6 +404,44 @@ describe("createProblemSet (generation)", () => { assert.equal(practiceStatuses().length, 0); }); + it("push S5: registerRow/phase-writes/finishPractice all announce activity-changed; reads stay silent", async () => { + const factory: FauxResponseStep = () => + fauxAssistantMessage( + [ + fauxToolCall( + "submit_problem", + { chains: ["b"], statement: "s", solution: "v", rubric: "r" } as ToolCall["arguments"], + { id: "gen-b" }, + ), + ], + { stopReason: "toolUse" }, + ); + const f = fixture({ + partitioner: partitionModel([{ chains: ["b"] }]), + problemGenerator: fauxEngine([factory]), + grader: echoGrader(1), + }); + const pipeline = f.make(); + + let announces = 0; + const off = subscribe((e) => { + if (e.type === "activity-changed") announces++; + }); + // Reads announce nothing. + practiceStatuses(); + assert.equal(announces, 0); + + const { done } = await pipeline.createProblemSet(["a", "b"], PARAMS); + await done; + // The lifecycle announced: partition row register + its finish, + // gen row register + the generating write + its finish. + assert.ok(announces >= 5, `register/phase/finish all announced (got ${announces})`); + const afterDone = announces; + practiceStatuses(); + assert.equal(announces, afterDone, "reads stay silent"); + off(); + }); + it("the learner's note threads to the partitioner + every generator and persists on the set (owner, 2026-09-29)", async () => { const partSeen: string[] = []; const genSeen: string[] = [];