test(history): bound batch drains by progress and total time

TESTING.md

@@ -200,8 +200,8 @@ loading a workflow or queueing generation.
 
 ## Durable execution ledger
 
-`npm run --prefix tools/vendor test:executions` runs 26 real execution cases
-(13 each in Classic and Nodes 2.0, retaining the original ten ledger cases),
+`npm run --prefix tools/vendor test:executions` runs 42 real execution cases
+(21 each in Classic and Nodes 2.0, retaining all 26 prior execution cases),
 alongside the existing 78 stock host cases and six pack compatibility cases. Run
 these suites serially because they share report cleanup and owned host resources. Use the same pinned disposable `PM_COMFY_ROOT`,
 `PM_TEST_PYTHON` and Playwright browser installation as `test:comfyui`:
@@ -217,8 +217,9 @@ This suite reserves `http://127.0.0.1:8196/` and never attaches to an existing
 server. Each renderer worker launches a CPU/offline host with independent
 configuration, database, user and output directories. The shared launcher installs
 tracked snapshots of PromptManager and a tiny temporary Git-tracked fixture pack.
-Only that owned test pack provides cached output, deliberate failure, an
-interruptible queue gate, deterministic CLIP/model/latent inputs and control routes.
+Only that owned test pack provides cached text, deliberate failure, an
+interruptible queue gate, deterministic CLIP/model/latent and CPU tensor inputs,
+and control routes.
 It replaces external sampler inference and trigger injection only inside the
 disposable host; real registered PromptManager classes, queue, executor and cache
 remain active. Its cache control resets the real executor on its worker to exercise
@@ -236,8 +237,24 @@ mode, exact UUID job/session filters and a 256-row limit; absent ledger schema i
 explicit, while missing/corrupt databases and other read errors fail the suite.
 Sensitive diagnostic and extra-data sentinels must never appear in lifecycle
 rows. Exact prompt fields are deliberately persisted only in context snapshots.
-The same read-only helper calls the internal `get_execution_contexts` reader for
-requested run IDs and inspects their snapshot identities and current usage counts.
+The same read-only helper calls the internal `get_execution_contexts` and
+`get_execution_outputs` readers for requested run IDs and inspects their snapshot
+identities and current usage counts. Its `outputs` map is keyed by immutable run
+ID, alongside the existing `rows`, `captures`, `usage` and `snapshotIds` results.
+
+`execution-outputs.spec.mjs` runs actual stock SaveImage and PreviewImage nodes
+from tiny CPU tensors. It compares each attempt's persisted descriptors with that
+attempt's final host history, including A,A,B batches, multiple output nodes,
+cached locators, reused public UUIDs, PNG metadata disabled, graphs without PM
+nodes and text-only outputs. The pinned SaveImage returns an IMAGE socket; linking
+it to the owned failure/gate node orders the partial-run tests explicitly. A held
+downstream gate proves completed image work does not finalize the run early.
+Other cases verify real output capture across startup in each renderer, switching
+both ways and workflow reload. PM usage counts retain their existing per-job
+behavior. References identify reported locators only: file identity is unverified,
+new generation and context membership are unknown, and temp references are
+transient. The stock suite excludes both execution files; discovery retains its
+78 cases and the six pack cases unchanged.
 
 For a meaningful historical failure, create a separate disposable detached
 checkout at `fdba6fd0126a8a40cbd6f3e50ed4c5de8991efd4` and pass its path to
@@ -285,8 +302,21 @@ Fault injection for restore fencing, inherited async capability revocation, queu
 finalization, epoch and generation revocation, provider-local cache entries, eviction/reset, malformed or
 oversized payloads and missing host interfaces lives in the focused Python
 `test_execution*.py` suites. These use disposable storage and mocked host interfaces;
-they are distinct from the pinned real executor/cache checks above. No output-file
-associations or chronological timeline are introduced. Deploying the Python changes
+they are distinct from the pinned real executor/cache checks above. Run-owned
+output descriptors do not create gallery image associations or a chronological
+timeline. Deploying the Python changes
 requires a ComfyUI restart. After restart, queue three new jobs A,A,B with identical
 positive fields and two negative recipes; existing counters should advance as usual,
 and internal context inspection should find three attempts and two combinations.
+For a live output check after that manual restart:
+
+- Queue SaveImage and PreviewImage with a batch of two, then submit an unchanged
+  cached repeat and a changed tensor/input under new jobs.
+- Inspect each run's internal output capture against its final ComfyUI history;
+  cached references may repeat, while the attempts remain distinct.
+- Switch Classic to Nodes 2.0 and back, reload the workflow and queue again.
+- Verify prompt/pair counters and gallery links retain their existing behavior;
+  output references alone do not prove file bytes or prompt membership.
+
+Automated checks use only managed disposable hosts. They never restart or upgrade
+the personal ComfyUI server.

database/execution_history_schema.py

@@ -4,7 +4,7 @@ import re
 import sqlite3
 from uuid import uuid4
 
-SCHEMA_VERSION = 2
+SCHEMA_VERSION = 3
 
 
 def _uuid_check(column: str) -> str:
