🌳 Retain and restore immutable Workspace roots

architecture.md

@@ -206,8 +206,31 @@ Workspace-root tables, current-root state, and a non-null Workspace-root
 association on every journal event. A fresh run starts with one
 content-addressed root manifest describing only the root directory, empty
 retained manifest and blob reference sets, and the corresponding root-only DOFS
-frontier. This storage layer recognizes that canonical empty frontier; it does
-not publish nonempty roots or Workspace mutations.
+frontier.
+
+A Workspace root is a complete, immutable filesystem checkpoint. Its canonical
+format-1 JSON includes `/` and every reachable absolute POSIX path in UTF-8 byte
+order, with kind, mode, observable mtime, file size and DOFS manifest identity,
+symbolic-link target, and deterministic file-hardlink groups. The root ID is
+the lowercase SHA-256 of `xmd-workspace-root\0v1\0` followed by those exact
+canonical bytes. Mutable inode numbers, revisions, tombstones and cache state
+do not participate in the identity.
+
+The root tables are authoritative checkpoints; the DOFS node, dirent and chunk
+tables are the live materialization of the current root. Files remain solely in
+DOFS content-addressed blobs. Normalized root-to-manifest and root-to-blob rows
+equal the root's transitive content exactly and prevent retained content from
+being deleted. Opening a run validates every root and referenced manifest and
+blob, then snapshots the live frontier read-only and requires it to equal
+`current_root`. The schema structure, retained content, live/current comparison
+and run row are read through one explicit SQLite snapshot. This recognition
+transaction is not a caller-owned Workspace transaction and enables no DOFS
+savepoints.
+
+Recognition also requires every file entry in every retained root, including a
+historical root, to declare the size held by its referenced DOFS manifest. A
+root whose XMD identity and reference rows are internally consistent but whose
+file size disagrees with DOFS is corrupt and not restorable.
 
 Which status transitions are legal, and what a caller may do to a run in each
 of them, is lifecycle policy applied above storage.
@@ -244,7 +267,9 @@ A transaction commits only once the work inside it has finished, including work
 that is still unwinding when the body returns. Cleanup belonging to that work
 appends through the same transaction, and a commit that happened first would
 leave those appends to publish themselves outside it. Failure and cancellation
-roll back everything the transaction did.
+roll back everything the transaction did. Adapter-private Workspace work also
+waits for its supplied scope to finish teardown before it performs the final
+live/current validation.
 
 A transaction opened inside another on the same storage is refused rather than
 nested, as is an ordinary operation called from inside a transaction body.
@@ -345,6 +370,11 @@ before their parent's effect begins. Declarative Git operations, including
 staging, switching and committing, operate on the same transactional Workspace
 rather than invoking an untracked native Git side effect.
 
+Successful effect coordination finishes the mutation scope, including child
+cleanup, before capturing the resulting root. The provider-level coordinator
+that performs this ordering and journal publication is not installed at this
+layer.
+
 An external provider cannot join that transaction. Prompt, Git push and pull
 request effects derive a stable identity from the run and expansion, ask the
 provider to perform or reconcile that identity, then append one local result
@@ -393,8 +423,25 @@ second long-lived DOFS connection is not a coherent reader because provider
 caches may retain negative entries across another connection's commit.
 
 Complete schema version 1 retains the canonical empty Workspace root and its
-root-only live frontier. The provider exposes no Workspace mutation operation
-and publishes no nonempty retained root at this layer.
+root-only live frontier initially. It retains arbitrary canonical roots and can
+materialize one privately through the authoritative connection. Capture runs
+inside the caller-owned transaction. Restoration runs in a nested savepoint,
+clears the authoritative resolution and blob caches, and resnapshots to the
+selected identity before release. Private Workspace transaction bodies finish
+their child teardown before final live/current validation; a later effect
+coordinator finishes its mutation scope before it invokes capture.
+
+Every unsuccessful caller-owned transaction attempts SQLite rollback and then
+invalidates both caches on the provider-owned DOFS wrapper while it still holds
+the serialized connection turn. This includes body failure, cancellation,
+teardown or final-validation failure, and commit failure. The surviving wrapper
+therefore cannot answer from uncommitted positive or negative cache entries
+after SQLite has restored the prior frontier.
+
+Retained roots, manifests and blobs remain indefinitely. Cloudflare garbage
+collection is not in the production closure and is never invoked. The provider
+exposes no public Workspace mutation effect, history selection or fork
+operation at this layer.
 
 The initial topology requires neither writable FUSE nor native subprocess
 access and does not bundle `workerd`. A Cloudflare-hosted or workerd-backed
