test(history): verify durable contexts on real ComfyUI hosts

TESTING.md

@@ -200,10 +200,10 @@ loading a workflow or queueing generation.
 
 ## Durable execution ledger
 
-`npm run --prefix tools/vendor test:executions` adds ten real execution cases
-(five each in Classic and Nodes 2.0) to 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`,
+`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),
+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`:
 
 ```sh
@@ -218,7 +218,11 @@ 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 and gate-control routes; production has no test nodes or
+interruptible queue gate, deterministic CLIP/model/latent 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
+Classic, LRU, RAM-pressure and disabled caching; production has no test nodes or
 endpoints. Cleanup always releases the gate, stops the host and removes both
 owned directories. Logs and Playwright failures appear under
 `tools/vendor/test-results/executions/` and the existing CI artifact upload.
@@ -230,8 +234,10 @@ removal before dequeue. Actual host history is checked alongside finite polling
 of SQLite after the background writer. The read helper uses SQLite read-only
 mode, exact UUID job/session filters and a 256-row limit; absent ledger schema is
 explicit, while missing/corrupt databases and other read errors fail the suite.
-Sensitive sentinels in graph inputs, extra data and error messages must never
-appear in any stored row.
+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.
 
 For a meaningful historical failure, create a separate disposable detached
 checkout at `fdba6fd0126a8a40cbd6f3e50ed4c5de8991efd4` and pass its path to
@@ -253,5 +259,34 @@ before restarting an existing deployment.
 
 The ledger records only new dequeued jobs and observed outcomes. Existing
 counters, prompts, pairs and image links are preserved, with no reconstructed old
-runs. Prompt contexts, output associations and the chronological timeline remain
-future work under #209. Deploying the Python changes requires a ComfyUI restart.
+runs. Schema v2 migrates automatically when the configured database opens, adding
+immutable prompt-context tables without backfilling old runs. The internal reader
+returns `not_captured` for preexisting runs; a new capture can instead be complete,
+partial, ambiguous or unavailable. Keep the existing database backup/restore
+procedure: restored history changes storage epoch, and old library navigation
+hints must never be rebound to unrelated current rows.
+
+Context cases verify A,A,B as three jobs/two combinations/one positive snapshot
+when every positive field matches; all five registered PM variants; exact raw
+whitespace and Unicode; known empty negatives; Text negative roles and crossed
+pairs; fully cached repeats after trigger configuration changes; successful source
+text before downstream failure/interruption; unknown cached dynamic raw; unsupported
+transforms and duplicate list invocation ambiguity. Fresh values use
+`runtime_return`, cached effective STRING outputs use `host_cache_output`, and
+submitted literals use graph inference. SQL NULL means unknown; empty text is
+known. Text is never trimmed or normalized by capture. Cache availability and
+inferred sampler roles prove neither sampler execution nor consumption, and list
+indices do not identify generated images. External trigger configuration changes
+must not rewrite the effective text of an existing cached result. The cache matrix
+also checks that positive/pair usage increments once per dequeued job while a
+negative-only component retains its existing zero standalone-use count.
+
+Fault injection for restore fencing, inherited async capability revocation, queued
+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
+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.

tests/comfyui/execution-history.spec.mjs

@@ -166,3 +166,185 @@ test('rejected and removed-before-dequeue jobs create no persisted attempts', as
         }
     } finally { await control(request, 'release'); }
 });