@@ -24,7 +24,7 @@ def _integer_check(column: str, *, nullable: bool = False) -> str:
 
 _METADATA_CHECKS = (
     "singleton_id = 1",
-    "schema_version = 2",
+    "schema_version = 3",
     _uuid_check("storage_epoch"),
 )
 _RUN_CHECKS = (
@@ -140,6 +140,17 @@ def history_schema_error(
     }
     from .execution_context_schema import TABLE_COLUMNS, context_schema_error
 
+    from .execution_output_schema import (
+        TABLE_COLUMNS as OUTPUT_COLUMNS,
+        output_schema_error,
+    )
+
+    output_present = any(
+        row[0] in OUTPUT_COLUMNS
+        for row in conn.execute("SELECT name FROM sqlite_master")
+    )
+    if not tables and output_present:
+        return "Execution outputs require execution history"
     context_present = any(
         row[0] in TABLE_COLUMNS
         for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
@@ -154,9 +165,11 @@ def history_schema_error(
         "SELECT CASE WHEN typeof(schema_version) = 'integer' "
         "THEN schema_version ELSE NULL END FROM execution_history_metadata LIMIT 2"
     ).fetchall()
-    if len(metadata) != 1 or len(metadata[0]) != 1 or metadata[0][0] not in (1, 2):
+    if len(metadata) != 1 or len(metadata[0]) != 1 or metadata[0][0] not in (1, 2, 3):
         return "Unsupported execution history metadata/version"
     version = metadata[0][0]
+    if version < 3 and output_present:
+        return "Unexpected execution outputs for history v1/v2"
     if version == 1 and context_present:
         return "Unexpected execution contexts for history v1"
     if not allow_missing and version != SCHEMA_VERSION:
@@ -170,9 +183,9 @@ def history_schema_error(
             return f"{table} has invalid execution history columns"
         sql = _compact(tables[table])
         checks = _TABLE_CHECKS[table]
-        if table == "execution_history_metadata" and version == 1:
+        if table == "execution_history_metadata":
             checks = tuple(
-                check.replace("schema_version = 2", "schema_version = 1")
+                check.replace("schema_version = 3", f"schema_version = {version}")
                 for check in checks
             )
         if sql.count("check(") != len(checks) or any(
@@ -222,9 +235,11 @@ def history_schema_error(
             item[3] or item[4] != "BINARY" for item in keys
         ):
             return f"Invalid execution history index {name}"
-    if version == 2:
-        return context_schema_error(conn)
-    return None
+    if version >= 2:
+        error = context_schema_error(conn)
+        if error:
+            return error
+    return output_schema_error(conn) if version == 3 else None
 
 
 def install_context_schema(conn: sqlite3.Connection) -> None:
@@ -233,8 +248,14 @@ def install_context_schema(conn: sqlite3.Connection) -> None:
     install(conn)
 
 
+def install_output_schema(conn: sqlite3.Connection) -> None:
+    from .execution_output_schema import install_output_schema as install
+
+    install(conn)
+
+
 def migrate_execution_history(conn: sqlite3.Connection) -> None:
-    """Install versioned history and context tables atomically, without backfill."""
+    """Install history, contexts and outputs atomically, without backfill."""
     conn.execute("SAVEPOINT execution_history_schema")
     try:
         error = history_schema_error(conn)
@@ -263,14 +284,15 @@ def migrate_execution_history(conn: sqlite3.Connection) -> None:
                 (SCHEMA_VERSION, str(uuid4())),
             )
             install_context_schema(conn)
+            install_output_schema(conn)
         else:
             version, epoch = conn.execute(
                 "SELECT schema_version,storage_epoch FROM execution_history_metadata"
             ).fetchone()
-            if version == 1:
+            if version < SCHEMA_VERSION:
                 conn.execute(
                     "ALTER TABLE execution_history_metadata RENAME TO "
-                    "execution_history_metadata_v1"
+                    "execution_history_metadata_old"
                 )
                 definitions = [
                     f"{name} {kind}{' NOT NULL' if required else ''}"
@@ -285,10 +307,13 @@ def migrate_execution_history(conn: sqlite3.Connection) -> None:
                     f"({', '.join(definitions)})"
                 )
                 conn.execute(
-                    "INSERT INTO execution_history_metadata VALUES(1,2,?)", (epoch,)
+                    "INSERT INTO execution_history_metadata VALUES(1,?,?)",
+                    (SCHEMA_VERSION, epoch),
                 )
-                conn.execute("DROP TABLE execution_history_metadata_v1")
-                install_context_schema(conn)
+                conn.execute("DROP TABLE execution_history_metadata_old")
+                if version == 1:
+                    install_context_schema(conn)
+                install_output_schema(conn)
         error = history_schema_error(conn, allow_missing=False)
         if error:
             raise sqlite3.DatabaseError(error)

database/execution_output_schema.py

@@ -0,0 +1,218 @@
+"""Owned schema and bounded restore validation for frozen output locators."""
+
+import sqlite3
+
+from .execution_history_schema import _compact, _uuid_check
+
+try:
+    from ..utils.execution_outputs import (
+        BUNDLE_OVERHEAD,
+        MAX_BUNDLE_BYTES,
+        MAX_IMAGE_POSITIONS,
+        MAX_REFERENCE_BYTES,
+        REFERENCE_OVERHEAD,
+        STATE_REASONS,
+        OutputBundle,
+        OutputReference,
+        validate_output_bundle,
+    )
+except ImportError:
+    from utils.execution_outputs import (
+        BUNDLE_OVERHEAD,
+        MAX_BUNDLE_BYTES,
+        MAX_IMAGE_POSITIONS,
+        MAX_REFERENCE_BYTES,
+        REFERENCE_OVERHEAD,
+        STATE_REASONS,
+        OutputBundle,
+        OutputReference,
+        validate_output_bundle,
+    )
+
+
+TABLE_COLUMNS = {
+    "execution_output_capture": (
+        ("run_id", "TEXT", 1, 1),
+        ("state", "TEXT", 1, 0),
+        ("reason", "TEXT", 1, 0),
+        ("format_version", "INTEGER", 1, 0),
+        ("source", "TEXT", 1, 0),
+    ),
+    "execution_run_outputs": (
+        ("run_id", "TEXT", 1, 1),
+        ("output_node_id", "TEXT", 1, 2),
+        ("image_ordinal", "INTEGER", 1, 3),
+        ("folder_type", "TEXT", 1, 0),
+        ("subfolder", "TEXT", 1, 0),
+        ("filename", "TEXT", 1, 0),
+        ("path_style", "TEXT", 1, 0),
+        ("root_key", "TEXT", 0, 0),
+    ),
+}
+TABLE_FOREIGN_KEYS = {
+    "execution_output_capture": (("run_id", "execution_runs", "run_id"),),
+    "execution_run_outputs": (("run_id", "execution_output_capture", "run_id"),),
+}
+
+
+def _text_check(column: str, maximum: int, *, characters: int | None = None) -> str:
+    return (
+        f"typeof({column}) = 'text' AND length({column}) <= "
+        f"{maximum if characters is None else characters} "
+        f"AND length(CAST({column} AS BLOB)) <= {maximum} "
+        f"AND instr(CAST({column} AS BLOB), x'00') = 0"
+    )
+
+
+def _enum_check(column: str, values: tuple | frozenset) -> str:
+    return (
+        f"{_text_check(column, 32)} AND {column} IN ("
+        + ",".join("'" + value + "'" for value in sorted(values))
+        + ")"
+    )
+
+
+_RUN_ID_CHECK = f"{_uuid_check('run_id')} AND {_text_check('run_id', 36)}"
+TABLE_CHECKS = {
+    "execution_output_capture": (
+        _RUN_ID_CHECK,
+        _enum_check("state", tuple(STATE_REASONS)),
+        _enum_check("reason", frozenset.union(*STATE_REASONS.values())),
+        "typeof(format_version) = 'integer' AND format_version = 1",
+        f"{_text_check('source', 18)} AND source = 'host_final_history'",
+    ),
+    "execution_run_outputs": (
+        _RUN_ID_CHECK,
+        f"{_text_check('output_node_id', 512, characters=128)} "
+        "AND length(output_node_id) >= 1",
+        "typeof(image_ordinal) = 'integer' AND image_ordinal BETWEEN 0 AND 255",
+        _enum_check("folder_type", ("output", "temp")),
+        _text_check("subfolder", 2048),
+        f"{_text_check('filename', 1024)} AND length(filename) >= 1",
+        _enum_check("path_style", ("posix", "windows")),
+        f"root_key IS NULL OR ({_text_check('root_key', 64)} "
+        "AND length(root_key) = 64 AND root_key NOT GLOB '*[^0-9a-f]*')",
+    ),
+}
+
+
+def _table_sql(table: str) -> str:
+    columns = TABLE_COLUMNS[table]
+    definitions = [
+        f"{name} {kind}{' NOT NULL' if required else ''}"
+        for name, kind, required, _ in columns
+    ]
+    primary = [name for name, _, _, order in columns if order]
+    definitions.append(f"PRIMARY KEY ({', '.join(primary)})")
+    definitions.extend(f"CHECK({check})" for check in TABLE_CHECKS[table])
+    definitions.extend(
+        f"FOREIGN KEY ({column}) REFERENCES {target}({key}) "
+        "ON UPDATE NO ACTION ON DELETE NO ACTION"
+        for column, target, key in TABLE_FOREIGN_KEYS[table]
+    )
+    return f"CREATE TABLE {table} ({', '.join(definitions)})"
+
+
+def install_output_schema(conn: sqlite3.Connection) -> None:
+    """Install the two owned tables within the caller's migration transaction."""
+    for table in TABLE_COLUMNS:
+        conn.execute(_table_sql(table))
+
+
+def output_schema_error(conn: sqlite3.Connection) -> str | None:
+    """Reject altered schemas and invalid rows before returning bounded DTOs."""
+    if conn.execute(
+        "SELECT 1 FROM sqlite_master WHERE type IN ('trigger','view') LIMIT 1"
+    ).fetchone():
+        return "Unsupported execution output trigger/view"
+    tables = {
+        row[0]: row[1]
+        for row in conn.execute("SELECT name,sql FROM sqlite_master WHERE type='table'")
+        if row[0] in TABLE_COLUMNS
+    }
+    if tables.keys() != TABLE_COLUMNS.keys():
+        return "Incomplete execution output schema"
+    for table, expected in TABLE_COLUMNS.items():
+        columns = tuple(
+            (row[1], row[2].upper(), row[3], row[5])
+            for row in conn.execute(f"PRAGMA table_xinfo({table})")
+        )
+        if columns != expected or _compact(tables[table]) != _compact(
+            _table_sql(table)
+        ):
+            return f"{table} has invalid execution output schema"
+        foreign = tuple(
+            sorted(
+                (row[3], row[2], row[4], row[5], row[6], row[7])
+                for row in conn.execute(f"PRAGMA foreign_key_list({table})")
+            )
+        )
+        if foreign != tuple(
+            (*key, "NO ACTION", "NO ACTION", "NONE")
+            for key in TABLE_FOREIGN_KEYS[table]
+        ):
+            return f"{table} has invalid execution output foreign keys"
+        for index in conn.execute(f"PRAGMA index_list({table})"):
+            name = index[1].replace('"', '""')
+            keys = tuple(
+                row for row in conn.execute(f'PRAGMA index_xinfo("{name}")') if row[5]
+            )
+            if (
+                not index[2]
+                or index[3] != "pk"
+                or index[4]
+                or tuple(key[2] for key in keys)
+                != tuple(column[0] for column in expected if column[3])
+                or any(key[3] or key[4] != "BINARY" for key in keys)
+            ):
+                return f"{table} has unexpected execution output index/uniqueness"
+        # Run every scalar guard in SQLite before any evidence text is fetched.
+        predicate = " AND ".join(f"({check})" for check in TABLE_CHECKS[table])
+        if conn.execute(
+            f"SELECT 1 FROM {table} WHERE NOT ({predicate}) LIMIT 1"
+        ).fetchone():
+            return f"{table} contains invalid execution output records"
+        if conn.execute(f"PRAGMA foreign_key_check({table})").fetchone():
+            return f"{table} contains orphan execution output records"
+    reference_bytes = f"{REFERENCE_OVERHEAD} + " + " + ".join(
+        f"coalesce(length(CAST({column} AS BLOB)), 0)"
+        for column in (
+            "output_node_id",
+            "folder_type",
+            "subfolder",
+            "filename",
+            "path_style",
+            "root_key",
+        )
+    )
+    if conn.execute(
+        "SELECT 1 FROM execution_run_outputs "
+        f"WHERE ({reference_bytes}) > {MAX_REFERENCE_BYTES} LIMIT 1"
+    ).fetchone():
+        return "Oversized execution output reference"
+    if conn.execute(
+        "SELECT 1 FROM execution_run_outputs GROUP BY run_id "
+        f"HAVING COUNT(*) > {MAX_IMAGE_POSITIONS} OR "
+        f"{BUNDLE_OVERHEAD} + SUM({reference_bytes}) > {MAX_BUNDLE_BYTES} LIMIT 1"
+    ).fetchone():
+        return "Oversized execution output bundle"
+    try:
+        for run_id, state, reason, version, source in conn.execute(
+            "SELECT run_id,state,reason,format_version,source "
+            "FROM execution_output_capture"
+        ):
+            references = tuple(
+                OutputReference(*row)
+                for row in conn.execute(
+                    "SELECT output_node_id,image_ordinal,folder_type,subfolder,"
+                    "filename,path_style,root_key FROM execution_run_outputs "
+                    "WHERE run_id=? ORDER BY output_node_id,image_ordinal",
+                    (run_id,),
+                )
+            )
+            validate_output_bundle(
+                OutputBundle(references, state, reason, version, source)
+            )
+    except (ValueError, TypeError, OverflowError, sqlite3.Error) as exc:
+        return f"Invalid execution output content: {type(exc).__name__}"
+    return None

database/execution_outputs.py

@@ -0,0 +1,153 @@
+"""Immutable run-owned output references, independent of mutable galleries."""
+
+import sqlite3
+from contextlib import ExitStack
+from typing import Callable, ContextManager
+
+try:
+    from ..utils.execution_history_events import RunFinish, encode_command
+    from ..utils.execution_outputs import OutputBundle, validate_output_bundle
+except ImportError:
+    from utils.execution_history_events import RunFinish, encode_command
+    from utils.execution_outputs import OutputBundle, validate_output_bundle
+
+
+_REFERENCE_FIELDS = (
+    "output_node_id",
+    "image_ordinal",
+    "folder_type",
+    "subfolder",
+    "filename",
+    "path_style",
+    "root_key",
+)
+_REFERENCE_COLUMNS = ",".join(_REFERENCE_FIELDS)
+
+
+def _capture(conn: sqlite3.Connection, run_id: str) -> sqlite3.Row | None:
+    capture = conn.execute(
+        "SELECT state,reason,format_version,source FROM execution_output_capture "
+        "WHERE run_id=?",
+        (run_id,),
+    ).fetchone()
+    # Legacy callers may disable FK enforcement; never adopt orphan evidence.
+    if capture is None:
+        if conn.execute(
+            "SELECT 1 FROM execution_run_outputs WHERE run_id=? LIMIT 1", (run_id,)
+        ).fetchone():
+            raise sqlite3.DatabaseError("Output references have no capture header")
+    elif (
+        conn.execute(
+            "SELECT 1 FROM execution_runs WHERE run_id=?", (run_id,)
+        ).fetchone()
+        is None
+    ):
+        raise sqlite3.DatabaseError("Output capture has no execution parent")
+    return capture
+
+
+def _references(conn: sqlite3.Connection, run_id: str) -> list[tuple]:
+    return [
+        tuple(row)
+        for row in conn.execute(
+            f"SELECT {_REFERENCE_COLUMNS} FROM execution_run_outputs "
+            "WHERE run_id=? ORDER BY output_node_id,image_ordinal",
+            (run_id,),
+        )
+    ]
+
+
+class ExecutionOutputOperations:
+    """Compose with ExecutionHistoryOperations on PromptDatabase."""
+
+    def write_execution_outputs_for_finish(
+        self,
+        command: RunFinish,
+        bundle: OutputBundle,
+        *,
+        generation_guard: Callable[[], ContextManager[bool]] | None = None,
+    ) -> bool:
+        encode_command(command)
+        if type(command) is not RunFinish:
+            raise ValueError("Expected execution finish command")
+        validate_output_bundle(bundle)
+        header = (bundle.state, bundle.reason, bundle.format_version, bundle.source)
+        prepared = [
+            tuple(getattr(reference, field) for field in _REFERENCE_FIELDS)
+            for reference in bundle.references
+        ]
+        # Enter after BEGIN and retain the guard until COMMIT/ROLLBACK finishes.
+        with ExitStack() as guards:
+            with self._execution_transaction(command.storage_epoch) as conn:
+                if generation_guard is not None and not guards.enter_context(
+                    generation_guard()
+                ):
+                    return False
+                parent = conn.execute(
+                    "SELECT run_id FROM execution_runs WHERE session_id=? "
+                    "AND queue_item_id=? AND storage_epoch=?",
+                    (command.session_id, command.queue_item_id, command.storage_epoch),
+                ).fetchone()
+                if parent is None:
+                    return False
+                run_id = parent[0]
+                capture = _capture(conn, run_id)
+                if capture is not None:
+                    if capture[0] != "conflicting" and (
+                        tuple(capture) != header
+                        or _references(conn, run_id) != prepared
+                    ):
+                        conn.execute(
+                            "UPDATE execution_output_capture SET "
+                            "state='conflicting',reason='conflicting' WHERE run_id=?",
+                            (run_id,),
+                        )
+                    return True
+                conn.execute(
+                    "INSERT INTO execution_output_capture "
+                    "(run_id,state,reason,format_version,source) VALUES(?,?,?,?,?)",
+                    (run_id, *header),
+                )
+                conn.executemany(
+                    f"INSERT INTO execution_run_outputs (run_id,{_REFERENCE_COLUMNS}) "
+                    "VALUES(?,?,?,?,?,?,?,?)",
+                    [(run_id, *reference) for reference in prepared],
+                )
+        return True
+
+    def get_execution_outputs(self, run_id: str) -> dict:
+        """Read frozen evidence in one snapshot without resolving its locators."""
+        conn = self.model.get_connection()
+        own_transaction = not conn.in_transaction
+        if own_transaction:
+            conn.execute("BEGIN")
+        try:
+            capture = _capture(conn, run_id)
+            if capture is None:
+                return {
+                    "state": "not_captured",
+                    "reason": None,
+                    "format_version": None,
+                    "source": None,
+                    "references": [],
+                }
+            references = []
+            for row in _references(conn, run_id):
+                reference = dict(zip(_REFERENCE_FIELDS, row))
+                reference.update(
+                    file_identity="unverified",
+                    newly_generated="unknown",
+                    context_membership="unknown",
+                    transient=reference["folder_type"] == "temp",
+                )
+                references.append(reference)
+            return {
+                "state": capture[0],
+                "reason": capture[1],
+                "format_version": capture[2],
+                "source": capture[3],
+                "references": references,
+            }
+        finally:
+            if own_transaction:
+                conn.rollback()

database/operations.py

@@ -13,6 +13,7 @@ from .pairs import PairDependencyError, PairOperations
 from .library import LibraryOperations
 from .execution_history import ExecutionHistoryOperations
 from .execution_contexts import ExecutionContextOperations
+from .execution_outputs import ExecutionOutputOperations
 
 # Import logging system
 try:
@@ -158,6 +159,7 @@ class PromptDatabase(
     LibraryOperations,
     ExecutionHistoryOperations,
     ExecutionContextOperations,
+    ExecutionOutputOperations,
 ):
     """Database operations class for managing prompts."""
 

docs/prompt-history-storage.md

@@ -1,9 +1,9 @@
 # Durable Prompt History
 
-Status: the durable job ledger is implemented as the first slice of
-[#209](https://github.com/ComfyAssets/ComfyUI_PromptManager/issues/209). Prompt
-contexts, output associations and the chronological timeline remain pending;
-#209 stays open.
+Status: the durable job ledger, immutable prompt contexts and run-owned output
+references are implemented for
+[#209](https://github.com/ComfyAssets/ComfyUI_PromptManager/issues/209). Gallery
+image associations and the chronological timeline remain pending; #209 stays open.
 
 ## Recorded jobs
 
@@ -21,7 +21,7 @@ outcomes. Missing or conflicting terminal evidence remains unknown; a process
 crash can leave an observed row without an invented completion time. Bounded
 safe diagnostics retain node ID/type and exception type, never graph text,
 arbitrary extra data, credentials, exception messages, traceback, inputs or
-output payloads. A bounded background writer keeps storage failures from
+arbitrary output payloads. A bounded background writer keeps storage failures from
 changing host execution and reports capture gaps rather than inventing rows.
 
 Existing databases begin with zero execution rows. Migration and backup restore
@@ -47,16 +47,38 @@ aggregate uses. It does not expose the ledger as chronological dots or repeat-ru
 branches. `PromptTracker` identities and PNG recipe evidence remain separate
 from durable host job identity.
 
+## Immutable contexts and output references
+
+Prompt contexts retain bounded raw/effective positive, negative and composition
+snapshots with nullable recipe navigation hints and explicit evidence provenance.
+Unknown text remains SQL NULL; a known empty negative remains empty text. Cached
+evidence belongs to its exact cache provider and attempt, rather than inheriting
+unrelated recent prompts. Deletion and restore can remove navigation hints without
+rewriting historical snapshots or rebinding them to current library rows.
+
+Schema v3 adds `execution_output_capture` and `execution_run_outputs` without
+backfilling older runs or modifying gallery image links. At final queue completion,
+the writer attaches bounded output evidence from that attempt's final host history
+to its immutable run ID. The internal `get_execution_outputs(run_id)` reader
+returns state, reason, format version, source and ordered references. A complete
+capture with `none_reported` means the final host history reported no images;
+`not_captured`, partial, unavailable and conflicting evidence remain explicit.
+Failed and interrupted jobs can still have complete evidence for images reported
+before termination. Running jobs do not gain a final capture merely because an
+image node finished.
+
+Each reference records output node ID, image ordinal, folder type, subfolder,
+filename, path style and an opaque configured-root key. A cached repeat can retain
+the same locators under a new run. Locators do not verify file bytes, continued
+existence, a newly generated file or membership in any prompt context: the reader
+reports file identity as `unverified`, generation and context membership as
+`unknown`, and temp outputs as transient. Capture never reads, resolves or serves
+the files and adds no public route. PNG metadata is not required. Same-path file
+replacement and existing gallery provenance retain their separate semantics.
+
 ## Remaining history work
 
-Design bounded execution contexts with nullable recipe references, immutable
-raw/effective positive, negative and composition snapshots and explicit
-inferred/recorded/unknown provenance. Cached nodes must not inherit unrelated
-recent prompt evidence. Define deletion and restore behavior before connecting
-saved recipes to historical snapshots.
-
-Add exact output references and a deduplicated run/image relationship without
-claiming cached outputs were newly created. Preserve existing image provenance
-and same-path replacement protection. Only after those semantics are tested can
-the timeline honestly show prompt-level execution chronology, batched outputs
-and repeat-run branches.
+Define a deduplicated run/gallery-image relationship that preserves existing image
+provenance and same-path replacement protection. Add a chronological timeline only
+after its navigation and evidence semantics can honestly represent prompt-level
+execution chronology, batched outputs and repeat-run branches.

tests/comfyui/execution-outputs.spec.mjs

@@ -0,0 +1,263 @@
+import { randomUUID } from 'node:crypto';
+import { expect } from '../../tools/vendor/node_modules/@playwright/test/index.mjs';
+import { test, readExecutionRuns } from './execution-fixture.mjs';
+
+async function resetCache({ execution }) {
+    expect((await execution.request.post('/pm_execution_fixture/cache', { data: { mode: 'RAM_PRESSURE' } })).ok()).toBe(true);
+}
+
+test.beforeEach(resetCache);
+test.afterAll(resetCache);
+
+function images(value, batch = 1) {
+    return {
+        '1': { class_type: 'PMHistoryImages', inputs: { value, batch } },
+        '2': { class_type: 'SaveImage', inputs: { images: ['1', 0], filename_prefix: 'output-evidence/nested/image' } },
+        '3': { class_type: 'PreviewImage', inputs: { images: ['1', 0] } },
+    };
+}
+
+async function submit(execution, prompt, id = randomUUID()) {
+    const response = await execution.request.post('/prompt', { data: { prompt, prompt_id: id } });
+    expect(response.ok()).toBe(true);
+    expect((await response.json()).prompt_id).toBe(id);
+    return id;
+}
+
+async function persisted(execution, id, outcomes) {
+    let result;
+    await expect.poll(async () => {
+        result = await readExecutionRuns(execution.databasePath, id);
+        return { outcomes: result.rows.map(row => row.outcome),
+            states: result.rows.map(row => result.outputs[row.run_id].state) };
+    }, { timeout: 15000, message: 'Exact attempts and final output evidence reach SQLite' }).toEqual({
+        outcomes, states: outcomes.map(() => 'complete'),
+    });
+    return result;
+}
+
+async function finalHistory(execution, id) {
+    const response = await execution.request.get(`/history/${id}`);
+    expect(response.ok()).toBe(true);
+    const item = (await response.json())[id];
+    expect(item.prompt[1]).toBe(id);
+    return item;
+}
+
+function expectHistory(capture, history) {
+    const reported = Object.entries(history.outputs).flatMap(([node, output]) =>
+        (output.images || []).map((image, ordinal) => ({ output_node_id: node, image_ordinal: ordinal,
+            folder_type: image.type, subfolder: image.subfolder, filename: image.filename })));
+    reported.sort((a, b) => a.output_node_id.localeCompare(b.output_node_id) || a.image_ordinal - b.image_ordinal);
+    expect(capture).toMatchObject({ state: 'complete', reason: reported.length ? 'reported' : 'none_reported',
+        source: 'host_final_history', format_version: 1 });
+    expect(capture.references.map(({ output_node_id, image_ordinal, folder_type, subfolder, filename }) =>
+        ({ output_node_id, image_ordinal, folder_type, subfolder, filename }))).toEqual(reported);
+    for (const reference of capture.references) {
+        expect(reference).toMatchObject({ path_style: 'posix', file_identity: 'unverified',
+            newly_generated: 'unknown', context_membership: 'unknown', transient: reference.folder_type === 'temp' });
+        expect(reference.root_key).toMatch(/^[a-f0-9]{64}$/);
+    }
+}
+
+async function gate(execution, action) {
+    expect((await execution.request.post(`/pm_execution_fixture/gate/${action}`)).ok()).toBe(true);
+}
+
+async function entered(execution) {
+    await expect.poll(async () => (await (await execution.request.get('/pm_execution_fixture/gate')).json()).entered,
+        { timeout: 10000 }).toBe(true);
+}
+
+function downstream(value, mode) {
+    const graph = images(value);
+    delete graph['3'];
+    graph['4'] = { class_type: 'PMHistoryImageOutput', inputs: { images: ['2', 0], mode } };
+    return graph;
+}
+
+test('headless stock SaveImage and PreviewImage retain A,A,B batches under a reused public UUID', async ({ execution }) => {
+    const id = randomUUID();
+    const graph = images(31, 2);
+    let result;
+    for (const [index, value] of [31, 31, 73].entries()) {
+        graph['1'].inputs.value = value;
+        await submit(execution, graph, id);
+        result = await persisted(execution, id, Array(index + 1).fill('succeeded'));
+        const item = await finalHistory(execution, id);
+        const row = result.rows[index];
+        expectHistory(result.outputs[row.run_id], item);
+        expect(result.outputs[row.run_id].references).toHaveLength(4);
+        expect(result.captures[row.run_id].contexts).toEqual([]);
+        expect(result.usage[row.run_id]).toEqual([]);
+        if (index === 1) expect(item.status.messages.find(([kind]) => kind === 'execution_cached')[1].nodes).toEqual(expect.arrayContaining(['1', '2', '3']));
+    }
+    const attempts = result.rows;
+    expect(new Set(attempts.map(row => row.run_id)).size).toBe(3);
+    expect(new Set(attempts.map(row => row.queue_item_id)).size).toBe(3);
+    const captures = attempts.map(row => result.outputs[row.run_id]);
+    expect(captures.map(capture => capture.state)).toEqual(['complete', 'complete', 'complete']);
+    expect(captures[1].references).toEqual(captures[0].references);
+    expect(captures[2].references).not.toEqual(captures[0].references);
+    expect(captures[0].references[0].context_membership).toBe('unknown');
+    expect(captures[0].references.filter(reference => reference.transient)).toHaveLength(2);
+});
+
+test('stock image evidence survives disabled PNG metadata', async ({ execution }) => {
+    const enabled = images(44);
+    delete enabled['3'];
+    const enabledId = await submit(execution, enabled);
+    const enabledResult = await persisted(execution, enabledId, ['succeeded']);
+    const enabledReference = enabledResult.outputs[enabledResult.rows[0].run_id].references[0];
+    const withMetadata = await execution.request.get('/view', { params: {
+        filename: enabledReference.filename, subfolder: enabledReference.subfolder, type: enabledReference.folder_type,
+    } });
+    expect(withMetadata.ok()).toBe(true);
+    expect((await withMetadata.body()).includes(Buffer.from('tEXtprompt'))).toBe(true);
+    expect((await execution.request.post('/pm_execution_fixture/metadata', { data: { disabled: true } })).ok()).toBe(true);
+    try {
+        const graph = images(45);
+        delete graph['3'];
+        const id = await submit(execution, graph);
+        const result = await persisted(execution, id, ['succeeded']);
+        const capture = result.outputs[result.rows[0].run_id];
+        expectHistory(capture, await finalHistory(execution, id));
+        expect(capture.references).toHaveLength(1);
+        const { filename, subfolder, folder_type: type } = capture.references[0];
+        const response = await execution.request.get('/view', { params: { filename, subfolder, type } });
+        expect(response.ok()).toBe(true);
+        const png = await response.body();
+        expect(png.subarray(1, 4).toString()).toBe('PNG');
+        expect(png.includes(Buffer.from('tEXtprompt'))).toBe(false);
+    } finally {
+        expect((await execution.request.post('/pm_execution_fixture/metadata', { data: { disabled: false } })).ok()).toBe(true);
+    }
+});
+
+test('text-only output reports a complete empty image set', async ({ execution }) => {
+    const id = await submit(execution, { '1': { class_type: 'PMHistoryCached', inputs: { value: 'text has no image locator' } } });
+    const result = await persisted(execution, id, ['succeeded']);
+    const capture = result.outputs[result.rows[0].run_id];
+    expectHistory(capture, await finalHistory(execution, id));
+    expect(capture.references).toEqual([]);
+    expect(capture.reason).toBe('none_reported');
+});
+
+test('batched outputs preserve PM usage while context membership stays unknown', async ({ execution }) => {
+    const graph = images(57, 2);
+    graph['10'] = { class_type: 'PMHistoryInputs', inputs: {} };
+    graph['20'] = { class_type: 'PromptManagerPair', inputs: { clip: ['10', 0], text: `output usage ${randomUUID()}`,
+        negative_text: `negative ${randomUUID()}`, prepend_text: '', append_text: '' } };
+    graph['30'] = { class_type: 'KSampler', inputs: { model: ['10', 1], latent_image: ['10', 2],
+        positive: ['20', 0], negative: ['20', 1], seed: 1, steps: 1, cfg: 1, sampler_name: 'euler', scheduler: 'normal', denoise: 1 } };
+    graph['1'].inputs.latent = ['30', 0];
+    const id = await submit(execution, graph);
+    const first = await persisted(execution, id, ['succeeded']);
+    const runId = first.rows[0].run_id;
+    expectHistory(first.outputs[runId], await finalHistory(execution, id));
+    expect(first.outputs[runId].references).toHaveLength(4);
+    expect(first.captures[runId].contexts).toHaveLength(1);
+    expect(first.usage[runId]).toEqual([{ role: 'positive', run_count: 1 }, { role: 'negative', run_count: 0 }, { role: 'pair', run_count: 1 }]);
+    await submit(execution, graph, id);
+    const second = await persisted(execution, id, ['succeeded', 'succeeded']);
+    expect(second.outputs[second.rows[1].run_id].references).toEqual(first.outputs[runId].references);
+    expect(second.usage[runId]).toEqual([{ role: 'positive', run_count: 2 }, { role: 'negative', run_count: 0 }, { role: 'pair', run_count: 2 }]);
+});
+
+test('failed work after stock SaveImage preserves exactly the final reported partial-run outputs', async ({ execution }) => {
+    const id = await submit(execution, downstream(81, 'failure'));
+    const result = await persisted(execution, id, ['failed']);
+    const item = await finalHistory(execution, id);
+    expect(item.status.messages.some(([kind]) => kind === 'execution_error')).toBe(true);
+    expectHistory(result.outputs[result.rows[0].run_id], item);
+    expect(result.outputs[result.rows[0].run_id].references).toHaveLength(1);
+});
+
+test('interruption after stock SaveImage preserves exactly the final reported outputs', async ({ execution }) => {
+    await gate(execution, 'hold');
+    try {
+        const id = await submit(execution, downstream(93, 'gate'));
+        await entered(execution);
+        expect((await execution.request.post('/interrupt', { data: {} })).ok()).toBe(true);
+        const result = await persisted(execution, id, ['interrupted']);
+        const item = await finalHistory(execution, id);
+        expect(item.status.messages.some(([kind]) => kind === 'execution_interrupted')).toBe(true);
+        expectHistory(result.outputs[result.rows[0].run_id], item);
+        expect(result.outputs[result.rows[0].run_id].references).toHaveLength(1);
+    } finally { await gate(execution, 'release'); }
+});
+
+test('completed image before pending gated work is not finalized early', async ({ execution }) => {
+    await gate(execution, 'hold');
+    try {
+        const id = await submit(execution, downstream(107, 'gate'));
+        await entered(execution);
+        let pending;
+        await expect.poll(async () => {
+            pending = await readExecutionRuns(execution.databasePath, id);
+            return pending.rows.map(row => row.outcome);
+        }, { timeout: 10000 }).toEqual(['observed']);
+        expect(pending.outputs[pending.rows[0].run_id]).toMatchObject({ state: 'not_captured', references: [] });
+        expect((await (await execution.request.get(`/history/${id}`)).json())[id]).toBeUndefined();
+        await gate(execution, 'release');
+        const result = await persisted(execution, id, ['succeeded']);
+        expectHistory(result.outputs[result.rows[0].run_id], await finalHistory(execution, id));
+        expect(result.outputs[result.rows[0].run_id].references).toHaveLength(1);
+    } finally { await gate(execution, 'release'); }
+});
+
+test('stock output capture survives both renderer switches and workflow reload', async ({ execution, page, context }, testInfo) => {
+    await context.route('**/*', async route => {
+        const path = new URL(route.request().url()).pathname.replace(/^\/api\//, '/');
+        if (path === '/settings' && route.request().method() === 'GET') {
+            const response = await route.fetch();
+            return route.fulfill({ json: { ...await response.json(),
+                'Comfy.VueNodes.Enabled': testInfo.project.metadata.initialRenderer,
+                'Comfy.TutorialCompleted': true, 'Comfy.Workflow.Persist': false,
+                'Comfy.Notification.ShowVersionUpdates': false } });
+        }
+        return route.continue();
+    });
+    const errors = [];
+    page.on('pageerror', error => errors.push(error.message));
+    await page.goto('/', { waitUntil: 'domcontentloaded' });
+    await expect(page.getByRole('status', { name: 'Loading ComfyUI' })).toBeHidden({ timeout: 30000 });
+    await page.waitForFunction(async () => {
+        const { app } = await import('/scripts/app.js');
+        return !!app.graph && !!globalThis.LiteGraph?.registered_node_types?.PMHistoryImages;
+    });
+    await expect.poll(() => page.evaluate(() => LiteGraph.vueNodesMode)).toBe(testInfo.project.metadata.initialRenderer);
+    await page.evaluate(async () => {
+        const { app } = await import('/scripts/app.js');
+        app.graph.clear();
+        const source = LiteGraph.createNode('PMHistoryImages');
+        const saved = LiteGraph.createNode('SaveImage');
+        const preview = LiteGraph.createNode('PreviewImage');
+        for (const node of [source, saved, preview]) app.graph.add(node);
+        source.widgets.find(widget => widget.name === 'value').value = 119;
+        source.widgets.find(widget => widget.name === 'batch').value = 1;
+        saved.widgets.find(widget => widget.name === 'filename_prefix').value = 'renderer-evidence/image';
+        source.connect(0, saved, 0);
+        source.connect(0, preview, 0);
+    });
+    const captures = [];
+    for (const renderer of [testInfo.project.metadata.initialRenderer, !testInfo.project.metadata.initialRenderer, testInfo.project.metadata.initialRenderer]) {
+        await page.evaluate(async renderer => {
+            const { app } = await import('/scripts/app.js');
+            await app.ui.settings.setSettingValue('Comfy.VueNodes.Enabled', renderer);
+            await app.loadGraphData(app.graph.serialize());
+        }, renderer);
+        await expect.poll(() => page.evaluate(() => LiteGraph.vueNodesMode)).toBe(renderer);
+        const graph = await page.evaluate(async () => (await (await import('/scripts/app.js')).app.graphToPrompt()).output);
+        expect(Object.values(graph).map(node => node.class_type).sort()).toEqual(['PMHistoryImages', 'PreviewImage', 'SaveImage']);
+        const id = await submit(execution, graph);
+        const result = await persisted(execution, id, ['succeeded']);
+        const capture = result.outputs[result.rows[0].run_id];
+        expectHistory(capture, await finalHistory(execution, id));
+        expect(capture.references).toHaveLength(2);
+        captures.push(capture);
+    }
+    expect(captures[1].references).toEqual(captures[0].references);
+    expect(captures[2].references).toEqual(captures[0].references);
+    expect(errors).toEqual([]);
+});

tests/comfyui/fixtures/ledger_nodes.py

@@ -5,6 +5,7 @@ import time
 
 import nodes
 import execution
+import torch
 
 from aiohttp import web
 import comfy.model_management
@@ -152,6 +153,62 @@ NODE_CLASS_MAPPINGS.update(
     {"PMHistoryList": PMHistoryList, "PMHistoryTransform": PMHistoryTransform}
 )
 
+
+class PMHistoryImages:
+    @classmethod
+    def INPUT_TYPES(cls) -> dict:
+        return {
+            "required": {
+                "value": ("INT", {"default": 0, "min": 0, "max": 255}),
+                "batch": ("INT", {"default": 1, "min": 1, "max": 2}),
+            },
+            "optional": {"latent": ("LATENT",)},
+        }
+
+    RETURN_TYPES = ("IMAGE",)
+    FUNCTION = "execute"
+    CATEGORY = "testing/execution-history"
+
+    def execute(self, value: int, batch: int, latent: dict | None = None) -> tuple:
+        return (torch.full((batch, 8, 8, 3), value / 255.0, device="cpu"),)
+
+
+class PMHistoryImageOutput:
+    @classmethod
+    def INPUT_TYPES(cls) -> dict:
+        return {
+            "required": {
+                "images": ("IMAGE",),
+                "mode": (["failure", "gate"],),
+            }
+        }
+
+    RETURN_TYPES = ()
+    FUNCTION = "execute"
+    CATEGORY = "testing/execution-history"
+    OUTPUT_NODE = True
+
+    def execute(self, images: torch.Tensor, mode: str) -> tuple:
+        if mode == "failure":
+            raise ValueError("owned failure after real image output")
+        PMHistoryGate().execute("owned gate after real image output")
+        return ()
+
+
+NODE_CLASS_MAPPINGS.update(
+    {"PMHistoryImages": PMHistoryImages, "PMHistoryImageOutput": PMHistoryImageOutput}
+)
+
+
+@PromptServer.instance.routes.post("/pm_execution_fixture/metadata")
+async def metadata(request: web.Request) -> web.Response:
+    disabled = (await request.json())["disabled"]
+    if type(disabled) is not bool:
+        raise web.HTTPBadRequest()
+    nodes.args.disable_metadata = disabled
+    return web.json_response({"disabled": disabled})
+
+
 _INJECTION_ORIGINALS = {}
 
 

tests/comfyui/read_execution_history.py

@@ -11,9 +11,12 @@ from uuid import UUID
 sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
 from database.execution_contexts import ExecutionContextOperations  # noqa: E402
 from database.execution_history import ExecutionHistoryOperations  # noqa: E402
+from database.execution_outputs import ExecutionOutputOperations  # noqa: E402
 
 
-class _ReadOnlyHistory(ExecutionContextOperations, ExecutionHistoryOperations):
+class _ReadOnlyHistory(
+    ExecutionContextOperations, ExecutionHistoryOperations, ExecutionOutputOperations
+):
     def __init__(self, connection: sqlite3.Connection) -> None:
         self.model = SimpleNamespace(get_connection=lambda: connection)
 
@@ -79,6 +82,10 @@ def read_runs(database: str, job_id: str, session_id: str | None = None) -> dict
             "captures": captures,
             "usage": usage,
             "snapshotIds": snapshot_ids,
+            "outputs": {
+                row["run_id"]: reader.get_execution_outputs(row["run_id"])
+                for row in rows
+            },
         }
 
 

tests/history_waits.py

@@ -0,0 +1,36 @@
+"""Finite test waits for real SQLite batches without a throughput requirement."""
+
+import threading
+import time
+from typing import Callable
+
+from utils.execution_history_writer import HistoryWriter
+
+
+def drain_writer(
+    writer: HistoryWriter, *, clock: Callable[[], float] = time.monotonic
+) -> None:
+    initial = current = writer.status()
+    started = last_progress = clock()
+    lowest_pending = initial.pending
+    while current.pending:
+        now = clock()
+        if current.pending < lowest_pending:
+            lowest_pending = current.pending
+            last_progress = now
+        reason = None
+        if not current.alive:
+            reason = "worker stopped"
+        elif now - started >= 30:
+            reason = "batch deadline"
+        elif now - last_progress >= 3:
+            reason = "worker stalled"
+        if reason is not None:
+            raise AssertionError(
+                f"history drain {reason}: elapsed={now - started:.3f}s, "
+                f"stalled={now - last_progress:.3f}s, "
+                f"initial={initial}, current={current}, "
+                f"charged_bytes={writer.context_budget.charged_bytes}"
+            )
+        threading.Event().wait(0.005)
+        current = writer.status()

tests/test_basic.py

@@ -247,7 +247,8 @@ class TestNodeIntegration(unittest.TestCase):
         # Import and test the node
         from prompt_manager import PromptManager
 
-        node = PromptManager()
+        with patch("prompt_manager_base.get_image_monitor"):
+            node = PromptManager()
 
         # Test encoding
         result = node.encode_prompt(clip=mock_clip, text="Test prompt", search_text="")

tests/test_execution_context_budget.py

@@ -1,6 +1,8 @@
 """Ownership and accounting behavior for shared context payloads."""
 
 import unittest
+from copy import copy
+from unittest.mock import patch
 
 
 class ContextBudgetTests(unittest.TestCase):
@@ -113,3 +115,111 @@ class ContextBudgetTests(unittest.TestCase):
                     contexts=(*sixteen.contexts, replace(small, anchor_id="17")),
                 )
             )
+
+    def test_reserve_up_to_uses_only_available_space_and_same_pool(self):
+        from utils.execution_context_budget import ContextBudget
+
+        pool = ContextBudget(100)
+        self.assertTrue(callable(getattr(pool, "reserve_up_to", None)))
+        first = pool.reserve(70)
+        second = pool.reserve_up_to(50)
+        self.assertEqual(second.byte_count, 30)
+        self.assertEqual(pool.charged_bytes, 100)
+        self.assertIsNone(pool.reserve_up_to(1))
+        first.release()
+        third = pool.reserve_up_to(20)
+        self.assertEqual(third.byte_count, 20)
+        self.assertIsNone(pool.reserve_up_to(0))
+        for invalid in (-1, True, 1.0):
+            with self.assertRaises(ValueError):
+                pool.reserve_up_to(invalid)
+        third.release()
+        second.release()
+        self.assertEqual(pool.charged_bytes, 0)
+
+    def test_reserve_up_to_is_atomic_under_competing_reservations(self):
+        from concurrent.futures import ThreadPoolExecutor
+        from threading import Barrier
+        from utils.execution_context_budget import ContextBudget
+
+        pool = ContextBudget(100)
+        self.assertTrue(callable(getattr(pool, "reserve_up_to", None)))
+        barrier = Barrier(8)
+
+        def reserve(_):
+            barrier.wait()
+            return pool.reserve_up_to(30)
+
+        with ThreadPoolExecutor(max_workers=8) as executor:
+            leases = list(executor.map(reserve, range(8)))
+        self.assertEqual(
+            sorted(lease.byte_count for lease in leases if lease), [10, 30, 30, 30]
+        )
+        self.assertEqual(pool.charged_bytes, 100)
+        for lease in leases:
+            if lease is not None:
+                lease.release()
+        self.assertEqual(pool.charged_bytes, 0)
+
+    def test_copied_and_forged_handles_cannot_spend_or_release_genuine_credits(self):
+        from utils.execution_context_budget import ContextBudget, PayloadLease
+
+        pool = ContextBudget(100)
+        genuine = pool.reserve(80)
+        forged = object.__new__(PayloadLease)
+        forged.__dict__.update(genuine.__dict__)
+        for impostor in (copy(genuine), forged):
+            with self.subTest(impostor=impostor):
+                with self.assertRaises(ValueError):
+                    _ = impostor.byte_count
+                with self.assertRaises(ValueError):
+                    impostor.transfer()
+                with self.assertRaises(ValueError):
+                    impostor.shrink_to(1)
+                impostor.release()
+                self.assertEqual(pool.charged_bytes, 80)
+                self.assertEqual(genuine.byte_count, 80)
+                self.assertIsNone(pool.reserve(21))
+        owner = genuine.transfer()
+        genuine.release()
+        owner.shrink_to(20)
+        self.assertEqual(pool.charged_bytes, 20)
+        owner.release()
+        self.assertEqual(pool.charged_bytes, 0)
+        last = pool.reserve(100)
+        self.assertIsNotNone(last)
+        last.release()
+        self.assertEqual(pool.charged_bytes, 0)
+
+    def test_failed_owner_allocation_or_registration_never_orphans_credits(self):
+        from utils.execution_context_budget import ContextBudget, PayloadLease
+
+        pool = ContextBudget(100)
+        genuine = pool.reserve(20)
+        with patch.object(PayloadLease, "_new", side_effect=MemoryError):
+            for operation in (
+                lambda: pool.reserve(10),
+                lambda: pool.reserve_up_to(10),
+                genuine.transfer,
+            ):
+                with self.assertRaises(MemoryError):
+                    operation()
+                self.assertEqual(pool.charged_bytes, 20)
+                self.assertEqual(genuine.byte_count, 20)
+
+        class DeniedOwners(set):
+            def add(self, owner):
+                raise MemoryError
+
+        pool._owners = DeniedOwners(pool._owners)
+        for operation in (
+            lambda: pool.reserve(10),
+            lambda: pool.reserve_up_to(10),
+            genuine.transfer,
+        ):
+            with self.assertRaises(MemoryError):
+                operation()
+            self.assertEqual(pool.charged_bytes, 20)
+            self.assertEqual(genuine.byte_count, 20)
+        genuine.release()
+        self.assertEqual(pool.charged_bytes, 0)

tests/test_execution_context_capture.py

@@ -73,6 +73,31 @@ class CaptureTests(unittest.TestCase):
             self.addCleanup(result.lease.release)
         return result
 
+    def test_unexpected_projection_error_releases_payload_but_keeps_capability(self):
+        with patch.object(
+            capture, "project_contexts", side_effect=RuntimeError("private")
+        ):
+            cap = self.begin()
+        self.assertTrue(self.manager.valid(cap))
+        self.assertEqual(self.pool.charged_bytes, 0)
+        result = self.seal(cap)
+        self.assertEqual(
+            (result.state, result.reason), ("unavailable", "invalid_evidence")
+        )
+        self.assertEqual(self.pool.charged_bytes, 0)
+
+    def test_unexpected_seal_error_discards_popped_attempt_and_lease(self):
+        cap = self.begin()
+        with patch.object(
+            capture, "bundle_payload_size", side_effect=RuntimeError("private")
+        ):
+            result = self.seal(cap)
+        self.assertEqual(
+            (result.state, result.reason), ("unavailable", "invalid_evidence")
+        )
+        self.assertFalse(self.manager.valid(cap))
+        self.assertEqual(self.pool.charged_bytes, 0)
+
     def test_runtime_keeps_raw_affix_whitespace_and_injected_effective_text(self):
         cap = self.begin()
         self.manager.observe(cap, SourceRef("p", "PromptManagerPair", 0), 0, snapshot())

tests/test_execution_context_schema.py

@@ -65,7 +65,7 @@ class ContextSchemaTests(unittest.TestCase):
             conn.execute(
                 "SELECT schema_version,storage_epoch FROM execution_history_metadata"
             ).fetchone(),
-            (2, EPOCH),
+            (history.SCHEMA_VERSION, EPOCH),
         )
         self.assertEqual(
             conn.execute("SELECT * FROM execution_runs").fetchall(), before
@@ -94,7 +94,7 @@ class ContextSchemaTests(unittest.TestCase):
             conn.execute(
                 "SELECT schema_version FROM execution_history_metadata"
             ).fetchone()[0],
-            2,
+            history.SCHEMA_VERSION,
         )
 
     def test_migration_failure_rolls_back_metadata_and_all_tables(self):

tests/test_execution_history_database.py

@@ -97,6 +97,12 @@ class HistoryDatabaseTests(unittest.TestCase):
                 with conn:
                     from database.execution_context_schema import TABLE_COLUMNS
 
+                    from database.execution_output_schema import (
+                        TABLE_COLUMNS as OUTPUT_COLUMNS,
+                    )
+
+                    for table in reversed(OUTPUT_COLUMNS):
+                        conn.execute(f"DROP TABLE {table}")
                     for table in reversed(TABLE_COLUMNS):
                         conn.execute(f"DROP TABLE {table}")
                     conn.execute("DROP TABLE execution_runs")
@@ -198,7 +204,10 @@ class HistoryDatabaseTests(unittest.TestCase):
         self.assertEqual(self.count(db), 1)
 
     def test_schema_rejects_extra_uniqueness_and_allows_nonunique_indexes(self):
-        from database.execution_history_schema import history_schema_error
+        from database.execution_history_schema import (
+            SCHEMA_VERSION,
+            history_schema_error,
+        )
 
         for table, constraint in (
             ("execution_runs", "job_id"),
@@ -236,8 +245,8 @@ class HistoryDatabaseTests(unittest.TestCase):
                             else:
                                 conn.execute(
                                     "INSERT INTO execution_history_metadata "
-                                    "VALUES(1,2,?)",
-                                    (epoch,),
+                                    "VALUES(1,?,?)",
+                                    (SCHEMA_VERSION, epoch),
                                 )
                         else:
                             conn.execute(
@@ -436,7 +445,7 @@ class HistoryDatabaseTests(unittest.TestCase):
             "sql=replace(sql, 'queue_item_id >= 0', 'queue_item_id >= -1') "
             "WHERE name='execution_runs'; PRAGMA writable_schema=OFF",
             "PRAGMA ignore_check_constraints=ON; "
-            "UPDATE execution_history_metadata SET schema_version=3",
+            "UPDATE execution_history_metadata SET schema_version=4",
             "PRAGMA writable_schema=ON; UPDATE sqlite_master SET "
             "sql=replace(sql, \"'observed'\", \"'OBSERVED'\") "
             "WHERE name='execution_runs'; PRAGMA writable_schema=OFF",

tests/test_execution_history_hooks.py

@@ -16,6 +16,7 @@ from types import SimpleNamespace
 import unittest
 from unittest.mock import patch
 
+from tests.history_waits import drain_writer
 from database import PromptDatabase
 from utils import usage_tracking
 from utils.execution_history_writer import HistoryWriter
@@ -117,7 +118,7 @@ class HistoryHooksTests(unittest.TestCase):
         ]
 
     def drain(self, registration):
-        self.wait_for(lambda: registration.writer.status().pending == 0)
+        drain_writer(registration.writer, clock=time.monotonic)
 
     def test_package_history_initializes_when_monitor_and_usage_setup_fail(self):
         try:

tests/test_execution_history_restore.py

@@ -22,6 +22,133 @@ class HistoryRestoreTests(unittest.TestCase):
     finish = history_tests.HistoryDatabaseTests.finish
     count = history_tests.HistoryDatabaseTests.count
 
+    def test_output_operations_restore_preserves_evidence_and_rejects_stale_finish(
+        self,
+    ):
+        from database.execution_history import HistoryEpochChanged
+        from tests.test_execution_output_database import bundle
+
+        db = self.database()
+        start = self.start(db)
+        db.write_execution_start(start)
+        finish = self.finish(start)
+        db.write_execution_outputs_for_finish(finish, bundle())
+        evidence = db.get_execution_outputs(start.run_id)
+        backup = str(Path(self.temp.name) / "output-operations.db")
+        self.assertTrue(db.model.backup_database(backup))
+        source = Path(backup).read_bytes()
+        db.model.restore_from_file(backup)
+        self.assertEqual(db.get_execution_outputs(start.run_id), evidence)
+        with self.assertRaises(HistoryEpochChanged):
+            db.write_execution_outputs_for_finish(finish, bundle("late.png"))
+        self.assertEqual(db.get_execution_outputs(start.run_id), evidence)
+        self.assertEqual(Path(backup).read_bytes(), source)
+
+    def test_output_wal_backup_restore_preserves_rows_and_historical_epoch(self):
+        try:
+            from tests.test_execution_output_schema import capture
+        except ImportError:
+            from test_execution_output_schema import capture
+        db = self.database()
+        start = self.start(db)
+        db.write_execution_start(start)
+        conn = db.model.get_connection()
+        conn.execute("PRAGMA wal_autocheckpoint=0")
+        with conn:
+            capture(conn, run=start.run_id)
+        expected = [
+            tuple(row) for row in conn.execute("SELECT * FROM execution_run_outputs")
+        ]
+        source = str(Path(self.temp.name) / "outputs-wal.db")
+        self.assertGreater(Path(self.path + "-wal").stat().st_size, 0)
+        self.assertTrue(db.model.backup_database(source))
+        original = Path(source).read_bytes()
+        for _ in range(2):
+            previous = db.get_execution_epoch()
+            db.model.restore_from_file(source)
+            self.assertNotEqual(db.get_execution_epoch(), previous)
+            self.assertNotEqual(db.get_execution_epoch(), start.storage_epoch)
+            conn = db.model.get_connection()
+            self.assertEqual(
+                [
+                    tuple(row)
+                    for row in conn.execute("SELECT * FROM execution_run_outputs")
+                ],
+                expected,
+            )
+            self.assertEqual(
+                db.get_execution_run(start.run_id)["storage_epoch"], start.storage_epoch
+            )
+            self.assertEqual(
+                tuple(
+                    conn.execute(
+                        "SELECT state,reason,format_version,source FROM "
+                        "execution_output_capture"
+                    ).fetchone()
+                ),
+                ("complete", "reported", 1, "host_final_history"),
+            )
+            self.assertEqual(Path(source).read_bytes(), original)
+
+    def test_malformed_outputs_preserve_source_live_and_existing_safety_backup(self):
+        try:
+            from tests.test_execution_output_schema import capture
+        except ImportError:
+            from test_execution_output_schema import capture
+        live = self.database()
+        start = self.start(live)
+        live.write_execution_start(start)
+        with live.model.get_connection() as conn:
+            capture(conn, run=start.run_id)
+        before = list(live.model.get_connection().iterdump())
+        safety = Path(self.path + ".backup_existing")
+        safety.write_bytes(b"existing safety backup")
+        mutations = (
+            "DROP TABLE execution_run_outputs",
+            "DROP TABLE execution_output_capture",
+            "DELETE FROM execution_output_capture",
+            "DELETE FROM execution_runs",
+            "CREATE UNIQUE INDEX bad_output ON execution_run_outputs(filename)",
+            "CREATE INDEX bad_output ON execution_run_outputs(filename)",
+            "CREATE VIEW bad_output AS SELECT * FROM execution_run_outputs",
+            "CREATE TRIGGER bad_output AFTER INSERT ON execution_run_outputs "
+            "BEGIN SELECT 1; END",
+            "PRAGMA ignore_check_constraints=ON; UPDATE execution_output_capture "
+            "SET source='filesystem_scan'",
+            "PRAGMA ignore_check_constraints=ON; UPDATE execution_output_capture "
+            "SET format_version=2",
+            "PRAGMA ignore_check_constraints=ON; UPDATE execution_output_capture "
+            "SET state='unavailable',reason='unavailable'",
+            "PRAGMA ignore_check_constraints=ON; UPDATE execution_run_outputs "
+            "SET filename='a' || char(0) || 'b'",
+            "PRAGMA ignore_check_constraints=ON; UPDATE execution_run_outputs "
+            "SET filename=replace(hex(zeroblob(1025)),'0','x')",
+            "DROP TABLE execution_runs; DROP TABLE execution_history_metadata",
+        )
+        for index, mutation in enumerate(mutations):
+            with self.subTest(mutation=mutation):
+                source = str(Path(self.temp.name) / f"bad-output-{index}.db")
+                self.assertTrue(live.model.backup_database(source))
+                with closing(sqlite3.connect(source)) as conn, conn:
+                    conn.executescript(mutation)
+                original = Path(source).read_bytes()
+                backups = {
+                    p.name: p.read_bytes()
+                    for p in Path(self.temp.name).glob("live.db.backup_*")
+                }
+                self.assertFalse(PromptModel.verify_database_file(source)[0])
+                with self.assertRaises(ValueError):
+                    live.model.restore_from_file(source)
+                self.assertEqual(list(live.model.get_connection().iterdump()), before)
+                self.assertEqual(Path(source).read_bytes(), original)
+                self.assertEqual(
+                    {
+                        p.name: p.read_bytes()
+                        for p in Path(self.temp.name).glob("live.db.backup_*")
+                    },
+                    backups,
+                )
+
     def test_legacy_identifier_and_diagnostic_bytes_rejected_before_fetch(self):
         try:
             from tests.test_execution_context_schema import legacy, RUN
@@ -200,6 +327,7 @@ class HistoryRestoreTests(unittest.TestCase):
 
     def test_wal_and_old_backup_staging_leave_source_unchanged(self):
         from database.execution_context_schema import TABLE_COLUMNS
+        from database.execution_output_schema import TABLE_COLUMNS as OUTPUT_COLUMNS
 
         source = str(Path(self.temp.name) / "v1.db")
         old = self.database(source)
@@ -207,6 +335,8 @@ class HistoryRestoreTests(unittest.TestCase):
         old.write_execution_start(start)
         with old.model.get_connection() as conn:
             conn.execute("INSERT INTO prompts(id,text,run_count) VALUES(7,'old',19)")
+            for table in reversed(OUTPUT_COLUMNS):
+                conn.execute(f"DROP TABLE {table}")
             for table in reversed(TABLE_COLUMNS):
                 conn.execute(f"DROP TABLE {table}")
             sql = (
@@ -215,7 +345,7 @@ class HistoryRestoreTests(unittest.TestCase):
                     "name='execution_history_metadata'"
                 )
                 .fetchone()[0]
-                .replace("schema_version = 2", "schema_version = 1")
+                .replace("schema_version = 3", "schema_version = 1")
             )
             conn.execute("DROP TABLE execution_history_metadata")
             conn.execute(sql)
@@ -272,6 +402,25 @@ class HistoryRestoreTests(unittest.TestCase):
                 source = str(Path(self.temp.name) / "bad-context.db")
                 self.assertTrue(live.model.backup_database(source))
                 with closing(sqlite3.connect(source)) as conn, conn:
+                    # Manufacture v2 explicitly so this still tests old restores.
+                    conn.execute("DROP TABLE execution_run_outputs")
+                    conn.execute("DROP TABLE execution_output_capture")
+                    sql = (
+                        conn.execute(
+                            "SELECT sql FROM sqlite_master WHERE "
+                            "name='execution_history_metadata'"
+                        )
+                        .fetchone()[0]
+                        .replace("schema_version = 3", "schema_version = 2")
+                    )
+                    epoch = conn.execute(
+                        "SELECT storage_epoch FROM execution_history_metadata"
+                    ).fetchone()[0]
+                    conn.execute("DROP TABLE execution_history_metadata")
+                    conn.execute(sql)
+                    conn.execute(
+                        "INSERT INTO execution_history_metadata VALUES(1,2,?)", (epoch,)
+                    )
                     conn.execute(mutation)
                 original = Path(source).read_bytes()
                 with self.assertRaises(ValueError):

tests/test_execution_history_writer.py

@@ -13,6 +13,7 @@ import unittest
 from unittest.mock import patch
 from uuid import uuid4
 
+from tests.history_waits import drain_writer
 from database import PromptDatabase
 from utils.execution_history_events import (
     LifecycleEvent,
@@ -245,11 +246,21 @@ class HistoryWriterTests(unittest.TestCase):
         backup = os.path.join(self.directory.name, "between.db")
         self.assertTrue(self.db.model.backup_database(backup))
         original = self.db.write_execution_finish
+        restore_started = threading.Event()
+        restore_finished = threading.Event()
+        restore_errors = []
 
         def finish(command):
-            result = original(command)
-            self.db.model.restore_from_file(backup)
-            return result
+            try:
+                result = original(command)
+                restore_started.set()
+                self.db.model.restore_from_file(backup)
+                return result
+            except Exception as error:
+                restore_errors.append(error)
+                raise
+            finally:
+                restore_finished.set()
 
         with patch.object(self.db, "write_execution_finish", finish):
             self.assertTrue(
@@ -257,7 +268,15 @@ class HistoryWriterTests(unittest.TestCase):
                     self.finish_command(start), self.sealed(writer)
                 )
             )
+            # Disk restore must finish before the worker's short drain deadline.
+            self.assertTrue(
+                restore_finished.wait(30),
+                "finish/restore operation did not complete: "
+                f"restore_started={restore_started.is_set()}, {writer.status()}",
+            )
+            self.assertEqual(restore_errors, [])
             self.wait_for(lambda: writer.status().pending == 0)
+        self.assertNotEqual(self.db.get_execution_epoch(), start.storage_epoch)
         self.assertEqual(
             self.db.get_execution_contexts(start.run_id)["state"], "not_captured"
         )
@@ -678,7 +697,7 @@ class HistoryWriterTests(unittest.TestCase):
                 "prompt_manager.execution_history_writer", "WARNING"
             ) as logs:
                 release.set()
-                self.wait_for(lambda: writer.status().pending == 0)
+                drain_writer(writer)
                 self.assertTrue(writer.close(2))
             self.assertEqual(len(logs.records), 1)
 

tests/test_execution_output_capture.py

@@ -0,0 +1,598 @@
+"""Host history is reduced to a bounded, scalar prefix under one payload pool."""
+
+import gc
+import unittest
+import weakref
+from dataclasses import FrozenInstanceError, replace
+from unittest.mock import patch
+
+from test_execution_contexts import load_module
+
+
+class Hostile:
+    def __iter__(self):
+        raise AssertionError("host protocol iteration")
+
+    def __str__(self):
+        raise AssertionError("host scalar coercion")
+
+    def __bool__(self):
+        raise AssertionError("host truth test")
+
+    def __eq__(self, other):
+        raise AssertionError("host equality")
+
+    def __hash__(self):
+        return 42
+
+
+class Text(str):
+    def encode(self, *args, **kwargs):
+        raise AssertionError("subclass encoding")
+
+
+class Mapping(dict):
+    def get(self, *args):
+        raise AssertionError("mapping protocol")
+
+    def items(self):
+        raise AssertionError("mapping iteration")
+
+
+class Sequence(list):
+    def __iter__(self):
+        raise AssertionError("sequence protocol")
+
+
+class OutputCaptureTests(unittest.TestCase):
+    def setUp(self):
+        self.m = load_module("execution_output_capture")
+        self.dto = load_module("execution_outputs")
+        self.budget = load_module("execution_context_budget")
+        self.pool = self.budget.ContextBudget()
+        self.roots = self.dto.snapshot_output_roots("posix", "/output", "/temp")
+
+    def image(self, **changes):
+        return dict(filename="same.png", subfolder="", type="output", **changes)
+
+    def capture(self, history, *, start=None, finish=None, **kwargs):
+        sealed = self.m.seal_outputs(
+            history,
+            self.roots if start is None else start,
+            self.roots if finish is None else finish,
+            budget=kwargs.pop("budget", self.pool),
+            capture_generation=kwargs.pop("capture_generation", 7),
+            context_bytes=kwargs.pop("context_bytes", 0),
+            **kwargs,
+        )
+        sealed.validate_header()
+        if sealed.lease is not None:
+            self.addCleanup(sealed.lease.release)
+        return sealed
+
+    def outputs(self, images):
+        return {"outputs": {"7": {"images": images}}}
+
+    def test_fresh_cached_duplicates_and_sorted_outer_identity(self):
+        history = {
+            "outputs": {
+                "7": {"images": [self.image(), self.image()]},
+                "10": {"images": [self.image()]},
+                "2": {
+                    "images": [
+                        dict(filename=" é.png ", subfolder=" 雪 / Café ", type="temp")
+                    ]
+                },
+            },
+            "meta": Hostile(),
+        }
+        first = self.capture(history)
+        second = self.capture(history)
+        self.assertEqual(first.bundle, second.bundle)
+        self.assertEqual((first.state, first.reason), ("complete", "reported"))
+        refs = first.bundle.references
+        self.assertEqual(
+            [(r.output_node_id, r.image_ordinal) for r in refs],
+            [("10", 0), ("2", 0), ("7", 0), ("7", 1)],
+        )
+        self.assertEqual(
+            (refs[1].filename, refs[1].subfolder, refs[1].folder_type),
+            (" é.png ", " 雪 / Café ", "temp"),
+        )
+        self.assertEqual(refs[1].root_key, self.roots.temp_key)
+        self.assertEqual(
+            first.lease.byte_count, self.dto.output_payload_size(first.bundle)
+        )
+        self.assertEqual(first.capture_generation, 7)
+        with self.assertRaises(FrozenInstanceError):
+            first.state = "partial"
+
+    def test_empty_and_text_only_are_positive_empty_evidence(self):
+        absent = self.dto.snapshot_output_roots("posix", None, None)
+        for outputs in ({}, {"7": {}}, {"7": {"text": Hostile(), "tensor": Hostile()}}):
+            with self.subTest(outputs=len(outputs)):
+                sealed = self.capture({"outputs": outputs}, start=absent, finish=absent)
+                self.assertEqual(
+                    (sealed.state, sealed.reason), ("complete", "none_reported")
+                )
+                self.assertEqual(sealed.bundle.references, ())
+                self.assertEqual(sealed.lease.byte_count, 128)
+                sealed.lease.release()
+        self.assertEqual(self.pool.charged_bytes, 0)
+
+    def test_missing_and_invalid_envelopes_have_lease_free_omission_headers(self):
+        for value, reason in (
+            (None, "invalid_evidence"),
+            ({}, "invalid_evidence"),
+            ({"outputs": None}, "invalid_evidence"),
+            (Hostile(), "invalid_evidence"),
+            (Mapping(outputs={}), "invalid_evidence"),
+            ({"outputs": Mapping()}, "invalid_evidence"),
+        ):
+            with self.subTest(kind=type(value).__name__):
+                sealed = self.capture(value)
+                self.assertEqual((sealed.state, sealed.reason), ("unavailable", reason))
+                self.assertIsNone(sealed.bundle)
+                self.assertIsNone(sealed.lease)
+        self.assertEqual(self.pool.charged_bytes, 0)
+
+    def test_wrong_container_or_scalar_types_never_invoke_host_protocols(self):
+        invalid_ui = [None, Hostile(), Mapping(images=[])]
+        for ui in invalid_ui:
+            sealed = self.capture({"outputs": {"7": ui}})
+            self.assertEqual(
+                (sealed.state, sealed.reason), ("partial", "invalid_evidence")
+            )
+        for images in (None, Hostile(), Sequence(), (), {}):
+            sealed = self.capture({"outputs": {"7": {"images": images}}})
+            self.assertEqual(sealed.reason, "invalid_evidence")
+        for entry in (
+            None,
+            True,
+            Hostile(),
+            Mapping(filename="x", subfolder="", type="output"),
+        ):
+            sealed = self.capture(self.outputs([entry, self.image()]))
+            self.assertEqual(sealed.reason, "invalid_evidence")
+            self.assertEqual(sealed.bundle.references[0].image_ordinal, 1)
+        for field in ("filename", "subfolder", "type"):
+            for value in (None, True, 1, Hostile(), Text("output")):
+                image = self.image()
+                image[field] = value
+                sealed = self.capture(self.outputs([image, self.image()]))
+                self.assertEqual(
+                    (sealed.state, sealed.reason), ("partial", "invalid_evidence")
+                )
+                self.assertEqual(len(sealed.bundle.references), 1)
+        for node_id in (None, True, 1, Hostile(), Text("7")):
+            sealed = self.capture(
+                {"outputs": {node_id: Hostile(), "good": {"images": [self.image()]}}}
+            )
+            self.assertEqual(sealed.reason, "invalid_evidence")
+            self.assertEqual(sealed.bundle.references[0].output_node_id, "good")
+
+    def test_missing_required_descriptor_fields_are_invalid(self):
+        for field in ("filename", "subfolder", "type"):
+            image = self.image()
+            del image[field]
+            sealed = self.capture(self.outputs([image]))
+            self.assertEqual(
+                (sealed.state, sealed.reason), ("partial", "invalid_evidence")
+            )
+            self.assertEqual(sealed.bundle.references, ())
+
+    def test_unsafe_paths_reject_entries_without_repair_or_coercion(self):
+        fixtures = {
+            "posix": {
+                "filename": ("", ".", "..", "a/b", "/absolute", "x\x00", "\ud800"),
+                "subfolder": (
+                    "/absolute",
+                    "../x",
+                    "x/../y",
+                    "x/./y",
+                    "x//y",
+                    "x/",
+                    ".",
+                    "..",
+                    "\ud800",
+                    "x\x00",
+                ),
+            },
+            "windows": {
+                "filename": ("a/b", "a\\b", "C:x", "x:stream"),
+                "subfolder": (
+                    "C:x",
+                    "C:\\x",
+                    "\\\\server\\share",
+                    "\\x",
+                    "x\\..\\y",
+                    "x\\.\\y",
+                    "x\\\\y",
+                    "x:stream",
+                ),
+            },
+        }
+        for style, fields in fixtures.items():
+            roots = self.dto.snapshot_output_roots(
+                style, "/output" if style == "posix" else "C:\\output", None
+            )
+            for field, values in fields.items():
+                for value in values:
+                    image = self.image()
+                    image[field] = value
+                    with self.subTest(style=style, field=field, value=repr(value)):
+                        sealed = self.capture(
+                            self.outputs([image, self.image()]),
+                            start=roots,
+                            finish=roots,
+                        )
+                        self.assertEqual(sealed.reason, "invalid_evidence")
+                        self.assertEqual(sealed.bundle.references[0].image_ordinal, 1)
+        image = dict(filename="a\\b:c.png ", subfolder=" x\\y /雪 ", type="output")
+        sealed = self.capture(self.outputs([image]))
+        self.assertEqual(sealed.reason, "reported")
+        self.assertEqual(sealed.bundle.references[0].filename, image["filename"])
+
+    def test_roots_missing_changed_or_style_changed_keep_unrooted_rows(self):
+        for finish in (
+            self.dto.snapshot_output_roots("posix", None, "/temp"),
+            self.dto.snapshot_output_roots("posix", "/changed", "/temp"),
+            self.dto.snapshot_output_roots("windows", "C:\\output", "C:\\temp"),
+        ):
+            sealed = self.capture(self.outputs([self.image()]), finish=finish)
+            self.assertEqual(
+                (sealed.state, sealed.reason), ("partial", "root_unavailable")
+            )
+            self.assertIsNone(sealed.bundle.references[0].root_key)
+            self.assertEqual(sealed.bundle.references[0].path_style, "posix")
+        missing = self.dto.snapshot_output_roots("posix", None, "/temp")
+        sealed = self.capture(self.outputs([self.image()]), start=missing)
+        self.assertIsNone(sealed.bundle.references[0].root_key)
+        sealed = self.capture(
+            {
+                "outputs": {
+                    "7": {"images": [dict(filename="x", subfolder="", type="temp")]}
+                }
+            },
+            finish=missing,
+        )
+        self.assertEqual(sealed.reason, "reported")
+
+    def test_node_boundary_counts_text_and_rejected_entries_before_sort(self):
+        for count, reason in ((500, "none_reported"), (501, "limit_reached")):
+            outputs = {str(i): {} for i in range(count)}
+            sealed = self.capture({"outputs": outputs})
+            self.assertEqual(sealed.reason, reason)
+        outputs = {i: Hostile() for i in range(500)}
+        outputs["accepted-if-overrun"] = {"images": [self.image()]}
+        sealed = self.capture({"outputs": outputs})
+        self.assertEqual((sealed.state, sealed.reason), ("partial", "limit_reached"))
+        self.assertEqual(sealed.bundle.references, ())
+        outputs = {
+            "z": {"images": [self.image()]},
+            **{str(i): {} for i in range(499)},
+            "a": {"images": [self.image()]},
+        }
+        sealed = self.capture({"outputs": outputs})
+        self.assertEqual(
+            [(r.output_node_id, r.image_ordinal) for r in sealed.bundle.references],
+            [("z", 0)],
+        )
+
+    def test_image_boundary_counts_malformed_positions_across_nodes(self):
+        for count, reason in ((256, "reported"), (257, "limit_reached")):
+            sealed = self.capture(self.outputs([self.image()] * count))
+            self.assertEqual(sealed.reason, reason)
+            self.assertEqual(len(sealed.bundle.references), 256)
+        sealed = self.capture(
+            {
+                "outputs": {
+                    "7": {"images": [None] * 255 + [self.image()]},
+                    "8": {"images": [self.image()]},
+                }
+            }
+        )
+        self.assertEqual(sealed.reason, "limit_reached")
+        self.assertEqual(
+            [(r.output_node_id, r.image_ordinal) for r in sealed.bundle.references],
+            [("7", 255)],
+        )
+        sealed = self.capture(self.outputs([None] * 256 + [Hostile()]))
+        self.assertEqual(sealed.reason, "limit_reached")
+        self.assertEqual(sealed.bundle.references, ())
+
+    def test_scalar_bounds_accept_exact_bytes_and_stop_before_later_entries(self):
+        fixtures = (
+            ("node", "😀" * 128, "😀" * 129),
+            ("filename", "é" * 512, "é" * 513),
+            ("subfolder", "é" * 1024, "é" * 1025),
+        )
+        for field, exact, over in fixtures:
+            for value, expected in ((exact, "reported"), (over, "limit_reached")):
+                image = self.image()
+                if field == "node":
+                    node = value
+                else:
+                    node = "7"
+                    image[field] = value
+                sealed = self.capture(
+                    {
+                        "outputs": {
+                            node: {"images": [image]},
+                            "tail": {"images": [self.image()]},
+                        }
+                    }
+                )
+                with self.subTest(field=field, expected=expected):
+                    self.assertEqual(sealed.reason, expected)
+                    self.assertEqual(
+                        len(sealed.bundle.references),
+                        2 if expected == "reported" else 0,
+                    )
+        sealed = self.capture(
+            {"outputs": {"x" * 129: {}, "tail": {"images": [self.image()]}}}
+        )
+        self.assertEqual(sealed.reason, "limit_reached")
+        self.assertEqual(sealed.bundle.references, ())
+
+    def test_bundle_byte_boundary_retains_only_prefix_and_shrinks_exactly(self):
+        node = "😀" * 128
+        image = dict(filename="x" * 1024, subfolder="x" * 2048, type="output")
+        # POSIX references charge 3,787 bytes; 34 + a 2,186-byte tail = 131,072.
+        tail = dict(filename="x", subfolder="x" * 1470, type="output")
+        exact = self.capture({"outputs": {node: {"images": [image] * 34 + [tail]}}})
+        self.assertEqual(exact.reason, "reported")
+        self.assertEqual(exact.lease.byte_count, 131072)
+        tail["subfolder"] += "x"
+        over = self.capture(
+            {"outputs": {node: {"images": [image] * 34 + [tail, self.image()]}}}
+        )
+        self.assertEqual((over.state, over.reason), ("partial", "limit_reached"))
+        self.assertEqual(len(over.bundle.references), 34)
+        self.assertEqual(over.lease.byte_count, 128886)
+
+    def test_pool_denial_too_small_and_attempt_remaining_space(self):
+        for available in (0, 1, 127):
+            pool = self.budget.ContextBudget(available)
+            sealed = self.capture(self.outputs([Hostile()]), budget=pool)
+            self.assertEqual(
+                (sealed.state, sealed.reason), ("unavailable", "budget_exhausted")
+            )
+            self.assertIsNone(sealed.bundle)
+            self.assertIsNone(sealed.lease)
+            self.assertEqual(pool.charged_bytes, 0)
+        for remaining, rows, reason in (
+            (127, None, "budget_exhausted"),
+            (128, 0, "budget_exhausted"),
+            (339, 0, "budget_exhausted"),
+            (340, 1, "reported"),
+        ):
+            sealed = self.capture(
+                self.outputs([self.image()]), context_bytes=1048576 - remaining
+            )
+            self.assertEqual(sealed.reason, reason)
+            if rows is None:
+                self.assertEqual(sealed.state, "unavailable")
+            else:
+                self.assertEqual(len(sealed.bundle.references), rows)
+                self.assertLessEqual(sealed.lease.byte_count, remaining)
+        empty = self.capture({"outputs": {}}, budget=self.budget.ContextBudget(128))
+        self.assertEqual((empty.state, empty.reason), ("complete", "none_reported"))
+
+    def test_pool_space_and_prior_fault_reason_precedence(self):
+        # A 340-byte rooted row plus the 128-byte header fits exactly.
+        pool = self.budget.ContextBudget(340)
+        sealed = self.capture(
+            self.outputs(
+                [
+                    None,
+                    dict(filename="x", subfolder="", type="input"),
+                    self.image(),
+                    self.image(),
+                ]
+            ),
+            budget=pool,
+        )
+        self.assertEqual((sealed.state, sealed.reason), ("partial", "budget_exhausted"))
+        self.assertEqual(len(sealed.bundle.references), 1)
+        for entries, reason in (
+            ([dict(filename="x", subfolder="", type="input")], "unsupported"),
+            (
+                [dict(filename="x", subfolder="", type="input"), None],
+                "invalid_evidence",
+            ),
+            ([None] * 257, "limit_reached"),
+        ):
+            sealed = self.capture(self.outputs(entries))
+            self.assertEqual((sealed.state, sealed.reason), ("partial", reason))
+        changed = self.dto.snapshot_output_roots("posix", "/changed", None)
+        sealed = self.capture(
+            self.outputs(
+                [self.image(), dict(filename="x", subfolder="", type="input")]
+            ),
+            finish=changed,
+        )
+        self.assertEqual(sealed.reason, "unsupported")
+
+    def test_reserve_precedes_copy_and_transfer_release_ownership(self):
+        pool = self.budget.ContextBudget(8388608)
+        context = pool.reserve(7000000)
+        old_writer = pool.reserve(1000000)
+        self.addCleanup(context.release)
+        self.addCleanup(old_writer.release)
+        sealed = self.capture(self.outputs([self.image()]), budget=pool)
+        self.assertEqual(pool.charged_bytes, 8000340)
+        owner = sealed.lease.transfer()
+        self.addCleanup(owner.release)
+        sealed.lease.release()
+        self.assertEqual(pool.charged_bytes, 8000340)
+        old_writer.release()
+        context.release()
+        self.assertEqual(pool.charged_bytes, 340)
+        owner.release()
+        owner.release()
+        self.assertEqual(pool.charged_bytes, 0)
+        original = self.dto.OutputReference
+
+        def check_reserved(*args, **kwargs):
+            self.assertEqual(pool.charged_bytes, 131072)
+            return original(*args, **kwargs)
+
+        # Construction is the first scalar-retention boundary.
+        with patch.object(self.m, "OutputReference", side_effect=check_reserved):
+            sealed = self.capture(self.outputs([self.image()]), budget=pool)
+        self.assertEqual(sealed.lease.byte_count, 340)
+
+    def test_exceptions_release_reservations_and_do_not_keep_host_envelope(self):
+        for target in ("OutputReference", "output_payload_size"):
+            with patch.object(
+                self.m, target, side_effect=RuntimeError("injected assembly failure")
+            ):
+                with self.assertRaises(RuntimeError):
+                    self.capture(self.outputs([self.image()]))
+            self.assertEqual(self.pool.charged_bytes, 0)
+        host = Hostile()
+        ref = weakref.ref(host)
+        history = self.outputs([self.image()])
+        history["tensor"] = host
+        history["outputs"]["7"]["metadata"] = host
+        sealed = self.capture(history)
+        history.clear()
+        del host, history
+        gc.collect()
+        self.assertIsNone(ref())
+        self.assertEqual(set(vars(sealed.lease)), {"_pool", "_bytes", "_active"})
+        self.assertEqual(
+            set(vars(sealed.bundle.references[0])),
+            {
+                "output_node_id",
+                "image_ordinal",
+                "folder_type",
+                "subfolder",
+                "filename",
+                "path_style",
+                "root_key",
+            },
+        )
+
+    def test_no_filesystem_or_sqlite_io(self):
+        with (
+            patch("builtins.open", side_effect=AssertionError("file read")),
+            patch("os.stat", side_effect=AssertionError("stat")),
+            patch("os.path.realpath", side_effect=AssertionError("resolve")),
+            patch("sqlite3.connect", side_effect=AssertionError("database")),
+        ):
+            sealed = self.capture(self.outputs([self.image()]))
+        self.assertEqual(sealed.reason, "reported")
+
+    def test_invalid_arguments_and_headers_cannot_leave_pool_charged(self):
+        for field, values in (
+            ("context_bytes", (-1, True, 1.0, 1048577)),
+            ("capture_generation", (-1, True, None, 2**63)),
+        ):
+            for value in values:
+                with (
+                    self.subTest(field=field, value=value),
+                    self.assertRaises(ValueError),
+                ):
+                    self.capture({"outputs": {}}, **{field: value})
+                self.assertEqual(self.pool.charged_bytes, 0)
+        sealed = self.capture({"outputs": {}})
+        for changes in (
+            {"state": Text("complete")},
+            {"reason": Hostile()},
+            {"state": "partial", "reason": "reported"},
+            {"capture_generation": True},
+            {"capture_generation": 2**63},
+        ):
+            with self.assertRaises(ValueError):
+                replace(sealed, **changes).validate_header()
+
+    def test_missing_history_has_distinct_omission_from_present_none(self):
+        self.assertTrue(hasattr(self.m, "MISSING_HISTORY"))
+        missing = self.capture(self.m.MISSING_HISTORY)
+        malformed = self.capture(None)
+        self.assertEqual(
+            (missing.state, missing.reason), ("unavailable", "unavailable")
+        )
+        self.assertEqual(
+            (malformed.state, malformed.reason), ("unavailable", "invalid_evidence")
+        )
+        self.assertEqual(self.pool.charged_bytes, 0)
+
+    def test_status_construction_failure_after_transfer_releases_new_owner(self):
+        with patch.object(
+            self.m.SealedOutputs,
+            "validate_header",
+            side_effect=RuntimeError("invalid header"),
+        ):
+            with self.assertRaises(RuntimeError):
+                self.capture(self.outputs([self.image()]))
+        self.assertEqual(self.pool.charged_bytes, 0)
+
+    def test_all_lease_assembly_failures_release_the_owned_reservation(self):
+        for operation in ("shrink_to", "transfer"):
+            with patch.object(
+                self.budget.PayloadLease,
+                operation,
+                side_effect=RuntimeError("lease operation"),
+            ):
+                with self.assertRaises(RuntimeError):
+                    self.capture(self.outputs([self.image()]))
+            self.assertEqual(self.pool.charged_bytes, 0)
+
+    def test_empty_pool_and_full_attempt_take_precedence_before_history_inspection(
+        self,
+    ):
+        for history in (None, {}, {"outputs": {}}, self.outputs([self.image()])):
+            for kwargs in (
+                {"budget": self.budget.ContextBudget(0)},
+                {"context_bytes": 1048576},
+            ):
+                sealed = self.capture(history, **kwargs)
+                self.assertEqual(
+                    (sealed.state, sealed.reason), ("unavailable", "budget_exhausted")
+                )
+                self.assertIsNone(sealed.bundle)
+                self.assertIsNone(sealed.lease)
+        sealed = self.capture(
+            self.outputs([self.image()]), budget=self.budget.ContextBudget(128)
+        )
+        self.assertEqual((sealed.state, sealed.reason), ("partial", "budget_exhausted"))
+        self.assertEqual(sealed.bundle.references, ())
+        self.assertEqual(sealed.lease.byte_count, 128)
+
+    def test_oversize_scalars_stop_on_character_precheck_without_host_encoding(self):
+        for field in ("filename", "subfolder"):
+            image = self.image()
+            image[field] = "x" * 1000000
+            sealed = self.capture(self.outputs([self.image(), image, self.image()]))
+            self.assertEqual(sealed.reason, "limit_reached")
+            self.assertEqual([r.image_ordinal for r in sealed.bundle.references], [0])
+        for node in ("x" * 129, "x" * 1000000):
+            sealed = self.capture(
+                {
+                    "outputs": {
+                        "first": {"images": [self.image()]},
+                        node: Hostile(),
+                        "tail": {"images": [self.image()]},
+                    }
+                }
+            )
+            self.assertEqual(sealed.reason, "limit_reached")
+            self.assertEqual(
+                [r.output_node_id for r in sealed.bundle.references], ["first"]
+            )
+
+    def test_sealed_references_outlive_later_history_mutations(self):
+        image = self.image()
+        images = [image, self.image()]
+        history = self.outputs(images)
+        sealed = self.capture(history)
+        image["filename"] = "replacement.png"
+        images.clear()
+        history.clear()
+        self.assertEqual(
+            [r.filename for r in sealed.bundle.references], ["same.png", "same.png"]
+        )
+        self.assertEqual([r.image_ordinal for r in sealed.bundle.references], [0, 1])

tests/test_execution_output_database.py

@@ -0,0 +1,428 @@
+"""Frozen output occurrences belong to an exact stored execution attempt."""
+
+import sqlite3
+import unittest
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import contextmanager
+from dataclasses import replace
+from threading import Barrier
+from unittest.mock import patch
+
+from database.execution_history import HistoryEpochChanged
+from utils.execution_outputs import OutputBundle, OutputReference
+
+try:
+    from tests import test_execution_history_database as history_tests
+except ImportError:
+    import test_execution_history_database as history_tests
+
+
+def bundle(
+    filename="first.png", *, root="a" * 64, node="9", ordinal=0, path_style="posix"
+):
+    return OutputBundle(
+        (OutputReference(node, ordinal, "output", "", filename, path_style, root),),
+        "complete" if root else "partial",
+        "reported" if root else "root_unavailable",
+    )
+
+
+class OutputDatabaseTests(unittest.TestCase):
+    setUp = history_tests.HistoryDatabaseTests.setUp
+    close_databases = history_tests.HistoryDatabaseTests.close_databases
+    database = history_tests.HistoryDatabaseTests.database
+    start = history_tests.HistoryDatabaseTests.start
+    finish = history_tests.HistoryDatabaseTests.finish
+
+    def capture(self, db, value=None, item=0):
+        start = self.start(db, item)
+        db.write_execution_start(start)
+        self.assertTrue(
+            db.write_execution_outputs_for_finish(self.finish(start), value or bundle())
+        )
+        return start
+
+    def rows(self, db):
+        return [
+            tuple(row)
+            for row in db.model.get_connection().execute(
+                "SELECT * FROM execution_run_outputs ORDER BY run_id,"
+                "output_node_id,image_ordinal"
+            )
+        ]
+
+    def test_exact_attempt_parent_not_recomputed_uuid(self):
+        db = self.database()
+        stored_id = "55555555-5555-4555-8555-555555555555"
+        start = replace(self.start(db), run_id=stored_id)
+        db.write_execution_start(start)
+        self.assertTrue(
+            db.write_execution_outputs_for_finish(self.finish(start), bundle())
+        )
+        self.assertEqual(self.rows(db)[0][0], stored_id)
+        self.assertEqual(db.get_execution_outputs(stored_id)["state"], "complete")
+        other = self.capture(db, item=1)
+        self.assertEqual(len(self.rows(db)), 2)
+        self.assertEqual(
+            db.get_execution_outputs(other.run_id)["references"][0]["filename"],
+            "first.png",
+        )
+        self.assertEqual(db.get_execution_run(stored_id)["outcome"], "observed")
+
+    def test_exact_replay_noop_and_conflict_sticky(self):
+        db = self.database()
+        first = bundle()
+        refs = (replace(first.references[0], output_node_id="2"),) + (
+            replace(first.references[0], image_ordinal=0),
+            replace(first.references[0], image_ordinal=1),
+        )
+        first = replace(first, references=refs)
+        start = self.capture(db, first)
+        finish = self.finish(start)
+        original = self.rows(db)
+        # Construct identical DTOs from dictionaries with different insertion orders.
+        reordered = tuple(
+            OutputReference(**dict(reversed(list(vars(ref).items())))) for ref in refs
+        )
+        self.assertTrue(
+            db.write_execution_outputs_for_finish(
+                finish, replace(first, references=reordered)
+            )
+        )
+        self.assertEqual(db.get_execution_outputs(start.run_id)["state"], "complete")
+        self.assertEqual(self.rows(db), original)
+        self.assertTrue(
+            db.write_execution_outputs_for_finish(finish, bundle("other.png"))
+        )
+        capture = db.get_execution_outputs(start.run_id)
+        self.assertEqual(
+            (capture["state"], capture["reason"]), ("conflicting", "conflicting")
+        )
+        self.assertEqual(
+            [r["filename"] for r in capture["references"]], ["first.png"] * 3
+        )
+        self.assertEqual(capture["references"][0]["context_membership"], "unknown")
+        for replay in (first, bundle("third.png")):
+            db.write_execution_outputs_for_finish(finish, replay)
+            self.assertEqual(db.get_execution_outputs(start.run_id), capture)
+            self.assertEqual(self.rows(db), original)
+
+    def test_header_and_every_reference_scalar_participate_in_replay(self):
+        variants = (
+            replace(bundle(), state="partial", reason="unsupported"),
+            bundle(root="b" * 64),
+            bundle(node="8"),
+            bundle(ordinal=1),
+            replace(
+                bundle(),
+                references=(replace(bundle().references[0], folder_type="temp"),),
+            ),
+            replace(
+                bundle(),
+                references=(replace(bundle().references[0], subfolder="nested"),),
+            ),
+            replace(
+                bundle(),
+                references=(replace(bundle().references[0], path_style="windows"),),
+            ),
+            OutputBundle((), "complete", "none_reported"),
+        )
+        db = self.database()
+        for item, variant in enumerate(variants):
+            with self.subTest(item=item):
+                start = self.capture(db, item=item)
+                db.write_execution_outputs_for_finish(self.finish(start), variant)
+                capture = db.get_execution_outputs(start.run_id)
+                self.assertEqual(capture["state"], "conflicting")
+                self.assertEqual(capture["references"][0]["root_key"], "a" * 64)
+                self.assertEqual(capture["references"][0]["output_node_id"], "9")
+
+    def test_concurrent_replay_keeps_first_rows(self):
+        db = self.database()
+        start = self.start(db)
+        db.write_execution_start(start)
+        finish = self.finish(start)
+        peers = [self.database() for _ in range(4)]
+        barrier = Barrier(len(peers))
+
+        def write(index):
+            barrier.wait(timeout=5)
+            return peers[index].write_execution_outputs_for_finish(
+                finish, bundle("first.png" if index % 2 else "second.png")
+            )
+
+        with ThreadPoolExecutor(max_workers=4) as executor:
+            self.assertEqual(list(executor.map(write, range(4))), [True] * 4)
+        capture = db.get_execution_outputs(start.run_id)
+        self.assertEqual(
+            (capture["state"], capture["reason"]), ("conflicting", "conflicting")
+        )
+        self.assertEqual(len(capture["references"]), 1)
+        frozen = self.rows(db)
+        for value in (bundle(), bundle("second.png")):
+            db.write_execution_outputs_for_finish(finish, value)
+            self.assertEqual(self.rows(db), frozen)
+
+    def test_epoch_and_generation_guard_cover_commit(self):
+        db = self.database()
+        start = self.start(db)
+        db.write_execution_start(start)
+        observations = []
+        conn = db.model.get_connection()
+        active = False
+
+        @contextmanager
+        def guard():
+            nonlocal active
+            self.assertTrue(conn.in_transaction)
+            active = True
+            try:
+                yield True
+            finally:
+                observations.append(("exit", conn.in_transaction))
+                active = False
+
+        def trace(sql):
+            if sql == "COMMIT":
+                observations.append(("commit", active))
+
+        conn.set_trace_callback(trace)
+        self.addCleanup(conn.set_trace_callback, None)
+        self.assertTrue(
+            db.write_execution_outputs_for_finish(
+                self.finish(start), bundle(), generation_guard=guard
+            )
+        )
+        self.assertEqual(observations, [("commit", True), ("exit", False)])
+        self.assertFalse(active)
+
+        @contextmanager
+        def stale():
+            yield False
+
+        other = self.start(db, item=1)
+        db.write_execution_start(other)
+        self.assertFalse(
+            db.write_execution_outputs_for_finish(
+                self.finish(other), bundle(), generation_guard=stale
+            )
+        )
+        self.assertEqual(
+            db.get_execution_outputs(other.run_id)["state"], "not_captured"
+        )
+        old_epoch = start.storage_epoch
+        with conn:
+            conn.execute(
+                "UPDATE execution_history_metadata SET storage_epoch=?",
+                ("66666666-6666-4666-8666-666666666666",),
+            )
+        with self.assertRaises(HistoryEpochChanged):
+            db.write_execution_outputs_for_finish(
+                self.finish(start), bundle("late.png")
+            )
+        self.assertEqual(db.get_execution_run(start.run_id)["storage_epoch"], old_epoch)
+        self.assertEqual(
+            db.get_execution_outputs(start.run_id)["references"][0]["filename"],
+            "first.png",
+        )
+
+    def test_failure_rolls_back_header_and_keeps_guard_until_rollback(self):
+        db = self.database()
+        start = self.start(db)
+        db.write_execution_start(start)
+        conn = db.model.get_connection()
+        with conn:
+            conn.execute(
+                "CREATE TRIGGER reject_output BEFORE INSERT ON execution_run_outputs "
+                "BEGIN SELECT RAISE(ABORT,'injected'); END"
+            )
+        states = []
+
+        @contextmanager
+        def guard():
+            states.append(conn.in_transaction)
+            try:
+                yield True
+            finally:
+                states.append(conn.in_transaction)
+
+        with self.assertRaises(sqlite3.IntegrityError):
+            db.write_execution_outputs_for_finish(
+                self.finish(start), bundle(), generation_guard=guard
+            )
+        self.assertEqual(states, [True, False])
+        self.assertEqual(
+            db.get_execution_outputs(start.run_id)["state"], "not_captured"
+        )
+        self.assertEqual(self.rows(db), [])
+
+    def test_missing_header_or_parent_rejected(self):
+        db = self.database()
+        start = self.start(db)
+        self.assertFalse(
+            db.write_execution_outputs_for_finish(self.finish(start), bundle())
+        )
+        self.assertEqual(
+            db.get_execution_outputs(start.run_id),
+            {
+                "state": "not_captured",
+                "reason": None,
+                "format_version": None,
+                "source": None,
+                "references": [],
+            },
+        )
+        db.write_execution_start(start)
+        conn = db.model.get_connection()
+        conn.execute("PRAGMA foreign_keys=OFF")
+        self.assertEqual(conn.execute("PRAGMA foreign_keys").fetchone()[0], 0)
+        with conn:
+            conn.execute(
+                "INSERT INTO execution_run_outputs "
+                "VALUES(?, '9',0,'output','','orphan.png','posix',?)",
+                (start.run_id, "a" * 64),
+            )
+        with self.assertRaises(sqlite3.DatabaseError):
+            db.get_execution_outputs(start.run_id)
+        with self.assertRaises(sqlite3.DatabaseError):
+            db.write_execution_outputs_for_finish(self.finish(start), bundle())
+        self.assertEqual(
+            conn.execute("SELECT count(*) FROM execution_output_capture").fetchone()[0],
+            0,
+        )
+        with conn:
+            conn.execute("DELETE FROM execution_run_outputs")
+        db.write_execution_outputs_for_finish(self.finish(start), bundle())
+        with conn:
+            conn.execute("DELETE FROM execution_runs")
+        with self.assertRaises(sqlite3.DatabaseError):
+            db.get_execution_outputs(start.run_id)
+
+    def test_reads_never_access_files_or_rebind(self):
+        db = self.database()
+        value = bundle(root=None)
+        value = replace(
+            value,
+            references=(
+                replace(
+                    value.references[0],
+                    folder_type="temp",
+                    subfolder=" Exact Ω ",
+                    filename="Keep Case.png",
+                ),
+            ),
+        )
+        start = self.capture(db, value)
+        conn = db.model.get_connection()
+        before = list(conn.iterdump())
+        with (
+            patch("builtins.open", side_effect=AssertionError("file read")),
+            patch("os.stat", side_effect=AssertionError("stat")),
+            patch("os.path.realpath", side_effect=AssertionError("resolve")),
+        ):
+            capture = db.get_execution_outputs(start.run_id)
+        self.assertEqual(
+            capture,
+            {
+                "state": "partial",
+                "reason": "root_unavailable",
+                "format_version": 1,
+                "source": "host_final_history",
+                "references": [
+                    {
+                        "output_node_id": "9",
+                        "image_ordinal": 0,
+                        "folder_type": "temp",
+                        "subfolder": " Exact Ω ",
+                        "filename": "Keep Case.png",
+                        "path_style": "posix",
+                        "root_key": None,
+                        "file_identity": "unverified",
+                        "newly_generated": "unknown",
+                        "context_membership": "unknown",
+                        "transient": True,
+                    }
+                ],
+            },
+        )
+        self.assertEqual(list(conn.iterdump()), before)
+        for table in (
+            "prompts",
+            "prompt_pairs",
+            "generated_images",
+            "execution_run_contexts",
+        ):
+            self.assertEqual(
+                conn.execute(f"SELECT count(*) FROM {table}").fetchone()[0], 0
+            )
+
+    def test_read_uses_consistent_snapshot_and_preserves_callers_transaction(self):
+        db = self.database()
+        start = self.capture(db)
+        peer = self.database()
+        conn = db.model.get_connection()
+        changed = False
+
+        def trace(sql):
+            nonlocal changed
+            if "FROM execution_run_outputs" in sql and not changed:
+                changed = True
+                with peer.model.get_connection() as writer:
+                    writer.execute(
+                        "UPDATE execution_run_outputs SET filename='later.png'"
+                    )
+
+        conn.set_trace_callback(trace)
+        self.addCleanup(conn.set_trace_callback, None)
+        self.assertEqual(
+            db.get_execution_outputs(start.run_id)["references"][0]["filename"],
+            "first.png",
+        )
+        conn.set_trace_callback(None)
+        self.assertTrue(changed)
+        self.assertFalse(conn.in_transaction)
+        self.assertEqual(
+            db.get_execution_outputs(start.run_id)["references"][0]["filename"],
+            "later.png",
+        )
+        conn.execute("BEGIN")
+        db.get_execution_outputs(start.run_id)
+        self.assertTrue(conn.in_transaction)
+        conn.rollback()
+
+    def test_empty_unavailable_and_partial_headers_remain_distinct(self):
+        db = self.database()
+        for item, (state, reason) in enumerate(
+            (
+                ("complete", "none_reported"),
+                ("unavailable", "unavailable"),
+                ("partial", "unsupported"),
+            )
+        ):
+            start = self.capture(db, OutputBundle((), state, reason), item)
+            capture = db.get_execution_outputs(start.run_id)
+            self.assertEqual(
+                (capture["state"], capture["reason"], capture["references"]),
+                (state, reason, []),
+            )
+
+    def test_invalid_scalars_rejected_before_transaction(self):
+        db = self.database()
+        start = self.start(db)
+        db.write_execution_start(start)
+        trace = []
+        conn = db.model.get_connection()
+        conn.set_trace_callback(trace.append)
+        self.addCleanup(conn.set_trace_callback, None)
+        for invalid in (
+            replace(
+                bundle(),
+                references=(replace(bundle().references[0], filename="../unsafe.png"),),
+            ),
+            replace(bundle(), source="filesystem_scan"),
+        ):
+            with self.assertRaises(ValueError):
+                db.write_execution_outputs_for_finish(self.finish(start), invalid)
+        with self.assertRaises(ValueError):
+            db.write_execution_outputs_for_finish(start, bundle())
+        self.assertEqual(trace, [])

tests/test_execution_output_gallery.py

@@ -0,0 +1,260 @@
+"""Mutable metadata galleries never own or retarget frozen output references."""
+
+from pathlib import PurePosixPath, PureWindowsPath
+
+from PIL import Image
+from PIL.PngImagePlugin import PngInfo
+
+from database.pairs import PairDependencyError
+from tests import test_execution_history_database as history_tests
+from tests.test_api_admin import AdminAPITestCase
+from tests.test_execution_output_database import bundle
+from utils.execution_outputs import snapshot_output_roots
+
+
+class OutputGalleryTests(AdminAPITestCase):
+    start = history_tests.HistoryDatabaseTests.start
+    finish = history_tests.HistoryDatabaseTests.finish
+
+    def capture(self, filename="first.png", item=0, root=None):
+        db = self.api.db
+        start = self.start(db, item)
+        db.write_execution_start(start)
+        output_root = root or self.output_dir
+        style = "windows" if isinstance(output_root, PureWindowsPath) else "posix"
+        key = snapshot_output_roots(style, str(output_root), None).output_key
+        db.write_execution_outputs_for_finish(
+            self.finish(start), bundle(filename, root=key, path_style=style)
+        )
+        return start.run_id, db.get_execution_outputs(start.run_id)
+
+    def assert_evidence(self, run, expected):
+        self.assertEqual(self.api.db.get_execution_outputs(run), expected)
+        reference = expected["references"][0]
+        self.assertEqual(reference["context_membership"], "unknown")
+        self.assertNotIn("image_id", reference)
+        self.assertNotIn("pair_id", reference)
+        self.assertNotIn("prompt_id", reference)
+
+    def gallery(self):
+        conn = self.api.db.model.get_connection()
+        return {
+            table: [
+                tuple(row) for row in conn.execute(f"SELECT * FROM {table} ORDER BY id")
+            ]
+            for table in (
+                "prompts",
+                "prompt_pairs",
+                "generated_images",
+                "image_prompt_pairs",
+            )
+        }
+
+    def test_capture_matches_root_path_style_on_each_platform(self) -> None:
+        roots = (
+            (PurePosixPath("/output"), "posix"),
+            (PureWindowsPath(r"D:\output"), "windows"),
+        )
+        for item, (root, style) in enumerate(roots):
+            with self.subTest(style=style):
+                run, evidence = self.capture(item=item, root=root)
+                self.assertEqual(evidence["state"], "complete")
+                reference = evidence["references"][0]
+                self.assertEqual(reference["path_style"], style)
+                self.assertEqual(len(reference["root_key"]), 64)
+                self.assert_evidence(run, evidence)
+
+    def test_metadata_disabled_or_malformed_still_captures_without_gallery_rows(self):
+        for item, metadata in enumerate(({}, {"prompt": "{malformed"})):
+            with self.subTest(metadata=metadata):
+                path = self.output_dir / f"metadata-{item}.png"
+                info = PngInfo()
+                for key, value in metadata.items():
+                    info.add_text(key, value)
+                Image.new("RGB", (2, 2)).save(path, pnginfo=info)
+                extracted = self.api._extract_comfyui_metadata(str(path))
+                self.assertEqual(
+                    self.api._ingest_scanned_file(path, extracted),
+                    None if not metadata else "unresolved",
+                )
+                before = self.gallery()
+                run, evidence = self.capture(path.name, item)
+                self.assertEqual(evidence["state"], "complete")
+                self.assertEqual(self.gallery(), before)
+                self.assertEqual(self.api.db.get_recent_prompts()["total"], 0)
+                self.assert_evidence(run, evidence)
+
+    def test_scans_before_after_delayed_repeated_and_overwrite_keep_exact_counters(
+        self,
+    ):
+        metadata = {
+            "parameters": "cat\nNegative prompt: blur\nSteps: 20, Sampler: Euler"
+        }
+        db = self.api.db
+        captures = []
+        for item, before_capture in enumerate((True, False)):
+            path = self.output_dir / f"scan-{item}.png"
+            path.write_bytes(b"first contents")
+            if before_capture:
+                first = self.api._ingest_scan_batch_sync([(path, metadata)])
+            before = self.gallery()
+            run, evidence = self.capture(path.name, item)
+            captures.append((run, evidence))
+            self.assertEqual(self.gallery(), before)
+            if not before_capture:
+                first = self.api._ingest_scan_batch_sync([(path, metadata)])
+            self.assertEqual(
+                first,
+                {
+                    "processed": 1,
+                    "found": 1,
+                    "added": 1,
+                    "linked": 0,
+                    "enriched": 0,
+                    "unchanged": 0,
+                    "ambiguous": 0,
+                    "unresolved": 0,
+                    "conflict": 0,
+                    "failed": 0,
+                    "pairs_added": int(before_capture),
+                    "pair_links_added": 1,
+                },
+            )
+            for _ in range(2):
+                repeated = self.api._ingest_scan_batch_sync([(path, metadata)])
+                self.assertEqual(
+                    repeated,
+                    {
+                        "processed": 1,
+                        "found": 1,
+                        "added": 0,
+                        "linked": 0,
+                        "enriched": 0,
+                        "unchanged": 1,
+                        "ambiguous": 0,
+                        "unresolved": 0,
+                        "conflict": 0,
+                        "failed": 0,
+                        "pairs_added": 0,
+                        "pair_links_added": 0,
+                    },
+                )
+            path.write_bytes(b"replacement bytes at the same relative path")
+            conflict = self.api._ingest_scan_batch_sync(
+                [
+                    (
+                        path,
+                        {"parameters": metadata["parameters"].replace("blur", "noise")},
+                    )
+                ]
+            )
+            self.assertEqual(conflict["conflict"], 1)
+            self.assertEqual(conflict["added"], 0)
+            self.assertEqual(db.list_prompt_pairs()["total"], 1)
+            self.assert_evidence(run, evidence)
+        pair = db.list_prompt_pairs()["pairs"][0]
+        self.assertEqual((pair["run_count"], pair["image_count"]), (0, 2))
+        self.assertEqual(
+            [p["run_count"] for p in db.get_recent_prompts()["prompts"]], [0, 0]
+        )
+        for run, evidence in captures:
+            self.assert_evidence(run, evidence)
+
+    def test_same_relative_path_across_roots_never_retargets(self):
+        second_root = self.output_dir / "second-root"
+        second_root.mkdir()
+        metadata = {
+            "parameters": "cat\nNegative prompt: blur\nSteps: 20, Sampler: Euler"
+        }
+        records = []
+        for item, root in enumerate((self.output_dir, second_root)):
+            path = root / "same.png"
+            path.write_bytes(b"image")
+            result = self.api._ingest_scanned_file(path, metadata, detailed=True)
+            self.assertEqual(result["status"], "added")
+            records.append(self.capture("same.png", item, root))
+        self.assertNotEqual(
+            records[0][1]["references"][0]["root_key"],
+            records[1][1]["references"][0]["root_key"],
+        )
+        self.assertEqual(self.api.db.list_prompt_pairs()["pairs"][0]["image_count"], 2)
+        for root in (self.output_dir, second_root):
+            (root / "same.png").unlink()
+        second_root.rmdir()
+        self.assertEqual(self.api.db.cleanup_missing_images(), 2)
+        for run, expected in records:
+            self.assert_evidence(run, expected)
+
+    def test_archive_delete_prune_cleanup_and_numeric_id_reuse_keep_evidence(self):
+        db = self.api.db
+        path = self.output_dir / "first.png"
+        path.write_bytes(b"image")
+        result = self.api._ingest_scanned_file(
+            path,
+            {"parameters": "cat\nNegative prompt: blur\nSteps: 20, Sampler: Euler"},
+            detailed=True,
+        )
+        self.assertEqual(
+            (result["status"], result["pairs_added"], result["links_added"]),
+            ("added", 1, 1),
+        )
+        image_id, pair_id = result["image_id"], result["pair_id"]
+        pair = db.get_prompt_pair(pair_id)
+        positive_id = pair["positive_prompt_id"]
+        run, evidence = self.capture()
+        with self.assertRaises(PairDependencyError):
+            db.delete_prompt(positive_id)
+        with self.assertRaises(PairDependencyError):
+            db.delete_prompt_pair(pair_id)
+        self.assertTrue(db.archive_prompt_pair(pair_id))
+        self.assertEqual(db.list_prompt_pairs()["total"], 0)
+        self.assertEqual(db.list_prompt_pairs(include_archived=True)["total"], 1)
+        self.assertEqual(db.get_prompt_pair(pair_id)["run_count"], 0)
+        self.assert_evidence(run, evidence)
+        self.assertTrue(db.delete_image(image_id))
+        self.assertFalse(db.delete_image(image_id))
+        self.assertEqual(db.get_image_prompt_pairs(image_id), [])
+        self.assertTrue(db.delete_prompt_pair(pair_id))
+        self.assertTrue(db.delete_prompt(positive_id))
+        self.assertEqual(db.prune_orphaned_prompts(), 1)
+        self.assertEqual(db.prune_orphaned_prompts(), 0)
+        self.assertEqual(db.cleanup_missing_images(), 0)
+        self.assert_evidence(run, evidence)
+
+        # Explicit IDs reproduce restored/imported numeric-ID reuse without assuming
+        # SQLite's allocator reuses AUTOINCREMENT identities.
+        conn = db.model.get_connection()
+        with conn:
+            conn.execute(
+                "INSERT INTO prompts(id,text,run_count) "
+                "VALUES(?, 'replacement prompt', 17)",
+                (positive_id,),
+            )
+            conn.execute(
+                "INSERT INTO prompt_pairs(id,positive_prompt_id,run_count) "
+                "VALUES(?,?,23)",
+                (pair_id, positive_id),
+            )
+            conn.execute(
+                "INSERT INTO generated_images "
+                "(id,prompt_id,image_path,filename,file_path) "
+                "VALUES(?,?,?,'first.png',?)",
+                (image_id, positive_id, str(path), str(path)),
+            )
+            conn.execute(
+                "INSERT INTO image_prompt_pairs(image_id,pair_id,provenance) "
+                "VALUES(?,?,'{}')",
+                (image_id, pair_id),
+            )
+        self.assert_evidence(run, evidence)
+        self.assertTrue(db.archive_prompt_pair(pair_id))
+        path.unlink()
+        self.assertEqual(db.cleanup_missing_images(), 1)
+        self.assertEqual(db.get_image_prompt_pairs(image_id), [])
+        self.assertTrue(db.delete_prompt_pair(pair_id))
+        # Standalone prompt deletion retains its original gallery cascade.
+        recreated = db.link_image_to_prompt(positive_id, str(path))
+        self.assertGreater(recreated, 0)
+        self.assertTrue(db.delete_prompt(positive_id))
+        self.assertIsNone(db.get_image_by_id(recreated))
+        self.assert_evidence(run, evidence)

tests/test_execution_output_hooks.py

@@ -0,0 +1,403 @@
+"""Exact successful host callbacks persist independently bounded output evidence."""
+
+from dataclasses import replace
+import os
+from pathlib import Path
+import threading
+import sys
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from utils.execution_history import HistoryRegistration
+from utils.execution_context_budget import ContextBudget
+from utils.execution_outputs import snapshot_output_roots
+from utils.queue_observers import subscribe_dequeue, subscribe_task_done
+
+try:
+    from tests import test_execution_history_hooks as fixtures
+except ImportError:
+    import test_execution_history_hooks as fixtures
+
+
+def history(filename="a.png"):
+    return {
+        "outputs": {
+            "save": {
+                "images": [{"filename": filename, "subfolder": "", "type": "output"}]
+            }
+        }
+    }
+
+
+class OutputHooksTests(fixtures.unittest.TestCase):
+    setUp = fixtures.HistoryHooksTests.setUp
+    restore_logger = fixtures.HistoryHooksTests.restore_logger
+    cleanup_registrations = fixtures.HistoryHooksTests.cleanup_registrations
+    wait_for = fixtures.HistoryHooksTests.wait_for
+    dequeue = fixtures.HistoryHooksTests.dequeue
+    rows = fixtures.HistoryHooksTests.rows
+    drain = fixtures.HistoryHooksTests.drain
+
+    def register(self, roots_reader=None, pool=None):
+        if roots_reader is None:
+
+            def roots_reader():
+                return snapshot_output_roots("posix", "/output", "/temp")
+
+        if pool is None:
+            registration = HistoryRegistration(
+                self.queue, fixtures.JOB, lambda: self.db, roots_reader=roots_reader
+            )
+        else:
+            from utils.execution_history_writer import HistoryWriter
+
+            with patch(
+                "utils.execution_history.HistoryWriter",
+                lambda factory, **kw: HistoryWriter(factory, context_budget=pool, **kw),
+            ):
+                registration = HistoryRegistration(
+                    self.queue, fixtures.JOB, lambda: self.db, roots_reader=roots_reader
+                )
+        self.registrations.append(registration)
+        self.wait_for(lambda: registration.writer.status().accepting)
+        return registration
+
+    def outputs(self, item_id):
+        row = next(row for row in self.rows() if row["queue_item_id"] == item_id)
+        return self.db.get_execution_outputs(row["run_id"])
+
+    def test_positional_keyword_and_text_only_outputs_use_exact_fresh_attempt(self):
+        registration = self.register()
+        for i in range(3):
+            self.dequeue(i)
+            payload = history("a.png" if i < 2 else "b.png")
+            if i == 1:
+                self.queue.task_done(
+                    item_id=i,
+                    history_result=payload,
+                    status=fixtures.status("execution_error"),
+                )
+            else:
+                self.assertIs(
+                    self.queue.task_done(i, payload, fixtures.status()),
+                    self.queue.done_result,
+                )
+        self.drain(registration)
+        self.assertEqual(
+            [self.outputs(i)["references"][0]["filename"] for i in range(3)],
+            ["a.png", "a.png", "b.png"],
+        )
+        self.assertEqual(
+            [r["outcome"] for r in self.rows()], ["succeeded", "failed", "succeeded"]
+        )
+        self.assertEqual(registration.writer.context_budget.charged_bytes, 0)
+
+    def test_budget_denied_context_still_captures_independently_affordable_outputs(
+        self,
+    ):
+        pool = ContextBudget(2048)
+        registration = self.register(pool=pool)
+        self.dequeue(0)
+        self.queue.task_done(0, history(), fixtures.status())
+        self.drain(registration)
+        self.assertEqual(self.outputs(0)["state"], "complete")
+        self.assertEqual(
+            self.db.get_execution_contexts(self.rows()[0]["run_id"])["reason"],
+            "budget_exhausted",
+        )
+        self.assertEqual(pool.charged_bytes, 0)
+
+    def test_missing_history_and_explicit_none_are_distinct_from_empty_evidence(self):
+        registration = self.register()
+        for i in range(4):
+            self.dequeue(i)
+            if i == 0:
+                self.queue.task_done(item_id=i, status=fixtures.status())
+            else:
+                self.queue.task_done(
+                    i,
+                    [None, {"outputs": {}}, {"outputs": {"text": {"text": ["ok"]}}}][
+                        i - 1
+                    ],
+                    fixtures.status(),
+                )
+        self.drain(registration)
+        self.assertEqual(
+            [(self.outputs(i)["state"], self.outputs(i)["reason"]) for i in range(4)],
+            [
+                ("unavailable", "unavailable"),
+                ("unavailable", "invalid_evidence"),
+                ("complete", "none_reported"),
+                ("complete", "none_reported"),
+            ],
+        )
+
+    def test_replay_old_dequeue_reused_uuid_and_forged_capability_cannot_rebind(self):
+        registration = self.register()
+        old = self.dequeue(1)
+        self.queue.task_done(1, history("first.png"), fixtures.status())
+        self.dequeue(2)
+        registration._dequeue(old)
+        self.queue.task_done(1, history("replayed.png"), fixtures.status())
+        self.queue.task_done(2, history("second.png"), fixtures.status())
+        self.dequeue(3)
+        cap = registration._capabilities[3]
+        registration._capabilities[3] = replace(cap)
+        self.queue.task_done(3, history("forged.png"), fixtures.status())
+        registration.capture.revoke(cap)
+        self.drain(registration)
+        self.assertEqual(self.outputs(1)["references"][0]["filename"], "first.png")
+        self.assertEqual(self.outputs(2)["references"][0]["filename"], "second.png")
+        self.assertEqual(self.outputs(3)["state"], "not_captured")
+        self.assertEqual(registration.writer.context_budget.charged_bytes, 0)
+
+    def test_root_snapshot_mutation_and_reader_errors_preserve_outcome_and_reference(
+        self,
+    ):
+        roots = [snapshot_output_roots("posix", "/before", "/temp")]
+
+        def reader():
+            if isinstance(roots[0], Exception):
+                raise roots[0]
+            return roots[0]
+
+        registration = self.register(reader)
+        self.dequeue(0)
+        roots[0] = snapshot_output_roots("posix", "/after", "/temp")
+        self.queue.task_done(0, history(), fixtures.status())
+        self.dequeue(1)
+        roots[0] = RuntimeError("private")
+        self.queue.task_done(1, history(), fixtures.status())
+        self.dequeue(2)
+        self.queue.task_done(2, history(), fixtures.status())
+        self.drain(registration)
+        for i in range(3):
+            self.assertEqual(self.outputs(i)["reason"], "root_unavailable")
+            self.assertIsNone(self.outputs(i)["references"][0]["root_key"])
+        self.assertEqual([r["outcome"] for r in self.rows()], ["succeeded"] * 3)
+
+    def test_unexpected_output_producer_error_preserves_sealed_context_and_outcome(
+        self,
+    ):
+        registration = self.register()
+        graph = {
+            "p": {
+                "class_type": "PromptManagerPair",
+                "inputs": {"text": "raw", "negative_text": "neg"},
+            },
+            "s": {
+                "class_type": "KSampler",
+                "inputs": {"positive": ["p", 0], "negative": ["p", 1]},
+            },
+        }
+        self.queue.results.append(((0, fixtures.JOB, graph, {}, ["s"]), 0))
+        self.queue.get()
+        with patch(
+            "utils.execution_output_capture._extract",
+            side_effect=RuntimeError("private"),
+        ):
+            self.queue.task_done(0, history(), fixtures.status())
+        self.drain(registration)
+        self.assertEqual(self.rows()[0]["outcome"], "succeeded")
+        self.assertEqual(self.outputs(0)["reason"], "invalid_evidence")
+        self.assertEqual(
+            self.db.get_execution_contexts(self.rows()[0]["run_id"])["reason"],
+            "partial",
+        )
+        self.assertEqual(
+            self.db.get_execution_contexts(self.rows()[0]["run_id"])["contexts"][0][
+                "positive"
+            ]["raw"],
+            "raw",
+        )
+        self.assertEqual(registration.writer.context_budget.charged_bytes, 0)
+
+    def test_unexpected_context_seal_and_projection_errors_do_not_lose_output(self):
+        registration = self.register()
+        with patch(
+            "utils.execution_context_capture.project_contexts",
+            side_effect=RuntimeError("private"),
+        ):
+            self.dequeue(0)
+        self.queue.task_done(0, history(), fixtures.status())
+        self.dequeue(1)
+        with patch(
+            "utils.execution_context_capture.bundle_payload_size",
+            side_effect=RuntimeError("private"),
+        ):
+            self.queue.task_done(1, history(), fixtures.status())
+        self.drain(registration)
+        self.assertEqual([self.outputs(i)["state"] for i in range(2)], ["complete"] * 2)
+        self.assertEqual([r["outcome"] for r in self.rows()], ["succeeded"] * 2)
+        self.assertEqual(registration.writer.context_budget.charged_bytes, 0)
+
+    def test_original_failure_missing_callback_disposal_and_observer_coexistence(self):
+        calls = []
+        before_get, before_done = self.queue.get, self.queue.task_done
+        unsubscribe_get = subscribe_dequeue(
+            self.queue, "other", lambda result: calls.append(result[1])
+        )
+        unsubscribe_done = subscribe_task_done(
+            self.queue, "other", lambda args, kwargs: calls.append(args[0])
+        )
+        self.addCleanup(unsubscribe_get)
+        self.addCleanup(unsubscribe_done)
+        registration = self.register()
+        self.dequeue(0)
+        late = registration._task_done
+        self.assertTrue(registration.dispose())
+        thread = threading.Thread(
+            target=lambda: late((0, history(), fixtures.status()), {})
+        )
+        thread.start()
+        thread.join(2)
+        self.assertFalse(thread.is_alive())
+        self.queue.task_done(0, history(), fixtures.status())
+        self.assertEqual(calls, [0, 0])
+        self.assertEqual(self.outputs(0)["state"], "not_captured")
+        unsubscribe_get()
+        unsubscribe_done()
+        self.assertEqual(self.queue.get, before_get)
+        self.assertEqual(self.queue.task_done, before_done)
+        self.assertEqual(registration.writer.context_budget.charged_bytes, 0)
+
+    def test_rejected_invalid_revoked_or_unfresh_dequeue_never_reads_roots(self):
+        reads = []
+
+        def reader():
+            reads.append(1)
+            return snapshot_output_roots("posix", "/output", "/temp")
+
+        registration = self.register(reader)
+        self.dequeue(10)
+        self.dequeue(9)
+        self.queue.task_done(9, history(), fixtures.status())
+        self.queue.task_done(10, history(), fixtures.status())
+        self.dequeue(11, job="invalid")
+        self.queue.task_done(11, history(), fixtures.status())
+        self.drain(registration)
+        self.assertEqual(reads, [1])  # 10 was revoked by the next actual dequeue.
+        self.assertEqual(self.outputs(9)["state"], "not_captured")
+        self.assertEqual(self.outputs(10)["state"], "not_captured")
+        self.assertEqual(registration.writer.context_budget.charged_bytes, 0)
+
+    def test_original_failures_preserve_exception_and_leave_outputs_missing(
+        self,
+    ):
+        error = RuntimeError("host error")
+
+        def fail(*args, **kwargs):
+            raise error
+
+        original_get = self.queue.get
+        self.queue.get = fail
+        registration = self.register()
+        with self.assertRaises(RuntimeError) as raised:
+            self.queue.get()
+        self.assertIs(raised.exception, error)
+        self.assertEqual(self.rows(), [])
+        self.assertTrue(registration.dispose())
+        self.queue.get = original_get
+        self.queue.task_done = fail
+        registration = self.register()
+        self.dequeue(1)
+        with self.assertRaises(RuntimeError) as raised:
+            self.queue.task_done(1, history(), fixtures.status())
+        self.assertIs(raised.exception, error)
+        self.drain(registration)
+        self.assertEqual(self.rows()[0]["outcome"], "observed")
+        self.assertEqual(self.outputs(1)["state"], "not_captured")
+        self.assertTrue(registration.dispose())
+        self.assertEqual(registration.writer.context_budget.charged_bytes, 0)
+
+    def test_default_adapter_uses_configured_output_and_temp_namespaces(self):
+        native_root = Path(self.directory.name)
+        cases = (
+            (
+                os.name,
+                str(native_root / "output"),
+                str(native_root / "temp"),
+                "windows" if os.name == "nt" else "posix",
+            ),
+            ("posix", "/configured", "/configured", "posix"),
+            ("nt", r"D:\configured", r"D:\configured", "windows"),
+        )
+        for item, (platform, output_root, temp_root, style) in enumerate(cases):
+            folder_paths = SimpleNamespace(
+                get_output_directory=lambda: output_root,
+                get_temp_directory=lambda: temp_root,
+            )
+            with (
+                self.subTest(platform=platform, root=output_root),
+                patch.dict(sys.modules, {"folder_paths": folder_paths}),
+                patch("utils.execution_history.os", SimpleNamespace(name=platform)),
+            ):
+                registration = HistoryRegistration(
+                    self.queue, fixtures.JOB, lambda: self.db
+                )
+                self.registrations.append(registration)
+                self.wait_for(lambda: registration.writer.status().accepting)
+                self.dequeue(item)
+                payload = history()
+                payload["outputs"]["save"]["images"].append(
+                    {"filename": "a.png", "subfolder": "", "type": "temp"}
+                )
+                self.queue.task_done(item, payload, fixtures.status())
+                self.drain(registration)
+                evidence = self.outputs(item)
+                self.assertEqual(evidence["state"], "complete")
+                references = evidence["references"]
+                self.assertEqual(len(references[0]["root_key"]), 64)
+                self.assertEqual(len(references[1]["root_key"]), 64)
+                self.assertEqual([r["path_style"] for r in references], [style] * 2)
+                self.assertEqual(
+                    [r["folder_type"] for r in references], ["output", "temp"]
+                )
+                self.assertNotEqual(
+                    references[0]["root_key"], references[1]["root_key"]
+                )
+                self.assertEqual(registration._roots, {})
+                self.assertTrue(registration.dispose())
+
+    def test_sealed_context_is_discarded_before_output_reservation(self):
+        pool = ContextBudget(1048576)
+        registration = self.register(pool=pool)
+        self.dequeue(0)
+        self.assertEqual(pool.charged_bytes, 1048576)
+        self.queue.task_done(0, history(), fixtures.status())
+        self.drain(registration)
+        self.assertEqual(self.outputs(0)["state"], "complete")
+        self.assertEqual(pool.charged_bytes, 0)
+
+    def test_saturated_start_cannot_capture_outputs_or_retain_previous_roots(self):
+        entered, release = threading.Event(), threading.Event()
+        self.releases.append(release)
+        original = self.db.write_execution_start
+
+        def blocked(command):
+            entered.set()
+            release.wait(3)
+            return original(command)
+
+        with patch.object(self.db, "write_execution_start", blocked):
+            registration = self.register()
+            self.dequeue(0)
+            self.assertTrue(entered.wait(1))
+            from utils.execution_history_events import make_start
+
+            epoch = registration.writer.status().epoch
+            for item in range(1023):
+                command = make_start(
+                    registration._session_id, fixtures.item(1000 + item), epoch, 0
+                )
+                self.assertTrue(registration.writer.submit(command))
+            self.dequeue(1)
+            self.assertEqual(registration._roots, {})
+            self.queue.task_done(0, history("revoked.png"), fixtures.status())
+            self.queue.task_done(1, history("rejected.png"), fixtures.status())
+            registration.writer.invalidate()
+            release.set()
+            self.assertTrue(registration.dispose(2))
+        self.assertEqual(self.outputs(0)["state"], "not_captured")
+        self.assertEqual(len(self.rows()), 1)
+        self.assertEqual(registration.writer.context_budget.charged_bytes, 0)

tests/test_execution_output_schema.py

@@ -0,0 +1,439 @@
+"""Atomic output schema upgrades and bounded validation of untrusted evidence."""
+
+import sqlite3
+import unittest
+from unittest.mock import patch
+
+from database import execution_history_schema as history
+from database.execution_context_schema import install_context_schema
+
+try:
+    from tests.test_execution_context_schema import EPOCH, RUN, legacy
+except ImportError:
+    from test_execution_context_schema import EPOCH, RUN, legacy
+
+OUTPUT_TABLES = ("execution_output_capture", "execution_run_outputs")
+ROOT = "a" * 64
+
+
+def v2(conn):
+    """Manufacture the supported pre-output metadata CHECK and context schema."""
+    legacy(conn)
+    sql = (
+        conn.execute(
+            "SELECT sql FROM sqlite_master WHERE name='execution_history_metadata'"
+        )
+        .fetchone()[0]
+        .replace("schema_version = 1", "schema_version = 2")
+    )
+    conn.execute("DROP TABLE execution_history_metadata")
+    conn.execute(sql)
+    conn.execute("INSERT INTO execution_history_metadata VALUES(1,2,?)", (EPOCH,))
+    install_context_schema(conn)
+    conn.commit()
+
+
+def capture(conn, run=RUN, state="complete", reason="reported", rows=True):
+    conn.execute(
+        "INSERT INTO execution_output_capture VALUES(?,?,?,1,'host_final_history')",
+        (run, state, reason),
+    )
+    if rows:
+        conn.execute(
+            "INSERT INTO execution_run_outputs VALUES(?, '保存😀', 0, 'output', "
+            "'Batch Å', ' image 😀.png ', 'posix', ?)",
+            (run, ROOT),
+        )
+
+
+class OutputSchemaTests(unittest.TestCase):
+    def connection(self):
+        conn = sqlite3.connect(":memory:")
+        self.addCleanup(conn.close)
+        return conn
+
+    def populated(self):
+        conn = self.connection()
+        legacy(conn)
+        history.migrate_execution_history(conn)
+        capture(conn)
+        conn.commit()
+        return conn
+
+    def test_legacy_v1_v2_upgrade_preserves_rows_ddl_epoch_and_counters(self):
+        for fixture in (None, legacy, v2):
+            with self.subTest(fixture=fixture):
+                conn = self.connection()
+                if fixture:
+                    fixture(conn)
+                before = list(conn.iterdump())
+                ddl = dict(conn.execute("SELECT name,sql FROM sqlite_master"))
+                self.assertIsNone(history.history_schema_error(conn))
+                self.assertIsNotNone(
+                    history.history_schema_error(conn, allow_missing=False)
+                )
+                history.migrate_execution_history(conn)
+                version, epoch = conn.execute(
+                    "SELECT schema_version,storage_epoch FROM "
+                    "execution_history_metadata"
+                ).fetchone()
+                self.assertEqual(version, 3)
+                if fixture:
+                    self.assertEqual(epoch, EPOCH)
+                    self.assertEqual(
+                        conn.execute("SELECT * FROM prompts").fetchall(),
+                        [(7, "kept", 19)],
+                    )
+                    self.assertEqual(
+                        conn.execute(
+                            "SELECT run_id,storage_epoch FROM execution_runs"
+                        ).fetchall(),
+                        [(RUN, EPOCH)],
+                    )
+                    after = list(conn.iterdump())
+                    self.assertTrue(
+                        all(line in after for line in before if "metadata" not in line)
+                    )
+                    for name, sql in ddl.items():
+                        if name != "execution_history_metadata":
+                            self.assertEqual(
+                                conn.execute(
+                                    "SELECT sql FROM sqlite_master WHERE name=?",
+                                    (name,),
+                                ).fetchone()[0],
+                                sql,
+                            )
+                for table in OUTPUT_TABLES:
+                    self.assertEqual(
+                        conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0], 0
+                    )
+                self.assertIsNone(
+                    history.history_schema_error(conn, allow_missing=False)
+                )
+                snapshot = list(conn.iterdump())
+                history.migrate_execution_history(conn)
+                self.assertEqual(list(conn.iterdump()), snapshot)
+
+    def test_v2_context_ids_and_content_survive_upgrade(self):
+        conn = self.connection()
+        try:
+            from tests.test_execution_context_schema import ContextSchemaTests
+        except ImportError:
+            from test_execution_context_schema import ContextSchemaTests
+        ContextSchemaTests().populate(conn)
+        # Explicit supported v2: remove every output-owned table and rebuild metadata.
+        for table in reversed(OUTPUT_TABLES):
+            conn.execute(f"DROP TABLE IF EXISTS {table}")
+        sql = (
+            conn.execute(
+                "SELECT sql FROM sqlite_master WHERE name='execution_history_metadata'"
+            )
+            .fetchone()[0]
+            .replace("schema_version = 3", "schema_version = 2")
+        )
+        conn.execute("DROP TABLE execution_history_metadata")
+        conn.execute(sql)
+        conn.execute("INSERT INTO execution_history_metadata VALUES(1,2,?)", (EPOCH,))
+        from database.execution_context_schema import TABLE_COLUMNS
+
+        before = {
+            table: conn.execute(f"SELECT * FROM {table}").fetchall()
+            for table in TABLE_COLUMNS
+        }
+        history.migrate_execution_history(conn)
+        for table, rows in before.items():
+            self.assertEqual(conn.execute(f"SELECT * FROM {table}").fetchall(), rows)
+        self.assertIsNone(history.history_schema_error(conn, allow_missing=False))
+
+    def test_creation_and_final_validation_failures_roll_back_and_retry(self):
+        for fixture in (None, legacy, v2):
+            for failure in ("creation", "validation"):
+                with self.subTest(fixture=fixture, failure=failure):
+                    conn = self.connection()
+                    if fixture:
+                        fixture(conn)
+                    before = list(conn.iterdump())
+                    if failure == "creation":
+
+                        def fail(connection):
+                            connection.execute(
+                                "CREATE TABLE execution_output_capture(x)"
+                            )
+                            raise sqlite3.OperationalError(
+                                "injected output DDL failure"
+                            )
+
+                        replacement = patch.object(
+                            history, "install_output_schema", fail
+                        )
+                    else:
+                        original = history.history_schema_error
+
+                        def fail(connection, *, allow_missing=True):
+                            return (
+                                original(connection)
+                                if allow_missing
+                                else "injected final output validation failure"
+                            )
+
+                        replacement = patch.object(
+                            history, "history_schema_error", fail
+                        )
+                    with (
+                        replacement,
+                        self.assertRaisesRegex(sqlite3.DatabaseError, "injected"),
+                    ):
+                        history.migrate_execution_history(conn)
+                    self.assertEqual(list(conn.iterdump()), before)
+                    history.migrate_execution_history(conn)
+                    self.assertIsNone(
+                        history.history_schema_error(conn, allow_missing=False)
+                    )
+
+    def test_outputs_without_history_or_under_old_versions_are_rejected(self):
+        for fixture in (None, legacy, v2):
+            for table in OUTPUT_TABLES:
+                with self.subTest(fixture=fixture, table=table):
+                    conn = self.connection()
+                    if fixture:
+                        fixture(conn)
+                    conn.execute(f"CREATE TABLE {table}(x TEXT)")
+                    before = list(conn.iterdump())
+                    self.assertIsNotNone(history.history_schema_error(conn))
+                    with self.assertRaises(sqlite3.DatabaseError):
+                        history.migrate_execution_history(conn)
+                    self.assertEqual(list(conn.iterdump()), before)
+
+    def test_missing_output_table_rejected_in_both_validation_modes(self):
+        for table in OUTPUT_TABLES:
+            conn = self.populated()
+            conn.execute(f"DROP TABLE {table}")
+            for allow in (True, False):
+                self.assertIsNotNone(
+                    history.history_schema_error(conn, allow_missing=allow)
+                )
+
+    def test_exact_owned_indexes_checks_columns_foreign_keys_and_objects(self):
+        mutations = (
+            "CREATE UNIQUE INDEX extra ON execution_run_outputs(filename)",
+            "CREATE INDEX extra ON execution_run_outputs(filename)",
+            "CREATE TRIGGER extra AFTER INSERT ON execution_run_outputs "
+            "BEGIN SELECT 1; END",
+            "CREATE VIEW extra AS SELECT * FROM execution_run_outputs",
+        )
+        for sql in mutations:
+            with self.subTest(sql=sql):
+                conn = self.populated()
+                conn.execute(sql)
+                self.assertIsNotNone(history.history_schema_error(conn))
+        replacements = (
+            (
+                "execution_run_outputs",
+                "filename TEXT NOT NULL",
+                "filename TEXT NOT NULL UNIQUE",
+            ),
+            (
+                "execution_run_outputs",
+                "filename TEXT NOT NULL",
+                "filename TEXT NOT NULL COLLATE NOCASE",
+            ),
+            (
+                "execution_run_outputs",
+                "PRIMARY KEY (run_id, output_node_id, image_ordinal)",
+                "PRIMARY KEY (run_id, image_ordinal, output_node_id)",
+            ),
+            (
+                "execution_output_capture",
+                "format_version = 1",
+                "format_version IN (1, 2)",
+            ),
+            (
+                "execution_output_capture",
+                "REFERENCES execution_runs(run_id)",
+                "REFERENCES execution_runs(job_id)",
+            ),
+            ("execution_run_outputs", "ON DELETE NO ACTION", "ON DELETE CASCADE"),
+            ("execution_run_outputs", "ON UPDATE NO ACTION", "ON UPDATE CASCADE"),
+            (
+                "execution_run_outputs",
+                "REFERENCES execution_output_capture(run_id)",
+                "REFERENCES execution_runs(run_id)",
+            ),
+            ("execution_run_outputs", "root_key TEXT", "root_key BLOB"),
+        )
+        for table, old, new in replacements:
+            with self.subTest(table=table, replacement=new):
+                conn = self.populated()
+                sql = conn.execute(
+                    "SELECT sql FROM sqlite_master WHERE name=?", (table,)
+                ).fetchone()[0]
+                self.assertIn(old, sql)
+                conn.execute(f"DROP TABLE {table}")
+                conn.execute(sql.replace(old, new))
+                self.assertIsNotNone(history.history_schema_error(conn))
+
+    def test_orphan_run_header_and_rows_rejected_without_fk_enforcement(self):
+        for mutation in (
+            "DELETE FROM execution_runs",
+            "DELETE FROM execution_output_capture",
+            "UPDATE execution_run_outputs "
+            "SET run_id='33333333-3333-4333-8333-333333333333'",
+        ):
+            with self.subTest(mutation=mutation):
+                conn = self.populated()
+                conn.execute(mutation)
+                self.assertIsNotNone(history.history_schema_error(conn))
+
+    def test_malformed_scalars_rejected_before_text_or_row_fetch(self):
+        cases = (
+            ("execution_output_capture", "run_id", RUN + "\0"),
+            ("execution_output_capture", "state", "complete\0"),
+            ("execution_output_capture", "reason", "reported\0"),
+            ("execution_output_capture", "source", "host_final_history\0"),
+            ("execution_output_capture", "format_version", "x" * 4096),
+            ("execution_run_outputs", "run_id", RUN + "\0"),
+            ("execution_run_outputs", "output_node_id", "x" * 129),
+            ("execution_run_outputs", "output_node_id", "😀" * 129),
+            ("execution_run_outputs", "output_node_id", "n\0"),
+            ("execution_run_outputs", "image_ordinal", "x" * 4096),
+            ("execution_run_outputs", "folder_type", "output\0"),
+            ("execution_run_outputs", "subfolder", "x" * 2049),
+            ("execution_run_outputs", "subfolder", "a\0b"),
+            ("execution_run_outputs", "filename", "x" * 1025),
+            ("execution_run_outputs", "filename", "a\0b"),
+            ("execution_run_outputs", "path_style", "posix\0"),
+            ("execution_run_outputs", "root_key", ROOT + "\0"),
+            ("execution_run_outputs", "filename", b"\xff"),
+        )
+        for table, column, malicious in cases:
+            with self.subTest(table=table, column=column, value=repr(malicious)[:30]):
+                conn = self.populated()
+                conn.execute("PRAGMA ignore_check_constraints=ON")
+                conn.execute(f"UPDATE {table} SET {column}=?", (malicious,))
+
+                def bounded_text(value):
+                    self.assertNotEqual(
+                        value,
+                        malicious.encode() if type(malicious) is str else malicious,
+                    )
+                    return value.decode()
+
+                def bounded_row(cursor, row):
+                    self.assertFalse(any(value == malicious for value in row))
+                    return row
+
+                conn.text_factory = bounded_text
+                conn.row_factory = bounded_row
+                self.assertIsNotNone(history.history_schema_error(conn))
+
+    def test_state_version_source_and_locator_corruption_rejected(self):
+        cases = (
+            ("execution_output_capture", "source", "scan"),
+            ("execution_output_capture", "format_version", 2),
+            ("execution_output_capture", "state", "not_captured"),
+            ("execution_output_capture", "reason", "none_reported"),
+            ("execution_run_outputs", "filename", "../image.png"),
+            ("execution_run_outputs", "filename", ""),
+            ("execution_run_outputs", "subfolder", "a/../b"),
+            ("execution_run_outputs", "folder_type", "input"),
+            ("execution_run_outputs", "path_style", "native"),
+            ("execution_run_outputs", "image_ordinal", 256),
+            ("execution_run_outputs", "image_ordinal", -1),
+            ("execution_run_outputs", "root_key", None),
+        )
+        for table, column, value in cases:
+            with self.subTest(table=table, column=column, value=value):
+                conn = self.populated()
+                conn.execute("PRAGMA ignore_check_constraints=ON")
+                conn.execute(f"UPDATE {table} SET {column}=?", (value,))
+                self.assertIsNotNone(history.history_schema_error(conn))
+
+    def test_state_row_consistency_and_conflicting_first_rows(self):
+        cases = (
+            ("complete", "reported", False, False),
+            ("complete", "none_reported", False, True),
+            ("complete", "none_reported", True, False),
+            ("partial", "invalid_evidence", False, True),
+            ("partial", "root_unavailable", False, False),
+            ("partial", "root_unavailable", True, True),
+            ("unavailable", "unavailable", True, False),
+            ("unavailable", "invalid_evidence", False, True),
+            ("partial", "reported", True, False),
+            ("conflicting", "conflicting", True, True),
+        )
+        for state, reason, rows, valid in cases:
+            with self.subTest(state=state, reason=reason, rows=rows):
+                conn = self.connection()
+                legacy(conn)
+                history.migrate_execution_history(conn)
+                conn.execute("PRAGMA ignore_check_constraints=ON")
+                capture(conn, state=state, reason=reason, rows=rows)
+                self.assertEqual(history.history_schema_error(conn) is None, valid)
+
+    def test_bundle_count_and_exact_payload_bounds(self):
+        for count, filename, valid in (
+            (256, "a.png", True),
+            (257, "a.png", False),
+            (128, "x" * 1024, False),
+        ):
+            with self.subTest(count=count, filename_bytes=len(filename)):
+                conn = self.populated()
+                conn.execute("DELETE FROM execution_run_outputs")
+                conn.executemany(
+                    "INSERT INTO execution_run_outputs "
+                    "VALUES(?,?,0,'temp','',?,'posix',?)",
+                    ((RUN, str(index), filename, ROOT) for index in range(count)),
+                )
+                self.assertEqual(history.history_schema_error(conn) is None, valid)
+
+    def test_exact_128_kib_charge_is_valid_but_one_extra_byte_is_rejected(self):
+        conn = self.populated()
+        conn.execute("DELETE FROM execution_run_outputs")
+        # 128-byte bundle header plus 128 references charged 1023 bytes each.
+        # Each row: 128 overhead + 1 node + 4 folder + 821 name + 5 style + 64 root.
+        conn.executemany(
+            "INSERT INTO execution_run_outputs VALUES(?,'n',?,'temp','',?,'posix',?)",
+            ((RUN, ordinal, "x" * 821, ROOT) for ordinal in range(128)),
+        )
+        self.assertIsNone(history.history_schema_error(conn))
+        conn.execute(
+            "UPDATE execution_run_outputs SET filename=? WHERE image_ordinal=0",
+            ("x" * 822,),
+        )
+        self.assertIsNotNone(history.history_schema_error(conn))
+
+    def test_oversized_bundle_is_rejected_before_fetching_its_scalar_payload(self):
+        conn = self.populated()
+        conn.execute("DELETE FROM execution_run_outputs")
+        filename = "x" * 1024
+        conn.executemany(
+            "INSERT INTO execution_run_outputs VALUES(?,?,0,'temp','',?,'posix',?)",
+            ((RUN, str(index), filename, ROOT) for index in range(128)),
+        )
+
+        def bounded_text(value):
+            self.assertNotEqual(value, filename.encode())
+            return value.decode()
+
+        conn.text_factory = bounded_text
+        self.assertIsNotNone(history.history_schema_error(conn))
+
+    def test_duplicate_locators_and_unicode_survive_validation_unchanged(self):
+        conn = self.populated()
+        conn.execute(
+            "INSERT INTO execution_run_outputs SELECT run_id,output_node_id,"
+            "1,folder_type,subfolder,filename,path_style,root_key "
+            "FROM execution_run_outputs"
+        )
+        conn.execute(
+            "INSERT INTO execution_run_outputs SELECT run_id,'other',"
+            "image_ordinal,folder_type,subfolder,filename,path_style,root_key "
+            "FROM execution_run_outputs"
+        )
+        before = conn.execute("SELECT * FROM execution_run_outputs").fetchall()
+        self.assertEqual(len(before), 4)
+        self.assertIsNone(history.history_schema_error(conn, allow_missing=False))
+        history.migrate_execution_history(conn)
+        self.assertEqual(
+            conn.execute("SELECT * FROM execution_run_outputs").fetchall(), before
+        )

