Remove completed Notarius cache roadmaps

docs/cli.md

@@ -212,6 +212,10 @@ artifacts and any prerequisites required by those targets; without
 never run. Common session/configuration options and repeatable artifact values
 pass through unchanged.
 
+The forced extraction rebuilds the complete configured Notarius bundle,
+disables Notarius checkpoint loading, and requests chunk-plan refresh.
+`--artifacts` narrows only the analysis work; it never selects Notarius lanes.
+
 Because the expansion owns `--force`, `--from`, and `--through`, callers cannot
 supply those options. The shared `run` parser reports them as duplicate
 singleton flags. The alias has no private execution options or behavior, and
@@ -254,6 +258,10 @@ Equivalent to:
 narratio run-stage analyze <session_id> --force [...common config flags]
 ```
 
+This selects only `analyze`: it does not invoke Notarius, replace extraction,
+or remove existing extract outputs. `--artifacts` narrows the forced analysis
+targets and their required analysis prerequisites.
+
 ### `publish`
 
 ```bash

docs/config.md

@@ -362,6 +362,8 @@ Rules:
 | `pipeline.notarius.pipeline_id` | string | Conditional | required when enabled |
 | `pipeline.notarius.timeout` | duration | No | `3h`; must be positive |
 | `pipeline.notarius.working_directory` | string | No | directory containing resolved `config_path`; relative paths resolve from the pipeline file directory |
+| `pipeline.notarius.resume` | bool | No | `false`; request compatible Notarius checkpoint reuse during ordinary executed extraction |
+| `pipeline.notarius.chunk_cache` | string | No | empty; when set, exactly `auto`, `bypass`, or `refresh` |
 | `pipeline.notarius.references` | map[string]string | No | empty; maps normalized Notarius selectors to supported prepared Narratio source IDs; maximum 256 entries |
 | `pipeline.notarius.outputs` | map | Conditional | at least one entry when enabled |
 | `pipeline.render.enabled` | bool | No | `true` |
@@ -406,6 +408,20 @@ configuration digest; rotating only its value does neither.
 See the [Notarius integration contract](./integrations/notarius.md) for the
 delivery and redaction boundary.
 
+### Notarius Cache Controls
+
+`pipeline.notarius.resume` defaults to `false`. When true, an executed ordinary
+extraction passes `--resume`; the deployed Notarius configuration must already
+enable checkpoint recording and should select an explicit persistent checkpoint
+root. Narratio does not inspect or generate that configuration.
+
+`pipeline.notarius.chunk_cache` is empty by default, leaving Notarius's
+configured mode in control. A nonempty value must be `auto`, `bypass`, or
+`refresh` and is passed as `--chunk_cache`. Forced extraction takes precedence
+over both fields: it omits `--resume` and passes `--chunk_cache refresh`.
+Neither field changes Narratio's semantic extraction fingerprint, so changing
+them alone does not rerun an otherwise complete durable extraction.
+
 ### Notarius Reference Bindings
 
 `pipeline.notarius.references` maps a Notarius CLI selector to a prepared
@@ -435,8 +451,8 @@ It validates only selector structure and the prepared source vocabulary;
 Notarius owns target-slot declarations and media compatibility.
 
 Before extraction, Narratio resolves every binding from the current prepared
-session manifest and streams it into a verified invocation-local snapshot whose
-absolute path is passed to Notarius. Missing, unsafe, empty,
+session manifest and streams it into a verified, content-addressed session
+snapshot whose absolute path is passed to Notarius. Missing, unsafe, empty,
 changed-during-copy, or checksum-inconsistent prepared evidence fails with
 guidance to force `prepare`. Bindings are sorted by normalized selector and are
 part of extraction fingerprint and resume identity. See the

docs/integrations/notarius.md

@@ -7,14 +7,14 @@ lanes from the final trimmed Seriatim transcript. Narratio owns invocation,
 safe bundle discovery, lane selection, and its own artifact metadata. Notarius
 owns pipeline definitions, lane schemas, the receipt, and bundle formats.
 
-Canonical Notarius v0.6.0 references:
+Canonical Notarius v0.7.0 references:
 