+
+function recipeGraph({ text = 'context positive', negative = '', kind = 'PromptManagerPair', prepend = '', append = '', dynamic = false } = {}) {
+    const graph = {
+        '10': { class_type: 'PMHistoryInputs', inputs: {} },
+        '20': { class_type: kind, inputs: { text: dynamic ? ['11', 0] : text,
+            prepend_text: prepend, append_text: append, ...(kind === 'PromptManagerText' ? {} : { clip: ['10', 0] }),
+            ...(kind === 'PromptManagerPair' ? { negative_text: negative } : {}) } },
+        '30': { class_type: 'CLIPTextEncode', inputs: { text: negative, clip: ['10', 0] } },
+        '40': { class_type: 'KSampler', inputs: { model: ['10', 1], latent_image: ['10', 2],
+            positive: ['20', 0], negative: kind === 'PromptManagerPair' ? ['20', 1] : ['30', 0],
+            seed: 1, steps: 1, cfg: 1, sampler_name: 'euler', scheduler: 'normal', denoise: 1 } },
+        '50': { class_type: 'PMHistoryOutput', inputs: { value: ['40', 0] } },
+    };
+    if (kind === 'PromptManagerText') {
+        graph['31'] = { class_type: 'CLIPTextEncode', inputs: { text: ['20', 0], clip: ['10', 0] } };
+        graph['40'].inputs.positive = ['31', 0];
+    }
+    if (dynamic) graph['11'] = { class_type: 'PMHistoryCached', inputs: { value: text } };
+    return graph;
+}
+
+async function capture(execution, id, outcome = 'succeeded') {
+    const [run] = await rows(execution, id, [outcome]);
+    let result;
+    await expect.poll(async () => {
+        result = await readExecutionRuns(execution.databasePath, id);
+        return result.captures[run.run_id].state;
+    }, { timeout: 10000, message: 'Background writer persisted context finalization' }).not.toBe('not_captured');
+    return result.captures[run.run_id];
+}
+
+test('actual PM A,A,B jobs persist two combinations and one identical positive snapshot', async ({ execution }) => {
+    const captures = [];
+    const positiveSnapshotIds = [];
+    for (const negative of ['negative A', 'negative A', 'negative B']) {
+        const id = await submit(execution.request, recipeGraph({ negative }));
+        await history(execution.request, id);
+        captures.push(await capture(execution, id));
+        const persisted = await readExecutionRuns(execution.databasePath, id);
+        positiveSnapshotIds.push(persisted.snapshotIds[persisted.rows[0].run_id][0].positive_snapshot_id);
+    }
+    expect(new Set(positiveSnapshotIds).size).toBe(1);
+    expect(captures.map(value => value.state)).toEqual(['complete', 'complete', 'complete']);
+    const contexts = captures.map(value => value.contexts[0]);
+    expect(new Set(contexts.map(value => value.context_id)).size).toBe(2);
+    expect(new Set(contexts.map(value => JSON.stringify(value.positive))).size).toBe(1);
+    expect(contexts[0].evidence.positive.effective.source).toBe('runtime_return');
+    expect(contexts[1].evidence.positive.effective.source).toBe('host_cache_output');
+    expect(contexts[1].evidence.positive.raw.source).toBe('graph_literal');
+    expect(contexts[1].evidence.role_provenance).toBe('inferred');
+    expect(contexts[2].negative.raw).toBe('negative B');
+});
+
+test('fresh registered PM variants preserve exact raw whitespace and known empty negative', async ({ execution }) => {
+    for (const kind of ['PromptManager', 'PromptManagerFocused', 'PromptManagerSplit', 'PromptManagerPair', 'PromptManagerText']) {
+        const graph = recipeGraph({ kind, text: `  ${kind} e\u0301\n`, prepend: ' pre \t', append: '\n post ' });
+        const id = await submit(execution.request, graph);
+        await history(execution.request, id);
+        const value = (await capture(execution, id)).contexts[0];
+        expect(value.positive).toEqual({ raw: `  ${kind} e\u0301\n`, prepend: ' pre \t', append: '\n post ', effective: `pre   ${kind} e\u0301\n post` });
+        expect(value.negative.raw).toBe('');
+        expect(value.negative.effective).toBe('');
+        expect(value.evidence.positive.raw).toMatchObject({ provenance: 'recorded', source: 'runtime_return' });
+    }
+});
+
+test('fully cached text uses old effective output after external trigger configuration changes', async ({ execution }) => {
+    const request = execution.request;
+    const configure = suffix => request.post('/pm_execution_fixture/injection', { data: { suffix } });
+    try {
+        expect((await configure(' trigger-A')).ok()).toBe(true);
+        const graph = recipeGraph({ text: 'cached trigger recipe', negative: 'trigger negative' });
+        const first = await submit(request, graph);
+        await history(request, first);
+        const original = (await capture(execution, first)).contexts[0];
+        expect(original.positive.effective).toBe('cached trigger recipe trigger-A');
+        expect((await configure(' trigger-B')).ok()).toBe(true);
+        const second = await submit(request, graph);
+        const item = await history(request, second);
+        expect(item.status.messages.find(([kind]) => kind === 'execution_cached')[1].nodes).toEqual(expect.arrayContaining(['20', '40', '50']));
+        const repeated = (await capture(execution, second)).contexts[0];
+        expect(repeated.context_id).toBe(original.context_id);
+        expect(repeated.positive.effective).toBe(original.positive.effective);
+        expect(repeated.evidence.positive.effective.source).toBe('host_cache_output');
+    } finally { expect((await configure('')).ok()).toBe(true); }
+});
+
+test('crossed Pair and Text negative roles remain independent declared contexts', async ({ execution }) => {
+    const graph = recipeGraph({ text: 'pair one positive', negative: 'pair one negative' });
+    graph['21'] = { class_type: 'PromptManagerPair', inputs: { text: 'pair two positive', negative_text: 'pair two negative', clip: ['10', 0] } };
+    graph['22'] = { class_type: 'PromptManagerText', inputs: { text: 'text negative role' } };
+    graph['32'] = { class_type: 'CLIPTextEncode', inputs: { text: ['22', 0], clip: ['10', 0] } };
+    graph['40'].inputs.negative = ['21', 1];
+    graph['41'] = structuredClone(graph['40']);
+    graph['41'].inputs.positive = ['21', 0];
+    graph['41'].inputs.negative = ['32', 0];
+    graph['51'] = { class_type: 'PMHistoryOutput', inputs: { value: ['41', 0] } };
+    const id = await submit(execution.request, graph);
+    await history(execution.request, id);
+    const result = await capture(execution, id);
+    expect(result.contexts.map(value => [value.anchor_id, value.positive.raw, value.negative.raw])).toEqual([
+        ['40', 'pair one positive', 'pair two negative'], ['41', 'pair two positive', 'text negative role'],
+    ]);
+    for (const value of result.contexts) expect(value.evidence.negative.effective.source).toBe('runtime_return');
+});
+
+test('successful source capture survives downstream failure and real interruption', async ({ execution }) => {
+    const request = execution.request;
+    for (const mode of ['failure', 'gate']) {
+        const graph = recipeGraph({ text: `source before ${mode}` });
+        graph['50'].inputs.mode = mode;
+        if (mode === 'gate') await control(request, 'hold');
+        try {
+            const id = await submit(request, graph);
+            if (mode === 'gate') {
+                await entered(request);
+                expect((await request.post('/interrupt', { data: {} })).ok()).toBe(true);
+            }
+            await history(request, id, mode === 'gate' ? 'execution_interrupted' : 'execution_error');
+            const result = await capture(execution, id, mode === 'gate' ? 'interrupted' : 'failed');
+            expect(result.contexts[0].positive.effective).toBe(`source before ${mode}`);
+            expect(result.contexts[0].evidence.positive.effective.source).toBe('runtime_return');
+        } finally { await control(request, 'release'); }
+    }
+});
+
+test('cached dynamic raw stays unknown while literal affixes and effective cache text survive', async ({ execution }) => {
+    const graph = recipeGraph({ dynamic: true, text: 'runtime dynamic recipe', prepend: ' pre ', append: ' post ' });
+    const first = await submit(execution.request, graph);
+    await history(execution.request, first);
+    expect((await capture(execution, first)).contexts[0].positive.raw).toBe('runtime dynamic recipe');
+    const second = await submit(execution.request, graph);
+    await history(execution.request, second);
+    const value = (await capture(execution, second)).contexts[0];
+    expect(value.positive).toEqual({ raw: null, prepend: ' pre ', append: ' post ', effective: 'pre runtime dynamic recipe post' });
+    expect(value.evidence.positive.raw.provenance).toBe('unknown');
+    expect(value.evidence.positive.effective.source).toBe('host_cache_output');
+});
+
+test('unsupported conditioning transforms and duplicate list invocations preserve uncertainty', async ({ execution }) => {
+    const transformed = recipeGraph({ text: 'unsupported transformed recipe' });
+    transformed['35'] = { class_type: 'PMHistoryTransform', inputs: { conditioning: ['20', 0] } };
+    transformed['40'].inputs.positive = ['35', 0];
+    const unsupported = await submit(execution.request, transformed);
+    await history(execution.request, unsupported);
+    const unknown = await capture(execution, unsupported);
+    expect(unknown.state).toBe('partial');
+    expect(unknown.contexts[0].positive.effective).toBeNull();
+    const graph = recipeGraph({ text: 'duplicate runtime list' });
+    graph['11'] = { class_type: 'PMHistoryList', inputs: { value: 'duplicate runtime list' } };
+    graph['20'].inputs.text = ['11', 0];
+    const id = await submit(execution.request, graph);
+    await history(execution.request, id);
+    const result = await capture(execution, id);
+    expect(result.state).toBe('ambiguous');
+    expect(result.reason).toBe('multiple_samples');
+    expect(result.contexts[0].positive.effective).toBeNull();
+});
+
+test('real executor Classic LRU RAM and disabled caches preserve honest evidence and usage', async ({ execution }) => {
+    const request = execution.request;
+    for (const mode of ['CLASSIC', 'LRU', 'RAM_PRESSURE', 'NONE']) {
+        expect((await request.post('/pm_execution_fixture/cache', { data: { mode } })).ok()).toBe(true);
+        const graph = recipeGraph({ text: `cache mode ${mode}`, negative: `cache mode negative ${mode}` });
+        const results = [];
+        for (let index = 0; index < 2; index++) {
+            const id = await submit(request, graph);
+            await history(request, id);
+            results.push(await capture(execution, id));
+            const persisted = await readExecutionRuns(execution.databasePath, id);
+            const [run] = persisted.rows;
+            expect(persisted.usage[run.run_id]).toEqual([
+                { role: 'positive', run_count: index + 1 },
+                { role: 'negative', run_count: 0 },
+                { role: 'pair', run_count: index + 1 },
+            ]);
+        }
+        expect(results[0].state).toBe('complete');
+        expect(results[1].contexts[0].context_id).toBe(results[0].contexts[0].context_id);
+        expect(results[1].contexts[0].evidence.positive.effective.source).toBe(mode === 'NONE' ? 'runtime_return' : 'host_cache_output');
+    }
+});

