wave 1 (platform stream): background-observability accuracy + owner-driven follow-ups
- observability: liveness-gated stale-job recovery, extraction run registry on globalThis with explicit 'embedding' phase, /api/jobs force-dynamic, client refresh nudges at every seam - headless: providerSafePng canonicalizer (multi-IDAT Chrome PNGs → single-IDAT) fixing qwen 400s - composer: Enter routes through pure enterAction — accepts a pending plan verdict (owner decision) - markdown chrome: session-scoped mermaid overlay + md-blocks component set + pan-zoom math, corner-copy (clipboard grant in electron main), math/table/code block fixes, heading weight 900 543/543 tests; typecheck/lint/build clean
This commit is contained in:
amosblomqvist
@@ -37,6 +37,18 @@ export function unregisterJobAbort(jobId: string): void {
|
||||
registry().delete(jobId);
|
||||
}
|
||||
|
||||
/** True while the job's loop is EXECUTING (a controller is registered) —
|
||||
* the liveness gate for the stale-job sweep: a "running" record with a
|
||||
* live registration belongs to a loop this process owns (hung or slow, it
|
||||
* is the one truth — never re-spawn beside it); without one the record is
|
||||
* an orphan of a process death and recovers immediately. The span is
|
||||
* loop-start to delivery-completion: the runner's unregistering finally
|
||||
* runs only after deliver() resolves, by which point finalizeJob has
|
||||
* flipped the record — so absence of registration is unambiguous. */
|
||||
export function isJobLive(jobId: string): boolean {
|
||||
return registry().has(jobId);
|
||||
}
|
||||
|
||||
/** Abort a running job's loop. Returns false when no controller is live —
|
||||
* the job's loop is not executing (crashed, or in the delivering-result
|
||||
* phase where the runner has already returned); the caller then owns the
|
||||
|
||||
@@ -478,8 +478,14 @@ function runNotesAuditor(
|
||||
|
||||
// ─── the pipeline status registry ─────────────────────────────────────────────
|
||||
|
||||
/** The pipeline phases, in order — what the home status endpoint shows. */
|
||||
export type ExtractionPhase = "extracting" | "auditing" | "merging";
|
||||
/** The pipeline phases, in order — what the home status endpoint shows.
|
||||
* `embedding` is its own phase (accuracy wave, 2026-09-19): the post-batch
|
||||
* tail — embed every uncovered proposition, re-cluster, settle the layout
|
||||
* — used to hide inside `merging`, so a run that had already merged every
|
||||
* batch sat on "merging" for the whole embedding tail (model load +
|
||||
* cold-start backfill on first runs = minutes). The phases must tell the
|
||||
* truth about what the loop is actually waiting on. */
|
||||
export type ExtractionPhase = "extracting" | "auditing" | "merging" | "embedding";
|
||||
|
||||
/** One in-flight pipeline run. Module-level: one process, one registry.
|
||||
* Exported for the status payload's consumer typing (the dock). */
|
||||
@@ -502,15 +508,34 @@ export interface RunningExtraction {
|
||||
controller: AbortController;
|
||||
}
|
||||
|
||||
const runningExtractions = new Map<string, RunningExtraction>();
|
||||
interface LhExtractionGlobal {
|
||||
__lhExtractionRuns?: {
|
||||
/** In-flight pipeline runs, keyed chatId (chat) / src id (notes). */
|
||||
running: Map<string, RunningExtraction>;
|
||||
/** The single-flight guard, shared by every pipeline instance. */
|
||||
inFlight: Set<string>;
|
||||
/** Last minted startedAt — kept monotonic so two runs starting in the
|
||||
* same millisecond never collide on the usage idempotency key. */
|
||||
lastStartedAt: number;
|
||||
};
|
||||
}
|
||||
|
||||
/** Last minted startedAt — kept monotonic so two runs starting in the same
|
||||
* millisecond never collide on the usage idempotency key. */
|
||||
let lastStartedAt = 0;
|
||||
/** THE RUN REGISTRIES live on globalThis, not module level — the
|
||||
* turn-liveness lesson: Next dev compiles each route entry as its own
|
||||
* module instance, so a module-level Map is a different object in the
|
||||
* /api/activity route than in the pipeline's chat-handler instance —
|
||||
* the status routes would read an always-empty registry. One process,
|
||||
* one truth — the same idiom as the turn, wake, and abort registries
|
||||
* (moved 2026-09-19, the observability-accuracy wave; also makes the
|
||||
* single-flight guard cross route instances). */
|
||||
function runs(): NonNullable<LhExtractionGlobal["__lhExtractionRuns"]> {
|
||||
const g = globalThis as typeof globalThis & LhExtractionGlobal;
|
||||
return (g.__lhExtractionRuns ??= { running: new Map(), inFlight: new Set(), lastStartedAt: 0 });
|
||||
}
|
||||
|
||||
/** In-flight pipeline runs (oldest first) — the /api/graph/status payload. */
|
||||
export function extractionStatuses(): RunningExtraction[] {
|
||||
return [...runningExtractions.values()].sort((a, b) => a.startedAt - b.startedAt);
|
||||
return [...runs().running.values()].sort((a, b) => a.startedAt - b.startedAt);
|
||||
}
|
||||
|
||||
/** True cancellation of the in-flight run for a chat (the dock's abort
|
||||
@@ -519,7 +544,7 @@ export function extractionStatuses(): RunningExtraction[] {
|
||||
* nothing merged, nothing checkpointed, partial usage still persisted.
|
||||
* Returns false when no run is in flight (idempotent no-op). */
|
||||
export function abortExtraction(chatId: string): boolean {
|
||||
const run = runningExtractions.get(chatId);
|
||||
const run = runs().running.get(chatId);
|
||||
if (!run) return false;
|
||||
run.controller.abort();
|
||||
return true;
|
||||
@@ -565,17 +590,22 @@ export interface ExtractionPipeline {
|
||||
|
||||
export function createExtractionPipeline(deps: ExtractionPipelineDeps): ExtractionPipeline {
|
||||
const { chatStore, graphStore, embedder: embed } = deps;
|
||||
const inFlight = new Set<string>();
|
||||
// THE SHARED REGISTRIES (runs() above — globalThis): every pipeline
|
||||
// instance (route entries in dev, tests) sees the same runs, the same
|
||||
// single-flight guard. No module-level state left in this factory.
|
||||
const runningExtractions = runs().running;
|
||||
const inFlight = runs().inFlight;
|
||||
|
||||
function setPhase(chatId: string, phase: ExtractionPhase, kind: "chat" | "notes", label?: string): void {
|
||||
const R = runs();
|
||||
const prev = runningExtractions.get(chatId);
|
||||
if (!prev) lastStartedAt = Math.max(Date.now(), lastStartedAt + 1);
|
||||
if (!prev) R.lastStartedAt = Math.max(Date.now(), R.lastStartedAt + 1);
|
||||
runningExtractions.set(chatId, {
|
||||
chatId,
|
||||
kind,
|
||||
label: prev?.label ?? label, // phase changes never lose the label
|
||||
phase,
|
||||
startedAt: prev?.startedAt ?? lastStartedAt,
|
||||
startedAt: prev?.startedAt ?? R.lastStartedAt,
|
||||
usage: prev?.usage ?? {}, // phase changes never lose usage
|
||||
controller: prev?.controller ?? new AbortController(),
|
||||
});
|
||||
@@ -742,6 +772,9 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
|
||||
);
|
||||
}
|
||||
const stats = total;
|
||||
// The embedding tail gets its own phase — the honest label for what
|
||||
// the run is waiting on here (see ExtractionPhase above).
|
||||
setPhase(chatId, "embedding", "chat");
|
||||
// Semantic embeddings (DECIDED 2026-09-13): topics come from
|
||||
// MEANING, so embed the new propositions BEFORE the settle and
|
||||
// the namer look at clusters. Failure is non-fatal — clustering
|
||||
@@ -868,7 +901,9 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
|
||||
const auditResult: AuditResult = { ...notesResult, attribution: { quiz: {} } };
|
||||
const stats = mergeFindings(id, auditResult, { chatStore, graphStore }, { messages: null, foldedAt: input.importedAt });
|
||||
// Same post-merge chain as the chat path: embeddings first
|
||||
// (topics from meaning), then the settle, then the namer.
|
||||
// (topics from meaning), then the settle, then the namer —
|
||||
// under their own phase, exactly like the chat path.
|
||||
setPhase(id, "embedding", "notes", input.filePath);
|
||||
await embed(graphStore).catch((err) => {
|
||||
console.error(`[extraction ${id}] embedding failed (non-fatal):`, err);
|
||||
});
|
||||
|
||||
@@ -179,6 +179,72 @@ function svgViewport(svg: string, opts: RenderSvgOptions): { width: number; heig
|
||||
};
|
||||
}
|
||||
|
||||
/** Canonicalize a raw Chrome screenshot for the model's eyes (2026-09-19 fix).
|
||||
*
|
||||
* WHY: Chrome writes PNGs with a SPLIT IDAT stream (4KB chunks) — and
|
||||
* Alibaba's vision decoder (behind OpenRouter's qwen3.8-flash, the
|
||||
* visual-maker's look-model) rejected those outright during the
|
||||
* 2026-09-18/19 window: "The image format is illegal and cannot be opened",
|
||||
* deterministically, at every size (50–800px) — while every single-IDAT
|
||||
* form passed (a handcrafted 1×1, and sharp re-encodes). On a later re-test
|
||||
* raw Chrome PNGs passed too (the upstream decoder was fixed or is flaky
|
||||
* per backend), but the canonical form stays: PNG spec says the IDAT chunks
|
||||
* are an arbitrary SPLIT of one zlib stream, so merging them is the more
|
||||
* standard shape, byte-identical stream, zero re-encoding.
|
||||
*
|
||||
* Pure Node on purpose (no sharp): the process already loads transformers'
|
||||
* nested sharp + libvips, and importing the root sharp (Next's optional
|
||||
* dependency) put a SECOND libvips into the process — macOS's ObjC runtime
|
||||
* warns (duplicate GNotificationCenterDelegate, "mysterious crashes").
|
||||
* One native image lib per process; app code carries none. */
|
||||
export function providerSafePng(png: Buffer): Buffer {
|
||||
const SIG_LEN = 8;
|
||||
if (png.length < SIG_LEN + 12 || png.subarray(0, SIG_LEN).toString("hex") !== "89504e470d0a1a0a") return png;
|
||||
// Walk the chunks once, merging the IDAT split into its canonical
|
||||
// single-chunk form (first IDAT's position, concatenated payload).
|
||||
const parts: Array<{ type: string; data: Buffer }> = [];
|
||||
let off = SIG_LEN;
|
||||
while (off + 12 <= png.length) {
|
||||
const len = png.readUInt32BE(off);
|
||||
if (off + 12 + len > png.length) break; // truncated tail — leave as-is
|
||||
parts.push({ type: png.subarray(off + 4, off + 8).toString("ascii"), data: png.subarray(off + 8, off + 8 + len) });
|
||||
off += 12 + len;
|
||||
}
|
||||
const idats = parts.filter((p) => p.type === "IDAT");
|
||||
if (idats.length < 2) return png; // already canonical
|
||||
const merged = Buffer.concat(idats.map((p) => p.data));
|
||||
let emittedIdat = false;
|
||||
const out: Buffer[] = [png.subarray(0, SIG_LEN)];
|
||||
for (const p of parts) {
|
||||
if (p.type !== "IDAT") out.push(chunk(p.type, p.data));
|
||||
else if (!emittedIdat) {
|
||||
out.push(chunk("IDAT", merged));
|
||||
emittedIdat = true;
|
||||
}
|
||||
}
|
||||
return Buffer.concat(out);
|
||||
}
|
||||
|
||||
/** One PNG chunk (length + type + data + CRC32) — the spec's exact layout. */
|
||||
function chunk(type: string, data: Buffer): Buffer {
|
||||
const out = Buffer.alloc(12 + data.length);
|
||||
out.writeUInt32BE(data.length, 0);
|
||||
out.write(type, 4, "ascii");
|
||||
data.copy(out, 8);
|
||||
out.writeUInt32BE(crc32(out.subarray(4, 8 + data.length)), 8 + data.length);
|
||||
return out;
|
||||
}
|
||||
|
||||
/** CRC-32 (PNG's polynomial, table-less) for chunk checksums. */
|
||||
function crc32(buf: Buffer): number {
|
||||
let c = ~0;
|
||||
for (const byte of buf) {
|
||||
c ^= byte;
|
||||
for (let k = 0; k < 8; k++) c = (c >>> 1) ^ (0xedb88320 & -(c & 1));
|
||||
}
|
||||
return ~c >>> 0;
|
||||
}
|
||||
|
||||
async function renderSvg(svg: string, opts: RenderSvgOptions = {}): Promise<Buffer> {
|
||||
const dir = tempDir("lh-visual-");
|
||||
const file = path.join(dir, "figure.html");
|
||||
@@ -194,8 +260,11 @@ async function renderSvg(svg: string, opts: RenderSvgOptions = {}): Promise<Buff
|
||||
const { width, height } = svgViewport(svg, opts);
|
||||
await page.setViewport({ width, height });
|
||||
await page.goto(pathToFileURL(file).href, { waitUntil: "load", timeout: 30_000 });
|
||||
const png = (await page.screenshot({ fullPage: true, type: "png" })) as Buffer;
|
||||
return png;
|
||||
const shot = (await page.screenshot({ fullPage: true, type: "png" })) as Buffer;
|
||||
// Chrome's raw split-IDAT PNG is canonicalized before it reaches a
|
||||
// provider (see providerSafePng) — the visual-maker feeds this exact
|
||||
// buffer back to the vision model as its only way to "look".
|
||||
return providerSafePng(Buffer.from(shot));
|
||||
} finally {
|
||||
await browser.close().catch(() => {});
|
||||
try {
|
||||
|
||||
+16
-8
@@ -42,7 +42,7 @@ import {
|
||||
type SteerMessage,
|
||||
} from "../core/contracts.ts";
|
||||
import { findPendingToolCall } from "../core/pending.ts";
|
||||
import { fireJobAbort } from "../agent/subagents/aborts.ts";
|
||||
import { fireJobAbort, isJobLive } from "../agent/subagents/aborts.ts";
|
||||
import { stores } from "./stores.ts";
|
||||
import { isTurnRunning } from "./turn-liveness.ts";
|
||||
|
||||
@@ -445,16 +445,24 @@ async function waitForIdle(chatId: string): Promise<void> {
|
||||
|
||||
// ─── crash recovery ──────────────────────────────────────────────────────────
|
||||
|
||||
/** Jobs this old with status "running" are orphans (crash/restart). */
|
||||
const STALE_JOB_MS = 15 * 60_000;
|
||||
const SWEEP_THROTTLE_MS = 30_000;
|
||||
|
||||
/**
|
||||
* Lazy sweep (called off the /api/jobs poll, throttled): every stale
|
||||
* "running" job either has a result (the wake never fired — re-deliver,
|
||||
* idempotently) or doesn't (the loop itself died — re-spawn it).
|
||||
* Lazy sweep (called off the /api/jobs poll, throttled), LIVENESS-GATED
|
||||
* (accuracy wave, 2026-09-19): a "running" record without a live loop
|
||||
* registration (subagents/aborts.ts) is an orphan of a process death —
|
||||
* re-deliver its stored result (idempotently) or re-spawn the loop, at
|
||||
* once (the old 15-minute age heuristic waited on ghosts AND re-ran
|
||||
* hung-but-alive loops beside themselves — the figure-maker incident,
|
||||
* 2026-09-18). A record WITH a live registration is provably executing
|
||||
* in THIS process — never touched, however old: a hung first model call
|
||||
* (zero usage ever billed) stays honestly "running" in the dock until
|
||||
* END kills it. The abort registration spans loop start to delivery
|
||||
* completion, so a live registration can never belong to a dead run.
|
||||
*/
|
||||
function recoverStaleJobs(): void {
|
||||
/** Re-run a job's sub-agent loop — exposed for the sweep's tests (the
|
||||
* /api/jobs poll is its only production caller, disabled under NODE_ENV=test). */
|
||||
export function recoverStaleJobs(): void {
|
||||
const s = state();
|
||||
const now = Date.now();
|
||||
if (now - s.lastSweepAt < SWEEP_THROTTLE_MS) return;
|
||||
@@ -463,7 +471,7 @@ function recoverStaleJobs(): void {
|
||||
for (const chat of stores.chatStore.listChats()) {
|
||||
for (const job of stores.jobStore.listJobs(chat.id)) {
|
||||
if (job.status !== "running") continue;
|
||||
if (now - Date.parse(job.updatedAt) < STALE_JOB_MS) continue;
|
||||
if (isJobLive(job.id)) continue; // provably executing here — never re-run beside it
|
||||
if (!stores.chatStore.getChat(job.chatId)) continue;
|
||||
if (isTurnRunning(job.chatId)) continue; // a wake turn may be delivering it right now
|
||||
if (job.kind === "graph-context") {
|
||||
|
||||
Reference in New Issue
Block a user