tests/test_execution_output_writer.py

@@ -0,0 +1,659 @@
+"""Real-thread combined payload ownership and independently fenced persistence."""
+
+from contextlib import nullcontext
+from dataclasses import replace
+from copy import copy
+import gc
+import os
+import sqlite3
+import threading
+import time
+import weakref
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from utils.execution_context_budget import ContextBudget, PayloadLease, SealedContext
+from utils.execution_contexts import MAX_ATTEMPT_BYTES
+from utils.execution_output_capture import seal_outputs
+from utils.execution_outputs import snapshot_output_roots
+from utils.execution_history_writer import HistoryWriter
+
+try:
+    from tests import test_execution_history_writer as fixtures
+except ImportError:
+    import test_execution_history_writer as fixtures
+
+
+class OutputWriterTests(fixtures.unittest.TestCase):
+    setUp = fixtures.HistoryWriterTests.setUp
+    cleanup_workers = fixtures.HistoryWriterTests.cleanup_workers
+    wait_for = fixtures.HistoryWriterTests.wait_for
+    writer = fixtures.HistoryWriterTests.writer
+    start_command = fixtures.HistoryWriterTests.start_command
+    finish_command = fixtures.HistoryWriterTests.finish_command
+    block = fixtures.HistoryWriterTests.block
+    sealed = fixtures.HistoryWriterTests.sealed
+
+    def outputs(self, writer, context_bytes=0):
+        roots = snapshot_output_roots("posix", "/output", "/temp")
+        return seal_outputs(
+            {
+                "outputs": {
+                    "save": {
+                        "images": [
+                            {"filename": "a.png", "subfolder": "", "type": "output"}
+                        ]
+                    }
+                }
+            },
+            roots,
+            roots,
+            budget=writer.context_budget,
+            capture_generation=writer.status().capture_generation,
+            context_bytes=context_bytes,
+        )
+
+    def finalization(self, writer, start):
+        context = self.sealed(writer)
+        outputs = self.outputs(writer, context.lease.byte_count)
+        self.assertTrue(
+            writer.submit_finalization(self.finish_command(start), context, outputs)
+        )
+        return context, outputs
+
+    def assert_idle_bundles_collectible(
+        self,
+        method: str | None = None,
+        error_type: type[Exception] = ValueError,
+        context_state: str = "complete",
+        output_state: str = "complete",
+        accepting: bool = True,
+    ) -> None:
+        writer = HistoryWriter(
+            lambda: self.db, clock=self.clock, context_budget=ContextBudget()
+        )
+        self.writers.append(writer)
+        writer.start()
+        self.wait_for(lambda: writer.status().accepting)
+        start = self.start_command(len(self.writers))
+        self.db.write_execution_start(start)
+        context = self.sealed(writer, "x" * 30000)
+        roots = snapshot_output_roots("posix", "/output", "/temp")
+        image = dict(filename="x" * 1024, subfolder="x" * 2048, type="output")
+        tail = dict(filename="x", subfolder="x" * 1470, type="output")
+        outputs = seal_outputs(
+            {"outputs": {"😀" * 128: {"images": [image] * 34 + [tail]}}},
+            roots,
+            roots,
+            budget=writer.context_budget,
+            capture_generation=writer.status().capture_generation,
+            context_bytes=context.lease.byte_count,
+        )
+        self.assertEqual(outputs.lease.byte_count, 131072)
+        references = (weakref.ref(context.bundle), weakref.ref(outputs.bundle))
+
+        def reject(_command: object, _bundle: object, **_kwargs: object) -> None:
+            raise error_type("supplement rejected")
+
+        write_patch = patch.object(self.db, method, reject) if method else nullcontext()
+        try:
+            with write_patch:
+                self.assertTrue(
+                    writer.submit_finalization(
+                        self.finish_command(start), context, outputs
+                    )
+                )
+                del context, outputs
+                self.wait_for(lambda: writer.status().pending == 0)
+                gc.collect()
+                self.assertTrue(writer.status().alive)
+                self.assertEqual(writer.status().pending, 0)
+                self.assertEqual(writer.context_budget.charged_bytes, 0)
+                self.assertFalse(
+                    any(reference() is not None for reference in references),
+                    "idle worker retained a bundle after releasing its payload lease",
+                )
+            self.assertEqual(writer.status().accepting, accepting)
+            self.assertEqual(
+                self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
+            )
+            self.assertEqual(
+                self.db.get_execution_contexts(start.run_id)["state"], context_state
+            )
+            self.assertEqual(
+                self.db.get_execution_outputs(start.run_id)["state"], output_state
+            )
+        finally:
+            self.assertTrue(writer.close(2))
+            self.assertFalse(writer.status().alive)
+
+    def test_successful_finalization_releases_bundles_while_worker_is_idle(
+        self,
+    ) -> None:
+        self.assert_idle_bundles_collectible()
+
+    def test_validation_omission_releases_bundles_while_worker_is_idle(self) -> None:
+        for method, context_state, output_state in (
+            ("write_execution_contexts_for_finish", "not_captured", "complete"),
+            ("write_execution_outputs_for_finish", "complete", "not_captured"),
+        ):
+            with self.subTest(method=method):
+                self.assert_idle_bundles_collectible(
+                    method, context_state=context_state, output_state=output_state
+                )
+
+    def test_early_storage_failure_releases_bundles_while_worker_is_idle(self) -> None:
+        self.assert_idle_bundles_collectible(
+            "write_execution_contexts_for_finish",
+            sqlite3.OperationalError,
+            context_state="not_captured",
+            output_state="not_captured",
+            accepting=False,
+        )
+
+    def test_fifo_outcome_context_outputs_commit_and_custom_parent_identity(self):
+        writer = self.writer()
+        start = replace(self.start_command(), run_id=str(fixtures.uuid4()))
+        self.assertTrue(writer.submit(start))
+        self.finalization(writer, start)
+        self.assertTrue(writer.close(2))
+        self.assertEqual(
+            self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
+        )
+        self.assertEqual(
+            self.db.get_execution_contexts(start.run_id)["state"], "complete"
+        )
+        self.assertEqual(
+            self.db.get_execution_outputs(start.run_id)["references"][0]["filename"],
+            "a.png",
+        )
+        self.assertEqual(writer.context_budget.charged_bytes, 0)
+
+    def test_supplement_validation_is_independent_and_storage_failure_invalidates(
+        self,
+    ):
+        for method, error, context_state, output_state, accepting in [
+            (
+                "write_execution_contexts_for_finish",
+                ValueError("invalid"),
+                "not_captured",
+                "complete",
+                True,
+            ),
+            (
+                "write_execution_outputs_for_finish",
+                ValueError("invalid"),
+                "complete",
+                "not_captured",
+                True,
+            ),
+            (
+                "write_execution_outputs_for_finish",
+                sqlite3.OperationalError("private"),
+                "complete",
+                "not_captured",
+                False,
+            ),
+        ]:
+            with self.subTest(method=method, error=type(error)):
+                writer = self.writer()
+                start = self.start_command(len(self.writers))
+                self.db.write_execution_start(start)
+                with patch.object(self.db, method, side_effect=error):
+                    self.finalization(writer, start)
+                    self.wait_for(lambda: writer.status().pending == 0)
+                self.assertEqual(
+                    self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
+                )
+                self.assertEqual(
+                    self.db.get_execution_contexts(start.run_id)["state"], context_state
+                )
+                self.assertEqual(
+                    self.db.get_execution_outputs(start.run_id)["state"], output_state
+                )
+                self.assertEqual(writer.status().accepting, accepting)
+                self.assertEqual(writer.context_budget.charged_bytes, 0)
+                self.assertTrue(writer.close())
+
+    def test_bad_output_owners_are_omitted_without_losing_context(self):
+        writer = self.writer()
+        foreign = ContextBudget(10000)
+        cases = [
+            "inactive",
+            "foreign",
+            "forged",
+            "generation",
+            "missing_generation",
+            "header",
+            "missing_lease",
+            "oversized_lease",
+        ]
+        for item_id, case in enumerate(cases, 1):
+            with self.subTest(case=case):
+                start = self.start_command(item_id)
+                self.db.write_execution_start(start)
+                context = self.sealed(writer)
+                outputs = self.outputs(writer)
+                if case == "inactive":
+                    outputs.lease.release()
+                elif case == "foreign":
+                    outputs.lease.release()
+                    outputs = replace(outputs, lease=foreign.reserve(10000))
+                elif case == "forged":
+                    outputs.lease.release()
+                    outputs = replace(outputs, lease=object.__new__(PayloadLease))
+                elif case == "generation":
+                    outputs = replace(outputs, capture_generation=99)
+                elif case == "missing_generation":
+                    outputs = replace(outputs, capture_generation=None)
+                elif case == "header":
+                    outputs = replace(outputs, reason="private" * 100)
+                elif case == "oversized_lease":
+                    outputs.lease.release()
+                    outputs = replace(
+                        outputs, lease=writer.context_budget.reserve(131073)
+                    )
+                else:
+                    outputs.lease.release()
+                    outputs = replace(outputs, lease=None)
+                self.assertTrue(
+                    writer.submit_finalization(
+                        self.finish_command(start), context, outputs
+                    )
+                )
+                self.wait_for(lambda: writer.status().pending == 0)
+                self.assertEqual(
+                    self.db.get_execution_contexts(start.run_id)["state"], "complete"
+                )
+                self.assertEqual(
+                    self.db.get_execution_outputs(start.run_id)["reason"],
+                    "invalid_evidence",
+                )
+                self.assertEqual(writer.context_budget.charged_bytes, 0)
+                self.assertEqual(foreign.charged_bytes, 0)
+
+    def test_combined_payload_limit_omits_outputs_and_releases_both_owners(self):
+        writer = self.writer()
+        start = self.start_command()
+        self.db.write_execution_start(start)
+        context = self.sealed(writer)
+        context.lease.release()
+        context = replace(
+            context, lease=writer.context_budget.reserve(MAX_ATTEMPT_BYTES)
+        )
+        outputs = self.outputs(writer)
+        self.assertTrue(
+            writer.submit_finalization(self.finish_command(start), context, outputs)
+        )
+        self.wait_for(lambda: writer.status().pending == 0)
+        self.assertEqual(
+            self.db.get_execution_contexts(start.run_id)["state"], "complete"
+        )
+        self.assertEqual(
+            self.db.get_execution_outputs(start.run_id)["reason"], "budget_exhausted"
+        )
+        self.assertEqual(writer.context_budget.charged_bytes, 0)
+
+    def test_restore_or_generation_rotation_between_transactions_discards_evidence(
+        self,
+    ):
+        for after in ("write_execution_finish", "write_execution_contexts_for_finish"):
+            for restore in (False, True):
+                with self.subTest(after=after, restore=restore):
+                    writer = self.writer()
+                    start = self.start_command(len(self.writers))
+                    self.db.write_execution_start(start)
+                    backup = os.path.join(self.directory.name, "restore.db")
+                    self.assertTrue(self.db.model.backup_database(backup))
+                    original = getattr(self.db, after)
+
+                    def write(*args, **kwargs):
+                        result = original(*args, **kwargs)
+                        if restore:
+                            self.db.model.restore_from_file(backup)
+                        else:
+                            writer.invalidate()
+                        return result
+
+                    with patch.object(self.db, after, write):
+                        self.finalization(writer, start)
+                        self.wait_for(lambda: writer.status().pending == 0)
+                    self.assertEqual(
+                        self.db.get_execution_outputs(start.run_id)["state"],
+                        "not_captured",
+                    )
+                    self.assertEqual(
+                        self.db.get_execution_contexts(start.run_id)["state"],
+                        (
+                            "complete"
+                            if after == "write_execution_contexts_for_finish"
+                            and not restore
+                            else "not_captured"
+                        ),
+                    )
+                    self.assertEqual(writer.context_budget.charged_bytes, 0)
+                    self.assertTrue(writer.close())
+
+    def test_invalidation_during_output_preparation_rechecks_actual_commit(self):
+        import database.execution_outputs as outputs_db
+
+        writer = self.writer()
+        start = self.start_command()
+        self.db.write_execution_start(start)
+        entered, release = self.block()
+        original = outputs_db.validate_output_bundle
+
+        def prepare(bundle):
+            value = original(bundle)
+            entered.set()
+            release.wait(3)
+            return value
+
+        with patch.object(outputs_db, "validate_output_bundle", prepare):
+            self.finalization(writer, start)
+            self.assertTrue(entered.wait(1))
+            writer.invalidate()
+            release.set()
+            self.wait_for(lambda: writer.status().pending == 0)
+        self.assertEqual(
+            self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
+        )
+        self.assertEqual(
+            self.db.get_execution_contexts(start.run_id)["state"], "complete"
+        )
+        self.assertEqual(
+            self.db.get_execution_outputs(start.run_id)["state"], "not_captured"
+        )
+        self.assertEqual(writer.context_budget.charged_bytes, 0)
+
+    def test_output_lock_failure_drops_backlog_and_never_retries_after_recovery(self):
+        writer = self.writer()
+        start, queued = self.start_command(), self.start_command(2)
+        self.db.write_execution_start(start)
+        entered, release = self.block()
+        original = self.db.write_execution_outputs_for_finish
+
+        def write(*args, **kwargs):
+            entered.set()
+            release.wait(3)
+            return original(*args, **kwargs)
+
+        lock = sqlite3.connect(self.db.model.db_path)
+        self.addCleanup(lock.close)
+        with patch.object(self.db, "write_execution_outputs_for_finish", write):
+            self.finalization(writer, start)
+            self.assertTrue(entered.wait(1))
+            lock.execute("BEGIN IMMEDIATE")
+            self.assertTrue(writer.submit(queued))
+            self.finalization(writer, queued)
+            release.set()
+            self.wait_for(lambda: not writer.status().accepting)
+        lock.rollback()
+        self.assertEqual(writer.context_budget.charged_bytes, 0)
+        self.clock.advance(5)
+        self.wait_for(lambda: writer.status().accepting)
+        self.assertIsNone(self.db.get_execution_run(queued.run_id))
+        self.assertEqual(
+            self.db.get_execution_outputs(start.run_id)["state"], "not_captured"
+        )
+        fresh = self.start_command(3)
+        self.assertTrue(writer.submit(fresh))
+        self.finalization(writer, fresh)
+        self.wait_for(lambda: writer.status().pending == 0)
+        self.assertEqual(
+            self.db.get_execution_outputs(fresh.run_id)["state"], "complete"
+        )
+        self.assertEqual(writer.context_budget.charged_bytes, 0)
+
+    def test_bounded_close_and_early_native_join_keep_inflight_combined_charge(self):
+        for early in (False, True):
+            with self.subTest(early_join=early):
+                writer = self.writer()
+                start = self.start_command(len(self.writers))
+                self.db.write_execution_start(start)
+                entered, release = self.block()
+                original = self.db.write_execution_outputs_for_finish
+
+                def write(*args, **kwargs):
+                    entered.set()
+                    release.wait(3)
+                    return original(*args, **kwargs)
+
+                with patch.object(self.db, "write_execution_outputs_for_finish", write):
+                    self.finalization(writer, start)
+                    self.assertTrue(entered.wait(1))
+                    charge = writer.context_budget.charged_bytes
+                    self.finalization(writer, start)
+                    clock = fixtures.ManualClock()
+                    native_join = writer._thread.join
+
+                    def return_early(timeout):
+                        native_join(0)
+                        clock.advance(min(timeout, 0.004))
+
+                    def sleep(delay):
+                        clock.advance(delay)
+
+                    clock_patch = (
+                        patch(
+                            "utils.execution_history_writer.time",
+                            SimpleNamespace(monotonic=clock, sleep=sleep),
+                        )
+                        if early
+                        else nullcontext()
+                    )
+                    join_patch = (
+                        patch.object(writer._thread, "join", return_early)
+                        if early
+                        else nullcontext()
+                    )
+                    with clock_patch, join_patch:
+                        before = clock() if early else time.monotonic()
+                        self.assertFalse(writer.close(0.02))
+                        elapsed = (clock() if early else time.monotonic()) - before
+                    self.assertGreaterEqual(elapsed, 0.02)
+                    self.assertLess(elapsed, 0.2)
+                    self.assertEqual(writer.context_budget.charged_bytes, charge)
+                    replacement = HistoryWriter(lambda: self.db)
+                    self.writers.append(replacement)
+                    self.assertIs(replacement.context_budget, writer.context_budget)
+                    self.assertIsNone(replacement.context_budget.reserve(8388608))
+                    release.set()
+                    self.assertTrue(writer.close(2))
+                self.assertEqual(writer.context_budget.charged_bytes, 0)
+
+    def test_pending_saturation_and_invalid_command_release_both_payloads(self):
+        writer = self.writer()
+        start = self.start_command()
+        context, outputs = self.sealed(writer), self.outputs(writer)
+        self.assertFalse(writer.submit_finalization(object(), context, outputs))
+        self.assertEqual(writer.context_budget.charged_bytes, 0)
+        entered, release = self.block()
+        original = self.db.write_execution_start
+
+        def write(command):
+            entered.set()
+            release.wait(3)
+            return original(command)
+
+        with patch.object(self.db, "write_execution_start", write):
+            self.assertTrue(writer.submit(start))
+            self.assertTrue(entered.wait(1))
+            for _ in range(1023):
+                context = SealedContext(None, None, "unavailable", "budget_exhausted")
+                self.assertTrue(
+                    writer.submit_finalization(
+                        self.finish_command(start), context, self.outputs(writer)
+                    )
+                )
+            self.assertEqual(writer.status().pending, 1024)
+            self.assertFalse(
+                writer.submit_finalization(
+                    self.finish_command(start),
+                    self.sealed(writer),
+                    self.outputs(writer),
+                )
+            )
+            writer.invalidate()
+            self.assertEqual(writer.context_budget.charged_bytes, 0)
+            release.set()
+        self.assertTrue(writer.close(2))
+
+    def test_malformed_context_omission_preserves_independent_valid_output(self):
+        writer = self.writer()
+        start = self.start_command()
+        self.db.write_execution_start(start)
+        context = self.sealed(writer)
+        context = replace(context, capture_generation=None)
+        self.assertTrue(
+            writer.submit_finalization(
+                self.finish_command(start), context, self.outputs(writer)
+            )
+        )
+        self.wait_for(lambda: writer.status().pending == 0)
+        self.assertEqual(
+            self.db.get_execution_contexts(start.run_id)["reason"], "invalid_evidence"
+        )
+        self.assertEqual(
+            self.db.get_execution_outputs(start.run_id)["state"], "complete"
+        )
+        self.assertEqual(writer.context_budget.charged_bytes, 0)
+
+    def test_generation_mismatch_after_same_epoch_recovery_cannot_write_old_outputs(
+        self,
+    ):
+        writer = self.writer()
+        start = self.start_command()
+        context, outputs = self.sealed(writer), self.outputs(writer)
+        writer.invalidate()
+        self.clock.advance(5)
+        self.wait_for(lambda: writer.status().accepting)
+        self.assertEqual(writer.status().epoch, start.storage_epoch)
+        self.assertTrue(writer.submit(start))
+        self.assertTrue(
+            writer.submit_finalization(self.finish_command(start), context, outputs)
+        )
+        self.wait_for(lambda: writer.status().pending == 0)
+        self.assertEqual(
+            self.db.get_execution_outputs(start.run_id)["reason"], "invalid_evidence"
+        )
+        self.assertEqual(
+            self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
+        )
+        self.assertEqual(writer.context_budget.charged_bytes, 0)
+
+    def test_output_commit_serializes_invalidation_without_blocking_close_or_queue(
+        self,
+    ):
+        import database.execution_outputs as outputs_db
+
+        writer = self.writer()
+        start = self.start_command()
+        self.db.write_execution_start(start)
+        entered, release = self.block()
+        invalidating, invalidated = threading.Event(), threading.Event()
+        original = outputs_db._capture
+
+        def capture(*args, **kwargs):
+            entered.set()
+            release.wait(3)
+            return original(*args, **kwargs)
+
+        def invalidate():
+            invalidating.set()
+            writer.invalidate()
+            invalidated.set()
+
+        with patch.object(outputs_db, "_capture", capture):
+            self.finalization(writer, start)
+            self.assertTrue(entered.wait(1))
+            invalidator = threading.Thread(target=invalidate)
+            invalidator.start()
+            try:
+                self.assertTrue(invalidating.wait(1))
+                self.assertFalse(invalidated.wait(0.03))
+                before = time.monotonic()
+                self.assertTrue(writer.submit(self.finish_command(start)))
+                self.assertFalse(writer.close(0))
+                self.assertLess(time.monotonic() - before, 0.1)
+                self.assertGreater(writer.context_budget.charged_bytes, 0)
+            finally:
+                release.set()
+                invalidator.join(3)
+            self.assertFalse(invalidator.is_alive())
+            self.assertTrue(writer.close(2))
+        self.assertEqual(
+            self.db.get_execution_outputs(start.run_id)["state"], "complete"
+        )
+        self.assertEqual(writer.context_budget.charged_bytes, 0)
+
+    def test_copied_payload_owner_is_rejected_without_spending_genuine_reservation(
+        self,
+    ):
+        writer = self.writer()
+        for item, copy_context in ((1, False), (2, True)):
+            start = self.start_command(item)
+            self.db.write_execution_start(start)
+            context, outputs = self.sealed(writer), self.outputs(writer)
+            genuine = context.lease if copy_context else outputs.lease
+            self.addCleanup(genuine.release)
+            charge = genuine.byte_count
+            if copy_context:
+                context = replace(context, lease=copy(genuine))
+            else:
+                outputs = replace(outputs, lease=copy(genuine))
+            self.assertTrue(
+                writer.submit_finalization(self.finish_command(start), context, outputs)
+            )
+            self.wait_for(lambda: writer.status().pending == 0)
+            self.assertEqual(writer.context_budget.charged_bytes, charge)
+            self.assertEqual(genuine.byte_count, charge)
+            self.assertEqual(
+                self.db.get_execution_outputs(start.run_id)["reason"],
+                "reported" if copy_context else "invalid_evidence",
+            )
+            genuine.release()
+            self.assertEqual(writer.context_budget.charged_bytes, 0)
+
+    def test_unexpected_validation_failure_keeps_other_evidence_and_outcome(self):
+        writer = self.writer()
+        for item, fail_context in ((1, False), (2, True)):
+            start = self.start_command(item)
+            self.db.write_execution_start(start)
+            context, outputs = self.sealed(writer), self.outputs(writer)
+            validator = "bundle_payload_size" if fail_context else "output_payload_size"
+            with patch(
+                "utils.execution_history_writer." + validator,
+                side_effect=RuntimeError("private"),
+            ):
+                self.assertTrue(
+                    writer.submit_finalization(
+                        self.finish_command(start), context, outputs
+                    )
+                )
+            self.wait_for(lambda: writer.status().pending == 0)
+            self.assertEqual(
+                self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
+            )
+            self.assertEqual(
+                self.db.get_execution_outputs(start.run_id)["reason"],
+                "reported" if fail_context else "invalid_evidence",
+            )
+            self.assertEqual(
+                self.db.get_execution_contexts(start.run_id)["reason"],
+                "invalid_evidence" if fail_context else "captured",
+            )
+            self.assertEqual(writer.context_budget.charged_bytes, 0)
+
+    def test_queue_failure_after_owner_transfer_releases_both_payloads(self):
+        writer = self.writer()
+        start = self.start_command()
+        self.db.write_execution_start(start)
+        context, outputs = self.sealed(writer), self.outputs(writer)
+        with patch.object(writer._queue, "put_nowait", side_effect=MemoryError):
+            self.assertFalse(
+                writer.submit_finalization(self.finish_command(start), context, outputs)
+            )
+        self.assertEqual(writer.context_budget.charged_bytes, 0)
+        self.assertEqual(writer.status().pending, 0)
+        self.assertEqual(self.db.get_execution_run(start.run_id)["outcome"], "observed")