tests/comfyui/fixtures/ledger_nodes.py

@@ -3,6 +3,9 @@
 import threading
 import time
 
+import nodes
+import execution
+
 from aiohttp import web
 import comfy.model_management
 from server import PromptServer
@@ -69,3 +72,129 @@ NODE_CLASS_MAPPINGS = {
     "PMHistoryFailure": PMHistoryFailure,
     "PMHistoryGate": PMHistoryGate,
 }
+
+# Only external inference is replaced; PM classes and host execution remain real.
+
+
+class _Clip:
+    def tokenize(self, text: str) -> str:
+        return text
+
+    def encode_from_tokens_scheduled(self, tokens: str) -> list:
+        return [[tokens, {}]]
+
+
+class PMHistoryInputs:
+    @classmethod
+    def INPUT_TYPES(cls) -> dict:
+        return {"required": {}}
+
+    RETURN_TYPES = ("CLIP", "MODEL", "LATENT")
+    FUNCTION = "execute"
+    CATEGORY = "testing/execution-history"
+
+    def execute(self) -> tuple:
+        return (_Clip(), object(), {"samples": "owned deterministic latent"})
+
+
+class PMHistoryOutput:
+    @classmethod
+    def INPUT_TYPES(cls) -> dict:
+        return {
+            "required": {"value": ("LATENT",)},
+            "optional": {"mode": (["normal", "failure", "gate"],)},
+        }
+
+    RETURN_TYPES = ()
+    FUNCTION = "execute"
+    CATEGORY = "testing/execution-history"
+    OUTPUT_NODE = True
+
+    def execute(self, value: dict, mode: str = "normal") -> tuple:
+        if mode == "failure":
+            raise ValueError("owned downstream failure")
+        if mode == "gate":
+            PMHistoryGate().execute("owned downstream gate")
+        return ()
+
+
+def _sample(self: object, **kwargs: object) -> tuple:
+    return (kwargs["latent_image"],)
+
+
+nodes.KSampler.sample = _sample
+NODE_CLASS_MAPPINGS.update(
+    {"PMHistoryInputs": PMHistoryInputs, "PMHistoryOutput": PMHistoryOutput}
+)
+
+
+class PMHistoryList(PMHistoryCached):
+    OUTPUT_IS_LIST = (True,)
+
+    def execute(self, value: str) -> tuple:
+        return ([value, value],)
+
+
+class PMHistoryTransform:
+    @classmethod
+    def INPUT_TYPES(cls) -> dict:
+        return {"required": {"conditioning": ("CONDITIONING",)}}
+
+    RETURN_TYPES = ("CONDITIONING",)
+    FUNCTION = "execute"
+    CATEGORY = "testing/execution-history"
+
+    def execute(self, conditioning: list) -> tuple:
+        return (conditioning,)
+
+
+NODE_CLASS_MAPPINGS.update(
+    {"PMHistoryList": PMHistoryList, "PMHistoryTransform": PMHistoryTransform}
+)
+
+_INJECTION_ORIGINALS = {}
+
+
+@PromptServer.instance.routes.post("/pm_execution_fixture/injection")
+async def injection(request: web.Request) -> web.Response:
+    suffix = (await request.json())["suffix"]
+    if not isinstance(suffix, str) or len(suffix) > 128:
+        raise web.HTTPBadRequest()
+    for name in ("PromptManager", "PromptManagerText"):
+        cls = nodes.NODE_CLASS_MAPPINGS[name]
+        if name not in _INJECTION_ORIGINALS:
+            _INJECTION_ORIGINALS[name] = cls._inject_lora_trigger_words
+        cls._inject_lora_trigger_words = (
+            (lambda self, text: text + suffix) if suffix else _INJECTION_ORIGINALS[name]
+        )
+    return web.json_response({"configured": True})
+
+
+# Apply a requested host cache configuration on its own worker before execution.
+# Retain only the mode scalar; never retain the worker/executor/cache objects.
+
+_CACHE_MODE = None
+_EXECUTE = execution.PromptExecutor.execute
+
+
+def _execute_with_cache_mode(self: object, *args: object, **kwargs: object) -> object:
+    global _CACHE_MODE
+    if _CACHE_MODE is not None:
+        self.cache_type = execution.CacheType[_CACHE_MODE]
+        self.cache_args = {**self.cache_args, "lru": 100, "ram": 0}
+        self.reset()
+        _CACHE_MODE = None
+    return _EXECUTE(self, *args, **kwargs)
+
+
+execution.PromptExecutor.execute = _execute_with_cache_mode
+
+
+@PromptServer.instance.routes.post("/pm_execution_fixture/cache")
+async def cache_mode(request: web.Request) -> web.Response:
+    global _CACHE_MODE
+    mode = (await request.json())["mode"]
+    if mode not in ("CLASSIC", "LRU", "RAM_PRESSURE", "NONE"):
+        raise web.HTTPBadRequest()
+    _CACHE_MODE = mode
+    return web.json_response({"configured": True})

