Improve Live polling responsiveness and reliability

Restore foreground/background polling as the primary harness architecture, add progressive publication and framework-safe previews, and harden quality and regression coverage. The experimental app-server runtime is intentionally excluded.\n\nPrepared with AI assistance under maintainer direction.
2026-09-17 16:46:31 +03:00 · 2026-07-15 17:05:30 -07:00
parent 8259c28209
commit c6ac34b929
74 changed files with 8511 additions and 333 deletions
@@ -29,7 +29,9 @@ import {
  resolveLiveBrowserScriptParts,
 } from './live/browser-script-parts.mjs';
 import { createLiveSessionStore } from './live/session-store.mjs';
+import { runGenerationPreflight } from './live/generation-preflight.mjs';
 import { validateEvent } from './live/event-validation.mjs';
+import { selectAvailablePendingEvent } from './live/poll-lanes.mjs';
 import { createManualEditRoutes } from './live/manual-edit-routes.mjs';
 import { LIVE_COMMANDS } from './live/vocabulary.mjs';
 import {
@@ -51,6 +53,7 @@ import {
  applyDeferredSvelteComponentAccepts,
  removeAllSvelteComponentSessions,
 } from './live/svelte-component.mjs';
+import { removeAllVueComponentSessions } from './live/vue-component.mjs';

 const __dirname = path.dirname(fileURLToPath(import.meta.url));
 // PRODUCT.md / DESIGN.md live wherever context.mjs resolves. The generated
@@ -156,29 +159,133 @@ function restorePendingEventsFromStore() {
  }
 }