tests/test_execution_outputs.py

@@ -0,0 +1,371 @@
+"""Exact, bounded output locators carry no authority to access files."""
+
+import hashlib
+import unittest
+from dataclasses import FrozenInstanceError, replace
+from unittest.mock import patch
+
+from test_execution_contexts import load_module
+
+
+class OutputContractTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.m = load_module("execution_outputs")
+
+    def reference(self, **changes) -> object:
+        reference = self.m.OutputReference(
+            "7", 0, "output", "", "same.png", "posix", "a" * 64
+        )
+        return replace(reference, **changes)
+
+    def test_exact_unicode_and_duplicate_locators_survive(self) -> None:
+        m = self.m
+        a = self.reference()
+        b = replace(a, image_ordinal=1)
+        bundle = m.OutputBundle((a, b), "complete", "reported")
+        self.assertIsNone(m.validate_output_bundle(bundle))
+        self.assertEqual(m.output_payload_size(bundle), 552)
+        self.assertEqual(
+            m.output_payload_size(bundle),
+            128
+            + sum(
+                128
+                + sum(
+                    load_module("execution_contexts").text_size(v, 4096)
+                    for v in (
+                        r.output_node_id,
+                        r.folder_type,
+                        r.subfolder,
+                        r.filename,
+                        r.path_style,
+                        r.root_key or "",
+                    )
+                )
+                for r in bundle.references
+            ),
+        )
+        variants = ("é.png", "e\u0301.png", "É.png", " é.png", "é.png ")
+        references = tuple(
+            self.reference(image_ordinal=i, filename=name, subfolder=" 雪 / Café ")
+            for i, name in enumerate(variants)
+        )
+        exact = m.OutputBundle(references, "complete", "reported")
+        self.assertIsNone(m.validate_output_bundle(exact))
+        self.assertEqual(tuple(r.filename for r in exact.references), variants)
+        self.assertEqual(exact.references[0].subfolder, " 雪 / Café ")
+        self.assertEqual(
+            m.output_payload_size(m.OutputBundle((), "complete", "none_reported")), 128
+        )
+        with self.assertRaises(FrozenInstanceError):
+            a.filename = "changed.png"
+
+    def test_posix_backslash_is_literal(self) -> None:
+        m = self.m
+        reference = self.reference(subfolder=r"a\b/雪", filename=r"a\b:c.png")
+        self.assertIsNone(m.validate_output_reference(reference))
+        self.assertEqual(reference.filename, r"a\b:c.png")
+        for subfolder in (
+            "/absolute",
+            "../escape",
+            "a/../b",
+            "./a",
+            "a//b",
+            "a/",
+            ".",
+            "..",
+        ):
+            with self.subTest(subfolder=subfolder), self.assertRaises(ValueError):
+                m.validate_output_reference(replace(reference, subfolder=subfolder))
+        for filename in ("", "/absolute", "a/b", ".", "..", "a\x00b", "\ud800"):
+            with self.subTest(filename=repr(filename)), self.assertRaises(ValueError):
+                m.validate_output_reference(replace(reference, filename=filename))
+
+    def test_windows_drive_unc_traversal_ads_rejected(self) -> None:
+        m = self.m
+        reference = self.reference(
+            path_style="windows", subfolder=r"a\雪/b", filename=" image.png "
+        )
+        self.assertIsNone(m.validate_output_reference(reference))
+        for subfolder in (
+            r"C:\absolute",
+            "C:relative",
+            r"\\server\share",
+            r"\rooted",
+            "/rooted",
+            "../escape",
+            r"a\..\b",
+            r"a\.\b",
+            r"a\\b",
+            "a//b",
+            r"a:stream",
+            "a/",
+            "a\\",
+            ".",
+            "..",
+        ):
+            with self.subTest(subfolder=subfolder), self.assertRaises(ValueError):
+                m.validate_output_reference(replace(reference, subfolder=subfolder))
+        for filename in (r"a\b.png", "a/b.png", "C:foo.png", "a.png:stream", ".", ".."):
+            with self.subTest(filename=filename), self.assertRaises(ValueError):
+                m.validate_output_reference(replace(reference, filename=filename))
+
+    def test_root_namespace_is_versioned_style_and_folder_specific(self) -> None:
+        m = self.m
+        roots = m.snapshot_output_roots("posix", "/a/../雪/root/", "/雪/root")
+        self.assertEqual(roots.path_style, "posix")
+        self.assertEqual(
+            roots.output_key,
+            hashlib.sha256('[1,"posix","output","/雪/root"]'.encode()).hexdigest(),
+        )
+        self.assertEqual(
+            roots.temp_key,
+            hashlib.sha256('[1,"posix","temp","/雪/root"]'.encode()).hexdigest(),
+        )
+        self.assertNotEqual(roots.output_key, roots.temp_key)
+        self.assertNotEqual(
+            roots.output_key,
+            m.snapshot_output_roots("posix", "/雪/Root", None).output_key,
+        )
+        windows = m.snapshot_output_roots(
+            "windows", r"C:\a\..\雪\root", r"\\server\share\a\.."
+        )
+        self.assertEqual(
+            windows.output_key,
+            hashlib.sha256(
+                r'[1,"windows","output","C:\\雪\\root"]'.encode()
+            ).hexdigest(),
+        )
+        self.assertIsNotNone(windows.temp_key)
+        invalid = (
+            None,
+            True,
+            b"/tmp",
+            "relative",
+            "",
+            "/a\x00b",
+            "/\ud800",
+            "/" + "x" * 8192,
+            "/" + "é" * 4096,
+        )
+        for root in invalid:
+            with self.subTest(root_type=type(root).__name__):
+                snapshot = m.snapshot_output_roots("posix", root, "/tmp")
+                self.assertIsNone(snapshot.output_key)
+                self.assertIsNotNone(snapshot.temp_key)
+        self.assertIsNotNone(
+            m.snapshot_output_roots("posix", "/" + "x" * 8191, None).output_key
+        )
+        self.assertIsNotNone(
+            m.snapshot_output_roots("posix", "/" + "é" * 4095 + "x", None).output_key
+        )
+        for root in (
+            r"\\server",
+            "\\\\server\\",
+            r"\\server\.",
+            r"1:\root",
+            r"C:relative",
+            r"\rooted",
+            "/rooted",
+            r"\\?\C:\root",
+            r"\\.\C:\root",
+        ):
+            with self.subTest(root=root):
+                self.assertIsNone(
+                    m.snapshot_output_roots("windows", root, None).output_key
+                )
+        for style in ("linux", "POSIX", None, True):
+            with self.assertRaises(ValueError):
+                m.snapshot_output_roots(style, "/tmp", "/tmp")
+
+    def test_bounds_and_state_row_consistency(self) -> None:
+        m = self.m
+        a = self.reference()
+        valid = {
+            "complete": ("reported", "none_reported"),
+            "partial": (
+                "invalid_evidence",
+                "unsupported",
+                "limit_reached",
+                "root_unavailable",
+                "budget_exhausted",
+            ),
+            "unavailable": ("unavailable", "invalid_evidence", "budget_exhausted"),
+            "conflicting": ("conflicting",),
+        }
+        reasons = (
+            "reported",
+            "none_reported",
+            "invalid_evidence",
+            "unsupported",
+            "limit_reached",
+            "root_unavailable",
+            "budget_exhausted",
+            "unavailable",
+            "conflicting",
+            "unknown",
+            "",
+        )
+        for state in (*valid, "not_captured", "unknown"):
+            for reason in reasons:
+                for references in ((), (a,)):
+                    accepted = reason in valid.get(state, ())
+                    if state == "complete":
+                        accepted = accepted and bool(references) == (
+                            reason == "reported"
+                        )
+                    if state == "unavailable":
+                        accepted = accepted and not references
+                    if reason == "root_unavailable":
+                        accepted = accepted and bool(references)
+                    bundle = m.OutputBundle(references, state, reason)
+                    with self.subTest(state=state, reason=reason, rows=len(references)):
+                        if accepted:
+                            self.assertIsNone(m.validate_output_bundle(bundle))
+                        else:
+                            with self.assertRaises(ValueError):
+                                m.validate_output_bundle(bundle)
+        null_root = replace(a, root_key=None)
+        for state, reason in (
+            ("partial", "root_unavailable"),
+            ("conflicting", "conflicting"),
+        ):
+            m.validate_output_bundle(m.OutputBundle((null_root,), state, reason))
+        with self.assertRaises(ValueError):
+            m.validate_output_bundle(
+                m.OutputBundle((null_root,), "complete", "reported")
+            )
+        for ordinal in (0, 255):
+            m.validate_output_reference(replace(a, image_ordinal=ordinal))
+        for changes in (
+            {"image_ordinal": -1},
+            {"image_ordinal": 256},
+            {"image_ordinal": True},
+            {"image_ordinal": 1.0},
+            {"output_node_id": ""},
+            {"output_node_id": "x" * 129},
+            {"output_node_id": "\ud800"},
+            {"output_node_id": "a\x00b"},
+            {"filename": "é" * 513},
+            {"subfolder": "雪" * 683},
+            {"root_key": "A" * 64},
+            {"root_key": "a" * 63},
+            {"root_key": "z" * 64},
+            {"folder_type": "input"},
+            {"path_style": "native"},
+        ):
+            with self.subTest(changes=changes), self.assertRaises(ValueError):
+                m.validate_output_reference(replace(a, **changes))
+        ordered = (
+            replace(a, output_node_id="10"),
+            replace(a, output_node_id="2"),
+            a,
+            replace(a, image_ordinal=1),
+        )
+        m.validate_output_bundle(m.OutputBundle(ordered, "complete", "reported"))
+        for references in (
+            (a, a),
+            tuple(reversed(ordered)),
+            tuple(replace(a, output_node_id=str(i).zfill(3)) for i in range(257)),
+        ):
+            with self.assertRaises(ValueError):
+                m.validate_output_bundle(
+                    m.OutputBundle(references, "complete", "reported")
+                )
+        maximum = replace(
+            a,
+            output_node_id="😀" * 128,
+            filename="x" * 1024,
+            subfolder="x" * 2048,
+            path_style="windows",
+        )
+        self.assertIsNone(m.validate_output_reference(maximum))
+        self.assertEqual(
+            m.output_payload_size(m.OutputBundle((maximum,), "complete", "reported")),
+            3917,
+        )
+        prefix = tuple(replace(maximum, image_ordinal=i) for i in range(34))
+        tail = replace(maximum, image_ordinal=34, filename="x", subfolder="x" * 1400)
+        exact = m.OutputBundle((*prefix, tail), "complete", "reported")
+        self.assertEqual(m.output_payload_size(exact), 131072)
+        with self.assertRaises(ValueError):
+            m.output_payload_size(
+                replace(
+                    exact, references=(*prefix, replace(tail, subfolder="x" * 1401))
+                )
+            )
+        for changes in (
+            {"format_version": True},
+            {"format_version": 2},
+            {"source": "host_cache"},
+            {"references": [a]},
+        ):
+            with self.subTest(changes=changes), self.assertRaises(ValueError):
+                m.validate_output_bundle(
+                    replace(m.OutputBundle((a,), "complete", "reported"), **changes)
+                )
+
+    def test_no_coercion_or_filesystem_calls(self) -> None:
+        m = self.m
+
+        class Hostile:
+            def __str__(self):
+                raise AssertionError("coerced arbitrary object")
+
+            def __fspath__(self):
+                raise AssertionError("read arbitrary path protocol")
+
+            def __iter__(self):
+                raise AssertionError("iterated arbitrary object")
+
+        class Text(str):
+            pass
+
+        class Sequence(tuple):
+            pass
+
+        hostile = Hostile()
+        for field in (
+            "output_node_id",
+            "folder_type",
+            "subfolder",
+            "filename",
+            "path_style",
+            "root_key",
+        ):
+            for value in (hostile, Text("output")):
+                with self.subTest(field=field), self.assertRaises(ValueError):
+                    m.validate_output_reference(self.reference(**{field: value}))
+        for value in (hostile, Sequence((self.reference(),))):
+            with self.assertRaises(ValueError):
+                m.validate_output_bundle(m.OutputBundle(value, "complete", "reported"))
+        for field in ("state", "reason", "source"):
+            with self.subTest(field=field), self.assertRaises(ValueError):
+                m.validate_output_bundle(
+                    replace(
+                        m.OutputBundle((), "complete", "none_reported"),
+                        **{field: hostile},
+                    )
+                )
+        for value in (hostile, Text("/tmp")):
+            self.assertIsNone(m.snapshot_output_roots("posix", value, None).output_key)
+        with (
+            patch("builtins.open", side_effect=AssertionError("filesystem open")),
+            patch("os.stat", side_effect=AssertionError("filesystem stat")),
+            patch("os.lstat", side_effect=AssertionError("filesystem lstat")),
+            patch(
+                "os.path.realpath", side_effect=AssertionError("filesystem resolution")
+            ),
+        ):
+            roots = m.snapshot_output_roots(
+                "posix", "/nonexistent/output", "/nonexistent/temp"
+            )
+            reference = self.reference(root_key=roots.output_key)
+            self.assertIsNone(
+                m.validate_output_bundle(
+                    m.OutputBundle((reference,), "complete", "reported")
+                )
+            )
+            self.assertEqual(
+                m.output_payload_size(m.OutputBundle((), "unavailable", "unavailable")),
+                128,
+            )