tests/comfyui/read_execution_history.py

@@ -4,8 +4,19 @@ import argparse
 import json
 from pathlib import Path
 import sqlite3
+import sys
+from types import SimpleNamespace
 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
+
+
+class _ReadOnlyHistory(ExecutionContextOperations, ExecutionHistoryOperations):
+    def __init__(self, connection: sqlite3.Connection) -> None:
+        self.model = SimpleNamespace(get_connection=lambda: connection)
+
 
 def read_runs(database: str, job_id: str, session_id: str | None = None) -> dict:
     job_id = str(UUID(job_id))
@@ -30,7 +41,45 @@ def read_runs(database: str, job_id: str, session_id: str | None = None) -> dict
         ).fetchall()
         if len(rows) > 256:
             raise ValueError("Execution query exceeds the 256-row bound")
-        return {"schemaAvailable": True, "rows": [dict(row) for row in rows]}
+        reader = _ReadOnlyHistory(connection)
+        captures = {
+            row["run_id"]: reader.get_execution_contexts(row["run_id"]) for row in rows
+        }
+        usage = {}
+        snapshot_ids = {}
+        for run_id, capture in captures.items():
+            usage[run_id] = []
+            snapshot_ids[run_id] = []
+            for context in capture["contexts"]:
+                for role in ("positive", "negative"):
+                    prompt_id = context[f"{role}_prompt_id"]
+                    if prompt_id is not None:
+                        count = connection.execute(
+                            "SELECT run_count FROM prompts WHERE id=?", (prompt_id,)
+                        ).fetchone()[0]
+                        usage[run_id].append({"role": role, "run_count": count})
+                pair_id = context["pair_id"]
+                if pair_id is not None:
+                    count = connection.execute(
+                        "SELECT run_count FROM prompt_pairs WHERE id=?", (pair_id,)
+                    ).fetchone()[0]
+                    usage[run_id].append({"role": "pair", "run_count": count})
+                snapshot_ids[run_id].append(
+                    dict(
+                        connection.execute(
+                            "SELECT positive_snapshot_id,negative_snapshot_id "
+                            "FROM execution_contexts WHERE context_id=?",
+                            (context["context_id"],),
+                        ).fetchone()
+                    )
+                )
+        return {
+            "schemaAvailable": True,
+            "rows": [dict(row) for row in rows],
+            "captures": captures,
+            "usage": usage,
+            "snapshotIds": snapshot_ids,
+        }
 
 
 if __name__ == "__main__":

