Merge pull request #276 from LeXwDeX/dev

release: current-revision DAG view + node-output file refs

Author: LeXwDeX

packages/core/schema.json

@@ -1,9 +1,9 @@
 {
   "version": "7",
   "dialect": "sqlite",
-  "id": "4142b961-0712-4834-b475-16ea4a74c43c",
+  "id": "874d8e74-d354-4dcb-b98c-c893660c9371",
   "prevIds": [
-    "7e8e00e9-7bbb-443e-996b-f646ec030c2b"
+    "4142b961-0712-4834-b475-16ea4a74c43c"
   ],
   "ddl": [
     {
@@ -682,6 +682,16 @@
       "entityType": "columns",
       "table": "workflow_node"
     },
+    {
+      "type": "integer",
+      "notNull": true,
+      "autoincrement": false,
+      "default": "false",
+      "generated": null,
+      "name": "superseded",
+      "entityType": "columns",
+      "table": "workflow_node"
+    },
     {
       "type": "integer",
       "notNull": true,
@@ -822,6 +832,16 @@
       "entityType": "columns",
       "table": "workflow"
     },
+    {
+      "type": "integer",
+      "notNull": true,
+      "autoincrement": false,
+      "default": "1",
+      "generated": null,
+      "name": "graph_rev",
+      "entityType": "columns",
+      "table": "workflow"
+    },
     {
       "type": "integer",
       "notNull": false,

packages/core/src/dag/projector.ts

@@ -141,33 +141,62 @@ export const layer = Layer.effectDiscard(
 
     yield* events.project(DagEvent.WorkflowReplanned, (event) =>
       Effect.gen(function* () {
-        // Atomic wake can reach the parent only after a leaf checkpoint has
-        // completed the current graph. An additive extend emits this event to
-        // reopen that completed workflow without changing completed nodes.
+        // Rev-view (v1.0.15 Train A): every replan opens a new graph
+        // revision. The two legs run in THIS order so each event bumps
+        // graph_rev exactly once: the seq-bump leg matches the active
+        // statuses first (status unchanged), and the reopen leg then matches
+        // completed rows still untouched by it — reversing the legs would let
+        // the reopened (now running) row match the seq-bump leg too and
+        // double-bump.
         yield* db
           .update(WorkflowTable)
           .set({
-            status: "running",
-            wake_reported: false,
-            completed_at: null,
+            graph_rev: sql`${WorkflowTable.graph_rev} + 1`,
             seq: event.durable!.seq,
             time_updated: toMillis(event.data.timestamp),
           })
           .where(and(
             eq(WorkflowTable.id, event.data.dagID),
-            inArray(WorkflowTable.status, [...WorkflowStatusProjection.replanReopen.from]),
+            inArray(WorkflowTable.status, ["pending", "running", "paused", "stepping"]),
           ))
           .run()
           .pipe(Effect.orDie)
+        // Atomic wake can reach the parent only after a leaf checkpoint has
+        // completed the current graph. An additive extend emits this event to
+        // reopen that completed workflow without changing completed nodes.
         yield* db
           .update(WorkflowTable)
-          .set({ seq: event.durable!.seq, time_updated: toMillis(event.data.timestamp) })
+          .set({
+            status: "running",
+            wake_reported: false,
+            completed_at: null,
+            graph_rev: sql`${WorkflowTable.graph_rev} + 1`,
+            seq: event.durable!.seq,
+            time_updated: toMillis(event.data.timestamp),
+          })
           .where(and(
             eq(WorkflowTable.id, event.data.dagID),
-            inArray(WorkflowTable.status, ["pending", "running", "paused", "stepping"]),
+            inArray(WorkflowTable.status, [...WorkflowStatusProjection.replanReopen.from]),
           ))
           .run()
           .pipe(Effect.orDie)
+        // Rev-view: mark the nodes this replan pushed out of the current
+        // revision — terminal rows the fragment bypassed (a failed node the
+        // new path routes around). plan.cancel rows are marked via the
+        // NodeCancelled projection instead. Idempotent fold: the marker is
+        // monotonic, replaying the event never resurrects a marked row.
+        const superseded = event.data.superseded
+        if (superseded && superseded.length > 0) {
+          yield* db
+            .update(WorkflowNodeTable)
+            .set({ superseded: true, seq: event.durable!.seq, time_updated: toMillis(event.data.timestamp) })
+            .where(and(
+              eq(WorkflowNodeTable.workflow_id, event.data.dagID),
+              inArray(WorkflowNodeTable.id, [...superseded]),
+            ))
+            .run()
+            .pipe(Effect.orDie)
+        }
       }),
     )
 
@@ -361,10 +390,16 @@ export const layer = Layer.effectDiscard(
     // therefore never hold status="cancelled"; see NodeStatusProjection.cancelled
     // above and the canonical proof in
     // packages/opencode/test/dag/dag-escalation-clear-flag.test.ts:130-148.
+    //
+    // Rev-view (v1.0.15 Train A): a cancelled node leaves the current graph
+    // revision — the projection also sets the superseded marker so view and
+    // aggregation reads (summaries, status, node lists, rebuild input, wake
+    // attribution) show only the current rev. This covers both plan.cancel
+    // rows and explicit dag.nodeCancelled publishes (U1: shared semantics).
     yield* events.project(DagEvent.NodeCancelled, (event) =>
       db
         .update(WorkflowNodeTable)
-        .set({ status: "failed", error_reason: "cancelled via replan", escalation_pending: false, seq: event.durable!.seq, time_updated: toMillis(event.data.timestamp) })
+        .set({ status: "failed", superseded: true, error_reason: "cancelled via replan", escalation_pending: false, seq: event.durable!.seq, time_updated: toMillis(event.data.timestamp) })
         .where(and(
           eq(WorkflowNodeTable.workflow_id, event.data.dagID),
           eq(WorkflowNodeTable.id, event.data.nodeID),

packages/core/src/dag/sql.ts

@@ -47,6 +47,11 @@ export const WorkflowTable = sqliteTable(
     config: text().notNull(), // YAML string
     seq: integer().notNull(), // latest durable event seq
     wake_reported: integer({ mode: "boolean" }).notNull().default(false), // D3: has workflow terminal been reported to parent?
+    // Rev-view (v1.0.15 Train A): the current graph-revision counter. Bumped
+    // by the WorkflowReplanned projection; audit/telemetry only — the view
+    // predicate is the per-node `superseded` marker below. Default 1: legacy
+    // rows predate the concept and render exactly as before.
+    graph_rev: integer().notNull().default(1),
     started_at: integer(),
     completed_at: integer(),
     ...Timestamps,
@@ -77,13 +82,20 @@ export const WorkflowNodeTable = sqliteTable(
     output: text({ mode: "json" }).$type<unknown>(),
     error_reason: text(),
     error_class: text(), // dag.node.failed trigger (timeout/exec_failed/verdict_fail/push_exhausted) for failure triage
-    captured_output: text({ mode: "json" }).$type<unknown>(), // durable payload from submit_result; survives a process crash, reset to null on a replan-restart via NodeStarted
+    captured_output: text({ mode: "json" }).$type<unknown>(), // durable payload from submit_result, or a Train B file-ref record ({content_ref, size, sha256, summary}); survives a process crash, reset to null on a replan-restart via NodeStarted
     deadline_ms: integer(), // absolute deadline (spawnedAt + timeout_ms) for D0 termination boundary
     wake_eligible: integer({ mode: "boolean" }).notNull().default(false), // D6: node has report_to_parent=true
     wake_reported: integer({ mode: "boolean" }).notNull().default(false), // D3: has this node's terminal event been injected into the parent session?
     replan_attempts: integer().notNull().default(0), // D4: per-node replan counter for circuit breaker
     timeout_extensions: integer().notNull().default(0), // timeout escalation count (node stays running; main agent adjudicates)
     escalation_pending: integer({ mode: "boolean" }).notNull().default(false), // set on escalate, cleared on adjudication (extend) or new attempt — "awaiting main-agent adjudication"
+    // Rev-view (v1.0.15 Train A): this node was pushed OUT of the current
+    // graph revision by a replan (cancelled via replan, or a terminal row the
+    // fragment bypassed). Durable data is untouched — the marker only filters
+    // VIEW/aggregation reads (summaries, status, node lists, rebuild input,
+    // wake attribution) to the current revision. Monotonic: once true, stays
+    // true. Default false: legacy rows render exactly as before.
+    superseded: integer({ mode: "boolean" }).notNull().default(false),
     seq: integer().notNull(), // latest durable event seq for this node
     started_at: integer(),
     completed_at: integer(),

packages/core/src/dag/store.ts

@@ -24,6 +24,8 @@ export interface WorkflowRow {
   config: string
   seq: number
   wakeReported: boolean
+  /** Rev-view (v1.0.15 Train A): current graph-revision counter (audit/telemetry). */
+  graphRev: number
   startedAt: number | null
   completedAt: number | null
   timeCreated: number
@@ -51,6 +53,8 @@ export interface NodeRow {
   replanAttempts: number
   timeoutExtensions: number
   escalationPending: boolean
+  /** Rev-view (v1.0.15 Train A): pushed out of the current graph revision by a replan. */
+  superseded: boolean
   seq: number
   startedAt: number | null
   completedAt: number | null
@@ -91,6 +95,7 @@ const mapWorkflow = (r: typeof WorkflowTable.$inferSelect): WorkflowRow => ({
   config: r.config,
   seq: r.seq,
   wakeReported: r.wake_reported,
+  graphRev: r.graph_rev,
   startedAt: r.started_at,
   completedAt: r.completed_at,
   timeCreated: r.time_created,
@@ -118,6 +123,7 @@ const mapNode = (r: typeof WorkflowNodeTable.$inferSelect): NodeRow => ({
   replanAttempts: r.replan_attempts,
   timeoutExtensions: r.timeout_extensions,
   escalationPending: r.escalation_pending,
+  superseded: r.superseded,
   seq: r.seq,
   startedAt: r.started_at,
   completedAt: r.completed_at,
@@ -156,6 +162,15 @@ export interface Interface {
   readonly getWorkflowSummaries: (sessionId: string) => Effect.Effect<WorkflowSummary[]>
 
   readonly getNodes: (workflowId: string) => Effect.Effect<NodeRow[]>
+  /**
+   * Rev-view (v1.0.15 Train A): the CURRENT graph revision only — rows the
+   * replan pushed out of the graph (superseded) are filtered out. This is the
+   * read for VIEW and terminal-aggregation consumers: summaries, status/node
+   * listings, the loop's rebuild/recovery/completion input, and wake failure
+   * attribution. Durable truth is untouched — getNodes still returns every
+   * row, and completed old-rev outputs stay resolvable for input mapping.
+   */
+  readonly getCurrentNodes: (workflowId: string) => Effect.Effect<NodeRow[]>
   readonly getNode: (workflowId: string, nodeId: string) => Effect.Effect<NodeRow | undefined>
   readonly getRunningNodes: (workflowId: string) => Effect.Effect<NodeRow[]>
   readonly setCapturedOutput: (childSessionID: string, payload: unknown) => Effect.Effect<void>
@@ -257,6 +272,9 @@ export const layer = Layer.effect(
         if (wfRows.length === 0) return []
         // P1-4: aggregate in SQL — pulling every node row into JS made each
         // dag.* event burst scale with total node count across the session.
+        // Rev-view (v1.0.15 Train A): superseded rows are filtered out so the
+        // counts reflect ONLY the current graph revision — a replaced segment
+        // neither counts toward nodeCount nor inflates failedNodes.
         const countRows = yield* db
           .select({
             workflowId: WorkflowNodeTable.workflow_id,
@@ -265,7 +283,7 @@ export const layer = Layer.effect(
           })
           .from(WorkflowNodeTable)
           .innerJoin(WorkflowTable, eq(WorkflowNodeTable.workflow_id, WorkflowTable.id))
-          .where(eq(WorkflowTable.session_id, sessionId))
+          .where(and(eq(WorkflowTable.session_id, sessionId), eq(WorkflowNodeTable.superseded, false)))
           .groupBy(WorkflowNodeTable.workflow_id, WorkflowNodeTable.status)
           .all()
           .pipe(Effect.orDie)
@@ -284,6 +302,7 @@ export const layer = Layer.effect(
             eq(WorkflowTable.session_id, sessionId),
             eq(WorkflowNodeTable.status, "running"),
             eq(WorkflowNodeTable.escalation_pending, true),
+            eq(WorkflowNodeTable.superseded, false),
           ))
           .groupBy(WorkflowNodeTable.workflow_id)
           .all()
@@ -320,6 +339,17 @@ export const layer = Layer.effect(
         return rows.map(mapNode)
       }),
 
+      getCurrentNodes: Effect.fn("DagStore.getCurrentNodes")(function* (workflowId) {
+        const rows = yield* db
+          .select()
+          .from(WorkflowNodeTable)
+          .where(and(eq(WorkflowNodeTable.workflow_id, workflowId), eq(WorkflowNodeTable.superseded, false)))
+          .orderBy(desc(WorkflowNodeTable.seq))
+          .all()
+          .pipe(Effect.orDie)
+        return rows.map(mapNode)
+      }),
+
       getNode: Effect.fn("DagStore.getNode")(function* (workflowId, nodeId) {
         const row = yield* db
           .select()

packages/core/src/database/migration.gen.ts

@@ -0,0 +1,23 @@
+// SPDX-FileCopyrightText: 2026 LeXwDeX
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import { Effect } from "effect"
+import type { DatabaseMigration } from "../migration"
+
+// Rev-view (v1.0.15 Train A, workflows/dag-engine-optimization.md). Legacy
+// policy: existing rows migrate in place with superseded=false / graph_rev=1,
+// so every pre-feature workflow renders EXACTLY as before — including its
+// cancelled-via-replan rows, which stay visible and counted (config
+// membership cannot be the current-rev predicate: <=v1.0.14 merged configs
+// already drop cancelled nodes, so it would hide rows that render today).
+// Marking only ever happens via the WorkflowReplanned and NodeCancelled
+// projections after this migration runs.
+export default {
+  id: "20260815044858_dag_graph_rev_view",
+  up(tx) {
+    return Effect.gen(function* () {
+      yield* tx.run(`ALTER TABLE \`workflow_node\` ADD \`superseded\` integer DEFAULT false NOT NULL;`)
+      yield* tx.run(`ALTER TABLE \`workflow\` ADD \`graph_rev\` integer DEFAULT 1 NOT NULL;`)
+    })
+  },
+} satisfies DatabaseMigration.Migration

packages/core/src/database/migration/20260815044858_dag_graph_rev_view.ts

@@ -91,6 +91,7 @@ export default {
           \`replan_attempts\` integer DEFAULT 0 NOT NULL,
           \`timeout_extensions\` integer DEFAULT 0 NOT NULL,
           \`escalation_pending\` integer DEFAULT false NOT NULL,
+          \`superseded\` integer DEFAULT false NOT NULL,
           \`seq\` integer NOT NULL,
           \`started_at\` integer,
           \`completed_at\` integer,
@@ -111,6 +112,7 @@ export default {
           \`config\` text NOT NULL,
           \`seq\` integer NOT NULL,
           \`wake_reported\` integer DEFAULT false NOT NULL,
+          \`graph_rev\` integer DEFAULT 1 NOT NULL,
           \`started_at\` integer,
           \`completed_at\` integer,
           \`time_created\` integer NOT NULL,