tests/test_history_waits.py

@@ -0,0 +1,90 @@
+"""Batch drain detects stalls without imposing a SQLite throughput target."""
+
+from dataclasses import replace
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+from tests import test_execution_history_hooks as hooks
+from utils.execution_history_writer import HistoryWriterStatus
+
+
+class StepClock:
+    def __init__(self, step: float):
+        self.value = 0.0
+        self.step = step
+
+    def __call__(self) -> float:
+        self.value += self.step
+        return self.value
+
+
+class ObservedWriter:
+    def __init__(self, pending: int, *, progress: bool = False, alive: bool = True):
+        self.pending = pending
+        self.progress = progress
+        self.alive = alive
+        self.context_budget = SimpleNamespace(charged_bytes=281)
+
+    def status(self) -> HistoryWriterStatus:
+        status = HistoryWriterStatus("epoch", True, self.pending, 0, self.alive, 0)
+        if self.progress and self.pending:
+            self.pending -= 1
+        return status
+
+
+class HistoryDrainTests(unittest.TestCase):
+    wait_for = hooks.HistoryHooksTests.wait_for
+
+    def drain(self, writer: ObservedWriter, clock: StepClock) -> None:
+        with patch.object(hooks, "time", SimpleNamespace(monotonic=clock)):
+            hooks.HistoryHooksTests.drain(self, SimpleNamespace(writer=writer))
+
+    def test_batch_with_progress_can_exceed_short_stall_deadline(self):
+        writer = ObservedWriter(128, progress=True)
+        clock = StepClock(0.04)
+        self.drain(writer, clock)
+        self.assertEqual(writer.pending, 0)
+        self.assertGreater(clock.value, 3)
+        self.assertLess(clock.value, 30)
+
+    def test_no_progress_fails_with_worker_and_lease_diagnostics(self):
+        writer = ObservedWriter(52)
+        clock = StepClock(0.5)
+        with self.assertRaisesRegex(
+            AssertionError, r"worker stalled.*pending=52.*charged_bytes=281"
+        ):
+            self.drain(writer, clock)
+        self.assertLessEqual(clock.value, 4)
+
+    def test_progress_cannot_extend_batch_past_hard_ceiling(self):
+        writer = ObservedWriter(1000, progress=True)
+        clock = StepClock(0.5)
+        with self.assertRaisesRegex(AssertionError, "batch deadline"):
+            self.drain(writer, clock)
+        self.assertGreater(writer.pending, 0)
+        self.assertGreaterEqual(clock.value, 30)
+        self.assertLessEqual(clock.value, 31)
+
+    def test_dead_worker_with_pending_work_fails_immediately(self):
+        writer = ObservedWriter(1, alive=False)
+        clock = StepClock(0.5)
+        with self.assertRaisesRegex(AssertionError, r"worker stopped.*alive=False"):
+            self.drain(writer, clock)
+        self.assertLess(clock.value, 3)
+
+    def test_oscillating_pending_is_not_new_progress(self):
+        writer = ObservedWriter(1)
+        original = writer.status
+        calls = 0
+
+        def oscillating() -> HistoryWriterStatus:
+            nonlocal calls
+            calls += 1
+            return replace(original(), pending=1 + calls % 2)
+
+        writer.status = oscillating
+        clock = StepClock(0.5)
+        with self.assertRaisesRegex(AssertionError, "worker stalled"):
+            self.drain(writer, clock)
+        self.assertLessEqual(clock.value, 5)