tests/test_execution_context_capture.py

@@ -373,6 +373,76 @@ class CaptureTests(unittest.TestCase):
 
         asyncio.run(scenario())
 
+    def test_writer_restore_or_close_revokes_callback_paused_during_encoding(self):
+        import asyncio
+
+        from utils.execution_history_writer import HistoryWriter
+
+        for action in ("invalidate", "close"):
+            with self.subTest(action=action):
+                pool = ContextBudget()
+                current = [HistoryWriterStatus(EPOCH, True, 0, 0, True, 3)]
+                manager = capture.ContextCapture(10, lambda: current[0], pool)
+                self.addCleanup(manager.revoke_all)
+                writer = HistoryWriter(
+                    lambda: self.fail("This callback test must not open storage"),
+                    context_budget=pool,
+                    on_invalidation=manager.revoke_all,
+                )
+                self.addCleanup(writer.close)
+                self.host_context()
+
+                async def scenario():
+                    old = manager.begin(self.start, 3, graph(), None)
+                    manager.observe(
+                        old, SourceRef("p", "PromptManagerPair", 0), 0, snapshot()
+                    )
+                    entered, release = asyncio.Event(), asyncio.Event()
+
+                    async def encoding_return():
+                        entered.set()
+                        await release.wait()
+                        capture.record_runtime_context(
+                            "PromptManagerPair", "late", "", "", "late", role="positive"
+                        )
+
+                    task = asyncio.create_task(encoding_return())
+                    await entered.wait()
+                    self.assertGreater(pool.charged_bytes, 0)
+                    if action == "invalidate":
+                        writer.invalidate()
+                    else:
+                        self.assertTrue(writer.close())
+                    self.assertEqual(pool.charged_bytes, 0)
+                    self.assertIsNone(capture.active_capture())
+                    new_epoch = "44444444-4444-4444-8444-444444444444"
+                    current[0] = HistoryWriterStatus(new_epoch, True, 0, 0, True, 4)
+                    new = manager.begin(
+                        replace(
+                            self.start,
+                            run_id=JOB,
+                            queue_item_id=2,
+                            storage_epoch=new_epoch,
+                        ),
+                        4,
+                        graph(),
+                        None,
+                    )
+                    release.set()
+                    await task
+                    self.assertIsNone(manager.seal(old).bundle)
+                    sealed = manager.seal(new)
+                    try:
+                        self.assertIsNone(
+                            sealed.bundle.contexts[0].positive.effective.value
+                        )
+                    finally:
+                        if sealed.lease:
+                            sealed.lease.release()
+                    self.assertEqual(pool.charged_bytes, 0)
+
+                asyncio.run(scenario())
+
     def test_oversized_runtime_field_is_unknown_without_erasing_other_fields(self):
         cap = self.begin()
         self.host_context()