feat: Phase 5 — CLI + Dashboard CAS adaptation, cleanup .data.jsonl

- Align REST API contracts for Dashboard (threads list, detail, SSE) - Add content resolution from CAS in thread show + API responses - Rename dataWatcher → threadsJsonWatcher in SSE routes - Update docs (CLAUDE.md, architecture.md, skill.ts) to reflect CAS storage - Zero .data.jsonl code paths in production code - All 166 tests pass, bun run check clean Refs #155, closes #160 小橘 <xiaoju@shazhou.work>
2026-05-09 08:16:04 +00:00
parent f3aedf8d6c
commit e29d1bf345
8 changed files with 114 additions and 62 deletions
@@ -136,6 +136,7 @@ async function emitRecordsForHead(params: {
    await params.stream.writeSSE({
      event: "record",
      data: JSON.stringify({
+        type: "role",
        role: fr.payload.role,
        contentHash: fr.payload.content,
        content,
@@ -309,7 +310,7 @@ export function createLiveRoutes(storageRoot: string): Hono {
      const controller = new AbortController();
      let completed = false;

-      const dataWatcher = watch(threadsJsonPath, async () => {
+      const threadsJsonWatcher = watch(threadsJsonPath, async () => {
        if (completed) {
          return;
        }
@@ -334,7 +335,7 @@ export function createLiveRoutes(storageRoot: string): Hono {

      stream.onAbort(() => {
        completed = true;
-        dataWatcher.close();
+        threadsJsonWatcher.close();
        infoWatcher?.close();
      });

@@ -347,7 +348,7 @@ export function createLiveRoutes(storageRoot: string): Hono {
        stream.onAbort(() => resolve());
      });

-      dataWatcher.close();
+      threadsJsonWatcher.close();
      infoWatcher?.close();
    });
  });
@@ -1,9 +1,12 @@
-import { createCasStore } from "@uncaged/workflow-cas";
+import { join } from "node:path";
+import { createCasStore, getContentMerklePayload } from "@uncaged/workflow-cas";
 import { FORK_BRANCH_ROLE, walkStateFramesNewestFirst } from "@uncaged/workflow-execute";
 import { END } from "@uncaged/workflow-runtime";
 import { getGlobalCasDir } from "@uncaged/workflow-util";
 import { Hono } from "hono";

+import { pathExists } from "../../fs-utils.js";
+import type { ResolvedThreadRecord } from "../../thread-scan.js";
 import {
  listHistoricalThreads,
  listRunningThreads,
@@ -12,13 +15,76 @@ import {
 import { cmdKill, cmdPause, cmdResume } from "../thread/control.js";
 import { cmdRun } from "../thread/run.js";

+async function buildThreadDetailRecords(
+  storageRoot: string,
+  resolved: ResolvedThreadRecord,
+): Promise<unknown[]> {
+  const cas = createCasStore(getGlobalCasDir(storageRoot));
+  const frames = await walkStateFramesNewestFirst(cas, resolved.head);
+  const chronological = [...frames].reverse();
+
+  const records: unknown[] = [
+    {
+      type: "thread-start",
+      threadId: resolved.threadId,
+      bundleHash: resolved.bundleHash,
+      head: resolved.head,
+      start: resolved.start,
+      source: resolved.source,
+    },
+  ];
+
+  for (const fr of chronological) {
+    if (fr.payload.role === FORK_BRANCH_ROLE) {
+      continue;
+    }
+    if (fr.payload.role === END) {
+      const returnCode = fr.payload.meta.returnCode;
+      const summary = fr.payload.meta.summary;
+      if (typeof returnCode === "number" && typeof summary === "string") {
+        records.push({ type: "workflow-result", returnCode, summary });
+      }
+      continue;
+    }
+    const payloadText = await getContentMerklePayload(cas, fr.payload.content);
+    const content =
+      payloadText !== null
+        ? payloadText
+        : `(content not in CAS; contentHash=${fr.payload.content})`;
+    records.push({
+      type: "role",
+      role: fr.payload.role,
+      contentHash: fr.payload.content,
+      content,
+      meta: fr.payload.meta,
+      timestamp: fr.payload.timestamp,
+    });
+  }
+
+  return records;
+}
+
 export function createThreadRoutes(storageRoot: string): Hono {
  const app = new Hono();

  app.get("/", async (c) => {
    const nameFilter = c.req.query("workflow") ?? null;
    const rows = await listHistoricalThreads(storageRoot, nameFilter);
-    return c.json({ threads: rows });
+    const threads = await Promise.all(
+      rows.map(async (r) => {
+        const runningPath = join(storageRoot, "logs", r.hash, `${r.threadId}.running`);
+        const isRunning = await pathExists(runningPath);
+        const status = r.source === "history" ? "completed" : isRunning ? "running" : "active";
+        return {
+          threadId: r.threadId,
+          workflow: r.workflowName,
+          hash: r.hash,
+          startedAt: new Date(r.activityTs).toISOString(),
+          status,
+        };
+      }),
+    );
+    return c.json({ threads });
  });

  app.get("/running", async (c) => {
@@ -32,42 +98,7 @@ export function createThreadRoutes(storageRoot: string): Hono {
    if (resolved === null) {
      return c.json({ error: `thread not found: ${threadId}` }, 404);
    }
-
-    const cas = createCasStore(getGlobalCasDir(storageRoot));
-    const frames = await walkStateFramesNewestFirst(cas, resolved.head);
-    const chronological = [...frames].reverse();
-
-    const records: unknown[] = [
-      {
-        type: "thread-start",
-        threadId: resolved.threadId,
-        bundleHash: resolved.bundleHash,
-        head: resolved.head,
-        start: resolved.start,
-        source: resolved.source,
-      },
-    ];
-
-    for (const fr of chronological) {
-      if (fr.payload.role === FORK_BRANCH_ROLE) {
-        continue;
-      }
-      if (fr.payload.role === END) {
-        const returnCode = fr.payload.meta.returnCode;
-        const summary = fr.payload.meta.summary;
-        if (typeof returnCode === "number" && typeof summary === "string") {
-          records.push({ type: "workflow-result", returnCode, summary });
-        }
-        continue;
-      }
-      records.push({
-        role: fr.payload.role,
-        contentHash: fr.payload.content,
-        meta: fr.payload.meta,
-        timestamp: fr.payload.timestamp,
-      });
-    }
-
+    const records = await buildThreadDetailRecords(storageRoot, resolved);
    return c.json({ threadId, records });
  });

@@ -1,4 +1,4 @@
-import { createCasStore } from "@uncaged/workflow-cas";
+import { createCasStore, getContentMerklePayload } from "@uncaged/workflow-cas";
 import { FORK_BRANCH_ROLE, walkStateFramesNewestFirst } from "@uncaged/workflow-execute";
 import { err, ok, type Result } from "@uncaged/workflow-protocol";
 import { END } from "@uncaged/workflow-runtime";
@@ -19,15 +19,20 @@ export async function cmdThreadShow(
  const frames = await walkStateFramesNewestFirst(cas, resolved.head);
  const chronological = [...frames].reverse();

-  const steps: Array<{ role: string; hash: string; timestamp: number }> = [];
+  const steps: Array<{ role: string; hash: string; timestamp: number; content: string }> = [];
  for (const fr of chronological) {
    if (fr.payload.role === END || fr.payload.role === FORK_BRANCH_ROLE) {
      continue;
    }
+    const payloadText = await getContentMerklePayload(cas, fr.payload.content);
    steps.push({
      role: fr.payload.role,
      hash: fr.hash,
      timestamp: fr.payload.timestamp,
+      content:
+        payloadText !== null
+          ? payloadText
+          : `(content not in CAS; contentHash=${fr.payload.content})`,
    });
  }

@@ -70,8 +70,8 @@ function formatSkillCli(): string {
 |---------|-------------|
 | **Workflow** | A single-file ESM bundle (\`.esm.js\`) that exports \`run\` and \`descriptor\`. Identified by name and XXH64 hash. |
 | **Bundle** | The physical \`.esm.js\` file stored in the bundles directory. Immutable once written. |
-| **Thread** | A single execution of a workflow, identified by a ULID. Persists state as JSONL files. |
-| **CAS** | Content-Addressable Storage. Per-thread key-value store keyed by content hash. |
+| **Thread** | A single execution of a workflow, identified by a ULID. CAS state chain; \`threads.json\` for active; \`history/*.jsonl\` when done; \`.info.jsonl\` for debug logs. |
+| **CAS** | Global content-addressable blob store (\`cas/\`), keyed by hash. |
 | **Registry** | \`workflow.yaml\` — maps workflow names to their current and historical bundle hashes. |

 ## Commands
@@ -94,6 +94,10 @@ export type HistoricalThreadRow = {
  threadId: string;
  hash: string;
  workflowName: string | null;
+  /** Active entry from `threads.json` vs completed line from `history/*.jsonl`. */
+  source: "active" | "history";
+  /** `updatedAt` for active threads; `completedAt` for history (ms since epoch). */
+  activityTs: number;
 };

 export type ResolvedThreadRecord = {
@@ -243,7 +247,13 @@ export async function listHistoricalThreads(
      if (workflowNameFilter !== null && workflowName !== workflowNameFilter) {
        continue;
      }
-      out.push({ threadId, hash: bundleHash, workflowName });
+      out.push({
+        threadId,
+        hash: bundleHash,
+        workflowName,
+        source: "active",
+        activityTs: entry.updatedAt,
+      });
    }

    const histDir = join(bundleDir, "history");
@@ -271,7 +281,13 @@ export async function listHistoricalThreads(
        if (workflowNameFilter !== null && workflowName !== workflowNameFilter) {
          continue;
        }
-        out.push({ threadId: e.threadId, hash: bundleHash, workflowName });
+        out.push({
+          threadId: e.threadId,
+          hash: bundleHash,
+          workflowName,
+          source: "history",
+          activityTs: e.completedAt,
+        });
      }
    }
  }