tests/vendor/comfyui-config.test.mjs

@@ -24,12 +24,15 @@ test('execution script discovers only the new matrix and preserves all 78 stock
     const stock = list('tools/vendor/comfyui.config.mjs');
     assert.match(stock, /Total: 78 tests in 7 files/);
     assert.doesNotMatch(stock, /execution-history/);
+    assert.doesNotMatch(stock, /execution-outputs/);
     const executions = list('tools/vendor/playwright.execution.config.mjs');
     assert.match(executions, /\[classic-start\]/);
     assert.match(executions, /\[nodes2-start\]/);
-    assert.match(executions, /Total: 26 tests in 1 file/);
+    assert.match(executions, /Total: 42 tests in 2 files/);
     for (const project of ['classic-start', 'nodes2-start']) {
-        assert.equal(executions.split('\n').filter(line => line.includes(`[${project}]`)).length, 13);
+        const cases = executions.split('\n').filter(line => line.includes(`[${project}]`));
+        assert.equal(cases.filter(line => line.includes('execution-history.spec.mjs')).length, 13);
+        assert.equal(cases.filter(line => line.includes('execution-outputs.spec.mjs')).length, 8);
     }
     assert.doesNotMatch(executions, /prompt-manager.spec/);
 });

tests/vendor/comfyui-host.test.mjs