-function findAvailablePendingEvent(now = Date.now()) {
-  for (const entry of state.pendingEvents) {
-    if (entry.leaseUntil && entry.leaseUntil > now) continue;
-    return entry;
-  }
-  return null;
+function findAvailablePendingEvent(now = Date.now(), types = null) {
+  return selectAvailablePendingEvent(state.pendingEvents, { now, types });
 }

 function leaseEvent(entry, leaseMs) {
+  prepareGenerateEventForLease(entry);
  if (!entry.event?.id) {
    const idx = state.pendingEvents.indexOf(entry);
    if (idx !== -1) state.pendingEvents.splice(idx, 1);
    return entry.event;
  }
  entry.leaseUntil = Date.now() + leaseMs;
+  recordGenerateDelivery(entry);
  scheduleLeaseFlush();
  broadcastAgentPollingIfChanged();
  return entry.event;
 }

-function acknowledgePendingEvent(id) {
+function recordGenerateDelivery(entry) {
+  const event = entry?.event;
+  if (!event || event.type !== 'generate' || event.generationReadyAt) return;
+  const at = Date.now();
+  entry.event = { ...event, generationReadyAt: at };
+  state.sessionStore?.appendEvent(entry.event);
+  recordAgentPhase(event.id, 'generation_ready', { at });
+}
+
+function prepareGenerateEventForLease(entry) {
+  const event = entry?.event;
+  if (!event || event.type !== 'generate' || event.scaffoldAttempted) return;
+
+  recordAgentPhase(event.id, 'picked_up');
+  recordAgentPhase(event.id, 'scaffolding');
+  const result = runGenerationPreflight(event, {
+    cwd: process.cwd(),
+    scriptsDir: __dirname,
+  });
+  entry.event = {
+    ...event,
+    scaffoldAttempted: true,
+    scaffoldDurationMs: result.durationMs ?? null,
+    ...(result.ok ? { scaffold: result.scaffold } : { scaffoldError: result.error || result.reason }),
+  };
+  state.sessionStore?.appendEvent(entry.event);
+  recordAgentPhase(event.id, result.ok ? 'source_ready' : 'scaffold_fallback', {
+    durationMs: result.durationMs ?? null,
+    previewMode: result.scaffold?.previewMode || 'source',
+  });
+}
+
+function recordAgentPhase(id, phase, details = {}) {
+  if (!id) return;
+  const event = {
+    type: 'agent_phase',
+    id,
+    phase,
+    at: Date.now(),
+    ...details,
+  };
+  state.sessionStore?.appendEvent(event);
+  broadcast(event);
+}
+
+function recordGenerationCheckpoint(event) {
+  if (!event?.id || event.type !== 'checkpoint') return;
+  if (generationIsFenced(event.id)) return;
+  const arrived = Number(event.arrivedVariants) || 0;
+  const expected = Number(event.expectedVariants) || 0;
+  if (arrived <= 0 || expected <= 0) return;
+  const previewMode = event.previewMode || 'source';
+  const previewFile = event.previewFile || event.file;
+  if (previewFile) {
+    broadcast({
+      type: 'variant_progress',
+      id: event.id,
+      file: previewFile,
+      sourceFile: event.sourceFile || (previewMode === 'source' ? previewFile : undefined),
+      previewFile,
+      previewMode,
+      arrivedVariants: arrived,
+      expectedVariants: expected,
+      publicationKind: event.publicationKind || 'variants',
+    });
+  }
+  const details = {
+    arrivedVariants: arrived,
+    expectedVariants: expected,
+    checkpointReason: event.reason || null,
+  };
+  const at = Date.now();
+  if (!generationPhaseAlreadyRecorded(event.id, 'first_reviewable')) {
+    recordAgentPhase(event.id, 'first_reviewable', { ...details, at });
+  }
+  if (arrived >= 2 && expected >= 3 && !generationPhaseAlreadyRecorded(event.id, 'second_reviewable')) {
+    recordAgentPhase(event.id, 'second_reviewable', { ...details, at });
+  }
+  if (arrived >= expected && !generationPhaseAlreadyRecorded(event.id, 'all_variants_ready')) {
+    recordAgentPhase(event.id, 'all_variants_ready', { ...details, at });
+  }
+}
+
+function generationIsFenced(id) {
+  if (!state.sessionStore || !id) return false;
+  try {
+    const snapshot = state.sessionStore.getSnapshot(id, { includeCompleted: true });
+    return snapshot?.generationCanceled === true;
+  } catch {
+    return false;
+  }
+}
+
+function generationPhaseAlreadyRecorded(id, phase) {
+  if (!state.sessionStore) return false;
+  try {
+    const snapshot = state.sessionStore.getSnapshot(id, { includeCompleted: true });
+    return !!snapshot?.generationTimings?.[phase];
+  } catch {
+    return false;
+  }
+}
+
+function acknowledgePendingEvent(id, sourceEventType) {
  if (!id) return false;
-  const idx = state.pendingEvents.findIndex((entry) => entry.event?.id === id);
+  const idx = state.pendingEvents.findIndex((entry) => (
+    entry.event?.id === id
+    && (!sourceEventType || entry.event?.type === sourceEventType)
+  ));
  if (idx === -1) return false;
  const acknowledged = state.pendingEvents[idx].event;
  state.pendingEvents.splice(idx, 1);
@@ -187,9 +294,39 @@ function acknowledgePendingEvent(id) {
  return acknowledged;
 }

-function findPendingEventById(id) {
+function releasePendingEvent(id, sourceEventType) {
+  const entry = state.pendingEvents.find((item) => (
+    item.event?.id === id
+    && (!sourceEventType || item.event?.type === sourceEventType)
+  ));
+  if (!entry) return null;
+  entry.leaseUntil = 0;
+  scheduleLeaseFlush();
+  return entry.event;
+}
+
+function retirePendingGeneration(id) {
+  if (!id) return 0;
+  let retired = 0;
+  for (let index = state.pendingEvents.length - 1; index >= 0; index -= 1) {
+    const event = state.pendingEvents[index]?.event;
+    if (event?.id !== id || event.type !== 'generate') continue;
+    state.pendingEvents.splice(index, 1);
+    retired += 1;
+  }
+  if (retired > 0) {
+    scheduleLeaseFlush();
+    broadcastAgentPollingIfChanged();
+  }
+  return retired;
+}
+
+function findPendingEventById(id, sourceEventType) {
  if (!id) return null;
-  const entry = state.pendingEvents.find((item) => item.event?.id === id);
+  const entry = state.pendingEvents.find((item) => (
+    item.event?.id === id
+    && (!sourceEventType || item.event?.type === sourceEventType)
+  ));
  return entry?.event || null;
 }

@@ -224,7 +361,13 @@ function summarizeActiveSessionForClient(snapshot = {}) {
    arrivedVariants: snapshot.arrivedVariants ?? 0,
    visibleVariant: snapshot.visibleVariant ?? null,
    checkpointRevision: snapshot.checkpointRevision ?? 0,
+    browserCheckpointRevision: snapshot.browserCheckpointRevision ?? snapshot.checkpointRevision ?? 0,
+    publicationCheckpointRevision: snapshot.publicationCheckpointRevision ?? 0,
    paramValues: snapshot.paramValues || {},
+    paramsPublished: snapshot.paramsPublished === true,
+    generationPhase: snapshot.generationPhase ?? null,
+    generationCanceled: snapshot.generationCanceled === true,
+    cancelReason: snapshot.cancelReason ?? null,
  };
 }

@@ -269,13 +412,21 @@ function scheduleLeaseFlush() {
 function flushPendingPolls() {
  let changed = false;
  while (state.pendingPolls.length > 0) {
-    const entry = findAvailablePendingEvent();
+    let pollIndex = -1;
+    let entry = null;
+    for (let index = 0; index < state.pendingPolls.length; index += 1) {
+      const candidate = findAvailablePendingEvent(Date.now(), state.pendingPolls[index].types);
+      if (!candidate) continue;
+      pollIndex = index;
+      entry = candidate;
+      break;
+    }
    if (!entry) {
      scheduleLeaseFlush();
      broadcastAgentPollingIfChanged();
      return;
    }
-    const poll = state.pendingPolls.shift();
+    const [poll] = state.pendingPolls.splice(pollIndex, 1);
    poll.resolve(leaseEvent(entry, poll.leaseMs));
    changed = true;
  }
@@ -284,9 +435,10 @@ function flushPendingPolls() {
 }

 function agentPollingConnected() {
-  const now = Date.now();
-  return state.pendingPolls.length > 0
-    || state.pendingEvents.some((entry) => entry.leaseUntil && entry.leaseUntil > now);
+  // A leased event only proves that a poll returned once. The foreground task
+  // may have ended immediately afterward, so only an actively waiting poll is
+  // evidence that steering can wake the task right now.
+  return state.pendingPolls.length > 0;
 }

 function broadcastAgentPollingIfChanged() {
@@ -689,6 +841,15 @@ function createRequestHandler({ detectScript, liveScriptParts }) {
          res.end(JSON.stringify({ error }));
          return;
        }
+        if (msg.type === 'agent_phase') {
+          recordAgentPhase(msg.id, msg.phase, {
+            ...(Number.isFinite(msg.durationMs) ? { durationMs: msg.durationMs } : {}),
+            owner: typeof msg.owner === 'string' ? msg.owner : undefined,
+          });
+          res.writeHead(200, { 'Content-Type': 'application/json' });
+          res.end(JSON.stringify({ ok: true }));
+          return;
+        }
        if (state.sessionStore && msg.id) {
          try {
            state.sessionStore.appendEvent(msg);
@@ -698,6 +859,10 @@ function createRequestHandler({ detectScript, liveScriptParts }) {
            return;
          }
        }
+        if (msg.type === 'accept' || msg.type === 'discard') {
+          retirePendingGeneration(msg.id);
+        }
+        recordGenerationCheckpoint(msg);
        if (msg.type === 'exit') {
          cleanupSvelteComponentSessionsBeforeExit();
        }
@@ -738,6 +903,12 @@ function createRequestHandler({ detectScript, liveScriptParts }) {
 // Agent poll endpoints (unchanged from WS version)
 // ---------------------------------------------------------------------------

+function parsePollTypes(value) {
+  if (!value) return null;
+  const types = String(value).split(',').map((type) => type.trim()).filter(Boolean);
+  return types.length > 0 ? new Set(types) : null;
+}
+
 function handlePollGet(req, res, url) {
  const token = url.searchParams.get('token');
  if (token !== state.token) {
@@ -748,13 +919,14 @@ function handlePollGet(req, res, url) {
  state.lastPollAt = Date.now();
  const timeout = parseInt(url.searchParams.get('timeout') || DEFAULT_POLL_TIMEOUT, 10);
  const leaseMs = parseInt(url.searchParams.get('leaseMs') || '30000', 10);
-  const available = findAvailablePendingEvent();
+  const types = parsePollTypes(url.searchParams.get('types'));
+  const available = findAvailablePendingEvent(Date.now(), types);
  if (available) {
    res.writeHead(200, { 'Content-Type': 'application/json' });
    res.end(JSON.stringify(leaseEvent(available, leaseMs)));
    return;
  }
-  const poll = { resolve, leaseMs };
+  const poll = { resolve, leaseMs, types };
  const timer = setTimeout(() => {
    const idx = state.pendingPolls.indexOf(poll);
    if (idx !== -1) state.pendingPolls.splice(idx, 1);
@@ -783,12 +955,20 @@ function sessionFileMetadataFromPollReply(file) {
  if (!file || typeof file !== 'string') return { file };
  const normalized = file.split(path.sep).join('/');
  const base = { file: normalized };
-  if (!normalized.endsWith('/manifest.json') && normalized !== 'manifest.json') return base;
-  if (!normalized.includes('node_modules/.impeccable-live/') && !normalized.includes('src/lib/impeccable/')) return base;
+  const sourceArtifactPreview = normalized.includes('.impeccable/live/previews/')
+    && !normalized.endsWith('/manifest.json');
+  const metadataFile = sourceArtifactPreview
+    ? normalized.slice(0, normalized.lastIndexOf('/') + 1) + 'manifest.json'
+    : normalized;
+  if (!metadataFile.endsWith('/manifest.json') && metadataFile !== 'manifest.json') return base;
+  if (!metadataFile.includes('node_modules/.impeccable-live/')
+      && !metadataFile.includes('src/lib/impeccable/')
+      && !metadataFile.includes('/.impeccable-live/')
+      && !metadataFile.includes('.impeccable/live/previews/')) return base;

  let full;
  try {
-    full = path.resolve(process.cwd(), normalized);
+    full = path.resolve(process.cwd(), metadataFile);
    const rel = path.relative(process.cwd(), full);
    if (!rel || rel.startsWith('..') || path.isAbsolute(rel)) return base;
  } catch {
@@ -797,18 +977,40 @@ function sessionFileMetadataFromPollReply(file) {

  try {
    const manifest = JSON.parse(fs.readFileSync(full, 'utf-8'));
-    if (manifest?.previewMode !== 'svelte-component' || !manifest.sourceFile) return base;
+    if (!['svelte-component', 'vue-component', 'source-artifact'].includes(manifest?.previewMode)
+        || !manifest.sourceFile) return base;
+    const previewFile = manifest.previewMode === 'source-artifact'
+      ? String(manifest.previewFile || normalized).split(path.sep).join('/')
+      : normalized;
    return {
      file: String(manifest.sourceFile).split(path.sep).join('/'),
      sourceFile: String(manifest.sourceFile).split(path.sep).join('/'),
-      previewFile: normalized,
-      previewMode: 'svelte-component',
+      previewFile,
+      previewMode: manifest.previewMode,
    };
  } catch {
    return base;
  }
 }

+function inferSourceEventType(msg = {}, pendingEvents = state.pendingEvents) {
+  const pendingTypes = new Set(
+    pendingEvents
+      .filter((entry) => entry.event?.id === msg.id)
+      .map((entry) => entry.event?.type),
+  );
+  if (msg.type === 'discarded' || msg.type === 'discard') return 'discard';
+  if (msg.type === 'complete') {
+    if (pendingTypes.has('carbonize_cleanup')) return 'carbonize_cleanup';
+    return pendingTypes.has('accept') ? 'accept' : (pendingTypes.has('generate') ? 'generate' : undefined);
+  }
+  if (msg.type === 'steer_done') return 'steer';
+  // `agent_done` can be the automatic acknowledgement for a carbonize Accept.
+  // New pollers send sourceEventType explicitly; default to generate only for
+  // older callers so a late worker cannot acknowledge a queued Accept.
+  return msg.type === 'agent_done' || msg.type === 'done' ? 'generate' : undefined;
+}
+
 function handlePollPost(req, res) {
  let body = '';
  req.on('data', (c) => { body += c; });
@@ -869,7 +1071,23 @@ function handlePollPost(req, res) {
      res.end(JSON.stringify({ error: 'stale_manual_edit_apply_reply', ...rollback }));
      return;
    }
-    const pendingEventBeforeAck = findPendingEventById(msg.id);
+    const sourceEventType = msg.sourceEventType || inferSourceEventType(msg);
+    if (msg.type === 'retry') {
+      const releasedEvent = releasePendingEvent(msg.id, sourceEventType);
+      if (!releasedEvent) {
+        res.writeHead(msg.id ? 404 : 400, { 'Content-Type': 'application/json' });
+        res.end(JSON.stringify({
+          error: msg.id ? 'unknown_poll_retry_id' : 'missing_poll_retry_id',
+          id: msg.id,
+        }));
+        return;
+      }
+      flushPendingPolls();
+      res.writeHead(200, { 'Content-Type': 'application/json' });
+      res.end(JSON.stringify({ ok: true, released: true }));
+      return;
+    }
+    const pendingEventBeforeAck = findPendingEventById(msg.id, sourceEventType);
    if (pendingEventBeforeAck?.type === 'steer' && msg.type === 'steer_done'
        && !msg.file && !(typeof msg.message === 'string' && msg.message.trim())) {
      res.writeHead(400, { 'Content-Type': 'application/json' });
@@ -879,7 +1097,7 @@ function handlePollPost(req, res) {
      }));
      return;
    }
-    const acknowledgedEvent = acknowledgePendingEvent(msg.id);
+    const acknowledgedEvent = acknowledgePendingEvent(msg.id, sourceEventType);
    let skipJournalReply = false;
    let existingSession = null;
    if (!acknowledgedEvent && state.sessionStore && msg.id) {
@@ -971,6 +1189,11 @@ function cleanupSvelteComponentSessionsBeforeExit() {
  } catch (err) {
    console.warn('[impeccable] Svelte component session cleanup failed:', err.message);
  }
+  try {
+    removeAllVueComponentSessions(process.cwd());
+  } catch (err) {
+    console.warn('[impeccable] Vue component session cleanup failed:', err.message);
+  }
 }

 function applyLegacyDeferredAcceptsOnStartup() {
@@ -1083,7 +1306,10 @@ if (args.includes('--background')) {
        process.exit(0);
      }
    } catch { /* not ready yet */ }
-    await new Promise(r => setTimeout(r, 200));
+    // The detached child is typically listening in 35-45ms. A 200ms polling
+    // floor dominated configured cold Live startup; poll cheaply and return
+    // as soon as the child has written its ready record.
+    await new Promise(r => setTimeout(r, 5));
  }
  console.error('Timed out waiting for live server to start.');
  process.exit(1);