@@ -683,7 +730,7 @@ Status is measured against main.
 | `Expansion` / `getExpansion()` | describes the current logical element expansion | built on main |
 | `useWorkflow()` / `getWorkflowRun()` | associates one document execution with a workflow run | built on main |
 | `Git.revParse()` | verifies and resolves one Git revision expression contextually | built on main |
-| workflow run storage | creates or compatibly finds one run by public run ID, and retains its identity, state, document executions, filtered journal and canonical empty Workspace root through one provider-owned connection entry | built on the #365 stack; Workspace mutation publication is unbuilt |
+| workflow run storage | creates or compatibly finds one run by public run ID, retains its identity, state, document executions and filtered journal, and validates immutable Workspace roots through one provider-owned connection entry | built on the #365 stack; Workspace effect publication is unbuilt |
 | caller-owned storage transaction | publishes several changes, including journal events, in one transaction nothing else enlists in | built on main |
 | `API.Service` / `startService()` | creates an authenticated, supervised loopback service attachment through a provider-neutral operation | built on main |
 | `service=<binding>` | publishes the attachment's endpoint into the live binding overlay for its invocation | built on main |
@@ -694,7 +741,7 @@ Status is measured against main.
 | Repository / Worktree / transactional Git effects | compose named checkouts and publish local mutations with their journal result | defined in `specs/workflow-workspace-spec.md`, unbuilt |
 | workflow inspection and history fork | reads status/history without advancing a run and creates a new run from a checkpoint | defined in `specs/workflow-workspace-spec.md`, unbuilt |
 | read-only workflow Agent / generated XMD | lets an Agent inspect a derived view and propose constrained executable changes | defined in `specs/workflow-workspace-spec.md`, unbuilt |