@@ -97,6 +97,7 @@ test('read-only ledger helper distinguishes absent schema from missing, corrupt
 from database.operations import PromptDatabase
 from utils.execution_history_events import RunStart, make_finish
 from utils.execution_contexts import ContextBundle, PromptSnapshot, RecipeContext, TextEvidence
+from utils.execution_outputs import OutputBundle, OutputReference
 from types import SimpleNamespace
 
 db = PromptDatabase(sys.argv[1])
@@ -123,6 +124,10 @@ try:
     db.write_execution_contexts(sys.argv[5], epoch, ContextBundle(
         (RecipeContext('4', snapshot(' positive '), snapshot('negative'), 'inferred'),),
         'complete', 'captured'))
+    db.write_execution_outputs_for_finish(make_finish(sys.argv[3], 1, epoch,
+        SimpleNamespace(messages=[('execution_success', {'prompt_id': sys.argv[2]})]), 200),
+        OutputBundle((OutputReference('9', 0, 'temp', 'nested', 'preview.png', 'posix', 'a' * 64),),
+                     'complete', 'reported'))
     print(epoch)
 finally:
     db.close_all()
@@ -138,6 +143,12 @@ finally:
         node_id: null, node_type: null, exception_type: null,
     }]);
     assert.deepEqual(Object.keys(result.captures), [runId]);
+    assert.deepEqual(result.outputs, { [runId]: {
+        state: 'complete', reason: 'reported', format_version: 1, source: 'host_final_history',
+        references: [{ output_node_id: '9', image_ordinal: 0, folder_type: 'temp',
+            subfolder: 'nested', filename: 'preview.png', path_style: 'posix', root_key: 'a'.repeat(64),
+            file_identity: 'unverified', newly_generated: 'unknown', context_membership: 'unknown', transient: true }],
+    } });
     const capture = result.captures[runId];
     assert.equal(capture.contexts.length, 1);
     const contextId = capture.contexts[0].context_id;
@@ -170,8 +181,11 @@ finally:
     });
     assert.deepEqual(unfiltered.usage[uncapturedId], []);
     assert.deepEqual(unfiltered.snapshotIds[uncapturedId], []);
+    assert.deepEqual(unfiltered.outputs[uncapturedId], {
+        state: 'not_captured', reason: null, format_version: null, source: null, references: [],
+    });
     assert.deepEqual(await readExecutionRuns(database, job, '60000000-0000-4000-8000-000000000001'), {
-        schemaAvailable: true, rows: [], captures: {}, usage: {}, snapshotIds: {},
+        schemaAvailable: true, rows: [], captures: {}, usage: {}, snapshotIds: {}, outputs: {},
     });
     assert.deepEqual(await readFile(database), before);
     await assert.rejects(readExecutionRuns(database, "' OR 1=1 --"));

tools/vendor/comfyui.config.mjs

