feat(dashboard): split OpenClaw Bay publishing lane

.gitignore

@@ -6,6 +6,7 @@ dist/
 .artifacts/
 artifacts/
 dashboard/.wrangler/
+.wrangler/
 .clawsweeper-repair/
 .claude/
 tmp/

dashboard/exact-review-publication-batches.ts

@@ -56,6 +56,10 @@ export type PublicationBatchStats = {
   expired: number;
   activeItems: number;
   activeItemKeys: string[];
+  // This contains only unfinished, currently leased membership. It lets the
+  // read-only Bay projection identify the bounded batch that owns an item
+  // without retaining a separate event history or looking it up through GitHub.
+  activeItemBatches: Array<{ itemKey: string; batchId: string }>;
   nextLeaseExpiresAt: number | null;
   oldestActiveAt: number | null;
   reclaimedItemsRetained: number;
@@ -424,6 +428,7 @@ export class ExactReviewPublicationBatchStore {
         expired: counts.get("expired") ?? 0,
         activeItems: activeLease.itemKeys.length,
         activeItemKeys: activeLease.itemKeys,
+        activeItemBatches: activeLease.items,
         nextLeaseExpiresAt: activeLease.nextLeaseExpiresAt,
         oldestActiveAt: leased ? Number(leased.oldest_at) : null,
         reclaimedItemsRetained,
@@ -470,6 +475,10 @@ export class ExactReviewPublicationBatchStore {
       ),
     );
     return {
+      items: rows.map((row) => ({
+        itemKey: String(row.item_key),
+        batchId: String(row.batch_id),
+      })),
       itemKeys: rows.map((row) => String(row.item_key)),
       activeBatches: new Set(rows.map((row) => String(row.batch_id))).size,
       nextLeaseExpiresAt: rows.length

dashboard/exact-review-queue.ts

@@ -1880,7 +1880,10 @@ export class ExactReviewQueue {
         stateWriterCoordinatorQueuedStaleMs(this.env),
       );
       const publicationBatches = this.batchStore.stats(now);
-      const batchOwnedItemKeys = new Set<string>(publicationBatches.activeItemKeys);
+      const batchByItemKey = new Map<string, ExactReviewBayBatchOwner>(
+        publicationBatches.activeItemBatches.map((batch) => [batch.itemKey, batch] as const),
+      );
+      const batchOwnedItemKeys = new Set<string>(batchByItemKey.keys());
       const freshPublicationItemKeys = this.freshPublicationItemKeysSync(state, now);
       const legacyExcludedItemKeys = new Set(batchOwnedItemKeys);
       if (exactReviewPublicationBatchingEnabled(this.env)) {
@@ -1920,7 +1923,11 @@ export class ExactReviewQueue {
       return json({
         ...stats,
         pressure: elevateExactReviewPressureForPublication(stats.pressure, publicationHealth),
-        bay_projection: exactReviewQueueBayProjection(Object.values(state.items), bayPriorityKeys),
+        bay_projection: exactReviewQueueBayProjection(
+          Object.values(state.items),
+          bayPriorityKeys,
+          batchByItemKey,
+        ),
         lanes: {
           review: {
             ...stats.lanes.review,
@@ -7257,7 +7264,9 @@ function exactReviewQueueLane(item: ExactReviewQueueItem) {
 // state. Keep this representation bounded and scrubbed: it is public dashboard
 // data, not a queue-inspection API. Live workers remain the authority for the
 // reviewing stage; these records only make the otherwise invisible admission,
-// setup, publication, and recovery phases visible.
+// setup, publication, and recovery phases visible. Publication is distinct
+// from the publisher workflow's deterministic follow-up, which the Bay shows
+// from the live worker as Applying.
 const EXACT_REVIEW_BAY_SAMPLE_LIMIT = 24;
 // The dashboard can retain both a terminal-buffer card and its washed card
 // while their live queue retry is pending. Accept all bounded Bay candidates
@@ -7267,6 +7276,7 @@ const EXACT_REVIEW_BAY_STAGES = [
   "arriving",
   "setting-up",
   "reviewing",
+  "publishing",
   "applying",
   "repairing",
 ] as const;
@@ -7280,10 +7290,28 @@ type ExactReviewBayProjectionItem = {
   created_at: string;
   updated_at: string;
   next_attempt_at: string;
+  batch_id?: string;
+  batch_created_at?: string;
+};
+
+type ExactReviewBayBatchOwner = {
+  batchId: string;
 };
 
-function exactReviewQueueBayStage(item: ExactReviewQueueItem): ExactReviewBayStage {
-  if (exactReviewQueueIsPublication(item)) return "applying";
+function exactReviewQueueBayStage(
+  item: ExactReviewQueueItem,
+  batchByItemKey: ReadonlyMap<string, ExactReviewBayBatchOwner> = new Map(),
+): ExactReviewBayStage {
+  // A parked item is deliberately no longer making normal queue progress. This
+  // includes bounded review-retry exhaustion, permanent dispatch rejection,
+  // and a publication that needs its dead-letter/recovery path. Keep it in the
+  // exception cove instead of making it look like an active setup or publisher.
+  if (item.state === "parked") return "repairing";
+  // The batch publisher's GitHub job is intentionally targetless. Its durable
+  // batch membership is the authoritative bounded source for the individual
+  // items it is currently applying, without another GitHub lookup.
+  if (batchByItemKey.has(item.key)) return "applying";
+  if (exactReviewQueueIsPublication(item)) return "publishing";
   if (isLowPriorityExactReviewDecision(item.decision)) return "repairing";
   return item.state === "pending" ? "arriving" : "setting-up";
 }
@@ -7306,22 +7334,31 @@ function exactReviewQueueBayPriorityKeys(values: string[]) {
 function exactReviewQueueBayProjection(
   items: ExactReviewQueueItem[],
   priorityItemKeys: string[] = [],
+  batchByItemKey: ReadonlyMap<string, ExactReviewBayBatchOwner> = new Map(),
 ) {
   const projected = new Map<string, ExactReviewBayProjectionItem>();
   for (const item of items) {
-    if (item.state === "parked") continue;
+    // Parked records are not terminal outcomes: they remain bounded durable
+    // queue work that needs recovery. Keep their already-scrubbed identity in
+    // the projection so Bay shows the exception rather than a false empty lane.
     const repository = String(item.decision.targetRepo || "").trim();
     const itemNumber = Number(item.decision.itemNumber);
     if (!repository || !Number.isSafeInteger(itemNumber) || itemNumber <= 0) continue;
+    const batch = batchByItemKey.get(item.key);
     const candidate: ExactReviewBayProjectionItem = {
       item_key: `${repository}#${itemNumber}`,
       repository,
       item_number: itemNumber,
-      stage: exactReviewQueueBayStage(item),
+      stage: exactReviewQueueBayStage(item, batchByItemKey),
       queue_state: item.state,
       created_at: new Date(item.createdAt).toISOString(),
       updated_at: new Date(item.updatedAt).toISOString(),
       next_attempt_at: new Date(item.nextAttemptAt).toISOString(),
+      ...(batch
+        ? {
+            batch_id: batch.batchId,
+          }
+        : {}),
     };
     const previous = projected.get(candidate.item_key);
     const candidateUpdatedAt = Date.parse(candidate.updated_at);

dashboard/worker.ts

@@ -3255,19 +3255,22 @@ async function activeWorkerSnapshot(
   const detailRuns: WorkflowRunSummary[] = runs.slice(0, detailRunLimit);
   const results = await mapWithConcurrency(detailRuns, fetchConcurrency, async (run) => {
     try {
-      const jobs = await workflowJobsForRun(env, repo, run.id, github);
+      const jobs = await workflowJobsForRun(env, repo, run.id, github, run);
+      const activeJobs = jobs.filter((job) => isActiveWorkflowJob(job));
       return {
         run,
-        workers: jobs
-          .filter((job) => isActiveWorkflowJob(job) && isCodexWorkerJob(job))
+        workers: activeJobs
+          .filter((job) => isDashboardWorkerJob(job, run))
           .map((job) => normalizeWorkerJob(run, job)),
-        hasWorkerJobs: jobs.some((job) => isCodexWorkerJob(job)),
+        codexWorkers: activeJobs.filter((job) => isCodexWorkerJob(job)).length,
+        hasWorkerJobs: jobs.some((job) => isDashboardWorkerJob(job, run)),
         error: null,
       };
     } catch (error) {
       return {
         run,
         workers: [],
+        codexWorkers: 0,
         hasWorkerJobs: false,
         error: error instanceof Error ? error.message : String(error),
       };
@@ -3276,12 +3279,15 @@ async function activeWorkerSnapshot(
   const workers = [];
   const errors = [];
   let fallbacks = 0;
+  let codexWorkers = 0;
   for (const result of results) {
+    codexWorkers += result.codexWorkers;
     if (result.error) {
       errors.push(`workflow jobs ${result.run.id}: ${result.error}`);
       if (isCodexWorkflowFallback(result.run)) {
         workers.push(normalizeFallbackWorker(result.run));
         fallbacks += 1;
+        codexWorkers += 1;
       }
       continue;
     }
@@ -3290,12 +3296,14 @@ async function activeWorkerSnapshot(
     } else if (!result.hasWorkerJobs && isCodexWorkflowFallback(result.run)) {
       workers.push(normalizeFallbackWorker(result.run));
       fallbacks += 1;
+      codexWorkers += 1;
     }
   }
   for (const run of runs.slice(detailRunLimit)) {
     if (!isCodexWorkflowFallback(run)) continue;
     workers.push(normalizeFallbackWorker(run));
     fallbacks += 1;
+    codexWorkers += 1;
   }
   workers.sort(
     (left, right) =>
@@ -3305,7 +3313,7 @@ async function activeWorkerSnapshot(
   );
   await attachWorkerTargets(env, workers, errors);
   return {
-    count: workers.length,
+    count: codexWorkers,
     workers,
     detailRuns: detailRuns.length,
     fallbacks,
@@ -3348,7 +3356,7 @@ async function recentWorkerHealth(
   const results = await mapWithConcurrency(completedRuns, fetchConcurrency, async (run) => {
     try {
       return {
-        attempts: (await workflowJobsForRun(env, repo, run.id, github))
+        attempts: (await workflowJobsForRun(env, repo, run.id, github, run))
           .filter((job) => isCodexWorkerJob(job))
           .map((job) => workerHealthAttempt(run, job))
           .filter(Boolean),
@@ -4244,6 +4252,7 @@ async function workflowJobsForRun(
   repo,
   runId,
   github: GithubJsonReader = (path) => githubJson(env, path),
+  run?: WorkflowRunSummary,
 ) {
   const key = `workflow-jobs:${repo}:${runId}`;
   const cached = await readStoredJson(env, key);
@@ -4263,7 +4272,9 @@ async function workflowJobsForRun(
       break;
     }
   }
-  const hasActiveWorker = jobs.some((job) => isActiveWorkflowJob(job) && isCodexWorkerJob(job));
+  const hasActiveWorker = jobs.some(
+    (job) => isActiveWorkflowJob(job) && isDashboardWorkerJob(job, run),
+  );
   await writeStoredJson(
     env,
     key,
@@ -4288,6 +4299,25 @@ function isCodexWorkerJob(job) {
   );
 }
 
+function isExactReviewPublicationJob(job, run?: WorkflowRunSummary) {
+  const name = String(job?.name || "");
+  const steps = Array.isArray(job?.steps) ? job.steps : [];
+  const workflow = `${run?.name || ""} ${run?.display_title || ""}`;
+  return (
+    /publish (?:exact )?review artifacts?/i.test(name) ||
+    (/publish exact review batch/i.test(workflow) && /^publish$/i.test(name)) ||
+    steps.some((step) =>
+      /claim durable exact review publication|claim one durable publication batch|finalize healthy members under a fenced heartbeat|publish event result and apply safe close|complete durable exact review publication|apply review artifacts|publish review artifact action ledger|commit review records/i.test(
+        String(step?.name || ""),
+      ),
+    )
+  );
+}
+
+function isDashboardWorkerJob(job, run?: WorkflowRunSummary) {
+  return isCodexWorkerJob(job) || isExactReviewPublicationJob(job, run);
+}
+
 function normalizeWorkerJob(run, job) {
   const runItem = classifyRun(run);
   const target = workerTargetFromJob(runItem, job.name);
@@ -4310,6 +4340,7 @@ function normalizeWorkerJob(run, job) {
   return {
     id: job.id,
     source: "job",
+    is_codex_worker: isCodexWorkerJob(job),
     name: String(job.name || runItem.title || "Codex worker"),
     mode,
     work_kind: workKind,
@@ -4369,6 +4400,7 @@ function normalizeFallbackWorker(run) {
   return {
     id: `run-${run.id}`,
     source: "workflow-fallback",
+    is_codex_worker: true,
     name: item.title || item.workflow || "Codex worker",
     mode: item.mode,
     work_kind: workerWorkKind(item, ""),
@@ -6212,7 +6244,7 @@ function workflowRunSummary(run) {
 function isCodexWorkflowFallback(run) {
   const name = `${run?.name || ""} ${run?.display_title || ""}`;
   if (
-    /repair comment router|clawsweeper_comment|@publish:|publish exact review artifact|exact.review publication|reconcile exact.review lease|sync codex review comments/i.test(
+    /repair comment router|clawsweeper_comment|@publish:|publish (?:exact )?review (?:artifacts?|batch)|exact.review publication|reconcile exact.review lease|sync codex review comments/i.test(
       name,
     )
   ) {
@@ -6231,13 +6263,16 @@ function controlPlaneSnapshot(runs) {
   };
   for (const run of runs) {
     const name = `${run?.name || ""} ${run?.display_title || ""}`;
-    const lane = /@publish:|publish exact review artifact|exact.review publication/i.test(name)
-      ? snapshot.publishers
-      : /repair comment router|clawsweeper_comment|sync codex review comments/i.test(name)
-        ? snapshot.comment_routers
-        : /reconcile exact.review lease/i.test(name)
-          ? snapshot.reconcilers
-          : null;
+    const lane =
+      /@publish:|publish (?:exact )?review (?:artifacts?|batch)|exact.review publication/i.test(
+        name,
+      )
+        ? snapshot.publishers
+        : /repair comment router|clawsweeper_comment|sync codex review comments/i.test(name)
+          ? snapshot.comment_routers
+          : /reconcile exact.review lease/i.test(name)
+            ? snapshot.reconcilers
+            : null;
     if (!lane) continue;
     if (run.status === "in_progress") lane.running += 1;
     else lane.waiting += 1;
@@ -8994,6 +9029,7 @@ function laneFlowDetails(laneKey, flow) {
 }
 function renderSystemMap(data) {
   const workers = data.workers || [];
+  const codexWorkers = workers.filter(worker => worker.is_codex_worker !== false);
   const pipeline = data.pipeline || [];
   const fleet = data.fleet || {};
   const workerRunIds = new Set(workers.map(worker => String(worker.run_id)));
@@ -9003,16 +9039,16 @@ function renderSystemMap(data) {
   const nodes = [
     ["01 · Intake", fleet.queued_workflow_runs || 0, "Events and scheduled sweeps waiting to start"],
     ["02 · Plan", planning, "Runs selecting work or expanding a matrix"],
-    ["03 · Workers", workers.length, "Codex jobs reviewing, repairing, or assisting"],
+    ["03 · Workers", codexWorkers.length, "Codex jobs reviewing, repairing, or assisting"],
     ["04 · Apply", applying, "Deterministic comment, close, merge, and publish lanes"],
     ["05 · Results", closed, (data.recent?.closed_stats?.window_hours || 24) + "h ClawSweeper closes"]
   ];
   document.getElementById("flow-map").innerHTML = nodes.map(node =>
     '<div class="flow-node"><span>' + esc(node[0]) + '</span><strong>' + fmt.format(node[1]) + '</strong><p>' + esc(node[2]) + '</p></div>'
   ).join("");
   const budget = Math.max(0, fleet.worker_budget || 0);
-  const running = workers.filter(worker => worker.status === "in_progress").length;
-  const waiting = workers.length - running;
+  const running = codexWorkers.filter(worker => worker.status === "in_progress").length;
+  const waiting = codexWorkers.length - running;
   const free = Math.max(0, budget - running - waiting);
   const overflow = Math.max(0, running + waiting - budget);
   const share = value => budget ? Math.min(100, (value / budget) * 100) : 0;
@@ -9532,7 +9568,7 @@ function renderDashboard(data, note) {
       );
   const severity = serverHealth?.severity ||
     (handoffStatus === "stalled" || operationalStatus === "stalled" ? "red" : needsAttention ? "amber" : "green");
-  const workerCount = (data.workers || []).length;
+  const workerCount = (data.workers || []).filter(worker => worker.is_codex_worker !== false).length;
   const repoCount = (data.source.target_repositories || []).length;
   document.getElementById("hero-dot").className = "hero-dot " + (severity === "green" ? "ok" : severity);
   document.getElementById("hero-headline").textContent =

docs/proof/openclaw-bay/README.md

@@ -10,15 +10,15 @@ The sequence proves:
 
 - visible partial-telemetry diagnostics;
 - the Bay timing badge naming its bounded **review trigger → final review** measurement, completed by the command-status update emitted after the durable review summary;
-- a 390px portrait layout that stacks Arriving through Applying vertically, keeps the terminal pools at the waterline, and has no horizontal page overflow;
+- a 390px portrait layout that stacks Arriving through Publishing and Applying vertically, keeps the terminal pools at the waterline, and has no horizontal page overflow;
 - advancing crustacean-claw and master-sweeper animations;
 - a READY flag followed by a physical forward sweep and landing;
 - a changed run ID using the retrigger tunnel and resurfacing path;
 - GitHub-reference search and focus;
 - repository filtering;
 - the read-only drawer's safe GitHub item, job, and workflow-run links;
 - readable overflow controls that open the known queue sample and explicitly explain when aggregate queue IDs are outside the bounded public projection;
-- compact review-admission and result-publication charts with labelled y-axes, exact point hover labels, and cached 6-hour, 24-hour, and 7-day range controls;
+- compact review-admission, result-publication, and State writer charts with labelled y-axes, exact point hover labels, and cached 6-hour, 24-hour, and 7-day range controls;
 - lightweight hover/focus explanations on the beach lane signs;
 - the local-only tide preview advancing through incoming, crest, backwash, and restored states while preserving terminal keys and count;
 - the short static reduced-motion tide cue preserving the same preview state;