-- [CLI reference](https://gitea.maximumdirect.net/eric/notarius/src/tag/v0.6.0/docs/cli.md)
-- [Subprocess consumer contract](https://gitea.maximumdirect.net/eric/notarius/src/tag/v0.6.0/docs/consumers/subprocess.md)
-- [D&D pipeline and lane contracts](https://gitea.maximumdirect.net/eric/notarius/src/tag/v0.6.0/docs/consumers/dnd-pipeline.md)
-- [Run-result receipt](https://gitea.maximumdirect.net/eric/notarius/src/tag/v0.6.0/docs/integrations/run-result.md)
-- [JSON output bundle](https://gitea.maximumdirect.net/eric/notarius/src/tag/v0.6.0/docs/integrations/json-output.md)
-- [D&D spell-catalog overlay](https://gitea.maximumdirect.net/eric/notarius/src/tag/v0.6.0/docs/integrations/dnd-spell-catalog-overlays.md)
+- [CLI reference](https://gitea.maximumdirect.net/eric/notarius/src/tag/v0.7.0/docs/cli.md)
+- [Subprocess consumer contract](https://gitea.maximumdirect.net/eric/notarius/src/tag/v0.7.0/docs/consumers/subprocess.md)
+- [D&D pipeline and lane contracts](https://gitea.maximumdirect.net/eric/notarius/src/tag/v0.7.0/docs/consumers/dnd-pipeline.md)
+- [Run-result receipt](https://gitea.maximumdirect.net/eric/notarius/src/tag/v0.7.0/docs/integrations/run-result.md)
+- [JSON output bundle](https://gitea.maximumdirect.net/eric/notarius/src/tag/v0.7.0/docs/integrations/json-output.md)
+- [D&D spell-catalog overlay](https://gitea.maximumdirect.net/eric/notarius/src/tag/v0.7.0/docs/integrations/dnd-spell-catalog-overlays.md)
 
 The [complete Narratio example](../../examples/pipeline.full.annotated.yml)
 records the exact current constraints for all ten D&D lanes. Treat the linked
@@ -25,19 +25,26 @@ duplicate the complete schemas.
 
 When `pipeline.notarius.enabled` is true, Narratio resolves the executable,
 configuration path, input path, output directory, and working directory to
-absolute paths. Narratio requires the Notarius v0.6.0 CLI contract when
-references are configured and invokes each binding as a separate argument
-before `--json`:
+absolute paths. Configured references require support for the `--reference`
+contract introduced in Notarius v0.6.0. Cache controls require support for the
+selected `--resume` and `--chunk_cache` flags. Notarius v0.7.0 is the tested
+cache integration baseline, not an unconditional minimum version; Narratio
+does not impose a runtime version gate. Each reference binding is a separate
+argument before `--json`:
 
 ```text
-notarius run <pipeline_id> --config <config_path> --input <trimmed_json> --output-dir <staging_dir> [--reference <selector>=<verified_snapshot_path>]... --json
+notarius run <pipeline_id> --config <config_path> --input <trimmed_json> --output-dir <staging_dir> [--resume] [--chunk_cache <mode>] [--reference <selector>=<verified_snapshot_path>]... --json
 ```
 
-Reference paths are absolute invocation-local snapshots streamed from the
-manifest-verified canonical files prepared inside the current Narratio session
-workspace. Narratio verifies snapshot checksum and size before and after the
-subprocess, and passes only configured bindings, ordered lexically by normalized
-selector, as direct argument-vector entries without shell interpretation. A CLI
+Reference paths are absolute stable snapshots streamed from manifest-verified
+canonical files prepared inside the current Narratio session workspace. They
+live at `cache/notarius/references/<sha256>/<prepared-basename>` and retain the
+full lowercase content digest. Narratio verifies snapshot checksum and size
+before and after the subprocess, reconstructs missing or corrupt regular files
+from a still-valid prepared source, and passes only configured bindings,
+ordered lexically by normalized selector, as direct argument-vector entries
+without shell interpretation. Symlinks, directories, other non-regular
+conflicts, and prepared-source mismatches fail rather than being deleted. A CLI
 binding takes precedence over a matching external path in Notarius
 configuration. Narratio never emits `--without-reference`.
 
@@ -75,6 +82,17 @@ It does not pass a Narratio session ID or run `notarius config validate`
 automatically; the configured working directory and Narratio's minimal child
 environment apply to the subprocess.
 
+Ordinary extraction maps `pipeline.notarius.resume: true` to `--resume` and a
+nonempty `pipeline.notarius.chunk_cache` to `--chunk_cache <mode>`. Forced
+extraction always omits `--resume` and supplies `--chunk_cache refresh`.
+Notarius owns checkpoint enablement, recording, identity, and both cache roots;
+Narratio neither reads nor synthesizes its configuration. Use explicit
+persistent roots in Notarius when reuse must span subprocess invocations.
+Changing a reference path/digest or other upstream identity input correctly
+causes a cold miss. Chunk refresh bypasses lookup and publishes the accepted
+replacement for the active source; it does not delete unrelated plans or
+checkpoint state.
+
 When `pipeline.notarius.llm_api_key` is absent or null, Narratio adds no
 credential to that environment. When configured, the field names an environment
 variable whose already-loaded nonempty value is forwarded under that exact
@@ -98,7 +116,7 @@ encoded credentials.
 
 ## Accepted Result
 
-Narratio's supported invocation baseline is Notarius v0.6.0. The accepted
+Narratio's tested invocation baseline is Notarius v0.7.0. The accepted
 receipt remains `notarius.run-result.v2`; reference flags do not change the
 receipt or ten-lane output contract. The receipt
 must identify the configured pipeline, and its `index_file` must be exactly
@@ -123,6 +141,8 @@ version, and, when configured, module key. Missing, duplicate, rejected, or
 incompatible required lanes fail extraction even if Notarius exited zero. A
 configured lane whose v2 validation summary is `rejected` or `incomplete` also
 fails extraction.
+An accepted reused producer may report `producer_attempt_count: 0`; Narratio
+preserves that valid summary rather than treating it as a missing field.
 Unconfigured lanes may remain in the preserved bundle but do not become
 selectable Narratio sources.
 
@@ -151,3 +171,18 @@ Configuration fields and defaults are in [Configuration](../config.md).
 Operator paths, rerun procedures, and bundle retention are in
 [Operations](../operations.md). See [Troubleshooting](../troubleshooting.md)
 for failure recovery.
+
+## Opt-in Integration Test
+
+The production-boundary cache test requires an absolute path to a locally
+built Notarius v0.7.0-compatible executable. It uses only a local
+OpenAI-compatible `httptest` provider and temporary cache/config/output roots:
+
+```sh
+NARRATIO_TEST_NOTARIUS_BINARY=/absolute/path/to/notarius \
+  GOWORK=off go test ./internal/stage \
+  -run '^TestNotariusCacheReuseIntegration$' -count=1 -v
+```
+
+When the variable is unset, the normal test suite skips this test and never
+searches `PATH`, a sibling checkout, the network, or a real provider.

docs/internal/adapters.md

@@ -65,6 +65,9 @@ Notarius is composed only when extraction is enabled; the extract stage owns
 prepared reference resolution, receipt, bundle, and configured-lane policy.
 The adapter validates the ordered selector/absolute-path pairs and is the sole
 owner of serializing them as repeated `--reference` arguments before `--json`.
+It also serializes an effective resume request as `--resume` and a nonempty
+validated chunk mode as `--chunk_cache <mode>`. The extract stage, not the
+adapter, resolves force precedence before constructing the request.
 Its request may also carry one already-resolved opaque ConfigKit secret
 reference. Only the Notarius adapter reads that value, rejects an unresolved or
 empty programmatic reference, and maps the exact configured name/value into the

docs/internal/stage-extract.md

@@ -19,13 +19,15 @@ procedures belong in [Operations](../operations.md).
 1. resolves the final trimmed transcript from the shared artifact catalog;
 2. resolves every configured prepared reference through the shared
    manifest-authoritative identity resolver before creating run-local output;
-3. streams each verified reference into an invocation-local snapshot and
-   rejects any source change observed while copying;
+3. resolves each reference to a stable content-addressed session snapshot,
+   reusing verified bytes or reconstructing a missing/corrupt regular entry,
+   and rejects any source change observed while copying;
 4. fingerprints the byte- and provenance-bearing Notarius invocation evidence,
    including sorted reference identities;
 5. creates a run-local staging directory and invokes the injected
    `notarius.Runner`, carrying the optional already-resolved credential
-   reference without exposing its value;
+   reference without exposing its value, plus effective resume and chunk-cache
+   controls after force precedence;
 6. revalidates the reference snapshots, then validates the v2 successful
    receipt, confined index, management documents, configured required lane
    descriptors, validation summaries, and regular payload files;
@@ -42,14 +44,22 @@ Narratio run ID, the resolved trimmed-input identity, and invocation
 fingerprint. The input identity binds the exact transcript bytes, canonical
 source ID, producer stage/output/run identity, and resolution provenance.
 Reference metadata contains only selector, source ID, canonical session-relative
-path, checksum, and size; adapter requests receive selector and absolute
-invocation-local snapshot path, never payload contents. Snapshot bytes must
+path, checksum, and size; adapter requests receive selector and absolute stable
+snapshot path, never payload contents. Snapshot bytes must
 match the prepared identity both before and after Notarius runs, so a concurrent
 prepared-file replacement cannot make recorded provenance describe different
 bytes from those supplied to Notarius.
 Validation completes before the single durable rename, so a rejected result
 cannot expose a partial durable bundle.
 
+Snapshot destinations are owned by `internal/artifacts` under
+`cache/notarius/references/<sha256>/<prepared-basename>`. They contain no run ID
+or timestamp and can be reused across invocations. Identical bytes and basename
+share a verified file; changed bytes select a different digest directory.
+Unsafe path components and non-regular conflicts fail without recursive
+deletion. Snapshot paths are delivery details and never become manifest output
+evidence or prerequisites for reuse of an already complete extract bundle.
+
 Any executed extraction outcome that replaces a different effective outcome
 marks succeeded analysis and delivery dependents stale. Render is an independent
 sibling and remains current. Repeating the same disabled self-skip with no
@@ -74,7 +84,10 @@ sorted prepared-reference identities, pipeline identity, and configured output
 contracts. The same reference helper and transcript identity are resolved again
 for artifact evidence, so changing current transcript bytes, reference bytes,
 or producer identity makes the prior extraction obsolete. Operational runner
-settings do not invalidate otherwise current durable evidence.
+settings do not invalidate otherwise current durable evidence. This includes
+`resume` and `chunk_cache`; changing them alone cannot make a complete
+extraction non-resumable. If extraction executes, ordinary runs forward the
+configured values, while force disables resume and selects `refresh`.
 
 A valid prepared-reference change makes extraction non-resumable. Missing,
 unsafe, or checksum-inconsistent prepared evidence is a hard validation error
@@ -115,9 +128,13 @@ recovery.
   `internal/stage/extract_resume.go`,
   `internal/stage/extract_test.go`,
   `internal/stage/semantic_contracts_delivery.go`
+- Stable snapshot materialization and verification:
+  `internal/stage/extract_references.go`
 - Subprocess boundary: `internal/adapters/notarius/subprocess.go`,
   `internal/adapters/notarius/subprocess_test.go`
 - Catalog hydration: `internal/artifacts/extraction_catalog.go`,
   `internal/artifacts/extraction_catalog_test.go`
 - Composition and downstream behavior: `internal/app/runner_test.go`,
   `internal/stage/analyze_test.go`, `internal/stage/publish_test.go`
+- Opt-in Notarius v0.7.0 production-boundary coverage:
+  `internal/stage/notarius_cache_integration_test.go`

docs/internal/workspace.md

@@ -37,6 +37,12 @@ validates and renames the external bundle to the unique immutable Notarius
 bundle path supplied by `internal/artifacts`. Exact physical paths belong in
 [Operations](../operations.md#extraction-workflow).
 
+Prepared references delivered to Notarius use session-durable paths owned by
+`internal/artifacts` beneath
+`cache/notarius/references/<sha256>/<prepared-basename>`. These verified
+content-addressed snapshots are outside run-local layout, survive ordinary
+published-run cleanup, and are excluded from publish and restore enumeration.
+
 ## Locking
 
 `artifacts.LocalStore` enforces the single-writer session lock via an

docs/operations.md

@@ -187,7 +187,20 @@ shared session lifecycle. In particular, render does not require Notarius or
 Scriptorium, extract does not require Scriptorium, and analyze does not require
 the transcription, Seriatim, Audita, or Notarius adapters.
 
-For the common post-transcript development loop, use:
+## Artifact Development Workflows
+
+Choose the narrowest workflow that matches the intended replacement:
+
+| Intent | Command | Preserved and rebuilt state |
+| --- | --- | --- |
+| Rebuild analysis only | [`narratio analyze 2026-04-04`](./cli.md#analyze) | Preserves transcripts and the current extraction bundle; force-runs selected analysis artifacts and their required analysis prerequisites. |
+| Rebuild extraction and analysis | [`narratio regenerate-artifacts 2026-04-04`](./cli.md#regenerate-artifacts) | Preserves transcripts; force-runs the complete configured extraction bundle, then selected analysis. `--artifacts` narrows analysis only. |
+| Continue bounded work normally | [`narratio run 2026-04-04 --from extract --through analyze`](./cli.md#run) | Uses ordinary resume decisions: current extraction and analysis can be skipped, while incomplete or obsolete work runs. |
+
+All three stop before publication. Inspect the results and run
+[`narratio publish`](./cli.md#publish) separately when delivery is intended.
+
+For the extraction-plus-analysis development loop, use:
 
 ```bash
 narratio regenerate-artifacts 2026-04-04
@@ -260,9 +273,14 @@ are not selectable or published implicitly.
 Configured Notarius references resolve only from the current manifest-backed
 prepared inputs. Their canonical locations are `inputs/party.yml`,
 `inputs/players.yml`, `inputs/glossary.yml`, and, when configured,
-`inputs/spell_catalog.json`. Extraction supplies Notarius with verified copies
-under `runs/<run_id>/extract/references/` so a concurrent refresh of canonical
-prepared files cannot change the bytes consumed by an in-flight invocation.
+`inputs/spell_catalog.json`. Extraction supplies Notarius with verified,
+content-addressed copies under
+`cache/notarius/references/<sha256>/<prepared-basename>`. The full lowercase
+SHA-256 digest identifies the reference bytes, so retries and later invocations
+reuse the same path while changed bytes select a new path. Older snapshots are
+retained and are not current artifact evidence. A concurrent refresh of
+canonical prepared files therefore cannot change the bytes consumed by an
+in-flight invocation.
 For a canonical party, preparation retains the validated authored party bytes
 at `inputs/party.yml` and generates `inputs/players.yml` from that roster.
 The manifest records their checksums separately, with the players input marked
@@ -297,6 +315,28 @@ Notarius extraction validates its run-local bundle and moves it once to the
 unique durable bundle path. A pre-existing durable destination is a collision;
 no duplicate promotion tree is created.
 
+`pipeline.notarius.resume` and `pipeline.notarius.chunk_cache` control work
+inside an executed Notarius invocation. Ordinary extraction forwards their
+configured values. Forced extraction always disables checkpoint loading and
+requests chunk-plan `refresh`, even when the configured mode is `bypass`.
+Refresh regenerates and replaces only the active input's accepted chunk plan;
+it is not cache deletion.
+
+Notarius checkpoint recording must be enabled in the deployed Notarius
+configuration. Narratio neither creates a Notarius pipeline/configuration nor
+reads it to infer checkpoint availability. Configure explicit persistent
+Notarius chunk-plan and checkpoint roots when reuse must survive separate
+Narratio subprocesses. A changed reference, transcript, selected profile,
+profile content, endpoint, or other Notarius checkpoint identity input can
+produce a legitimate cold miss.
+
+There are two independent resume boundaries: Narratio may skip a completed
+`extract` stage without invoking Notarius, while `--resume` allows Notarius to
+reuse compatible accepted work only after Narratio has decided extraction must
+execute. Deleting Narratio's local reference snapshots does not invalidate a
+complete durable extraction; missing snapshots are reconstructed only when a
+later extraction actually runs.
+
 ## External Command Lifecycle
 
 When an external command is cancelled or times out, Narratio terminates its
@@ -509,6 +549,7 @@ Durable session paths:
 - `logs/**`
 - `config/**`
 - `runs/**`
+- `cache/notarius/references/<sha256>/<prepared-basename>`
 
 `manifest.json` is the versioned session-progress document. Narratio accepts
 only `narratio.session-manifest.v1`. Do not edit unsupported session state to
@@ -519,6 +560,13 @@ relevant workflow requires.
 Validated Notarius bundles live below `artifacts/notarius/{run_id}/`; receipt,
 stderr, and pre-promotion output remain in the producing run's `extract`
 directory as described in [Extraction Workflow](#extraction-workflow).
+Stable reference snapshots live in the session cache path above. Ordinary
+published-run cleanup removes its selected run-local target but retains this
+cache, and publish/restore never enumerate it as an artifact. Remove a snapshot
+or digest directory manually only while no command is active for the session;
+regular missing or corrupt content is reconstructed on the next executed
+extraction, while symlinks, directories, and other unsafe conflicts must be
+investigated and removed deliberately.
 
 Run-local layout:
 

docs/troubleshooting.md

@@ -323,6 +323,59 @@ Do not point Notarius directly at the original source path or edit the manifest
 checksum. Relevant references: [Notarius reference configuration](./config.md#notarius-reference-bindings)
 and [Operations: Extraction Workflow](./operations.md#extraction-workflow).
 
+## Notarius reference snapshot cache conflict
+
+Symptom:
+
+- extraction reports that a path beneath
+  `cache/notarius/references/<sha256>/` is a symlink, directory, or another
+  non-regular or unsafe entry.
+
+Safe fix:
+
+- stop commands for the affected session and inspect the exact reported path;
+- investigate unexpected links or special files as possible tampering; and
+- remove only the reported conflicting leaf or digest directory after
+  confirming it belongs to the reconstructible reference cache.
+
+A missing snapshot or corrupt regular snapshot is reconstructed automatically
+from currently verified prepared input. Do not recursively delete the session
+root, edit the manifest checksum, or remove cache entries while extraction is
+active.
+
+Relevant reference: [Operations: Extraction Workflow](./operations.md#extraction-workflow).
+
+## Notarius resume executes work instead of reusing it
+
+Symptom:
+
+- Narratio executes `extract`, and Notarius makes provider calls despite
+  `pipeline.notarius.resume: true`.
+
+Likely causes:
+
+- checkpoint recording is disabled in the Notarius configuration;
+- Notarius is using a nonpersistent or different checkpoint root;
+- transcript/reference bytes, reference paths, pipeline/profile content,
+  local-backend endpoint, selected lanes, or another checkpoint identity input
+  changed; or
+- Narratio extraction was forced, which deliberately omits `--resume`.
+
+Safe fix:
+
+- validate the deployed Notarius configuration and configure explicit
+  persistent chunk-plan and checkpoint roots;
+- inspect Notarius checkpoint decision reason codes for a legitimate cold miss;
+  and
+- retry without force only when reuse is intended.
+
+Remember that Narratio skipping a completed `extract` stage is separate from
+Notarius reusing work inside an extraction that does execute. Narratio does not
+read the Notarius configuration to infer checkpoint enablement.
+
+Relevant references: [Notarius cache controls](./config.md#notarius-cache-controls)
+and [Notarius integration](./integrations/notarius.md).
+
 ## Notarius reference selector or generated-handoff collision
 
 Symptom:

examples/README.md

@@ -60,7 +60,7 @@ with the sample campaign and a compatible local- or S3-audio session.
   [glossary](campaigns/sample-campaign/glossary.yml), and canonical
   [party](campaigns/sample-campaign/party.yml) fixture, plus an optional
   [spell-catalog overlay](campaigns/sample-campaign/spell_catalog.json) that
-  follows the Notarius v0.6 contract. Narratio derives the players projection
+  follows the Notarius v0.7 contract. Narratio derives the players projection
   from this party source.
 - [Sample speaker audio](audio/sample-speaker.flac) is a text placeholder that
   reserves the expected filename and directory shape. Replace it with a real

examples/pipeline.full.annotated.yml

@@ -128,7 +128,7 @@ trim:
     report: false
 
 notarius:
-  # Optional structured extraction between trim and render.
+  # Optional structured extraction after render and before analyze.
   enabled: true
   binary: notarius
   # Optional environment-variable reference. Match this name to api_key_env in
@@ -138,6 +138,12 @@ notarius:
   pipeline_id: dnd-session
   timeout: 3h
   working_directory: /usr/local/etc/notarius
+  # Reuse compatible accepted Notarius checkpoints during executed extraction.
+  # The Notarius config must enable recording and provide persistent cache roots.
+  resume: true
+  # auto uses or creates a source-addressed chunk plan. Forced Narratio
+  # extraction overrides this with refresh and disables checkpoint loading.
+  chunk_cache: auto
   # External campaign references use prepared Narratio source IDs. Omit an
   # optional binding when the selected Notarius pipeline does not need it.
   references:

internal/adapters/notarius/runner.go

@@ -35,6 +35,8 @@ type RunRequest struct {
 	ReceiptPath      string
 	LogPath          string
 	Timeout          time.Duration
+	Resume           bool
+	ChunkCache       string
 	References       []ReferenceBinding
 }
 

internal/adapters/notarius/subprocess.go

@@ -73,6 +73,12 @@ func (r *SubprocessRunner) Run(ctx context.Context, req RunRequest) (RunResult,
 		"--input", req.InputPath,
 		"--output-dir", req.OutputRoot,
 	}
+	if req.Resume {
+		args = append(args, "--resume")
+	}
+	if req.ChunkCache != "" {
+		args = append(args, "--chunk_cache", req.ChunkCache)
+	}
 	for _, reference := range references {
 		args = append(args, "--reference", reference.Selector+"="+reference.Path)
 	}
@@ -159,6 +165,9 @@ func validateRunRequest(req RunRequest) ([]ReferenceBinding, error) {
 	if req.Timeout <= 0 {
 		return nil, fmt.Errorf("notarius timeout must be positive")
 	}
+	if !validChunkCacheMode(req.ChunkCache) {
+		return nil, fmt.Errorf("notarius chunk cache mode must be empty, auto, bypass, or refresh")
+	}
 	for label, path := range map[string]string{
 		"config":            req.ConfigPath,
 		"input":             req.InputPath,
@@ -217,6 +226,15 @@ func validateRunRequest(req RunRequest) ([]ReferenceBinding, error) {
 	return references, nil
 }
 
+func validChunkCacheMode(mode string) bool {
+	switch mode {
+	case "", "auto", "bypass", "refresh":
+		return true
+	default:
+		return false
+	}
+}
+
 type receiptDocument struct {
 	SchemaVersion             string                      `json:"schema_version"`
 	RunID                     string                      `json:"run_id"`
@@ -264,7 +282,7 @@ func validateValidationSummaries(documents []validationSummaryDocument) ([]Valid
 		if document.Status != "complete" && document.Status != "rejected" && document.Status != "incomplete" {
 			return nil, fmt.Errorf("notarius validation summary status %q is invalid", document.Status)
 		}
-		if document.ProducerAttemptCount == nil || *document.ProducerAttemptCount <= 0 || !validTerminalAction(document.TerminalAction) {
+		if document.ProducerAttemptCount == nil || *document.ProducerAttemptCount < 0 || !validTerminalAction(document.TerminalAction) {
 			return nil, fmt.Errorf("notarius validation summary is missing required fields")
 		}
 		summaries = append(summaries, ValidationSummary{

internal/adapters/notarius/subprocess_test.go

@@ -77,6 +77,68 @@ func TestSubprocessRunnerBuildsExactInvocationAndDiscoversBundle(t *testing.T) {
 	}
 }
 
+func TestSubprocessRunnerMapsCacheControlsToArguments(t *testing.T) {
+	tests := []struct {
+		name       string
+		resume     bool
+		chunkCache string
+		want       []string
+	}{
+		{name: "resume", resume: true, want: []string{"--resume"}},
+		{name: "auto", chunkCache: "auto", want: []string{"--chunk_cache", "auto"}},
+		{name: "bypass", chunkCache: "bypass", want: []string{"--chunk_cache", "bypass"}},
+		{name: "refresh", chunkCache: "refresh", want: []string{"--chunk_cache", "refresh"}},
+		{name: "resume and mode", resume: true, chunkCache: "auto", want: []string{"--resume", "--chunk_cache", "auto"}},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			req := validRunRequest(t)
+			req.Resume = test.resume
+			req.ChunkCache = test.chunkCache
+			var captured sharedsubprocess.RunRequest
+			runner := &SubprocessRunner{run: func(_ context.Context, processReq sharedsubprocess.RunRequest) (sharedsubprocess.RunResult, error) {
+				captured = processReq
+				writeValidBundleAndReceipt(t, req, false)
+				return sharedsubprocess.RunResult{ExitCode: 0}, nil
+			}}
+
+			if _, err := runner.Run(context.Background(), req); err != nil {
+				t.Fatalf("Run() error = %v", err)
+			}
+			want := []string{
+				"run", req.PipelineID, "--config", req.ConfigPath, "--input", req.InputPath,
+				"--output-dir", req.OutputRoot,
+			}
+			want = append(want, test.want...)
+			want = append(want, "--json")
+			if !reflect.DeepEqual(captured.Args, want) {
+				t.Fatalf("subprocess args = %#v, want %#v", captured.Args, want)
+			}
+		})
+	}
+}
+
+func TestSubprocessRunnerRejectsInvalidChunkCacheBeforeLaunch(t *testing.T) {
+	for _, mode := range []string{"AUTO", "invalid", " auto "} {
+		t.Run(mode, func(t *testing.T) {
+			req := validRunRequest(t)
+			req.ChunkCache = mode
+			started := false
+			runner := &SubprocessRunner{run: func(context.Context, sharedsubprocess.RunRequest) (sharedsubprocess.RunResult, error) {
+				started = true
+				return sharedsubprocess.RunResult{}, nil
+			}}
+			_, err := runner.Run(context.Background(), req)
+			if err == nil || !strings.Contains(err.Error(), "chunk cache mode") {
+				t.Fatalf("Run() error = %v, want chunk cache validation failure", err)
+			}
+			if started {
+				t.Fatal("subprocess started with invalid chunk cache mode")
+			}
+		})
+	}
+}
+
 func TestSubprocessRunnerForwardsLoadedCredentialWithoutRereadingEnvironment(t *testing.T) {
 	const envName = "NARRATIO_NOTARIUS_CUSTOM_LLM_KEY"
 	const loadedValue = "loaded-notarius-credential"
@@ -704,6 +766,26 @@ func assertRedactedCredentialBytes(t *testing.T, data []byte, canary string) {
 	}
 }
 
+func TestValidateValidationSummariesAcceptsReusedProducerWithoutAttempts(t *testing.T) {
+	zero := 0
+	summaries, err := validateValidationSummaries([]validationSummaryDocument{{
+		Stage: "chunk", Status: "complete", ProducerAttemptCount: &zero, TerminalAction: "accepted",
+	}})
+	if err != nil {
+		t.Fatalf("validateValidationSummaries() error = %v", err)
+	}
+	if len(summaries) != 1 || summaries[0].ProducerAttemptCount != 0 {
+		t.Fatalf("summaries = %#v, want one reused producer with zero attempts", summaries)
+	}
+
+	negative := -1
+	if _, err := validateValidationSummaries([]validationSummaryDocument{{
+		Stage: "chunk", Status: "complete", ProducerAttemptCount: &negative, TerminalAction: "accepted",
+	}}); err == nil {
+		t.Fatal("validateValidationSummaries() negative producer attempts error = nil")
+	}
+}
+
 func writeValidBundleAndReceipt(t *testing.T, req RunRequest, includeUnknown bool) {
 	t.Helper()
 	bundle := filepath.Join(req.OutputRoot, "notarius-run-1")

internal/app/extract_lifecycle_test.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"reflect"
 	"sort"
 	"strings"
 	"testing"
@@ -182,12 +183,16 @@ func TestExtractLifecyclePrepareBindsVerifiedReferenceSnapshots(t *testing.T) {
 	for index, expected := range want {
 		binding := request.References[index]
 		canonical := filepath.Join(paths.InputsDir, expected.filename)
-		snapshot := filepath.Join(
-			artifacts.SessionRunNotariusReferencesDirForCampaign(
-				cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, loaded.RunID,
-			),
-			expected.filename,
+		descriptor, ok := artifactpolicy.DescribePreparedInputSource(expected.sourceID)
+		if !ok {
+			t.Fatalf("DescribePreparedInputSource(%q) ok = false", expected.sourceID)
+		}
+		snapshot, err := artifacts.SessionNotariusReferenceSnapshotPath(
+			paths, lifecycleInputChecksum(t, loaded, descriptor.ManifestKind), expected.filename,
 		)
+		if err != nil {
+			t.Fatalf("SessionNotariusReferenceSnapshotPath(%q) error = %v", expected.sourceID, err)
+		}
 		if binding.Selector != expected.selector || binding.Path != snapshot || binding.Path == canonical || binding.Path == originalPaths[expected.sourceID] {
 			t.Fatalf("reference[%d] = %#v, want selector %q snapshot %q and not prepared/source paths", index, binding, expected.selector, snapshot)
 		}
@@ -394,6 +399,140 @@ func TestExtractLifecycleFailureInvalidatesAndOrdinaryRetryRerunsDownstream(t *t
 	}
 }
 
+func TestExtractLifecycleRetryReusesStableSnapshotsAndChangedReferenceSelectsNewPath(t *testing.T) {
+	cfg, env, runner := extractionLifecycleFixture(t, true)
+	configureLifecycleReferences(t, cfg)
+	runner.failuresRemaining = 1
+	plan := prepareExtractLifecyclePlan(t)
+
+	if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err == nil {
+		t.Fatal("first executeStages() error = nil, want extraction failure")
+	}
+	if len(runner.requests) != 1 {
+		t.Fatalf("failed requests = %d, want 1", len(runner.requests))
+	}
+	firstPaths := lifecycleReferencePaths(runner.requests[0])
+	if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err != nil {
+		t.Fatalf("retry executeStages() error = %v", err)
+	}
+	if len(runner.requests) != 2 {
+		t.Fatalf("retry requests = %d, want 2", len(runner.requests))
+	}
+	if secondPaths := lifecycleReferencePaths(runner.requests[1]); !reflect.DeepEqual(secondPaths, firstPaths) {
+		t.Fatalf("snapshot paths changed on retry:\nfirst %#v\nsecond %#v", firstPaths, secondPaths)
+	}
+
+	partyDocument, err := config.ParseParty([]byte(`schema_version: narratio.party.v1
+characters:
+  changed:
+    player: {name: Eric}
+    character:
+      name: Changed
+      classes: [{name: wizard}]
+`))
+	if err != nil {
+		t.Fatalf("ParseParty(changed party) error = %v", err)
+	}
+	cfg.Party.Canonical = partyDocument
+	persisted := loadLifecycleManifest(t, cfg)
+	prepare, err := stage.Select("prepare")
+	if err != nil {
+		t.Fatalf("stage.Select(prepare) error = %v", err)
+	}
+	if _, err := prepare.Run(context.Background(), env, persisted); err != nil {
+		t.Fatalf("prepare.Run() error = %v", err)
+	}
+	if err := (&manifest.LocalStore{}).Save(context.Background(), manifestPathFor(cfg), persisted); err != nil {
+		t.Fatalf("Save(reprepared manifest) error = %v", err)
+	}
+	extractPlan, _ := BuildSingleStagePlan("extract")
+	if _, err := executeStages(context.Background(), cfg, extractPlan, RunOptions{Env: env}); err != nil {
+		t.Fatalf("changed-reference executeStages() error = %v", err)
+	}
+	if len(runner.requests) != 3 {
+		t.Fatalf("changed-reference requests = %d, want 3", len(runner.requests))
+	}
+	thirdPaths := lifecycleReferencePaths(runner.requests[2])
+	if thirdPaths["party"] == firstPaths["party"] {
+		t.Fatalf("changed party retained snapshot path %q", thirdPaths["party"])
+	}
+	if thirdPaths["players"] == firstPaths["players"] {
+		t.Fatalf("changed party retained derived players snapshot path %q", thirdPaths["players"])
+	}
+	for selector, firstPath := range firstPaths {
+		if selector != "party" && selector != "players" && thirdPaths[selector] != firstPath {
+			t.Fatalf("unchanged %s snapshot path = %q, want %q", selector, thirdPaths[selector], firstPath)
+		}
+		if _, err := os.Stat(firstPath); err != nil {
+			t.Fatalf("old %s snapshot %q was not retained: %v", selector, firstPath, err)
+		}
+	}
+}
+
+func TestExtractLifecycleCacheControlsAndSnapshotLossDoNotInvalidateCompleteResult(t *testing.T) {
+	cfg, env, runner := extractionLifecycleFixture(t, true)
+	configureLifecycleReferences(t, cfg)
+	if _, err := executeStages(context.Background(), cfg, prepareExtractLifecyclePlan(t), RunOptions{Env: env}); err != nil {
+		t.Fatalf("initial executeStages() error = %v", err)
+	}
+	paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)
+	cacheRoot := artifacts.SessionNotariusReferenceCacheDir(paths)
+	if err := os.RemoveAll(cacheRoot); err != nil {
+		t.Fatalf("RemoveAll(snapshot cache) error = %v", err)
+	}
+	cfg.Pipeline.Notarius.Resume = true
+	cfg.Pipeline.Notarius.ChunkCache = "bypass"
+	extractPlan, _ := BuildSingleStagePlan("extract")
+	summary, err := executeStages(context.Background(), cfg, extractPlan, RunOptions{Env: env})
+	if err != nil {
+		t.Fatalf("resume executeStages() error = %v", err)
+	}
+	if len(summary.Executed) != 0 || !reflect.DeepEqual(summary.Skipped, []string{"extract"}) || len(runner.requests) != 1 {
+		t.Fatalf("resume summary = %#v requests=%d", summary, len(runner.requests))
+	}
+	if _, err := os.Stat(cacheRoot); !errors.Is(err, os.ErrNotExist) {
+		t.Fatalf("skipped extraction recreated snapshot cache: %v", err)
+	}
+}
+
+func TestAnalyzeOnlyLifecyclePreservesExtractionWithoutInvokingNotarius(t *testing.T) {
+	cfg, env, runner := extractionLifecycleFixture(t, true)
+	extractPlan, _ := BuildSingleStagePlan("extract")
+	if _, err := executeStages(context.Background(), cfg, extractPlan, RunOptions{Env: env}); err != nil {
+		t.Fatalf("extract executeStages() error = %v", err)
+	}
+	before := loadLifecycleManifest(t, cfg).Stages["extract"]
+	beforeJSON, _ := json.Marshal(before)
+	analyzeRuns := 0
+	if _, err := executeStages(context.Background(), cfg, []stage.Stage{countingStage{name: "analyze", runs: &analyzeRuns}}, RunOptions{Env: env, Force: true}); err != nil {
+		t.Fatalf("analyze-only executeStages() error = %v", err)
+	}
+	after := loadLifecycleManifest(t, cfg).Stages["extract"]
+	afterJSON, _ := json.Marshal(after)
+	if analyzeRuns != 1 || len(runner.requests) != 1 || string(afterJSON) != string(beforeJSON) {
+		t.Fatalf("analyze-only lifecycle changed extraction: analyze=%d requests=%d\nbefore=%s\nafter=%s", analyzeRuns, len(runner.requests), beforeJSON, afterJSON)
+	}
+	for _, output := range after.Outputs {
+		if _, err := os.Stat(output.LocalPath); err != nil {
+			t.Fatalf("analyze-only lifecycle removed extract output %q: %v", output.LocalPath, err)
+		}
+	}
+}
+
+func TestForcedArtifactRegenerationUsesOrdinaryExtractAndAnalyzePath(t *testing.T) {
+	cfg, env, runner := extractionLifecycleFixture(t, true)
+	cfg.Pipeline.Notarius.Resume = true
+	cfg.Pipeline.Notarius.ChunkCache = "bypass"
+	analyzeRuns := 0
+	plan := extractionLifecyclePlan(t, &analyzeRuns)
+	if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env, Force: true, SelectedArtifacts: []string{"session_recap"}}); err != nil {
+		t.Fatalf("forced artifact regeneration error = %v", err)
+	}
+	if len(runner.requests) != 1 || runner.requests[0].Resume || runner.requests[0].ChunkCache != "refresh" || analyzeRuns != 1 {
+		t.Fatalf("forced path request=%#v analyze=%d", runner.requests, analyzeRuns)
+	}
+}
+
 func TestExtractLifecycleForcedSelfSkipInvalidatesDownstream(t *testing.T) {
 	cfg, env, _ := extractionLifecycleFixture(t, false)
 	markLifecycleStageSucceeded(t, cfg, "analyze")
@@ -783,6 +922,14 @@ func lifecycleInputChecksum(t *testing.T, m *manifest.Manifest, kind string) str
 	return ""
 }
 
+func lifecycleReferencePaths(request notarius.RunRequest) map[string]string {
+	paths := make(map[string]string, len(request.References))
+	for _, reference := range request.References {
+		paths[reference.Selector] = reference.Path
+	}
+	return paths
+}
+
 func lifecycleDndOutputs() map[string]config.NotariusOutputConfig {
 	return map[string]config.NotariusOutputConfig{
 		"item_registry":        {LaneID: "item-registry", MediaType: "application/json", SchemaID: "notarius.dnd.item_registry", SchemaVersion: "v1", ModuleKey: "dnd/item-registry"},

internal/app/restore_plan_test.go

@@ -24,6 +24,7 @@ func TestRestorePlanDefaultScope(t *testing.T) {
 	seedCurrentRestoreObject(t, store, current, current.SessionPrefix+"audio/alice.flac", []byte("audio"))
 	seedRestoreObject(store, current.SessionPrefix+"runs/20260519T010203Z-a1b2/manifest.json", []byte("{}"))
 	seedRestoreObject(store, current.SessionPrefix+"logs/publish.log", []byte("log"))
+	seedRestoreObject(store, current.SessionPrefix+"cache/notarius/references/ignored/party.yml", []byte("cache"))
 
 	plan, err := buildRestorePlan(context.Background(), cfg, current, store, RestorePlanOptions{})
 	if err != nil {

internal/artifacts/paths.go

@@ -10,6 +10,8 @@ import (
 	"gitea.maximumdirect.net/eric/narratio/internal/pathsafe"
 )
 
+const sha256HexLength = 64
+
 // SessionPaths contains canonical local paths for one session work directory.
 type SessionPaths struct {
 	WorkspaceRoot         string
@@ -28,6 +30,7 @@ type SessionPaths struct {
 	CurrentDir            string
 	RunsDir               string
 	PreviousDir           string
+	CacheDir              string
 	PreviousManifestPath  string
 	PreviousArtifactsDir  string
 	ManifestPath          string
@@ -108,17 +111,37 @@ func SessionRunNotariusLogPathForCampaign(rootDir, campaign, sessionID, runID st
 	return filepath.Join(SessionRunExtractDirForCampaign(rootDir, campaign, sessionID, runID), "notarius.stderr.log")
 }
 
-// SessionRunNotariusReferencesDirForCampaign returns the invocation-local
-// directory containing verified reference snapshots supplied to Notarius.
-func SessionRunNotariusReferencesDirForCampaign(rootDir, campaign, sessionID, runID string) string {
-	return filepath.Join(SessionRunExtractDirForCampaign(rootDir, campaign, sessionID, runID), "references")
-}
-
 // SessionRunNotariusOutputRootForCampaign returns the invocation-local Notarius output root.
 func SessionRunNotariusOutputRootForCampaign(rootDir, campaign, sessionID, runID string) string {
 	return filepath.Join(SessionRunExtractDirForCampaign(rootDir, campaign, sessionID, runID), "notarius-output")
 }
 
+// SessionNotariusReferenceCacheDir returns the stable cache root for verified
+// reference snapshots supplied to Notarius.
+func SessionNotariusReferenceCacheDir(paths SessionPaths) string {
+	return filepath.Join(paths.CacheDir, "notarius", "references")
+}
+
+// SessionNotariusReferenceSnapshotPath returns the content-addressed path for
+// one verified prepared reference snapshot.
+func SessionNotariusReferenceSnapshotPath(paths SessionPaths, checksum, preparedBasename string) (string, error) {
+	if len(checksum) != sha256HexLength {
+		return "", fmt.Errorf("reference checksum must be a full lowercase SHA-256 digest")
+	}
+	for _, character := range checksum {
+		if (character < '0' || character > '9') && (character < 'a' || character > 'f') {
+			return "", fmt.Errorf("reference checksum must be a full lowercase SHA-256 digest")
+		}
+	}
+	if err := pathsafe.ValidateOpaqueSegment(preparedBasename); err != nil {
+		return "", fmt.Errorf("prepared reference basename: %w", err)
+	}
+	return pathsafe.JoinSlashRelativeUnderRoot(
+		SessionNotariusReferenceCacheDir(paths),
+		checksum+"/"+preparedBasename,
+	)
+}
+
 // SessionNotariusBundleDirForCampaign returns one immutable durable bundle destination.
 func SessionNotariusBundleDirForCampaign(rootDir, campaign, sessionID, runID string) string {
 	return filepath.Join(
@@ -268,6 +291,7 @@ func buildSessionPathsFromRoot(workspaceRoot, campaign, sessionID, root string)
 		CurrentDir:            filepath.Join(root, config.PathCurrentDirSegment),
 		RunsDir:               filepath.Join(root, config.PathRunsDirSegment),
 		PreviousDir:           filepath.Join(root, config.PathPreviousDirSegment),
+		CacheDir:              filepath.Join(root, config.PathCacheDirSegment),
 		PreviousManifestPath:  filepath.Join(root, config.PathPreviousDirSegment, config.PathManifestFile),
 		PreviousArtifactsDir:  filepath.Join(root, config.PathPreviousDirSegment, config.PathArtifactsDirSegment),
 		ManifestPath:          filepath.Join(root, config.PathManifestFile),

internal/artifacts/paths_model_test.go

@@ -2,6 +2,7 @@ package artifacts
 
 import (
 	"path/filepath"
+	"strings"
 	"testing"
 )
 
@@ -54,7 +55,6 @@ func TestSessionNotariusPathsForCampaign(t *testing.T) {
 		{name: "extract directory", got: SessionRunExtractDirForCampaign(root, campaign, sessionID, runID), want: extractDir},
 		{name: "receipt", got: SessionRunNotariusReceiptPathForCampaign(root, campaign, sessionID, runID), want: filepath.Join(extractDir, "notarius.receipt.json")},
 		{name: "stderr", got: SessionRunNotariusLogPathForCampaign(root, campaign, sessionID, runID), want: filepath.Join(extractDir, "notarius.stderr.log")},
-		{name: "references", got: SessionRunNotariusReferencesDirForCampaign(root, campaign, sessionID, runID), want: filepath.Join(extractDir, "references")},
 		{name: "output root", got: SessionRunNotariusOutputRootForCampaign(root, campaign, sessionID, runID), want: filepath.Join(extractDir, "notarius-output")},
 		{name: "durable bundle", got: SessionNotariusBundleDirForCampaign(root, campaign, sessionID, runID), want: filepath.Join(root, "work", campaign, sessionID, "artifacts", "notarius", runID)},
 	}
@@ -67,6 +67,43 @@ func TestSessionNotariusPathsForCampaign(t *testing.T) {
 	}
 }
 
+func TestSessionNotariusReferenceSnapshotPath(t *testing.T) {
+	paths := buildSessionPaths("/tmp/workspace", "forsaken", "2026-04-19")
+	checksum := strings.Repeat("a", 64)
+	got, err := SessionNotariusReferenceSnapshotPath(paths, checksum, "party.yml")
+	if err != nil {
+		t.Fatalf("SessionNotariusReferenceSnapshotPath() error = %v", err)
+	}
+	wantRoot := filepath.Join(paths.Root, "cache", "notarius", "references")
+	if root := SessionNotariusReferenceCacheDir(paths); root != wantRoot {
+		t.Fatalf("SessionNotariusReferenceCacheDir() = %q, want %q", root, wantRoot)
+	}
+	if want := filepath.Join(wantRoot, checksum, "party.yml"); got != want {
+		t.Fatalf("SessionNotariusReferenceSnapshotPath() = %q, want %q", got, want)
+	}
+}
+
+func TestSessionNotariusReferenceSnapshotPathRejectsUnsafeIdentity(t *testing.T) {
+	paths := buildSessionPaths("/tmp/workspace", "forsaken", "2026-04-19")
+	validChecksum := strings.Repeat("a", 64)
+	for _, test := range []struct {
+		name, checksum, basename string
+	}{
+		{name: "short digest", checksum: "abc", basename: "party.yml"},
+		{name: "uppercase digest", checksum: strings.Repeat("A", 64), basename: "party.yml"},
+		{name: "nonhex digest", checksum: strings.Repeat("g", 64), basename: "party.yml"},
+		{name: "empty basename", checksum: validChecksum},
+		{name: "nested basename", checksum: validChecksum, basename: "nested/party.yml"},
+		{name: "parent basename", checksum: validChecksum, basename: ".."},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			if got, err := SessionNotariusReferenceSnapshotPath(paths, test.checksum, test.basename); err == nil {
+				t.Fatalf("SessionNotariusReferenceSnapshotPath() = %q, want error", got)
+			}
+		})
+	}
+}
+
 func TestSessionPreviousPathsForCampaign(t *testing.T) {
 	root := "/tmp/workspace"
 	previousDir := SessionPreviousDirForCampaign(root, "forsaken", "2026-04-19")

internal/config/config.go

@@ -310,6 +310,8 @@ type NotariusConfig struct {
 	Enabled          bool                            `yaml:"enabled" json:"enabled"`
 	Binary           string                          `yaml:"binary" json:"binary"`
 	LLMAPIKey        *configkit.Secret               `yaml:"llm_api_key,omitempty" json:"llm_api_key,omitempty"`
+	Resume           bool                            `yaml:"resume,omitempty" json:"resume,omitempty"`
+	ChunkCache       string                          `yaml:"chunk_cache,omitempty" json:"chunk_cache,omitempty"`
 	ConfigPath       string                          `yaml:"config_path" json:"config_path"`
 	PipelineID       string                          `yaml:"pipeline_id" json:"pipeline_id"`
 	Timeout          string                          `yaml:"timeout" json:"timeout"`

internal/config/credential_projection.go

@@ -56,6 +56,8 @@ func (cfg NotariusConfig) MarshalYAML() (any, error) {
 		Enabled          bool                            `yaml:"enabled"`
 		Binary           string                          `yaml:"binary"`
 		LLMAPIKey        *string                         `yaml:"llm_api_key,omitempty"`
+		Resume           bool                            `yaml:"resume,omitempty"`
+		ChunkCache       string                          `yaml:"chunk_cache,omitempty"`
 		ConfigPath       string                          `yaml:"config_path"`
 		PipelineID       string                          `yaml:"pipeline_id"`
 		Timeout          string                          `yaml:"timeout"`
@@ -65,6 +67,7 @@ func (cfg NotariusConfig) MarshalYAML() (any, error) {
 	}
 	return safeNotarius{
 		Enabled: cfg.Enabled, Binary: cfg.Binary, LLMAPIKey: secretReferenceName(cfg.LLMAPIKey),
+		Resume: cfg.Resume, ChunkCache: cfg.ChunkCache,
 		ConfigPath: cfg.ConfigPath, PipelineID: cfg.PipelineID, Timeout: cfg.Timeout,
 		WorkingDirectory: cfg.WorkingDirectory, References: cfg.References, Outputs: cfg.Outputs,
 	}, nil

internal/config/defaults.go

@@ -78,6 +78,7 @@ const (
 	PathCurrentDirSegment      = "current"
 	PathRunsDirSegment         = "runs"
 	PathPreviousDirSegment     = "previous"
+	PathCacheDirSegment        = "cache"
 	PathManifestFile           = "manifest.json"
 	PathLockFile               = ".lock"
 	PathTranscriptBase         = artifactmodel.TranscriptPathBase

internal/config/notarius_test.go

@@ -1,12 +1,15 @@
 package config
 
 import (
+	"encoding/json"
 	"fmt"
 	"os"
 	"path/filepath"
 	"reflect"
 	"strings"
 	"testing"
+
+	"gopkg.in/yaml.v3"
 )
 
 func TestNotariusOmittedAndDisabledBehavior(t *testing.T) {
@@ -45,6 +48,9 @@ notarius:
 	if disabled.Notarius.Binary != DefaultNotariusBinary || disabled.Notarius.Timeout != DefaultNotariusTimeout {
 		t.Fatalf("disabled defaults = %#v", disabled.Notarius)
 	}
+	if disabled.Notarius.Resume || disabled.Notarius.ChunkCache != "" {
+		t.Fatalf("disabled cache-control defaults = %#v", disabled.Notarius)
+	}
 	if disabled.Notarius.ConfigPath != "relative/notarius.yml" || disabled.Notarius.WorkingDirectory != "" {
 		t.Fatalf("disabled paths were resolved unexpectedly: %#v", disabled.Notarius)
 	}
@@ -187,6 +193,89 @@ notarius:
 	}
 }
 
+func TestNotariusCacheControlNormalizationAndValidation(t *testing.T) {
+	for _, mode := range []string{"", "auto", "bypass", "refresh"} {
+		t.Run("mode "+mode, func(t *testing.T) {
+			cfg := validNotariusConfig()
+			cfg.Resume = true
+			if mode != "" {
+				cfg.ChunkCache = "  " + mode + "  "
+			}
+			if err := NormalizePipelineConfig(&PipelineConfig{Notarius: cfg}); err != nil {
+				t.Fatalf("NormalizePipelineConfig() error = %v", err)
+			}
+			if cfg.ChunkCache != mode {
+				t.Fatalf("ChunkCache = %q, want %q", cfg.ChunkCache, mode)
+			}
+			if !cfg.Resume {
+				t.Fatal("Resume = false, want true")
+			}
+			if err := validateNotarius(cfg, nil); err != nil {
+				t.Fatalf("validateNotarius() error = %v", err)
+			}
+		})
+	}
+
+	for _, mode := range []string{"AUTO", "invalid"} {
+		t.Run("invalid "+mode, func(t *testing.T) {
+			cfg := validNotariusConfig()
+			cfg.ChunkCache = mode
+			err := validateNotarius(cfg, nil)
+			if err == nil || !strings.Contains(err.Error(), "pipeline.notarius.chunk_cache") {
+				t.Fatalf("validateNotarius() error = %v, want chunk_cache error", err)
+			}
+		})
+	}
+
+	disabled := &NotariusConfig{Enabled: false, Resume: true, ChunkCache: " INVALID "}
+	if err := NormalizePipelineConfig(&PipelineConfig{Notarius: disabled}); err != nil {
+		t.Fatalf("NormalizePipelineConfig(disabled) error = %v", err)
+	}
+	if disabled.ChunkCache != " INVALID " {
+		t.Fatalf("disabled ChunkCache = %q, want unchanged", disabled.ChunkCache)
+	}
+	if err := validateNotarius(disabled, nil); err != nil {
+		t.Fatalf("validateNotarius(disabled) error = %v", err)
+	}
+}
+
+func TestNotariusCacheControlsSafeSerialization(t *testing.T) {
+	cfg := validNotariusConfig()
+	cfg.Resume = true
+	cfg.ChunkCache = "refresh"
+
+	yamlData, err := yaml.Marshal(cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	jsonData, err := json.Marshal(cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(string(yamlData), "resume: true") || !strings.Contains(string(yamlData), "chunk_cache: refresh") {
+		t.Fatalf("YAML omitted cache controls: %s", yamlData)
+	}
+	if !strings.Contains(string(jsonData), `"resume":true`) || !strings.Contains(string(jsonData), `"chunk_cache":"refresh"`) {
+		t.Fatalf("JSON omitted cache controls: %s", jsonData)
+	}
+
+	cfg.Resume = false
+	cfg.ChunkCache = ""
+	yamlData, err = yaml.Marshal(cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	jsonData, err = json.Marshal(cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, serialized := range []string{string(yamlData), string(jsonData)} {
+		if strings.Contains(serialized, "resume") || strings.Contains(serialized, "chunk_cache") {
+			t.Fatalf("zero-value cache controls were serialized: %s", serialized)
+		}
+	}
+}
+
 func TestNotariusRelativePathsDoNotMakeEffectiveDigestLocationDependent(t *testing.T) {
 	pipelineYAML := testPipelineBaseYAML + `
 notarius:

internal/config/pipeline_configkit_test.go

@@ -182,6 +182,57 @@ func TestNotariusCredentialRootProvenance(t *testing.T) {
 	t.Fatal("Notarius credential provenance did not identify the root pipeline")
 }
 
+func TestNotariusCacheControlProfileReplacementAndNullReset(t *testing.T) {
+	tests := []struct {
+		name       string
+		overlay    string
+		wantResume bool
+		wantMode   string
+		wantSource bool
+		wantNil    bool
+	}{
+		{name: "replacement", overlay: "notarius:\n  resume: true\n  chunk_cache: refresh\n", wantResume: true, wantMode: "refresh", wantSource: true},
+		{name: "null reset", overlay: "notarius: null\n", wantNil: true},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			dir := t.TempDir()
+			root := writePipelineFixture(t, dir, "pipeline.yml", "whisperx:\n  transcribe_url: https://transcription.example.test/transcribe\nnotarius:\n  enabled: false\n  resume: false\n  chunk_cache: bypass\n")
+			overlay := writePipelineFixture(t, dir, "profiles/testing.yml", test.overlay)
+			profile := "testing"
+			cfg, err := LoadPipelineWithOptions(root, PipelineLoadOptions{Profile: &profile})
+			if err != nil {
+				t.Fatal(err)
+			}
+			if test.wantNil {
+				if cfg.Notarius != nil {
+					t.Fatalf("Notarius = %#v, want null reset", cfg.Notarius)
+				}
+				return
+			}
+			if cfg.Notarius == nil || cfg.Notarius.Resume != test.wantResume || cfg.Notarius.ChunkCache != test.wantMode {
+				t.Fatalf("cache controls = resume %v, mode %q; want resume %v, mode %q", cfg.Notarius.Resume, cfg.Notarius.ChunkCache, test.wantResume, test.wantMode)
+			}
+			if !test.wantSource {
+				return
+			}
+			sources, err := EffectivePipelineSources(cfg, ResolvedParty{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			found := map[string]bool{}
+			for _, source := range sources {
+				if source.Path == "notarius.resume" || source.Path == "notarius.chunk_cache" {
+					found[source.Path] = source.Source == overlay
+				}
+			}
+			if !found["notarius.resume"] || !found["notarius.chunk_cache"] {
+				t.Fatalf("cache-control provenance did not identify profile: %#v", found)
+			}
+		})
+	}
+}
+
 func TestPipelineConfigKitAllowsOneBaseDeclarationWithProfileOverride(t *testing.T) {
 	dir := t.TempDir()
 	writePipelineFixture(t, dir, "pipeline.yml", `configkit:

internal/config/pipeline_normalize.go

@@ -14,6 +14,7 @@ func NormalizePipelineConfig(cfg *PipelineConfig) error {
 	}
 	notarius := *cfg.Notarius
 	notarius.Binary = strings.TrimSpace(notarius.Binary)
+	notarius.ChunkCache = strings.TrimSpace(notarius.ChunkCache)
 	notarius.ConfigPath = cleanNotariusPath(notarius.ConfigPath)
 	notarius.PipelineID = strings.TrimSpace(notarius.PipelineID)
 	notarius.Timeout = strings.TrimSpace(notarius.Timeout)

internal/config/validate_notarius.go

@@ -39,10 +39,12 @@ func validateNotarius(cfg *NotariusConfig, scriptorium *ScriptoriumConfig) error
 	if len(cfg.References) > MaxNotariusReferenceBindings {
 		return fmt.Errorf("pipeline.notarius.references must contain at most %d bindings", MaxNotariusReferenceBindings)
 	}
-
 	if err := validateNotariusCanonicalScalars(cfg); err != nil {
 		return err
 	}
+	if !validNotariusChunkCacheMode(cfg.ChunkCache) {
+		return fmt.Errorf("pipeline.notarius.chunk_cache must be empty, auto, bypass, or refresh")
+	}
 	if err := validateNotariusReferences(cfg.References); err != nil {
 		return err
 	}
@@ -56,6 +58,7 @@ func validateNotariusCanonicalScalars(cfg *NotariusConfig) error {
 		want string
 	}{
 		{name: "binary", got: cfg.Binary, want: strings.TrimSpace(cfg.Binary)},
+		{name: "chunk_cache", got: cfg.ChunkCache, want: strings.TrimSpace(cfg.ChunkCache)},
 		{name: "config_path", got: cfg.ConfigPath, want: cleanNotariusPath(cfg.ConfigPath)},
 		{name: "pipeline_id", got: cfg.PipelineID, want: strings.TrimSpace(cfg.PipelineID)},
 		{name: "timeout", got: cfg.Timeout, want: strings.TrimSpace(cfg.Timeout)},
@@ -69,6 +72,15 @@ func validateNotariusCanonicalScalars(cfg *NotariusConfig) error {
 	return nil
 }
 
+func validNotariusChunkCacheMode(mode string) bool {
+	switch mode {
+	case "", "auto", "bypass", "refresh":
+		return true
+	default:
+		return false
+	}
+}
+
 func normalizeNotariusReferences(references map[string]string) (map[string]string, error) {
 	selectors := make([]string, 0, len(references))
 	for selector := range references {

internal/stage/cleanup_test.go

@@ -58,9 +58,12 @@ func TestCleanupStageRemovesOnlySelectedPublishedRunTargets(t *testing.T) {
 			otherSpool := filepath.Join(env.Config.Pipeline.Spool.Root, m.Campaign, m.SessionID, "other-run", "audio")
 			otherWorkspace := filepath.Join(env.Config.Pipeline.Workspace.Root, "work", m.Campaign, m.SessionID, "runs", "other-run")
 			cacheFile := filepath.Join(env.Config.Pipeline.Cache.Root, "s3", "audio.cache")
+			sessionPaths := artifacts.NewLocalStore(env.Config.Pipeline.Workspace.Root).SessionPathsFor(m.Campaign, m.SessionID)
+			referenceCacheFile := filepath.Join(artifacts.SessionNotariusReferenceCacheDir(sessionPaths), strings.Repeat("a", 64), "party.yml")
 			writeStageTestFile(t, filepath.Join(otherSpool, "audio.wav"), "other spool")
 			writeStageTestFile(t, filepath.Join(otherWorkspace, "result.txt"), "other workspace")
 			writeStageTestFile(t, cacheFile, "cache")
+			writeStageTestFile(t, referenceCacheFile, "reference cache")
 
 			result, err := (cleanupStage{}).Run(context.Background(), env, m)
 			if err != nil {
@@ -81,7 +84,7 @@ func TestCleanupStageRemovesOnlySelectedPublishedRunTargets(t *testing.T) {
 					t.Fatalf("unselected workspace target changed: %v", err)
 				}
 			}
-			for _, preserved := range []string{otherSpool, otherWorkspace, cacheFile} {
+			for _, preserved := range []string{otherSpool, otherWorkspace, cacheFile, referenceCacheFile} {
 				if _, err := os.Stat(preserved); err != nil {
 					t.Fatalf("preserved path %q: %v", preserved, err)
 				}

internal/stage/extract.go

@@ -124,20 +124,13 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (re
 	if err != nil {
 		return nil, fmt.Errorf("extract: resolve durable bundle path: %w", err)
 	}
-	referenceSnapshotRoot, err := absolutePath(artifacts.SessionRunNotariusReferencesDirForCampaign(workspaceRoot, campaign, sessionID, runID))
-	if err != nil {
-		return nil, fmt.Errorf("extract: resolve reference snapshot path: %w", err)
-	}
 	directories := []string{filepath.Dir(receiptPath), outputRoot, filepath.Dir(durableBundle)}
-	if len(references.Bindings) > 0 {
-		directories = append(directories, referenceSnapshotRoot)
-	}
 	for _, directory := range directories {
 		if err := fileops.EnsureWorkspaceDirectory(directory); err != nil {
 			return nil, fmt.Errorf("extract: create directory %q: %w", directory, err)
 		}
 	}
-	referenceBindings, err := materializeExtractReferenceSnapshots(paths, references, referenceSnapshotRoot)
+	referenceBindings, err := materializeExtractReferenceSnapshots(paths, references)
 	if err != nil {
 		return nil, fmt.Errorf("extract: materialize Notarius reference snapshots: %w", err)
 	}
@@ -146,10 +139,16 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (re
 	if err != nil {
 		return nil, fmt.Errorf("extract: build configuration fingerprint: %w", err)
 	}
+	resume := notariusConfig.Resume
+	chunkCache := notariusConfig.ChunkCache
+	if env.Force {
+		resume = false
+		chunkCache = "refresh"
+	}
 	request := notarius.RunRequest{
 		Binary: resolvedBinary, LLMAPIKey: notariusConfig.LLMAPIKey, ConfigPath: configPath, PipelineID: notariusConfig.PipelineID,
 		InputPath: inputPath, OutputRoot: outputRoot, WorkingDirectory: workingDirectory,
-		ReceiptPath: receiptPath, LogPath: logPath, Timeout: timeout, References: referenceBindings,
+		ReceiptPath: receiptPath, LogPath: logPath, Timeout: timeout, Resume: resume, ChunkCache: chunkCache, References: referenceBindings,
 	}
 	stageSucceeded := false
 	bundleMoved := false
@@ -176,7 +175,7 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (re
 	if adapterResult.Receipt.RunID == "" || adapterResult.Receipt.PipelineID != notariusConfig.PipelineID {
 		return nil, fmt.Errorf("extract: notarius receipt identity is missing or incompatible")
 	}
-	if err := verifyExtractReferenceSnapshots(referenceSnapshotRoot, references); err != nil {
+	if err := verifyExtractReferenceSnapshots(paths, references); err != nil {
 		return nil, fmt.Errorf("extract: verify Notarius reference snapshots: %w", err)
 	}
 

internal/stage/extract_references.go

@@ -3,6 +3,7 @@ package stage
 import (
 	"crypto/sha256"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -16,6 +17,7 @@ import (
 	"gitea.maximumdirect.net/eric/narratio/internal/fileops"
 	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
 	"gitea.maximumdirect.net/eric/narratio/internal/notariusref"
+	"gitea.maximumdirect.net/eric/narratio/internal/pathsafe"
 )
 
 type extractReferenceIdentity struct {
@@ -29,83 +31,204 @@ type extractReferenceIdentity struct {
 func materializeExtractReferenceSnapshots(
 	paths artifacts.SessionPaths,
 	references resolvedExtractReferences,
-	snapshotRoot string,
 ) ([]notarius.ReferenceBinding, error) {
+	var err error
+	paths, err = absoluteExtractSnapshotPaths(paths)
+	if err != nil {
+		return nil, err
+	}
 	if len(references.Bindings) != len(references.Identities) {
 		return nil, fmt.Errorf("reference bindings and identities have different lengths")
 	}
 	bindings := make([]notarius.ReferenceBinding, len(references.Bindings))
-	copied := make(map[string]string, len(references.Bindings))
-	destinationOwners := make(map[string]string, len(references.Bindings))
+	materialized := make(map[string]extractReferenceIdentity, len(references.Bindings))
 	for index, binding := range references.Bindings {
 		identity := references.Identities[index]
-		destination, ok := copied[identity.SourceID]
-		if !ok {
-			filename := filepath.Base(filepath.FromSlash(identity.Path))
-			if filename == "" || filename == "." || filename == string(filepath.Separator) {
-				return nil, fmt.Errorf("reference %q source %q has invalid prepared filename", identity.Selector, identity.SourceID)
-			}
-			destination = filepath.Join(snapshotRoot, filename)
-			if owner, duplicate := destinationOwners[destination]; duplicate && owner != identity.SourceID {
-				return nil, fmt.Errorf("reference sources %q and %q resolve to the same snapshot path", owner, identity.SourceID)
-			}
-			destinationOwners[destination] = identity.SourceID
-
-			file, err := os.Open(filepath.Join(paths.Root, filepath.FromSlash(identity.Path)))
-			if err != nil {
-				return nil, fmt.Errorf("open reference %q source %q for snapshot: %w", identity.Selector, identity.SourceID, err)
-			}
-			checksum, copyErr := fileops.WriteReaderAtomicWithChecksum(destination, file, fileops.WorkspaceFileMode)
-			closeErr := file.Close()
-			if copyErr != nil {
-				return nil, fmt.Errorf("snapshot reference %q source %q: %w", identity.Selector, identity.SourceID, copyErr)
-			}
-			if closeErr != nil {
-				return nil, fmt.Errorf("close reference %q source %q: %w", identity.Selector, identity.SourceID, closeErr)
-			}
-			info, err := os.Lstat(destination)
-			if err != nil {
-				return nil, fmt.Errorf("inspect reference %q snapshot: %w", identity.Selector, err)
-			}
-			if !info.Mode().IsRegular() || !strings.EqualFold(checksum, identity.Checksum) || info.Size() != identity.SizeBytes {
-				return nil, fmt.Errorf("reference %q source %q changed while its verified snapshot was created", identity.Selector, identity.SourceID)
-			}
-			copied[identity.SourceID] = destination
+		destination, err := extractReferenceSnapshotDestination(paths, identity)
+		if err != nil {
+			return nil, fmt.Errorf("reference %q source %q snapshot path: %w", identity.Selector, identity.SourceID, err)
 		}
+		if previous, ok := materialized[destination]; ok {
+			if previous.Checksum != identity.Checksum || previous.SizeBytes != identity.SizeBytes {
+				return nil, fmt.Errorf("references %q and %q have conflicting identities for snapshot %q", previous.Selector, identity.Selector, destination)
+			}
+			bindings[index] = notarius.ReferenceBinding{Selector: binding.Selector, Path: destination}
+			continue
+		}
+		if err := materializeExtractReferenceSnapshot(paths, identity, destination); err != nil {
+			return nil, err
+		}
+		materialized[destination] = identity
 		bindings[index] = notarius.ReferenceBinding{Selector: binding.Selector, Path: destination}
 	}
 	return bindings, nil
 }
 
-func verifyExtractReferenceSnapshots(snapshotRoot string, references resolvedExtractReferences) error {
-	verified := make(map[string]struct{}, len(references.Identities))
-	for _, identity := range references.Identities {
-		if _, ok := verified[identity.SourceID]; ok {
-			continue
+func materializeExtractReferenceSnapshot(paths artifacts.SessionPaths, identity extractReferenceIdentity, destination string) error {
+	if err := checkExtractSnapshotDirectory(paths.Root, filepath.Dir(destination), true); err != nil {
+		return fmt.Errorf("reference %q snapshot cache: %w", identity.Selector, err)
+	}
+	info, err := os.Lstat(destination)
+	if err == nil && !info.Mode().IsRegular() {
+		return unsafeExtractSnapshotPath(destination, "the destination is not a regular file")
+	}
+	if err != nil && !errors.Is(err, os.ErrNotExist) {
+		return fmt.Errorf("inspect reference %q snapshot %q: %w", identity.Selector, destination, err)
+	}
+	if err == nil {
+		if verifyErr := verifyExtractReferenceFile(destination, identity); verifyErr == nil {
+			return nil
 		}
-		filename := filepath.Base(filepath.FromSlash(identity.Path))
-		file, err := os.Open(filepath.Join(snapshotRoot, filename))
-		if err != nil {
-			return fmt.Errorf("reopen reference %q source %q snapshot: %w", identity.Selector, identity.SourceID, err)
-		}
-		digest := sha256.New()
-		size, readErr := io.Copy(digest, file)
-		closeErr := file.Close()
-		if readErr != nil {
-			return fmt.Errorf("revalidate reference %q source %q snapshot: %w", identity.Selector, identity.SourceID, readErr)
-		}
-		if closeErr != nil {
-			return fmt.Errorf("close reference %q source %q snapshot: %w", identity.Selector, identity.SourceID, closeErr)
-		}
-		checksum := hex.EncodeToString(digest.Sum(nil))
-		if !strings.EqualFold(checksum, identity.Checksum) || size != identity.SizeBytes {
-			return fmt.Errorf("reference %q source %q snapshot changed while Notarius was running", identity.Selector, identity.SourceID)
-		}
-		verified[identity.SourceID] = struct{}{}
+	}
+
+	source, err := pathsafe.JoinSlashRelativeUnderRoot(paths.Root, identity.Path)
+	if err != nil {
+		return fmt.Errorf("resolve reference %q prepared source: %w", identity.Selector, err)
+	}
+	if err := verifyExtractReferenceFile(source, identity); err != nil {
+		return fmt.Errorf("reference %q prepared source %q no longer matches its manifest identity; rerun prepare with --force: %w", identity.Selector, source, err)
+	}
+	checksum, err := fileops.CopyFileAtomicWithChecksum(source, destination, fileops.WorkspaceFileMode)
+	if err != nil {
+		return fmt.Errorf("snapshot reference %q source %q to %q: %w", identity.Selector, identity.SourceID, destination, err)
+	}
+	if checksum != identity.Checksum {
+		return cleanupInvalidExtractSnapshot(paths, identity, destination,
+			fmt.Errorf("copied checksum %q does not match manifest checksum %q", checksum, identity.Checksum))
+	}
+	if err := verifyExtractReferenceFile(destination, identity); err != nil {
+		return cleanupInvalidExtractSnapshot(paths, identity, destination, err)
 	}
 	return nil
 }
 
+func cleanupInvalidExtractSnapshot(paths artifacts.SessionPaths, identity extractReferenceIdentity, destination string, verifyErr error) error {
+	err := fmt.Errorf("reference %q source %q changed while snapshot %q was created; rerun prepare with --force: %w", identity.Selector, identity.SourceID, destination, verifyErr)
+	if cleanupErr := fileops.RemoveFileUnderRoot(artifacts.SessionNotariusReferenceCacheDir(paths), destination); cleanupErr != nil {
+		return errors.Join(err, fmt.Errorf("remove invalid newly written snapshot %q: %w", destination, cleanupErr))
+	}
+	return err
+}
+
+func verifyExtractReferenceSnapshots(paths artifacts.SessionPaths, references resolvedExtractReferences) error {
+	var err error
+	paths, err = absoluteExtractSnapshotPaths(paths)
+	if err != nil {
+		return err
+	}
+	verified := make(map[string]extractReferenceIdentity, len(references.Identities))
+	for _, identity := range references.Identities {
+		destination, err := extractReferenceSnapshotDestination(paths, identity)
+		if err != nil {
+			return fmt.Errorf("reference %q source %q snapshot path: %w", identity.Selector, identity.SourceID, err)
+		}
+		if previous, ok := verified[destination]; ok {
+			if previous.Checksum != identity.Checksum || previous.SizeBytes != identity.SizeBytes {
+				return fmt.Errorf("references %q and %q have conflicting identities for snapshot %q", previous.Selector, identity.Selector, destination)
+			}
+			continue
+		}
+		if err := checkExtractSnapshotDirectory(paths.Root, filepath.Dir(destination), false); err != nil {
+			return fmt.Errorf("reference %q snapshot cache: %w", identity.Selector, err)
+		}
+		if err := verifyExtractReferenceFile(destination, identity); err != nil {
+			return fmt.Errorf("reference %q source %q snapshot changed while Notarius was running at %q: %w", identity.Selector, identity.SourceID, destination, err)
+		}
+		verified[destination] = identity
+	}
+	return nil
+}
+
+// Resolve both roots against the caller's working directory before comparing
+// paths. A session-prefixed relative destination is not relative to Root.
+func absoluteExtractSnapshotPaths(paths artifacts.SessionPaths) (artifacts.SessionPaths, error) {
+	var err error
+	paths.Root, err = filepath.Abs(paths.Root)
+	if err != nil {
+		return paths, fmt.Errorf("resolve snapshot session root: %w", err)
+	}
+	paths.CacheDir, err = filepath.Abs(paths.CacheDir)
+	if err != nil {
+		return paths, fmt.Errorf("resolve snapshot cache root: %w", err)
+	}
+	return paths, nil
+}
+
+func extractReferenceSnapshotDestination(paths artifacts.SessionPaths, identity extractReferenceIdentity) (string, error) {
+	basename := filepath.Base(filepath.FromSlash(identity.Path))
+	return artifacts.SessionNotariusReferenceSnapshotPath(paths, identity.Checksum, basename)
+}
+
+func verifyExtractReferenceFile(path string, identity extractReferenceIdentity) error {
+	info, err := os.Lstat(path)
+	if err != nil {
+		return err
+	}
+	if !info.Mode().IsRegular() {
+		return unsafeExtractSnapshotPath(path, "the path is not a regular file")
+	}
+	file, err := os.Open(path)
+	if err != nil {
+		return err
+	}
+	digest := sha256.New()
+	size, readErr := io.Copy(digest, file)
+	closeErr := file.Close()
+	if readErr != nil {
+		return readErr
+	}
+	if closeErr != nil {
+		return closeErr
+	}
+	checksum := hex.EncodeToString(digest.Sum(nil))
+	if checksum != identity.Checksum || size != identity.SizeBytes || info.Size() != identity.SizeBytes {
+		return fmt.Errorf("content identity is checksum %q size %d, want checksum %q size %d", checksum, size, identity.Checksum, identity.SizeBytes)
+	}
+	return nil
+}
+
+func checkExtractSnapshotDirectory(sessionRoot, directory string, create bool) error {
+	relative, err := pathsafe.SlashRelativeFromRoot(sessionRoot, directory)
+	if err != nil {
+		return fmt.Errorf("resolve snapshot directory %q below session root: %w", directory, err)
+	}
+	root, err := os.OpenRoot(sessionRoot)
+	if err != nil {
+		return fmt.Errorf("open session root %q: %w", sessionRoot, err)
+	}
+	defer func() { _ = root.Close() }()
+
+	current := ""
+	for _, component := range strings.Split(relative, "/") {
+		if current == "" {
+			current = component
+		} else {
+			current += "/" + component
+		}
+		info, err := root.Lstat(current)
+		if errors.Is(err, os.ErrNotExist) && create {
+			if err := root.Mkdir(current, fileops.WorkspaceDirectoryMode.Perm()); err != nil && !errors.Is(err, os.ErrExist) {
+				return fmt.Errorf("create snapshot cache directory %q: %w", filepath.Join(sessionRoot, filepath.FromSlash(current)), err)
+			}
+			info, err = root.Lstat(current)
+		}
+		if err != nil {
+			return fmt.Errorf("inspect snapshot cache directory %q: %w", filepath.Join(sessionRoot, filepath.FromSlash(current)), err)
+		}
+		if !info.IsDir() || info.Mode()&os.ModeSymlink != 0 {
+			return unsafeExtractSnapshotPath(filepath.Join(sessionRoot, filepath.FromSlash(current)), "a path component is not a real directory")
+		}
+		if err := root.Chmod(current, fileops.WorkspaceDirectoryMode); err != nil {
+			return fmt.Errorf("set snapshot cache directory permissions %q: %w", filepath.Join(sessionRoot, filepath.FromSlash(current)), err)
+		}
+	}
+	return nil
+}
+
+func unsafeExtractSnapshotPath(path, reason string) error {
+	return fmt.Errorf("unsafe Notarius reference snapshot cache path %q: %s; remove or replace it only while the session is inactive", path, reason)
+}
+
 type resolvedExtractReferences struct {
 	Bindings   []notarius.ReferenceBinding
 	Identities []extractReferenceIdentity

internal/stage/extract_test.go

@@ -96,6 +96,9 @@ notarius:
 	if len(req.References) != 0 {
 		t.Fatalf("references = %#v, want none", req.References)
 	}
+	if req.Resume || req.ChunkCache != "" {
+		t.Fatalf("default cache controls = resume %v, mode %q", req.Resume, req.ChunkCache)
+	}
 	if req.LLMAPIKey != loadedPipeline.Notarius.LLMAPIKey || req.LLMAPIKey.EnvName() != credentialName || !req.LLMAPIKey.IsResolved() {
 		t.Fatal("request did not retain the loaded Notarius credential reference")
 	}
@@ -114,6 +117,41 @@ notarius:
 	}
 }
 
+func TestExtractStageAppliesNotariusCacheControlsAndForcePrecedence(t *testing.T) {
+	tests := []struct {
+		name             string
+		configuredResume bool
+		configuredMode   string
+		force            bool
+		wantResume       bool
+		wantMode         string
+	}{
+		{name: "ordinary defaults"},
+		{name: "ordinary configured", configuredResume: true, configuredMode: "bypass", wantResume: true, wantMode: "bypass"},
+		{name: "force overrides configured", configuredResume: true, configuredMode: "bypass", force: true, wantMode: "refresh"},
+		{name: "force overrides defaults", force: true, wantMode: "refresh"},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			env, m, fake := setupExtractEnv(t)
+			env.Config.Pipeline.Notarius.Resume = test.configuredResume
+			env.Config.Pipeline.Notarius.ChunkCache = test.configuredMode
+			env.Force = test.force
+
+			if _, err := (extractStage{}).Run(context.Background(), env, m); err != nil {
+				t.Fatalf("Run() error = %v", err)
+			}
+			if len(fake.Requests) != 1 {
+				t.Fatalf("requests = %d, want 1", len(fake.Requests))
+			}
+			request := fake.Requests[0]
+			if request.Resume != test.wantResume || request.ChunkCache != test.wantMode {
+				t.Fatalf("cache controls = resume %v, mode %q; want resume %v, mode %q", request.Resume, request.ChunkCache, test.wantResume, test.wantMode)
+			}
+		})
+	}
+}
+
 func TestExtractStageResolvesAllPreparedReferencesInSelectorOrder(t *testing.T) {
 	env, m, fake := setupExtractEnv(t)
 	configured := []struct {
@@ -149,12 +187,16 @@ func TestExtractStageResolvesAllPreparedReferencesInSelectorOrder(t *testing.T)
 	}
 	for index, selector := range wantSelectors {
 		binding := fake.Requests[0].References[index]
-		wantPath := filepath.Join(
-			artifacts.SessionRunNotariusReferencesDirForCampaign(
-				env.Config.Pipeline.Workspace.Root, m.Campaign, m.SessionID, m.RunID,
-			),
-			filepath.Base(preparedPaths[selector]),
+		checksum, err := artifacts.SHA256File(preparedPaths[selector])
+		if err != nil {
+			t.Fatalf("SHA256File(%q) error = %v", selector, err)
+		}
+		wantPath, err := artifacts.SessionNotariusReferenceSnapshotPath(
+			sessionPathsForEnv(env, m.SessionID), checksum, filepath.Base(preparedPaths[selector]),
 		)
+		if err != nil {
+			t.Fatalf("SessionNotariusReferenceSnapshotPath(%q) error = %v", selector, err)
+		}
 		if binding.Selector != selector || binding.Path != wantPath || binding.Path == preparedPaths[selector] || !filepath.IsAbs(binding.Path) {
 			t.Fatalf("reference[%d] = %#v, want selector %q verified snapshot %q", index, binding, selector, wantPath)
 		}
@@ -191,11 +233,11 @@ func TestExtractStageResolvesAllPreparedReferencesInSelectorOrder(t *testing.T)
 }
 
 type referenceMutationRunner struct {
-	delegate       notarius.Runner
-	preparedPath   string
-	preparedBytes  []byte
-	mutateSnapshot bool
-	snapshotBytes  []byte
+	delegate      notarius.Runner
+	preparedPath  string
+	preparedBytes []byte
+	mutate        func(string) error
+	snapshotBytes []byte
 }
 
 func (r *referenceMutationRunner) Run(ctx context.Context, req notarius.RunRequest) (notarius.RunResult, error) {
@@ -210,8 +252,8 @@ func (r *referenceMutationRunner) Run(ctx context.Context, req notarius.RunReque
 		return notarius.RunResult{}, err
 	}
 	r.snapshotBytes = data
-	if r.mutateSnapshot {
-		if err := os.WriteFile(req.References[0].Path, []byte("tampered snapshot\n"), 0o664); err != nil {
+	if r.mutate != nil {
+		if err := r.mutate(req.References[0].Path); err != nil {
 			return notarius.RunResult{}, err
 		}
 	}
@@ -234,21 +276,292 @@ func TestExtractStageInvokesNotariusWithVerifiedReferenceSnapshot(t *testing.T)
 		t.Fatalf("Notarius reference bytes = %q, want verified snapshot", runner.snapshotBytes)
 	}
 	if len(fake.Requests) != 1 || fake.Requests[0].References[0].Path == preparedPath {
-		t.Fatalf("adapter requests = %#v, want a run-local reference snapshot", fake.Requests)
+		t.Fatalf("adapter requests = %#v, want a stable reference snapshot", fake.Requests)
 	}
 }
 
 func TestExtractStageRejectsReferenceSnapshotChangedDuringNotariusRun(t *testing.T) {
-	env, m, fake := setupExtractEnv(t)
+	for _, test := range []struct {
+		name   string
+		mutate func(string) error
+	}{
+		{name: "content", mutate: func(path string) error { return os.WriteFile(path, []byte("tampered snapshot\n"), 0o664) }},
+		{name: "missing", mutate: os.Remove},
+		{name: "symlink", mutate: func(path string) error {
+			if err := os.Remove(path); err != nil {
+				return err
+			}
+			return os.Symlink(path+".outside", path)
+		}},
+		{name: "directory", mutate: func(path string) error {
+			if err := os.Remove(path); err != nil {
+				return err
+			}
+			return os.Mkdir(path, 0o775)
+		}},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			env, m, fake := setupExtractEnv(t)
+			env.Config.Pipeline.Notarius.References = map[string]string{"party": artifactpolicy.SourceInputParty}
+			preparedPath := recordPreparedExtractInput(t, env, m, artifactpolicy.SourceInputParty, "verified party\n")
+			env.Notarius = &referenceMutationRunner{
+				delegate: fake, preparedPath: preparedPath, preparedBytes: []byte("changed after snapshot\n"), mutate: test.mutate,
+			}
+
+			result, err := (extractStage{}).Run(context.Background(), env, m)
+			if err == nil || !strings.Contains(err.Error(), "snapshot changed while Notarius was running") || result != nil {
+				t.Fatalf("Run() result = %#v error = %v, want changed-snapshot failure", result, err)
+			}
+		})
+	}
+}
+
+func TestExtractReferenceSnapshotsAreStableReusableAndRecoverable(t *testing.T) {
+	env, m, _ := setupExtractEnv(t)
 	env.Config.Pipeline.Notarius.References = map[string]string{"party": artifactpolicy.SourceInputParty}
-	preparedPath := recordPreparedExtractInput(t, env, m, artifactpolicy.SourceInputParty, "verified party\n")
-	env.Notarius = &referenceMutationRunner{
-		delegate: fake, preparedPath: preparedPath, preparedBytes: []byte("changed after snapshot\n"), mutateSnapshot: true,
+	preparedPath := recordPreparedExtractInput(t, env, m, artifactpolicy.SourceInputParty, "stable party\n")
+	paths := sessionPathsForEnv(env, m.SessionID)
+	references, err := resolveExtractReferences(paths, m, env.Config.Pipeline.Notarius, m.SessionID)
+	if err != nil {
+		t.Fatalf("resolveExtractReferences() error = %v", err)
+	}
+	first, err := materializeExtractReferenceSnapshots(paths, references)
+	if err != nil {
+		t.Fatalf("materializeExtractReferenceSnapshots(first) error = %v", err)
+	}
+	if len(first) != 1 {
+		t.Fatalf("first bindings = %#v", first)
+	}
+	firstPath := first[0].Path
+
+	oldTime := time.Unix(123, 0)
+	if err := os.Chtimes(firstPath, oldTime, oldTime); err != nil {
+		t.Fatalf("Chtimes(snapshot) error = %v", err)
+	}
+	m.RunID = "20260810T020304Z-fedcba98"
+	second, err := materializeExtractReferenceSnapshots(paths, references)
+	if err != nil {
+		t.Fatalf("materializeExtractReferenceSnapshots(second run) error = %v", err)
+	}
+	if second[0].Path != firstPath {
+		t.Fatalf("snapshot path changed across run IDs: %q != %q", second[0].Path, firstPath)
+	}
+	info, err := os.Stat(firstPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !info.ModTime().Equal(oldTime) {
+		t.Fatalf("verified snapshot was rewritten: modtime = %v, want %v", info.ModTime(), oldTime)
 	}
 
-	result, err := (extractStage{}).Run(context.Background(), env, m)
-	if err == nil || !strings.Contains(err.Error(), "snapshot changed while Notarius was running") || result != nil {
-		t.Fatalf("Run() result = %#v error = %v, want changed-snapshot failure", result, err)
+	if err := os.Remove(firstPath); err != nil {
+		t.Fatalf("Remove(snapshot) error = %v", err)
+	}
+	if _, err := materializeExtractReferenceSnapshots(paths, references); err != nil {
+		t.Fatalf("reconstruct missing snapshot error = %v", err)
+	}
+	assertFileContents(t, firstPath, "stable party\n")
+
+	if err := os.WriteFile(firstPath, []byte("corrupt snapshot\n"), 0o664); err != nil {
+		t.Fatalf("WriteFile(corrupt snapshot) error = %v", err)
+	}
+	if _, err := materializeExtractReferenceSnapshots(paths, references); err != nil {
+		t.Fatalf("repair corrupt snapshot error = %v", err)
+	}
+	assertFileContents(t, firstPath, "stable party\n")
+
+	if err := os.WriteFile(preparedPath, []byte("changed party bytes\n"), 0o664); err != nil {
+		t.Fatalf("WriteFile(changed prepared source) error = %v", err)
+	}
+	if err := os.WriteFile(firstPath, []byte("corrupt before source mismatch\n"), 0o664); err != nil {
+		t.Fatalf("WriteFile(corrupt snapshot) error = %v", err)
+	}
+	if _, err := materializeExtractReferenceSnapshots(paths, references); err == nil || !strings.Contains(err.Error(), "prepared source") {
+		t.Fatalf("materialize with changed prepared source error = %v, want prepared-source mismatch", err)
+	}
+	assertFileContents(t, firstPath, "corrupt before source mismatch\n")
+
+	changedChecksum, err := artifacts.SHA256File(preparedPath)
+	if err != nil {
+		t.Fatalf("SHA256File(changed prepared source) error = %v", err)
+	}
+	descriptor, _ := artifactpolicy.DescribePreparedInputSource(artifactpolicy.SourceInputParty)
+	for index := range m.Inputs {
+		if m.Inputs[index].Kind == descriptor.ManifestKind {
+			m.Inputs[index].Checksum = changedChecksum
+		}
+	}
+	changed, err := resolveExtractReferences(paths, m, env.Config.Pipeline.Notarius, m.SessionID)
+	if err != nil {
+		t.Fatalf("resolveExtractReferences(changed) error = %v", err)
+	}
+	changedBindings, err := materializeExtractReferenceSnapshots(paths, changed)
+	if err != nil {
+		t.Fatalf("materializeExtractReferenceSnapshots(changed) error = %v", err)
+	}
+	if changedBindings[0].Path == firstPath || filepath.Base(changedBindings[0].Path) != filepath.Base(firstPath) {
+		t.Fatalf("changed bytes paths = old %q new %q, want new digest path with same basename", firstPath, changedBindings[0].Path)
+	}
+	assertFileContents(t, firstPath, "corrupt before source mismatch\n")
+	assertFileContents(t, changedBindings[0].Path, "changed party bytes\n")
+}
+
+func TestExtractReferenceSnapshotsDeduplicateEquivalentDestinations(t *testing.T) {
+	env, m, _ := setupExtractEnv(t)
+	path := recordPreparedExtractInput(t, env, m, artifactpolicy.SourceInputParty, "shared bytes\n")
+	checksum, err := artifacts.SHA256File(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	identity := extractReferenceIdentity{
+		Selector: "party", SourceID: artifactpolicy.SourceInputParty, Path: "inputs/party.yml",
+		Checksum: checksum, SizeBytes: int64(len("shared bytes\n")),
+	}
+	references := resolvedExtractReferences{
+		Bindings: []notarius.ReferenceBinding{{Selector: "party"}, {Selector: "alternate"}},
+		Identities: []extractReferenceIdentity{
+			identity,
+			{Selector: "alternate", SourceID: artifactpolicy.SourceInputGlossary, Path: identity.Path, Checksum: checksum, SizeBytes: identity.SizeBytes},
+		},
+	}
+	bindings, err := materializeExtractReferenceSnapshots(sessionPathsForEnv(env, m.SessionID), references)
+	if err != nil {
+		t.Fatalf("materializeExtractReferenceSnapshots() error = %v", err)
+	}
+	if len(bindings) != 2 || bindings[0].Path != bindings[1].Path {
+		t.Fatalf("equivalent bindings = %#v, want shared destination", bindings)
+	}
+}
+
+func TestExtractReferenceSnapshotsWithRelativeWorkspace(t *testing.T) {
+	env, m, _ := setupExtractEnv(t)
+	prepared := recordPreparedExtractInput(t, env, m, artifactpolicy.SourceInputParty, "party bytes\n")
+	env.Config.Pipeline.Notarius.References = map[string]string{"party": artifactpolicy.SourceInputParty}
+	absPaths := sessionPathsForEnv(env, m.SessionID)
+	references, err := resolveExtractReferences(absPaths, m, env.Config.Pipeline.Notarius, m.SessionID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Chdir(env.Config.Pipeline.Workspace.Root)
+	paths := artifacts.NewLocalStore(".").SessionPathsFor(m.Campaign, m.SessionID)
+	first, err := materializeExtractReferenceSnapshots(paths, references)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !filepath.IsAbs(first[0].Path) {
+		t.Fatalf("adapter reference remains relative: %q", first[0].Path)
+	}
+	m.RunID = "20260809T020203Z-abcdef13"
+	second, err := materializeExtractReferenceSnapshots(paths, references)
+	if err != nil || !reflect.DeepEqual(first, second) {
+		t.Fatalf("second invocation bindings = %#v, error = %v", second, err)
+	}
+	if err := verifyExtractReferenceSnapshots(paths, references); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := os.Stat(filepath.Join(absPaths.Root, paths.Root)); !errors.Is(err, os.ErrNotExist) {
+		t.Fatalf("duplicated session prefix exists or cannot be inspected: %v", err)
+	}
+
+	// A controlled executable proves the production adapter accepts the paths.
+	// It intentionally produces no receipt; receipt validation is covered elsewhere.
+	marker := filepath.Join(t.TempDir(), "started")
+	binary := filepath.Join(t.TempDir(), "notarius")
+	if err := os.WriteFile(binary, []byte("#!/bin/sh\nprintf started > '"+marker+"'\n"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	diagnostics := t.TempDir()
+	_, err = notarius.NewSubprocessRunner().Run(context.Background(), notarius.RunRequest{
+		Binary: binary, PipelineID: "dnd-session", ConfigPath: env.Config.Pipeline.Notarius.ConfigPath,
+		InputPath: prepared, OutputRoot: t.TempDir(), WorkingDirectory: env.Config.Pipeline.Workspace.Root,
+		Timeout: time.Second * 10, References: first,
+		ReceiptPath: filepath.Join(diagnostics, "receipt.json"), LogPath: filepath.Join(diagnostics, "stderr.log"),
+	})
+	if err == nil {
+		t.Fatal("expected missing receipt error")
+	}
+	if _, err := os.Stat(marker); err != nil {
+		t.Fatalf("adapter did not launch with the snapshot bindings: %v", err)
+	}
+
+	// A symlink on the actual cache path must not be missed by a duplicated walk.
+	if err := os.Remove(first[0].Path); err != nil {
+		t.Fatal(err)
+	}
+	digestDir := filepath.Dir(first[0].Path)
+	if err := os.Remove(digestDir); err != nil {
+		t.Fatal(err)
+	}
+	outside := t.TempDir()
+	if err := os.Symlink(outside, digestDir); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := materializeExtractReferenceSnapshots(paths, references); err == nil {
+		t.Fatal("accepted symlink on actual cache path")
+	}
+	if _, err := os.Stat(filepath.Join(outside, filepath.Base(first[0].Path))); !errors.Is(err, os.ErrNotExist) {
+		t.Fatalf("unsafe cache path wrote outside session: %v", err)
+	}
+}
+
+func TestExtractReferenceSnapshotsRejectUnsafeConflicts(t *testing.T) {
+	for _, test := range []struct {
+		name  string
+		setup func(*testing.T, artifacts.SessionPaths, string)
+	}{
+		{name: "leaf symlink", setup: func(t *testing.T, _ artifacts.SessionPaths, destination string) {
+			if err := os.MkdirAll(filepath.Dir(destination), 0o775); err != nil {
+				t.Fatal(err)
+			}
+			if err := os.Symlink(filepath.Join(t.TempDir(), "outside"), destination); err != nil {
+				t.Fatal(err)
+			}
+		}},
+		{name: "leaf directory", setup: func(t *testing.T, _ artifacts.SessionPaths, destination string) {
+			if err := os.MkdirAll(destination, 0o775); err != nil {
+				t.Fatal(err)
+			}
+		}},
+		{name: "digest symlink", setup: func(t *testing.T, paths artifacts.SessionPaths, destination string) {
+			cacheRoot := artifacts.SessionNotariusReferenceCacheDir(paths)
+			if err := os.MkdirAll(cacheRoot, 0o775); err != nil {
+				t.Fatal(err)
+			}
+			if err := os.Symlink(t.TempDir(), filepath.Dir(destination)); err != nil {
+				t.Fatal(err)
+			}
+		}},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			env, m, _ := setupExtractEnv(t)
+			env.Config.Pipeline.Notarius.References = map[string]string{"party": artifactpolicy.SourceInputParty}
+			recordPreparedExtractInput(t, env, m, artifactpolicy.SourceInputParty, "party bytes\n")
+			paths := sessionPathsForEnv(env, m.SessionID)
+			references, err := resolveExtractReferences(paths, m, env.Config.Pipeline.Notarius, m.SessionID)
+			if err != nil {
+				t.Fatal(err)
+			}
+			destination, err := extractReferenceSnapshotDestination(paths, references.Identities[0])
+			if err != nil {
+				t.Fatal(err)
+			}
+			test.setup(t, paths, destination)
+			_, err = materializeExtractReferenceSnapshots(paths, references)
+			if err == nil || !strings.Contains(err.Error(), artifacts.SessionNotariusReferenceCacheDir(paths)) || !strings.Contains(err.Error(), "session is inactive") {
+				t.Fatalf("materializeExtractReferenceSnapshots() error = %v, want actionable unsafe-path failure for %q", err, destination)
+			}
+		})
+	}
+}
+
+func assertFileContents(t *testing.T, path, want string) {
+	t.Helper()
+	data, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("ReadFile(%q) error = %v", path, err)
+	}
+	if string(data) != want {
+		t.Fatalf("ReadFile(%q) = %q, want %q", path, data, want)
 	}
 }
 

internal/stage/notarius_cache_integration_test.go

@@ -0,0 +1,379 @@
+package stage
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync/atomic"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/adapters/notarius"
+	"gitea.maximumdirect.net/eric/narratio/internal/artifactmodel"
+	"gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy"
+	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
+	"gitea.maximumdirect.net/eric/narratio/internal/config"
+	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
+)
+
+const notariusIntegrationBinaryEnv = "NARRATIO_TEST_NOTARIUS_BINARY"
+
+// Run with:
+// NARRATIO_TEST_NOTARIUS_BINARY=/absolute/path/to/notarius GOWORK=off go test ./internal/stage -run '^TestNotariusCacheReuseIntegration$' -count=1 -v
+func TestNotariusCacheReuseIntegration(t *testing.T) {
+	binary := strings.TrimSpace(os.Getenv(notariusIntegrationBinaryEnv))
+	if binary == "" {
+		t.Skipf("set %s to an absolute Notarius v0.7.0-compatible executable to run this integration test", notariusIntegrationBinaryEnv)
+	}
+	if !filepath.IsAbs(binary) {
+		t.Fatalf("%s = %q, want an absolute path", notariusIntegrationBinaryEnv, binary)
+	}
+	info, err := os.Stat(binary)
+	if err != nil {
+		t.Fatalf("inspect %s: %v", notariusIntegrationBinaryEnv, err)
+	}
+	if !info.Mode().IsRegular() || info.Mode().Perm()&0o111 == 0 {
+		t.Fatalf("%s = %q, want an executable regular file", notariusIntegrationBinaryEnv, binary)
+	}
+
+	t.Run("reuse reference identity and forced refresh", func(t *testing.T) {
+		provider := newNotariusIntegrationProvider(t)
+		env, runManifest, fixture := newNotariusIntegrationFixture(t, binary, provider.URL(), "dnd-session")
+
+		first := runNotariusIntegrationExtraction(t, env, runManifest, "20260916T010101Z-a0000001", false)
+		if got := provider.Calls(); got != 2 {
+			t.Fatalf("first provider calls = %d, want 2", got)
+		}
+		if first.ChunkPlan == nil || first.ChunkPlan.Action != "generated" {
+			t.Fatalf("first chunk plan = %#v, want generated", first.ChunkPlan)
+		}
+
+		env.Config.Pipeline.Notarius.Resume = true
+		second := runNotariusIntegrationExtraction(t, env, runManifest, "20260916T010102Z-a0000002", false)
+		if got := provider.Calls(); got != 2 {
+			t.Fatalf("resumed provider calls = %d, want unchanged count 2", got)
+		}
+		assertNotariusIntegrationDecision(t, second, "extract", "spells-first", "reused", "checkpoint_reused")
+		assertNotariusIntegrationDecision(t, second, "extract", "spells-second", "reused", "checkpoint_reused")
+
+		oldSnapshot := notariusIntegrationSnapshotPath(t, env, runManifest, artifactpolicy.SourceInputParty)
+		setNotariusIntegrationReference(t, runManifest, fixture.partyPath, "The party now travels with a silver dragon.\n")
+		third := runNotariusIntegrationExtraction(t, env, runManifest, "20260916T010103Z-a0000003", false)
+		if got := provider.Calls(); got != 4 {
+			t.Fatalf("changed-reference provider calls = %d, want 4", got)
+		}
+		assertNotariusIntegrationDecision(t, third, "extract", "spells-first", "executed", "checkpoint_missing")
+		newSnapshot := notariusIntegrationSnapshotPath(t, env, runManifest, artifactpolicy.SourceInputParty)
+		if newSnapshot == oldSnapshot {
+			t.Fatalf("changed reference retained snapshot %q", newSnapshot)
+		}
+		for _, path := range []string{oldSnapshot, newSnapshot} {
+			if _, err := os.Stat(path); err != nil {
+				t.Fatalf("snapshot %q is unavailable: %v", path, err)
+			}
+		}
+
+		setNotariusIntegrationTranscript(t, runManifest, fixture.transcriptPath, notariusIntegrationTranscript("session-unrelated", "The wizard studies Magic Missile."))
+		unrelated := runNotariusIntegrationExtraction(t, env, runManifest, "20260916T010104Z-a0000004", false)
+		if unrelated.ChunkPlan == nil || unrelated.ChunkPlan.Action != "generated" {
+			t.Fatalf("unrelated chunk plan = %#v, want generated", unrelated.ChunkPlan)
+		}
+		unrelatedPlan := filepath.Join(fixture.chunkRoot, strings.TrimPrefix(unrelated.ChunkPlan.SourceDigest, "sha256:"), "plan.json")
+		unrelatedBefore, err := os.ReadFile(unrelatedPlan)
+		if err != nil {
+			t.Fatalf("read unrelated chunk plan: %v", err)
+		}
+
+		setNotariusIntegrationTranscript(t, runManifest, fixture.transcriptPath, notariusIntegrationTranscript("session-cache", "Aria casts Cure Wounds."))
+		forced := runNotariusIntegrationExtraction(t, env, runManifest, "20260916T010105Z-a0000005", true)
+		if got := provider.Calls(); got != 8 {
+			t.Fatalf("forced provider calls = %d, want 8", got)
+		}
+		if len(forced.CheckpointDecisions) != 0 {
+			t.Fatalf("forced extraction loaded checkpoint state: %#v", forced.CheckpointDecisions)
+		}
+		if forced.ChunkPlan == nil || forced.ChunkPlan.Action != "refreshed" {
+			t.Fatalf("forced chunk plan = %#v, want refreshed", forced.ChunkPlan)
+		}
+		unrelatedAfter, err := os.ReadFile(unrelatedPlan)
+		if err != nil {
+			t.Fatalf("read unrelated chunk plan after refresh: %v", err)
+		}
+		if string(unrelatedAfter) != string(unrelatedBefore) {
+			t.Fatal("forced refresh changed the unrelated input's cached plan")
+		}
+	})
+
+	t.Run("partial failure resumes accepted predecessor", func(t *testing.T) {
+		provider := newNotariusIntegrationProvider(t)
+		provider.FailRequest(2)
+		env, runManifest, _ := newNotariusIntegrationFixture(t, binary, provider.URL(), "dnd-recovery")
+		env.Config.Pipeline.Notarius.Resume = true
+
+		runManifest.RunID = "20260916T020101Z-b0000001"
+		env.Force = false
+		if result, err := (extractStage{}).Run(context.Background(), env, runManifest); err == nil || result != nil {
+			t.Fatalf("partial extraction result = %#v error = %v, want failure", result, err)
+		}
+		if got := provider.Calls(); got != 2 {
+			t.Fatalf("failed invocation provider calls = %d, want 2", got)
+		}
+
+		provider.FailRequest(0)
+		recovered := runNotariusIntegrationExtraction(t, env, runManifest, "20260916T020102Z-b0000002", false)
+		if got := provider.Calls(); got != 3 {
+			t.Fatalf("recovery provider calls = %d, want one additional call (3 total)", got)
+		}
+		assertNotariusIntegrationDecision(t, recovered, "extract", "spells-first", "reused", "checkpoint_reused")
+		assertNotariusIntegrationDecision(t, recovered, "extract", "spells-second", "executed", "")
+	})
+}
+
+type notariusIntegrationProvider struct {
+	server *httptest.Server
+	calls  atomic.Int64
+	failAt atomic.Int64
+}
+
+func newNotariusIntegrationProvider(t *testing.T) *notariusIntegrationProvider {
+	t.Helper()
+	provider := &notariusIntegrationProvider{}
+	provider.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) {
+		call := provider.calls.Add(1)
+		if request.Method != http.MethodPost || request.URL.Path != "/v1/chat/completions" {
+			t.Errorf("provider request = %s %s, want POST /v1/chat/completions", request.Method, request.URL.Path)
+			http.Error(w, "unexpected request", http.StatusNotFound)
+			return
+		}
+		if provider.failAt.Load() == call {
+			http.Error(w, "controlled provider failure", http.StatusInternalServerError)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{"id":"local-completion","object":"chat.completion","created":1,"model":"local-model","choices":[{"index":0,"message":{"role":"assistant","content":"{\"spell_casts\":[]}"},"finish_reason":"stop"}],"usage":{"prompt_tokens":10,"completion_tokens":4,"total_tokens":14}}`))
+	}))
+	t.Cleanup(provider.server.Close)
+	return provider
+}
+
+func (p *notariusIntegrationProvider) URL() string               { return p.server.URL }
+func (p *notariusIntegrationProvider) Calls() int64              { return p.calls.Load() }
+func (p *notariusIntegrationProvider) FailRequest(request int64) { p.failAt.Store(request) }
+
+type notariusIntegrationFixture struct {
+	transcriptPath string
+	partyPath      string
+	chunkRoot      string
+}
+
+func newNotariusIntegrationFixture(t *testing.T, binary, endpoint, pipelineID string) (*Env, *manifest.Manifest, notariusIntegrationFixture) {
+	t.Helper()
+	env, runManifest, _ := setupExtractEnv(t)
+	initialOutput := artifacts.SessionRunNotariusOutputRootForCampaign(
+		env.Config.Pipeline.Workspace.Root, runManifest.Campaign, runManifest.SessionID, runManifest.RunID,
+	)
+	if err := os.RemoveAll(initialOutput); err != nil {
+		t.Fatalf("remove synthetic output fixture: %v", err)
+	}
+
+	root := env.Config.Pipeline.Workspace.Root
+	profilePath := filepath.Join(root, "local-profile.yml")
+	if err := os.WriteFile(profilePath, []byte("id: local-test\nbackend: local\nmodel: local-model\ntimeout_seconds: 30\n"), 0o600); err != nil {
+		t.Fatalf("write local profile: %v", err)
+	}
+	chunkRoot := filepath.Join(root, "notarius-cache", "chunk-plans")
+	checkpointRoot := filepath.Join(root, "notarius-cache", "checkpoints")
+	configPath := filepath.Join(root, "notarius-integration.yml")
+	content := fmt.Sprintf(`version: 5
+promptkit:
+  profile_file: %q
+  local_backend:
+    endpoint: %q
+    concurrency_limit: 1
+concurrency:
+  total_llm: 1
+  stage_workers:
+    extract: 1
+output:
+  directory: %q
+cache:
+  chunk_plans:
+    mode: auto
+    directory: %q
+  checkpoints:
+    enabled: true
+    directory: %q
+pipelines:
+  %s:
+    llm_profile: local-test
+    input:
+      module: seriatim
+    chunk:
+      module: generic
+    output:
+      module: json
+    steps:
+      - id: collect-first
+        artifacts:
+          spells-first:
+            extract:
+              module: dnd/spells
+            normalize:
+              module: dnd/spells
+      - id: collect-second
+        artifacts:
+          spells-second:
+            extract:
+              module: dnd/spells
+            normalize:
+              module: dnd/spells
+`, profilePath, endpoint+"/v1", filepath.Join(root, "unused-output"), chunkRoot, checkpointRoot, pipelineID)
+	if err := os.WriteFile(configPath, []byte(content), 0o600); err != nil {
+		t.Fatalf("write Notarius integration config: %v", err)
+	}
+
+	transcriptPath := filepath.Join(sessionPathsForEnv(env, runManifest.SessionID).ArtifactsDir, "trimmed.from-manifest.json")
+	setNotariusIntegrationTranscript(t, runManifest, transcriptPath, notariusIntegrationTranscript("session-cache", "Aria casts Cure Wounds."))
+	partyPath := recordPreparedExtractInput(t, env, runManifest, artifactpolicy.SourceInputParty, "The party includes Aria the cleric.\n")
+	env.Notarius = notarius.NewSubprocessRunner()
+	env.Config.Pipeline.Notarius = &config.NotariusConfig{
+		Enabled: true, Binary: binary, ConfigPath: configPath, PipelineID: pipelineID,
+		Timeout: "2m", WorkingDirectory: root, ChunkCache: "auto",
+		References: map[string]string{"party": artifactpolicy.SourceInputParty},
+		Outputs: map[string]config.NotariusOutputConfig{
+			"spells_first": {
+				LaneID: "spells-first", MediaType: "application/json", ModuleKey: "dnd/spells",
+				SchemaID: "notarius.dnd.spells", SchemaVersion: "v1",
+			},
+			"spells_second": {
+				LaneID: "spells-second", MediaType: "application/json", ModuleKey: "dnd/spells",
+				SchemaID: "notarius.dnd.spells", SchemaVersion: "v1",
+			},
+		},
+	}
+	return env, runManifest, notariusIntegrationFixture{transcriptPath: transcriptPath, partyPath: partyPath, chunkRoot: chunkRoot}
+}
+
+func notariusIntegrationTranscript(id, text string) []byte {
+	return []byte(fmt.Sprintf(`{"metadata":{"id":%q,"title":"Cache integration"},"segments":[{"id":1,"start":0,"end":4,"speaker":"Aria","text":%q}]}`, id, text))
+}
+
+func setNotariusIntegrationTranscript(t *testing.T, runManifest *manifest.Manifest, path string, content []byte) {
+	t.Helper()
+	if err := os.WriteFile(path, content, 0o644); err != nil {
+		t.Fatalf("write integration transcript: %v", err)
+	}
+	checksum, err := artifacts.SHA256File(path)
+	if err != nil {
+		t.Fatalf("checksum integration transcript: %v", err)
+	}
+	record := runManifest.Stages["trim"]
+	if record == nil || len(record.Outputs) != 1 {
+		t.Fatalf("trim record = %#v, want one output", record)
+	}
+	record.Outputs[0].Kind = artifactmodel.TranscriptOutputKindFinalTrimmed
+	record.Outputs[0].SourceID = artifactmodel.SourceTranscriptFinalTrimmed
+	record.Outputs[0].LocalPath = path
+	record.Outputs[0].Checksum = checksum
+}
+
+func setNotariusIntegrationReference(t *testing.T, runManifest *manifest.Manifest, path, content string) {
+	t.Helper()
+	descriptor, ok := artifactpolicy.DescribePreparedInputSource(artifactpolicy.SourceInputParty)
+	if !ok {
+		t.Fatal("prepared party input descriptor is unavailable")
+	}
+	if err := os.WriteFile(path, []byte(content), 0o644); err != nil {
+		t.Fatalf("write integration reference: %v", err)
+	}
+	checksum, err := artifacts.SHA256File(path)
+	if err != nil {
+		t.Fatalf("checksum integration reference: %v", err)
+	}
+	for index := range runManifest.Inputs {
+		if runManifest.Inputs[index].Kind == descriptor.ManifestKind {
+			runManifest.Inputs[index].Checksum = checksum
+			return
+		}
+	}
+	t.Fatalf("prepared party input not found in %#v", runManifest.Inputs)
+}
+
+func notariusIntegrationSnapshotPath(t *testing.T, env *Env, runManifest *manifest.Manifest, sourceID string) string {
+	t.Helper()
+	descriptor, ok := artifactpolicy.DescribePreparedInputSource(sourceID)
+	if !ok {
+		t.Fatalf("prepared input descriptor %q is unavailable", sourceID)
+	}
+	for _, input := range runManifest.Inputs {
+		if input.Kind != descriptor.ManifestKind {
+			continue
+		}
+		path, err := artifacts.SessionNotariusReferenceSnapshotPath(
+			sessionPathsForEnv(env, runManifest.SessionID), input.Checksum, filepath.Base(input.Path),
+		)
+		if err != nil {
+			t.Fatalf("resolve integration snapshot: %v", err)
+		}
+		return path
+	}
+	t.Fatalf("prepared input %q not found", sourceID)
+	return ""
+}
+
+func runNotariusIntegrationExtraction(t *testing.T, env *Env, runManifest *manifest.Manifest, runID string, force bool) notariusIntegrationRunManifest {
+	t.Helper()
+	runManifest.RunID = runID
+	env.Force = force
+	result, err := (extractStage{}).Run(context.Background(), env, runManifest)
+	if err != nil {
+		receiptPath := artifacts.SessionRunNotariusReceiptPathForCampaign(env.Config.Pipeline.Workspace.Root, runManifest.Campaign, runManifest.SessionID, runID)
+		if data, readErr := os.ReadFile(receiptPath); readErr == nil {
+			t.Logf("failed Notarius receipt: %s", data)
+		}
+		t.Fatalf("extract run %q: %v", runID, err)
+	}
+	bundleRoot, ok := result.Metadata["bundle_root"].(string)
+	if !ok || bundleRoot == "" {
+		t.Fatalf("extract metadata bundle_root = %#v", result.Metadata["bundle_root"])
+	}
+	data, err := os.ReadFile(filepath.Join(bundleRoot, "manifest.json"))
+	if err != nil {
+		t.Fatalf("read Notarius manifest: %v", err)
+	}
+	var published notariusIntegrationRunManifest
+	if err := json.Unmarshal(data, &published); err != nil {
+		t.Fatalf("decode Notarius manifest: %v", err)
+	}
+	return published
+}
+
+type notariusIntegrationRunManifest struct {
+	ChunkPlan *struct {
+		Action       string `json:"action"`
+		SourceDigest string `json:"source_digest"`
+	} `json:"chunk_plan"`
+	CheckpointDecisions []struct {
+		Stage      string `json:"stage"`
+		LaneID     string `json:"lane_id"`
+		Category   string `json:"category"`
+		ReasonCode string `json:"reason_code"`
+	} `json:"checkpoint_decisions"`
+}
+
+func assertNotariusIntegrationDecision(t *testing.T, published notariusIntegrationRunManifest, stageName, laneID, category, reasonCode string) {
+	t.Helper()
+	for _, decision := range published.CheckpointDecisions {
+		if decision.Stage == stageName && decision.LaneID == laneID {
+			if decision.Category != category || reasonCode != "" && decision.ReasonCode != reasonCode {
+				t.Fatalf("checkpoint decision for %s/%s = %#v, want category %q reason %q", stageName, laneID, decision, category, reasonCode)
+			}
+			return
+		}
+	}
+	t.Fatalf("checkpoint decision for %s/%s not found in %#v", stageName, laneID, published.CheckpointDecisions)
+}

internal/stage/publish_test.go

@@ -324,6 +324,9 @@ func TestPublishRejectsDeclaredDirectoryBeforeUploading(t *testing.T) {
 func TestPublishDoesNotUploadUndeclaredRunFile(t *testing.T) {
 	env, m, runRoot := publishFixture(t)
 	writeStageTestFile(t, filepath.Join(runRoot, "unlisted.txt"), "outside of manifest\n")
+	paths := artifacts.NewLocalStore(env.Config.Pipeline.Workspace.Root).SessionPathsFor(m.Campaign, m.SessionID)
+	cacheFile := filepath.Join(artifacts.SessionNotariusReferenceCacheDir(paths), strings.Repeat("a", 64), "party.yml")
+	writeStageTestFile(t, cacheFile, "local Notarius reference cache\n")
 
 	if _, err := (publishStage{}).Run(context.Background(), env, m); err != nil {
 		t.Fatalf("Run() error = %v", err)
@@ -333,8 +336,8 @@ func TestPublishDoesNotUploadUndeclaredRunFile(t *testing.T) {
 		t.Fatal("undeclared run file was uploaded")
 	}
 	for _, object := range fake.Objects {
-		if strings.Contains(string(object.Data), "outside of manifest") {
-			t.Fatalf("undeclared bytes were uploaded in %q", object.Key)
+		if strings.Contains(string(object.Data), "outside of manifest") || strings.Contains(string(object.Data), "local Notarius reference cache") {
+			t.Fatalf("undeclared or cache bytes were uploaded in %q", object.Key)
 		}
 	}
 }

internal/stage/semantic_contracts_delivery_test.go

@@ -76,6 +76,26 @@ func TestExtractSemanticConfigSensitivityAndOperationalExclusions(t *testing.T)
 	if !want.Equal(got) {
 		t.Fatalf("operational settings changed fingerprint: got %#v want %#v", got, want)
 	}
+
+	for _, test := range []struct {
+		name   string
+		mutate func(*config.NotariusConfig)
+	}{
+		{name: "resume", mutate: func(cfg *config.NotariusConfig) { cfg.Resume = true }},
+		{name: "chunk cache", mutate: func(cfg *config.NotariusConfig) { cfg.ChunkCache = "refresh" }},
+	} {
+		t.Run("operational "+test.name, func(t *testing.T) {
+			env := downstreamSemanticEnv()
+			test.mutate(env.Config.Pipeline.Notarius)
+			got, err := (extractStage{}).SemanticConfigFingerprint(env)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if !want.Equal(got) {
+				t.Fatalf("%s changed semantic fingerprint: got %#v want %#v", test.name, got, want)
+			}
+		})
+	}
 }
 
 func TestExtractResumeValidationExcludesOperationalSettings(t *testing.T) {
@@ -85,6 +105,8 @@ func TestExtractResumeValidationExcludesOperationalSettings(t *testing.T) {
 	env.Config.Pipeline.Notarius.ConfigPath = "/not/present/private.yml"
 	env.Config.Pipeline.Notarius.Timeout = "not-a-duration"
 	env.Config.Pipeline.Notarius.WorkingDirectory = "/not/present/work"
+	env.Config.Pipeline.Notarius.Resume = true
+	env.Config.Pipeline.Notarius.ChunkCache = "bypass"
 
 	validation, err := (extractStage{}).ValidateResume(context.Background(), env, m)
 	if err != nil {