@@ -14,7 +14,7 @@ const servers = [{
 
 export default defineConfig({
     testDir: '../../tests/comfyui',
-    testIgnore: 'execution-history.spec.mjs',
+    testIgnore: ['execution-history.spec.mjs', 'execution-outputs.spec.mjs'],
     outputDir: './test-results/comfyui',
     timeout: 60000,
     workers: 1,

tools/vendor/playwright.execution.config.mjs

@@ -6,7 +6,7 @@ if (host.href !== 'http://127.0.0.1:8196/') throw new Error('Execution tests req
 
 export default defineConfig({
     testDir: '../../tests/comfyui',
-    testMatch: 'execution-history.spec.mjs',
+    testMatch: ['execution-history.spec.mjs', 'execution-outputs.spec.mjs'],
     outputDir: './test-results/executions/cases',
     timeout: 60000,
     workers: 1,

utils/execution_context_budget.py

@@ -17,6 +17,7 @@ class ContextBudget:
         _byte_count(limit_bytes)
         self._limit = limit_bytes
         self._charged = 0
+        self._owners: set[PayloadLease] = set()
         self._lock = threading.Lock()
 
     @property
@@ -29,8 +30,22 @@ class ContextBudget:
         with self._lock:
             if byte_count > self._limit - self._charged:
                 return None
+            owner = PayloadLease._new(self, byte_count)
+            self._owners.add(owner)
             self._charged += byte_count
-            return PayloadLease._new(self, byte_count)
+            return owner
+
+    def reserve_up_to(self, max_bytes: int) -> "PayloadLease | None":
+        """Atomically own up to the remaining capacity in this same pool."""
+        _byte_count(max_bytes)
+        with self._lock:
+            byte_count = min(max_bytes, self._limit - self._charged)
+            if not byte_count:
+                return None
+            owner = PayloadLease._new(self, byte_count)
+            self._owners.add(owner)
+            self._charged += byte_count
+            return owner
 
 
 class PayloadLease:
@@ -52,13 +67,15 @@ class PayloadLease:
             return self._bytes
 
     def _require_owner(self) -> None:
-        if not self._active:
-            raise ValueError("Inactive payload lease")
+        if self not in self._pool._owners or not self._active:
+            raise ValueError("Inactive or unissued payload lease")
 
     def transfer(self) -> "PayloadLease":
         with self._pool._lock:
             self._require_owner()
             owner = PayloadLease._new(self._pool, self._bytes)
+            self._pool._owners.add(owner)
+            self._pool._owners.remove(self)
             self._active = False
             return owner
 
@@ -73,7 +90,8 @@ class PayloadLease:
 
     def release(self) -> None:
         with self._pool._lock:
-            if self._active:
+            if self in self._pool._owners:
+                self._pool._owners.remove(self)
                 self._pool._charged -= self._bytes
                 self._active = False
 

utils/execution_context_capture.py

@@ -187,6 +187,12 @@ class ContextCapture:
                     projection = GraphProjection((), (), "")
                     reason = "graph_limit"
                     retained = 0
+                except Exception:
+                    projection = GraphProjection((), (), "")
+                    reason = "invalid_evidence"
+                    retained = 0
+                    lease.release()
+                    lease = None
             cap = replace(cap, sources=projection.sources)
             self._attempts[cap.nonce] = _Attempt(
                 cap, projection, lease, {}, retained, reason
@@ -396,6 +402,10 @@ class ContextCapture:
                 self._discard_payload(attempt)
                 lease.release()
                 return SealedContext(None, None, "unavailable", "limit_reached")
+            except Exception:
+                self._discard_payload(attempt)
+                lease.release()
+                return SealedContext(None, None, "unavailable", "invalid_evidence")
 
 
 def record_runtime_context(

utils/execution_history.py

@@ -2,6 +2,7 @@
 
 from collections import OrderedDict
 import atexit
+import os
 import threading
 import time
 from typing import Callable
@@ -13,7 +14,15 @@ except ImportError:
     from database import PromptDatabase
 
 from .execution_history_events import make_finish, make_start
-from .execution_context_capture import ContextCapture, active_capture, unbind_current
+from .execution_context_capture import (
+    AttemptCapability,
+    ContextCapture,
+    active_capture,
+    unbind_current,
+)
+from .execution_context_budget import SealedContext
+from .execution_output_capture import MISSING_HISTORY, SealedOutputs, seal_outputs
+from .execution_outputs import OutputRoots, snapshot_output_roots
 from .execution_context_host import observe_cached_context, subscribe_executor_messages
 from .execution_history_writer import HistoryWriter
 from .logging_config import get_logger
@@ -25,6 +34,16 @@ _registration_lock = threading.Lock()
 _MAX_ACTIVE = 128
 
 
+def _configured_output_roots() -> OutputRoots:
+    import folder_paths
+
+    return snapshot_output_roots(
+        "windows" if os.name == "nt" else "posix",
+        folder_paths.get_output_directory(),
+        folder_paths.get_temp_directory(),
+    )
+
+
 class HistoryRegistration:
     """Own subscriptions, one asynchronous writer and bounded attempt identities."""
 
@@ -35,6 +54,7 @@ class HistoryRegistration:
         db_factory: Callable[[], PromptDatabase],
         *,
         high_water: int = -1,
+        roots_reader: Callable[[], OutputRoots] | None = None,
     ) -> None:
         self.writer = HistoryWriter(
             db_factory, on_invalidation=self._invalidate_capture
@@ -43,6 +63,8 @@ class HistoryRegistration:
             id(queue), self.writer.status, self.writer.context_budget
         )
         self._capabilities = {}
+        self._roots = {}
+        self._roots_reader = roots_reader or _configured_output_roots
         self._accepted = OrderedDict()
         self._high_water = high_water
         self._session_id = session_id
@@ -82,7 +104,35 @@ class HistoryRegistration:
         observe_cached_context(executor, event, data)
 
     def _invalidate_capture(self) -> None:
-        self.capture.revoke_all()
+        with self._lock:
+            self._active.clear()
+            self._capabilities.clear()
+            self._roots.clear()
+            self.capture.revoke_all()
+
+    def _read_roots(self, path_style: str | None = None) -> OutputRoots:
+        try:
+            roots = self._roots_reader()
+            if (
+                type(roots) is not OutputRoots
+                or type(roots.path_style) is not str
+                or roots.path_style not in ("posix", "windows")
+            ):
+                raise ValueError("Invalid output roots")
+            for key in (roots.output_key, roots.temp_key):
+                if key is not None and (
+                    type(key) is not str
+                    or len(key) != 64
+                    or any(char not in "0123456789abcdef" for char in key)
+                ):
+                    raise ValueError("Invalid output root key")
+            return roots
+        except Exception:
+            return OutputRoots(
+                path_style or ("windows" if os.name == "nt" else "posix"),
+                None,
+                None,
+            )
 
     def _ready_epoch(self) -> str | None:
         status = self.writer.status()
@@ -92,6 +142,7 @@ class HistoryRegistration:
         ):
             self._active.clear()
             self._capabilities.clear()
+            self._roots.clear()
             self.capture.revoke_all()
         self._capture_generation = status.capture_generation
         return epoch
@@ -117,7 +168,12 @@ class HistoryRegistration:
                     command.storage_epoch,
                 ):
                     return
+                binding = active_capture()
                 unbind_current()
+                if binding is not None and binding[0] is self.capture:
+                    previous = binding[1]
+                    self._roots.pop(previous.queue_item_id, None)
+                    self._capabilities.pop(previous.queue_item_id, None)
                 if self._disposed:
                     return
                 if command is None or command.queue_item_id in self._active:
@@ -144,13 +200,18 @@ class HistoryRegistration:
                     cap = self.capture.begin(
                         command, self.writer.status().capture_generation, graph, outputs
                     )
-                    self._capabilities[command.queue_item_id] = cap
+                    if self.capture.valid(cap):
+                        self._capabilities[command.queue_item_id] = cap
+                        self._roots[command.queue_item_id] = (cap, self._read_roots())
         except Exception:
             unbind_current()
             self.writer.record_gap()
 
     def _task_done(self, args: tuple, kwargs: dict) -> None:
         sealed = None
+        outputs = None
+        cap = None
+        roots_owner = None
         try:
             with self._lock:
                 if self._disposed:
@@ -160,11 +221,21 @@ class HistoryRegistration:
                     self.writer.record_gap()
                     return
                 item_id = args[0] if args else kwargs.get("item_id")
+                owned = False
                 if type(item_id) is int:
                     cap = self._capabilities.pop(item_id, None)
-                    self._active.pop(item_id, None)
-                    if cap is not None:
-                        sealed = self.capture.seal(cap)
+                    roots_owner = self._roots.pop(item_id, None)
+                    active_epoch = self._active.pop(item_id, None)
+                    owned = (
+                        type(cap) is AttemptCapability
+                        and roots_owner is not None
+                        and roots_owner[0] is cap
+                        and active_epoch == epoch
+                        and cap.session_id == self._session_id
+                        and cap.queue_item_id == item_id
+                        and cap.storage_epoch == epoch
+                        and self.capture.valid(cap)
+                    )
                 status = args[2] if len(args) > 2 else kwargs.get("status")
                 command = make_finish(
                     self._session_id,
@@ -176,16 +247,54 @@ class HistoryRegistration:
                 if command is None:
                     self.writer.record_gap()
                     return
-                if sealed is not None:
-                    self.writer.submit_finalization(command, sealed)
-                    sealed = None
+                if owned:
+                    # Capture authority is consumed once; the local identity remains
+                    # trusted after sealing discards the recorder and its samples.
+                    try:
+                        sealed = self.capture.seal(cap)
+                    except Exception:
+                        sealed = SealedContext(
+                            None, None, "unavailable", "invalid_evidence"
+                        )
+                    finally:
+                        self.capture.revoke(cap)
+                    try:
+                        context_bytes = sealed.lease.byte_count if sealed.lease else 0
+                        history = (
+                            args[1]
+                            if len(args) > 1
+                            else kwargs.get("history_result", MISSING_HISTORY)
+                        )
+                        outputs = seal_outputs(
+                            history,
+                            roots_owner[1],
+                            self._read_roots(roots_owner[1].path_style),
+                            budget=self.writer.context_budget,
+                            capture_generation=cap.capture_generation,
+                            context_bytes=context_bytes,
+                        )
+                    except Exception:
+                        outputs = SealedOutputs(
+                            None,
+                            None,
+                            "unavailable",
+                            "invalid_evidence",
+                            cap.capture_generation,
+                        )
+                    self.writer.submit_finalization(command, sealed, outputs)
+                    sealed = outputs = None
                 else:
                     self.writer.submit(command)
         except Exception:
             self.writer.record_gap()
         finally:
-            if sealed is not None and sealed.lease is not None:
-                sealed.lease.release()
+            if type(cap) is AttemptCapability:
+                self.capture.revoke(cap)
+            if roots_owner is not None:
+                self.capture.revoke(roots_owner[0])
+            for payload in (sealed, outputs):
+                if payload is not None and payload.lease is not None:
+                    payload.lease.release()
 
     def dispose(self, timeout: float = 2.0) -> bool:
         """Stop capture, release subscriptions and drain the owned writer boundedly."""
@@ -194,6 +303,7 @@ class HistoryRegistration:
                 self._disposed = True
                 self._active.clear()
                 self._capabilities.clear()
+                self._roots.clear()
                 self.capture.revoke_all()
                 for unsubscribe in self._unsubscribers:
                     unsubscribe()

utils/execution_history_writer.py

@@ -3,6 +3,7 @@
 from contextlib import contextmanager
 from dataclasses import dataclass
 from functools import partial
+import json
 import math
 import queue
 import threading
@@ -28,7 +29,9 @@ from .execution_context_budget import (
     PayloadLease,
     SealedContext,
 )
-from .execution_contexts import ContextBundle, bundle_payload_size
+from .execution_contexts import MAX_ATTEMPT_BYTES, ContextBundle, bundle_payload_size
+from .execution_output_capture import SealedOutputs
+from .execution_outputs import MAX_BUNDLE_BYTES, OutputBundle, output_payload_size
 from .execution_history_events import RunCommand, RunFinish, RunStart, encode_command
 from .logging_config import get_logger
 
@@ -52,6 +55,7 @@ class HistoryWriterStatus:
 class _Finalization:
     command: RunFinish
     context: SealedContext
+    outputs: SealedOutputs | None = None
 
 
 class HistoryWriter:
@@ -120,11 +124,18 @@ class HistoryWriter:
         return accepted
 
     @staticmethod
-    def _release_context(context: SealedContext) -> None:
-        if isinstance(context, SealedContext) and type(context.lease) is PayloadLease:
-            context.lease.release()
+    def _release_payload(payload: SealedContext | SealedOutputs | None) -> None:
+        if type(payload) in (SealedContext, SealedOutputs):
+            lease = payload.lease
+            if type(lease) is PayloadLease:
+                try:
+                    lease.release()
+                except (AttributeError, TypeError, ValueError):
+                    # Invalid leases may lack reservations; keep independent evidence.
+                    pass
 
     def _consume_context_locked(self, context: SealedContext) -> SealedContext:
+        owner = None
         try:
             if type(context) is not SealedContext:
                 raise ValueError("Invalid sealed context")
@@ -139,7 +150,7 @@ class HistoryWriter:
                 or context.capture_generation is None
                 or type(context.lease) is not PayloadLease
                 or context.lease._pool is not self.context_budget
-                or context.lease.byte_count < size
+                or not size <= context.lease.byte_count <= MAX_ATTEMPT_BYTES
                 or context.state != context.bundle.state
                 or context.reason != context.bundle.reason
             ):
@@ -152,17 +163,66 @@ class HistoryWriter:
                 context.reason,
                 context.capture_generation,
             )
-        except ValueError:
-            self._release_context(context)
+        except Exception:
+            if owner is not None:
+                owner.release()
+            self._release_payload(context)
             return SealedContext(None, None, "unavailable", "invalid_evidence")
 
-    def submit_finalization(self, command: RunFinish, context: SealedContext) -> bool:
+    def _consume_outputs_locked(
+        self, outputs: SealedOutputs, context_bytes: int
+    ) -> SealedOutputs:
+        reason = "invalid_evidence"
+        owner = None
+        try:
+            if type(outputs) is not SealedOutputs:
+                raise ValueError("Invalid sealed outputs")
+            outputs.validate_header()
+            if outputs.bundle is None:
+                if outputs.lease is not None:
+                    raise ValueError("Unexpected output lease")
+                return outputs
+            size = output_payload_size(outputs.bundle)
+            if (
+                outputs.capture_generation != self._generation
+                or outputs.capture_generation is None
+                or type(outputs.lease) is not PayloadLease
+                or outputs.lease._pool is not self.context_budget
+                or not size <= outputs.lease.byte_count <= MAX_BUNDLE_BYTES
+                or outputs.state != outputs.bundle.state
+                or outputs.reason != outputs.bundle.reason
+            ):
+                raise ValueError("Invalid output ownership or generation")
+            if outputs.lease.byte_count + context_bytes > MAX_ATTEMPT_BYTES:
+                reason = "budget_exhausted"
+                raise ValueError("Combined payload limit")
+            owner = outputs.lease.transfer()
+            return SealedOutputs(
+                outputs.bundle,
+                owner,
+                outputs.state,
+                outputs.reason,
+                outputs.capture_generation,
+            )
+        except Exception:
+            if owner is not None:
+                owner.release()
+            self._release_payload(outputs)
+            return SealedOutputs(None, None, "unavailable", reason)
+
+    def submit_finalization(
+        self,
+        command: RunFinish,
+        context: SealedContext,
+        outputs: SealedOutputs | None = None,
+    ) -> bool:
         try:
             if type(command) is not RunFinish:
                 raise ValueError("Finalization requires finish")
             encode_command(command)
         except ValueError:
-            self._release_context(context)
+            self._release_payload(context)
+            self._release_payload(outputs)
             self.record_gap()
             return False
         with self._lock:
@@ -171,12 +231,48 @@ class HistoryWriter:
                 or command.storage_epoch != self._epoch
                 or self._pending >= _MAX_PENDING
             ):
-                self._release_context(context)
+                self._release_payload(context)
+                self._release_payload(outputs)
                 self._record_gap_locked()
                 accepted = False
             else:
                 owned = self._consume_context_locked(context)
-                self._queue.put_nowait(_Finalization(command, owned))
+                owned_outputs = None
+                try:
+                    owned_outputs = (
+                        self._consume_outputs_locked(
+                            outputs, owned.lease.byte_count if owned.lease else 0
+                        )
+                        if outputs is not None
+                        else None
+                    )
+                    header = json.dumps(
+                        [
+                            [payload.state, payload.reason, payload.capture_generation]
+                            for payload in (owned, owned_outputs)
+                            if payload is not None
+                        ],
+                        separators=(",", ":"),
+                    ).encode("utf-8")
+                    if len(header) > 256:
+                        self._release_payload(owned)
+                        self._release_payload(owned_outputs)
+                        owned = SealedContext(
+                            None, None, "unavailable", "invalid_evidence"
+                        )
+                        owned_outputs = (
+                            SealedOutputs(None, None, "unavailable", "invalid_evidence")
+                            if outputs is not None
+                            else None
+                        )
+                    self._queue.put_nowait(_Finalization(command, owned, owned_outputs))
+                except Exception:
+                    self._release_payload(owned)
+                    self._release_payload(owned_outputs)
+                    self._release_payload(outputs)
+                    self._record_gap_locked()
+                    self._wake.set()
+                    return False
                 self._pending += 1
                 accepted = True
         self._wake.set()
@@ -268,7 +364,8 @@ class HistoryWriter:
             except queue.Empty:
                 return
             if isinstance(item, _Finalization):
-                self._release_context(item.context)
+                self._release_payload(item.context)
+                self._release_payload(item.outputs)
             self._pending -= 1
             self._record_gap_locked()
 
@@ -326,9 +423,9 @@ class HistoryWriter:
                             self._record_gap_locked()
                 if command is not None:
                     failed = False
-                    context = None
+                    context = outputs = None
                     if isinstance(command, _Finalization):
-                        context = command.context
+                        context, outputs = command.context, command.outputs
                         command = command.command
                     try:
                         # Restore may have reopened this thread's managed connection.
@@ -337,37 +434,50 @@ class HistoryWriter:
                             db.write_execution_start(command)
                         else:
                             db.write_execution_finish(command)
-                        if context is not None:
+                        for payload, bundle_type, write in (
+                            (
+                                context,
+                                ContextBundle,
+                                db.write_execution_contexts_for_finish,
+                            ),
+                            (
+                                outputs,
+                                OutputBundle,
+                                db.write_execution_outputs_for_finish,
+                            ),
+                        ):
+                            if payload is None:
+                                continue
                             with self._lock:
                                 current = (
                                     generation == self._generation
                                     and command.storage_epoch == self._epoch
                                     and (
-                                        context.capture_generation is None
-                                        or context.capture_generation
-                                        == self._generation
+                                        payload.capture_generation is None
+                                        or payload.capture_generation == generation
                                     )
                                 )
                             if current:
                                 try:
-                                    db.write_execution_contexts_for_finish(
+                                    write(
                                         command,
                                         (
-                                            context.bundle
-                                            if context.bundle is not None
-                                            else ContextBundle(
-                                                (), context.state, context.reason
+                                            payload.bundle
+                                            if payload.bundle is not None
+                                            else bundle_type(
+                                                (), payload.state, payload.reason
                                             )
                                         ),
                                         generation_guard=partial(
                                             self._context_generation_guard,
                                             generation,
                                             command.storage_epoch,
-                                            context.capture_generation,
+                                            payload.capture_generation,
                                         ),
                                     )
                                 except ValueError:
-                                    # Validation/collision cannot revoke the outcome.
+                                    # One supplemental omission cannot revoke earlier
+                                    # commits or suppress independent valid evidence.
                                     with self._lock:
                                         self._record_gap_locked()
                     except HistoryEpochChanged:
@@ -377,9 +487,9 @@ class HistoryWriter:
                     except Exception:
                         failed = True
                     finally:
-                        if context is not None:
-                            self._release_context(context)
-                            context = None
+                        self._release_payload(context)
+                        self._release_payload(outputs)
+                        context = outputs = payload = None
                     with self._lock:
                         self._pending -= 1
                         if failed:

utils/execution_output_capture.py

@@ -0,0 +1,230 @@
+"""Pure, bounded final-history extraction under the shared payload budget."""
+
+from dataclasses import dataclass
+import json
+
+from .execution_context_budget import ContextBudget, PayloadLease
+from .execution_contexts import MAX_ATTEMPT_BYTES
+from .execution_outputs import (
+    BUNDLE_OVERHEAD,
+    MAX_BUNDLE_BYTES,
+    MAX_IMAGE_POSITIONS,
+    OutputBundle,
+    OutputReference,
+    OutputRoots,
+    STATE_REASONS,
+    _reference_size,
+    output_payload_size,
+)
+
+MISSING_HISTORY = object()
+
+
+@dataclass(frozen=True)
+class SealedOutputs:
+    bundle: OutputBundle | None
+    lease: PayloadLease | None
+    state: str
+    reason: str
+    capture_generation: int | None = None
+
+    def validate_header(self) -> None:
+        if (
+            type(self.state) is not str
+            or type(self.reason) is not str
+            or self.reason not in STATE_REASONS.get(self.state, ())
+            or (
+                self.capture_generation is not None
+                and (
+                    type(self.capture_generation) is not int
+                    or not 0 <= self.capture_generation <= 2**63 - 1
+                )
+            )
+        ):
+            raise ValueError("Invalid output capture status")
+        header = json.dumps(
+            [self.state, self.reason, self.capture_generation], separators=(",", ":")
+        ).encode("utf-8")
+        if len(header) > 256:
+            raise ValueError("Oversized output capture status")
+
+
+def _scalar_status(value: object, limit: int, *, char_limit: int | None = None) -> str:
+    """Check before retaining/encoding, distinguishing rejection from a stop bound."""
+    if type(value) is not str:
+        return "invalid_evidence"
+    if len(value) > (limit if char_limit is None else char_limit):
+        return "limit_reached"
+    size = 0
+    for char in value:
+        code = ord(char)
+        if code == 0 or 0xD800 <= code <= 0xDFFF:
+            return "invalid_evidence"
+        size += 1 if code < 128 else 2 if code < 2048 else 3 if code < 65536 else 4
+        if size > limit:
+            return "limit_reached"
+    return ""
+
+
+def _root_key(start: OutputRoots, finish: OutputRoots, folder: str) -> str | None:
+    if start.path_style != finish.path_style:
+        return None
+    before = start.output_key if folder == "output" else start.temp_key
+    after = finish.output_key if folder == "output" else finish.temp_key
+    return before if before == after else None
+
+
+def _extract(
+    outputs: dict,
+    start: OutputRoots,
+    finish: OutputRoots,
+    allowance: int,
+) -> OutputBundle:
+    references = []
+    reasons = set()
+    used = BUNDLE_OVERHEAD
+    positions = 0
+    stopped = False
+    for node_count, (node_id, ui) in enumerate(outputs.items()):
+        if node_count == 500:
+            reasons.add("limit_reached")
+            break
+        status = _scalar_status(node_id, 512, char_limit=128)
+        if status:
+            reasons.add(status)
+            if status == "limit_reached":
+                break
+            continue
+        if not node_id or type(ui) is not dict:
+            reasons.add("invalid_evidence")
+            continue
+        if "images" not in ui:
+            continue
+        images = ui["images"]
+        if type(images) is not list:
+            reasons.add("invalid_evidence")
+            continue
+        for ordinal, image in enumerate(images):
+            if positions == MAX_IMAGE_POSITIONS:
+                reasons.add("limit_reached")
+                stopped = True
+                break
+            positions += 1
+            if type(image) is not dict:
+                reasons.add("invalid_evidence")
+                continue
+            folder = image.get("type")
+            if type(folder) is not str:
+                reasons.add("invalid_evidence")
+                continue
+            if folder not in ("output", "temp"):
+                reasons.add("unsupported")
+                continue
+            filename, subfolder = image.get("filename"), image.get("subfolder")
+            statuses = (
+                _scalar_status(filename, 1024),
+                _scalar_status(subfolder, 2048),
+            )
+            if "limit_reached" in statuses:
+                reasons.add("limit_reached")
+                stopped = True
+                break
+            if "invalid_evidence" in statuses:
+                reasons.add("invalid_evidence")
+                continue
+            root = _root_key(start, finish, folder)
+            reference = OutputReference(
+                node_id, ordinal, folder, subfolder, filename, start.path_style, root
+            )
+            try:
+                size = _reference_size(reference)
+            except ValueError:
+                reasons.add("invalid_evidence")
+                continue
+            if used + size > allowance:
+                reasons.add(
+                    "limit_reached"
+                    if allowance == MAX_BUNDLE_BYTES
+                    else "budget_exhausted"
+                )
+                stopped = True
+                break
+            references.append(reference)
+            used += size
+            if root is None:
+                reasons.add("root_unavailable")
+        if stopped:
+            break
+    reason = next(
+        (
+            candidate
+            for candidate in (
+                "budget_exhausted",
+                "limit_reached",
+                "invalid_evidence",
+                "unsupported",
+                "root_unavailable",
+            )
+            if candidate in reasons
+        ),
+        "reported" if references else "none_reported",
+    )
+    references.sort(
+        key=lambda reference: (reference.output_node_id, reference.image_ordinal)
+    )
+    return OutputBundle(tuple(references), "partial" if reasons else "complete", reason)
+
+
+def seal_outputs(
+    history_result: object,
+    roots_at_start: OutputRoots,
+    roots_at_finish: OutputRoots,
+    *,
+    budget: ContextBudget,
+    capture_generation: int,
+    context_bytes: int,
+) -> SealedOutputs:
+    """Return one immutable prefix and its lease, or a lease-free omission header."""
+    if type(capture_generation) is not int or not 0 <= capture_generation <= 2**63 - 1:
+        raise ValueError("Invalid output capture generation")
+    if type(context_bytes) is not int or not 0 <= context_bytes <= MAX_ATTEMPT_BYTES:
+        raise ValueError("Invalid retained context bytes")
+    for roots in (roots_at_start, roots_at_finish):
+        if (
+            type(roots) is not OutputRoots
+            or type(roots.path_style) is not str
+            or roots.path_style not in ("posix", "windows")
+        ):
+            raise ValueError("Invalid output roots")
+
+    def omission(reason: str) -> SealedOutputs:
+        return SealedOutputs(None, None, "unavailable", reason, capture_generation)
+
+    lease = budget.reserve_up_to(
+        min(MAX_BUNDLE_BYTES, MAX_ATTEMPT_BYTES - context_bytes)
+    )
+    if lease is None:
+        return omission("budget_exhausted")
+    handed_off = False
+    try:
+        if lease.byte_count < BUNDLE_OVERHEAD:
+            return omission("budget_exhausted")
+        if history_result is MISSING_HISTORY:
+            return omission("unavailable")
+        if type(history_result) is not dict:
+            return omission("invalid_evidence")
+        outputs = history_result.get("outputs")
+        if type(outputs) is not dict:
+            return omission("invalid_evidence")
+        bundle = _extract(outputs, roots_at_start, roots_at_finish, lease.byte_count)
+        lease.shrink_to(output_payload_size(bundle))
+        lease = lease.transfer()
+        sealed = SealedOutputs(
+            bundle, lease, bundle.state, bundle.reason, capture_generation
+        )
+        sealed.validate_header()
+        handed_off = True
+        return sealed
+    finally:
+        if not handed_off:
+            lease.release()

utils/execution_outputs.py

@@ -0,0 +1,211 @@
+"""Bounded scalar output locators, without filesystem resolution or authority."""
+
+import hashlib
+import json
+import ntpath
+import posixpath
+import re
+from dataclasses import dataclass
+from types import ModuleType
+
+from .execution_contexts import text_size
+
+
+@dataclass(frozen=True)
+class OutputReference:
+    output_node_id: str
+    image_ordinal: int
+    folder_type: str
+    subfolder: str
+    filename: str
+    path_style: str
+    root_key: str | None
+
+
+@dataclass(frozen=True)
+class OutputBundle:
+    references: tuple[OutputReference, ...]
+    state: str
+    reason: str
+    format_version: int = 1
+    source: str = "host_final_history"
+
+
+@dataclass(frozen=True)
+class OutputRoots:
+    path_style: str
+    output_key: str | None
+    temp_key: str | None
+
+
+MAX_REFERENCE_BYTES = 4096
+MAX_BUNDLE_BYTES = 128 * 1024
+MAX_IMAGE_POSITIONS = 256
+MAX_ROOT_BYTES = 8192
+REFERENCE_OVERHEAD = 128
+BUNDLE_OVERHEAD = 128
+STATE_REASONS = {
+    "complete": frozenset({"reported", "none_reported"}),
+    "partial": frozenset(
+        {
+            "invalid_evidence",
+            "unsupported",
+            "limit_reached",
+            "root_unavailable",
+            "budget_exhausted",
+        }
+    ),
+    "unavailable": frozenset({"unavailable", "invalid_evidence", "budget_exhausted"}),
+    "conflicting": frozenset({"conflicting"}),
+}
+
+
+def _path_module(path_style: str) -> ModuleType:
+    if type(path_style) is not str or path_style not in ("posix", "windows"):
+        raise ValueError("Invalid output path style")
+    return posixpath if path_style == "posix" else ntpath
+
+
+def _scalar_size(value: str, limit: int) -> int:
+    size = text_size(value, limit)
+    if "\x00" in value:
+        raise ValueError("NUL in output scalar")
+    return size
+
+
+def _root_key(path_style: str, folder_type: str, root: object) -> str | None:
+    try:
+        _scalar_size(root, MAX_ROOT_BYTES)
+    except ValueError:
+        return None
+    paths = _path_module(path_style)
+    if not paths.isabs(root):
+        return None
+    if path_style == "windows":
+        drive, _ = ntpath.splitdrive(root)
+        portable = root.replace("/", "\\")
+        if not drive or portable.startswith(("\\\\?\\", "\\\\.\\")):
+            return None
+        portable_drive = drive.replace("/", "\\")
+        if portable_drive.startswith("\\\\"):
+            parts = portable_drive[2:].split("\\")
+            if len(parts) != 2 or any(part in ("", ".", "..") for part in parts):
+                return None
+        elif re.fullmatch(r"[a-zA-Z]:", drive) is None:
+            return None
+    normalized = paths.normpath(root)
+    encoded = json.dumps(
+        [1, path_style, folder_type, normalized],
+        ensure_ascii=False,
+        allow_nan=False,
+        separators=(",", ":"),
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def snapshot_output_roots(
+    path_style: str, output_root: object, temp_root: object
+) -> OutputRoots:
+    """Namespace bounded, lexically absolute configured roots without reading files."""
+    _path_module(path_style)
+    return OutputRoots(
+        path_style,
+        _root_key(path_style, "output", output_root),
+        _root_key(path_style, "temp", temp_root),
+    )
+
+
+def _reference_size(reference: OutputReference) -> int:
+    if type(reference) is not OutputReference:
+        raise ValueError("Invalid output reference")
+    size = REFERENCE_OVERHEAD
+    if (
+        type(reference.output_node_id) is not str
+        or not reference.output_node_id
+        or len(reference.output_node_id) > 128
+    ):
+        raise ValueError("Invalid output node ID")
+    size += _scalar_size(reference.output_node_id, 512)
+    if type(reference.image_ordinal) is not int or not (
+        0 <= reference.image_ordinal < MAX_IMAGE_POSITIONS
+    ):
+        raise ValueError("Invalid output image ordinal")
+    size += _scalar_size(reference.folder_type, 6)
+    if reference.folder_type not in ("output", "temp"):
+        raise ValueError("Unsupported output folder type")
+    size += _scalar_size(reference.path_style, 7)
+    paths = _path_module(reference.path_style)
+    size += _scalar_size(reference.subfolder, 2048)
+    size += _scalar_size(reference.filename, 1024)
+    if not reference.filename or reference.filename in (".", ".."):
+        raise ValueError("Invalid output filename")
+    separators = ("/", "\\") if reference.path_style == "windows" else ("/",)
+    if any(separator in reference.filename for separator in separators):
+        raise ValueError("Output filename must be a basename")
+    if paths.isabs(reference.subfolder) or paths.splitdrive(reference.subfolder)[0]:
+        raise ValueError("Output subfolder must be relative")
+    if reference.path_style == "windows" and (
+        ":" in reference.filename or ":" in reference.subfolder
+    ):
+        raise ValueError("Windows output colon or alternate stream")
+    if reference.subfolder:
+        parts = (
+            re.split(r"[/\\]", reference.subfolder)
+            if reference.path_style == "windows"
+            else reference.subfolder.split("/")
+        )
+        if any(part in ("", ".", "..") for part in parts):
+            raise ValueError("Unsafe or ambiguous output subfolder")
+    if reference.root_key is not None:
+        size += _scalar_size(reference.root_key, 64)
+        if re.fullmatch(r"[a-f0-9]{64}", reference.root_key) is None:
+            raise ValueError("Invalid output root key")
+    if size > MAX_REFERENCE_BYTES:
+        raise ValueError("Output reference payload limit reached")
+    return size
+
+
+def validate_output_reference(reference: OutputReference) -> None:
+    _reference_size(reference)
+
+
+def validate_output_bundle(bundle: OutputBundle) -> None:
+    output_payload_size(bundle)
+
+
+def output_payload_size(bundle: OutputBundle) -> int:
+    """Validate the immutable capture and measure its charged scalar payload."""
+    if (
+        type(bundle) is not OutputBundle
+        or type(bundle.format_version) is not int
+        or bundle.format_version != 1
+        or type(bundle.references) is not tuple
+        or len(bundle.references) > MAX_IMAGE_POSITIONS
+    ):
+        raise ValueError("Invalid output bundle")
+    for value in (bundle.state, bundle.reason, bundle.source):
+        _scalar_size(value, 128)
+    if bundle.source != "host_final_history" or bundle.reason not in STATE_REASONS.get(
+        bundle.state, ()
+    ):
+        raise ValueError("Invalid output state, reason or source")
+    if bundle.state == "complete":
+        if bool(bundle.references) != (bundle.reason == "reported"):
+            raise ValueError("Inconsistent complete output rows")
+    if bundle.state == "unavailable" and bundle.references:
+        raise ValueError("Unavailable output capture has rows")
+    if bundle.reason == "root_unavailable" and not bundle.references:
+        raise ValueError("Root unavailability requires output rows")
+    size = BUNDLE_OVERHEAD
+    previous = None
+    for reference in bundle.references:
+        size += _reference_size(reference)
+        coordinate = (reference.output_node_id, reference.image_ordinal)
+        if previous is not None and coordinate <= previous:
+            raise ValueError("Output coordinates must be unique and sorted")
+        previous = coordinate
+        if bundle.state == "complete" and reference.root_key is None:
+            raise ValueError("Complete output rows require root keys")
+        if size > MAX_BUNDLE_BYTES:
+            raise ValueError("Output bundle payload limit reached")
+    return size