-| Deno-local DOFS provider | owns one authoritative SQLite/DOFS connection per run path and recognizes the complete-v1 canonical empty frontier | built on the #365 stack; nonempty roots and effect-transaction integration are unbuilt |
+| Deno-local DOFS provider | owns one authoritative SQLite/DOFS connection per run path, captures arbitrary canonical retained roots, and privately restores them with cache-coherent savepoints | built on the #365 stack; public mutation and effect-transaction integration are unbuilt |
 | scoped Worker Shell | executes `just-bash` through the Workspace adapter inside a Deno Worker | containment and effect-transaction POCs complete (#351, #357); production integration unbuilt |
 | `<Retry max timeout>` | retry a region until it completes | defined, unbuilt |
 | suspension effect | suspend durably | defined, unbuilt |

packages/workflow/src/deno/connections.ts

@@ -2,6 +2,8 @@ import { DatabaseSync } from "node:sqlite";
 import { resolve } from "node:path";
 import { Database as CloudflareDatabase } from "../../vendor/cloudflare-computer-dofs/generated/storage.js";
 import { WorkspaceFilesystem } from "../../vendor/cloudflare-computer-dofs/generated/fs/filesystem.js";
+import { clearBlobCache } from "../../vendor/cloudflare-computer-dofs/generated/fs/blobCache.js";
+import { clearResolveCache } from "../../vendor/cloudflare-computer-dofs/generated/fs/resolveCache.js";
 import type {
   DurableObjectStorageLike,
   SQLCursorLike,
@@ -18,6 +20,8 @@ export interface RunConnection {
   readonly lock: ConnectionLock;
   readonly savepoints: SavepointManager;
   transactionOpen: boolean;
+  invalidateDofsCaches(): void;
+  setClock(now: () => number): void;
   close(): void;
 }
 
@@ -80,12 +84,13 @@ function createConnection(path: string): RunConnection {
   const dofs = new CloudflareDatabase(durableStorage);
   const savepoints = createSavepointManager(database, () => connection.transactionOpen);
   connection.savepoints = savepoints;
+  let clock = Date.now;
 
   return {
     path,
     database,
     dofs,
-    filesystem: new WorkspaceFilesystem(dofs),
+    filesystem: new WorkspaceFilesystem(dofs, { now: () => clock() }),
     lock: createConnectionLock(),
     savepoints,
     get transactionOpen() {
@@ -94,6 +99,13 @@ function createConnection(path: string): RunConnection {
     set transactionOpen(value: boolean) {
       connection.transactionOpen = value;
     },
+    invalidateDofsCaches(): void {
+      clearResolveCache(dofs);
+      clearBlobCache(dofs);
+    },
+    setClock(now: () => number): void {
+      clock = now;
+    },
     close() {
       if (open) {
         open = false;

packages/workflow/src/deno/database.ts

@@ -31,7 +31,7 @@
  */
 
 import { randomUUID } from "node:crypto";
-import type { DatabaseSync, StatementSync } from "node:sqlite";
+import type { DatabaseSync } from "node:sqlite";
 import { ensure, Err, Ok, type Operation, resource, type Result, scoped } from "effection";
 import type { DurableEvent, DurableStream, Json } from "@executablemd/durable-streams";
 import type { JournalEntry, WorkflowRunDatabase, WorkflowRunTransaction } from "../storage/api.ts";
@@ -62,6 +62,7 @@ import {
   useTransactionSavepoints,
 } from "./transaction.ts";
 import { readDocumentExecution, readRetrieval, readRunRecord, stopReasonColumns } from "./rows.ts";
+import { reading } from "./reading.ts";
 import { isSqliteForeignKeyConstraint, translateSqliteError } from "./schema.ts";
 
 const SELECT_RUN = "SELECT * FROM workflow_run WHERE id = 1";
@@ -110,6 +111,27 @@ interface Handle {
   close(): void;
 }
 
+const DENO_CONNECTION = Symbol("executablemd.workflow.deno.connection");
+
+interface DenoWorkflowRunDatabase extends WorkflowRunDatabase {
+  readonly [DENO_CONNECTION]: RunConnection;
+}
+
+export function workflowRunConnection(database: WorkflowRunDatabase): RunConnection {
+  if (!isDenoWorkflowRunDatabase(database)) {
+    throw new WorkflowTransactionError(
+      "the WorkflowRun database is not owned by this Deno storage provider.",
+    );
+  }
+  return database[DENO_CONNECTION];
+}
+
+function isDenoWorkflowRunDatabase(
+  database: WorkflowRunDatabase,
+): database is DenoWorkflowRunDatabase {
+  return DENO_CONNECTION in database;
+}
+
 function createHandle(connection: OpenConnection): Handle {
   const { database, path, lock } = connection.connection;
 
@@ -198,6 +220,7 @@ function createHandle(connection: OpenConnection): Handle {
         connection.connection.transactionOpen = false;
         if (!committed) {
           rollback(database);
+          connection.connection.invalidateDofsCaches();
         }
       });
 
@@ -245,7 +268,9 @@ function createHandle(connection: OpenConnection): Handle {
     },
   };
 
-  const handle: WorkflowRunDatabase = {
+  const handle: DenoWorkflowRunDatabase = {
+    [DENO_CONNECTION]: connection.connection,
+
     get record() {
       return record;
     },
@@ -504,20 +529,6 @@ function readRetrievalRow(database: DatabaseSync): DefinitionRetrieval | undefin
   return row === undefined ? undefined : readRetrieval(row);
 }
 
-/**
- * A statement that answers with `bigint` rather than refusing to answer.
- *
- * `node:sqlite` throws a `RangeError` when a column holds a 64-bit value —
- * and quotes the value in the message. Reading integers as `bigint` puts the
- * decision back where every other stored value is decided, in a parser that
- * refuses without repeating what it refused.
- */
-function reading(database: DatabaseSync, sql: string): StatementSync {
-  const statement = database.prepare(sql);
-  statement.setReadBigInts(true);
-  return statement;
-}
-
 function readExecution(database: DatabaseSync, executionId: string): DocumentExecutionRecord {
   const row = reading(database, SELECT_EXECUTION).get(executionId);
   if (row === undefined) {

packages/workflow/src/deno/provider.ts

@@ -26,8 +26,9 @@
  */
 
 import { dirname, isAbsolute } from "node:path";
+import type { DatabaseSync } from "node:sqlite";
 import { ensureDir, exists } from "@effectionx/fs";
-import { ensure, Err, Ok, type Operation, type Result, scoped } from "effection";
+import { createContext, ensure, Err, Ok, type Operation, type Result, scoped } from "effection";
 import {
   type CreateWorkflowRunRequest,
   type WorkflowRunDatabase,
@@ -62,6 +63,7 @@ import {
   type WorkflowRunConnections,
 } from "./connections.ts";
 import { workflowRunPath } from "./path.ts";
+import { readTransaction } from "./reading.ts";
 import { initializeSchema, isUninitialized, translateSqliteError, verifySchema } from "./schema.ts";
 
 const INSERT_RUN = `INSERT INTO workflow_run
@@ -80,6 +82,14 @@ export interface WorkflowRunStorageOptions {
   readonly root: string;
 }
 
+export type WorkflowRunRecognitionProbe = (database: DatabaseSync) => void;
+
+/** Adapter-private observation seam for deterministic snapshot tests. */
+export const WorkflowRunRecognition = createContext<WorkflowRunRecognitionProbe>(
+  "executablemd.workflow.deno.recognition",
+  () => {},
+);
+
 /**
  * Install this host's run storage for the current scope and its descendants.
  *
@@ -156,14 +166,15 @@ function* createWorkflowRun(
   }
 
   try {
+    const inspectRecognition = yield* WorkflowRunRecognition.expect();
     const connection = connections.at(path);
     const { lock } = connection;
     // Held across initialization, so a second caller creating the same run
     // waits here rather than inside a synchronous `BEGIN IMMEDIATE` that
     // would stop the host while the first one is still committing.
     const stored = yield* scoped(function* () {
       yield* lock.hold();
-      return establish(connection, path, wanted);
+      return establish(connection, path, wanted, inspectRecognition);
     });
     if (!stored.ok) {
       return stored;
@@ -205,13 +216,13 @@ function* lookupWorkflowRun(
   }
 
   try {
+    const inspectRecognition = yield* WorkflowRunRecognition.expect();
     const connection = connections.at(path);
     const { database, lock } = connection;
     const record = yield* scoped(function* (): Operation<Result<WorkflowRunRecord>> {
       yield* lock.hold();
       try {
-        verifySchema(database, path);
-        return Ok(readRunRow(database, path));
+        return Ok(recognizeExisting(connection, path, inspectRecognition));
       } catch (error) {
         return refusal(error, path);
       }
@@ -242,12 +253,16 @@ function establish(
   connection: RunConnection,
   path: string,
   request: CheckedRequest,
+  inspectRecognition: WorkflowRunRecognitionProbe,
 ): Result<WorkflowRunRecord> {
   const { database } = connection;
   try {
-    if (!isUninitialized(database, path)) {
-      verifySchema(database, path);
-    }
+    readTransaction(database, () => {
+      if (!isUninitialized(database, path)) {
+        verifySchema(database, path, connection.dofs);
+        inspectRecognition(database);
+      }
+    });
 
     database.exec("BEGIN IMMEDIATE");
     connection.transactionOpen = true;
@@ -267,10 +282,10 @@ function establish(
             );
         });
       } else {
-        verifySchema(database, path);
+        verifySchema(database, path, connection.dofs);
       }
 
-      verifySchema(database, path);
+      verifySchema(database, path, connection.dofs);
       const record = readRunRow(database, path);
       connection.transactionOpen = false;
       database.exec("COMMIT");
@@ -285,6 +300,18 @@ function establish(
   }
 }
 
+function recognizeExisting(
+  connection: RunConnection,
+  path: string,
+  inspectRecognition: WorkflowRunRecognitionProbe,
+): WorkflowRunRecord {
+  return readTransaction(connection.database, () => {
+    verifySchema(connection.database, path, connection.dofs);
+    inspectRecognition(connection.database);
+    return readRunRow(connection.database, path);
+  });
+}
+
 /**
  * Report a storage refusal as itself, and let anything else propagate.
  *

packages/workflow/src/deno/reading.ts

@@ -0,0 +1,32 @@
+import type { DatabaseSync, StatementSync } from "node:sqlite";
+
+/**
+ * A statement that returns SQLite integers as `bigint` rather than throwing.
+ *
+ * A plain node:sqlite read raises a RangeError, including the stored value in
+ * its message, when an INTEGER exceeds JavaScript's safe range. Adapter
+ * parsers need to receive that value so they can refuse it without disclosing
+ * it.
+ */
+export function reading(database: DatabaseSync, sql: string): StatementSync {
+  const statement = database.prepare(sql);
+  statement.setReadBigInts(true);
+  return statement;
+}
+
+/** One consistent SQLite snapshot, without advertising a caller-owned write transaction. */
+export function readTransaction<T>(database: DatabaseSync, body: () => T): T {
+  database.exec("BEGIN");
+  try {
+    const value = body();
+    database.exec("COMMIT");
+    return value;
+  } catch (error) {
+    try {
+      database.exec("ROLLBACK");
+    } catch (rollbackError) {
+      throw rollbackError;
+    }
+    throw error;
+  }
+}