11 Commits

Author SHA1 Message Date
c45e042462 Prepare the v2.1.0 release
All checks were successful
ci/woodpecker/push/verify Pipeline was successful
ci/woodpecker/tag/release Pipeline was successful
2026-09-16 18:50:30 +00:00
207da1b6ef Remove completed Notarius cache roadmaps
Some checks failed
ci/woodpecker/push/verify Pipeline was canceled
2026-09-16 18:48:33 +00:00
9a1609435e Clarify Notarius compatibility requirements 2026-09-16 18:48:23 +00:00
13b83d7ee2 Fix Notarius snapshots in relative workspaces 2026-09-16 18:48:23 +00:00
a0312ed034 Document Notarius cache reuse workflow 2026-09-16 18:14:40 +00:00
8ff74e1251 Prove Notarius cache reuse end to end 2026-09-16 18:07:37 +00:00
1b480d4ed5 Verify Notarius cache lifecycle boundaries 2026-09-16 17:59:52 +00:00
e13d87a077 Use stable verified Notarius reference snapshots 2026-09-16 17:53:29 +00:00
5202f16972 Forward Notarius cache controls during extraction 2026-09-16 17:44:28 +00:00
5086596f94 Add typed Notarius cache controls 2026-09-16 17:42:16 +00:00
f7fc64db10 Plan Notarius cache reuse 2026-09-16 17:34:21 +00:00
32 changed files with 1711 additions and 137 deletions

View File

@@ -212,6 +212,10 @@ artifacts and any prerequisites required by those targets; without
never run. Common session/configuration options and repeatable artifact values never run. Common session/configuration options and repeatable artifact values
pass through unchanged. 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 Because the expansion owns `--force`, `--from`, and `--through`, callers cannot
supply those options. The shared `run` parser reports them as duplicate supply those options. The shared `run` parser reports them as duplicate
singleton flags. The alias has no private execution options or behavior, and 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] 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` ### `publish`
```bash ```bash

View File

@@ -362,6 +362,8 @@ Rules:
| `pipeline.notarius.pipeline_id` | string | Conditional | required when enabled | | `pipeline.notarius.pipeline_id` | string | Conditional | required when enabled |
| `pipeline.notarius.timeout` | duration | No | `3h`; must be positive | | `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.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.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.notarius.outputs` | map | Conditional | at least one entry when enabled |
| `pipeline.render.enabled` | bool | No | `true` | | `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 See the [Notarius integration contract](./integrations/notarius.md) for the
delivery and redaction boundary. 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 ### Notarius Reference Bindings
`pipeline.notarius.references` maps a Notarius CLI selector to a prepared `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. Notarius owns target-slot declarations and media compatibility.
Before extraction, Narratio resolves every binding from the current prepared Before extraction, Narratio resolves every binding from the current prepared
session manifest and streams it into a verified invocation-local snapshot whose session manifest and streams it into a verified, content-addressed session
absolute path is passed to Notarius. Missing, unsafe, empty, snapshot whose absolute path is passed to Notarius. Missing, unsafe, empty,
changed-during-copy, or checksum-inconsistent prepared evidence fails with changed-during-copy, or checksum-inconsistent prepared evidence fails with
guidance to force `prepare`. Bindings are sorted by normalized selector and are guidance to force `prepare`. Bindings are sorted by normalized selector and are
part of extraction fingerprint and resume identity. See the part of extraction fingerprint and resume identity. See the

View File

@@ -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 safe bundle discovery, lane selection, and its own artifact metadata. Notarius
owns pipeline definitions, lane schemas, the receipt, and bundle formats. 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) - [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.6.0/docs/consumers/subprocess.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.6.0/docs/consumers/dnd-pipeline.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.6.0/docs/integrations/run-result.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.6.0/docs/integrations/json-output.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.6.0/docs/integrations/dnd-spell-catalog-overlays.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) The [complete Narratio example](../../examples/pipeline.full.annotated.yml)
records the exact current constraints for all ten D&D lanes. Treat the linked 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, When `pipeline.notarius.enabled` is true, Narratio resolves the executable,
configuration path, input path, output directory, and working directory to configuration path, input path, output directory, and working directory to
absolute paths. Narratio requires the Notarius v0.6.0 CLI contract when absolute paths. Configured references require support for the `--reference`
references are configured and invokes each binding as a separate argument contract introduced in Notarius v0.6.0. Cache controls require support for the
before `--json`: 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 ```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 Reference paths are absolute stable snapshots streamed from manifest-verified
manifest-verified canonical files prepared inside the current Narratio session canonical files prepared inside the current Narratio session workspace. They
workspace. Narratio verifies snapshot checksum and size before and after the live at `cache/notarius/references/<sha256>/<prepared-basename>` and retain the
subprocess, and passes only configured bindings, ordered lexically by normalized full lowercase content digest. Narratio verifies snapshot checksum and size
selector, as direct argument-vector entries without shell interpretation. A CLI 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 binding takes precedence over a matching external path in Notarius
configuration. Narratio never emits `--without-reference`. 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 automatically; the configured working directory and Narratio's minimal child
environment apply to the subprocess. 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 When `pipeline.notarius.llm_api_key` is absent or null, Narratio adds no
credential to that environment. When configured, the field names an environment credential to that environment. When configured, the field names an environment
variable whose already-loaded nonempty value is forwarded under that exact variable whose already-loaded nonempty value is forwarded under that exact
@@ -98,7 +116,7 @@ encoded credentials.
## Accepted Result ## 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 remains `notarius.run-result.v2`; reference flags do not change the
receipt or ten-lane output contract. The receipt receipt or ten-lane output contract. The receipt
must identify the configured pipeline, and its `index_file` must be exactly 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 incompatible required lanes fail extraction even if Notarius exited zero. A
configured lane whose v2 validation summary is `rejected` or `incomplete` also configured lane whose v2 validation summary is `rejected` or `incomplete` also
fails extraction. 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 Unconfigured lanes may remain in the preserved bundle but do not become
selectable Narratio sources. 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 Operator paths, rerun procedures, and bundle retention are in
[Operations](../operations.md). See [Troubleshooting](../troubleshooting.md) [Operations](../operations.md). See [Troubleshooting](../troubleshooting.md)
for failure recovery. 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.

View File

@@ -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. prepared reference resolution, receipt, bundle, and configured-lane policy.
The adapter validates the ordered selector/absolute-path pairs and is the sole The adapter validates the ordered selector/absolute-path pairs and is the sole
owner of serializing them as repeated `--reference` arguments before `--json`. 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 Its request may also carry one already-resolved opaque ConfigKit secret
reference. Only the Notarius adapter reads that value, rejects an unresolved or reference. Only the Notarius adapter reads that value, rejects an unresolved or
empty programmatic reference, and maps the exact configured name/value into the empty programmatic reference, and maps the exact configured name/value into the

View File

@@ -19,13 +19,15 @@ procedures belong in [Operations](../operations.md).
1. resolves the final trimmed transcript from the shared artifact catalog; 1. resolves the final trimmed transcript from the shared artifact catalog;
2. resolves every configured prepared reference through the shared 2. resolves every configured prepared reference through the shared
manifest-authoritative identity resolver before creating run-local output; manifest-authoritative identity resolver before creating run-local output;
3. streams each verified reference into an invocation-local snapshot and 3. resolves each reference to a stable content-addressed session snapshot,
rejects any source change observed while copying; 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, 4. fingerprints the byte- and provenance-bearing Notarius invocation evidence,
including sorted reference identities; including sorted reference identities;
5. creates a run-local staging directory and invokes the injected 5. creates a run-local staging directory and invokes the injected
`notarius.Runner`, carrying the optional already-resolved credential `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 6. revalidates the reference snapshots, then validates the v2 successful
receipt, confined index, management documents, configured required lane receipt, confined index, management documents, configured required lane
descriptors, validation summaries, and regular payload files; 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 fingerprint. The input identity binds the exact transcript bytes, canonical
source ID, producer stage/output/run identity, and resolution provenance. source ID, producer stage/output/run identity, and resolution provenance.
Reference metadata contains only selector, source ID, canonical session-relative Reference metadata contains only selector, source ID, canonical session-relative
path, checksum, and size; adapter requests receive selector and absolute path, checksum, and size; adapter requests receive selector and absolute stable
invocation-local snapshot path, never payload contents. Snapshot bytes must snapshot path, never payload contents. Snapshot bytes must
match the prepared identity both before and after Notarius runs, so a concurrent match the prepared identity both before and after Notarius runs, so a concurrent
prepared-file replacement cannot make recorded provenance describe different prepared-file replacement cannot make recorded provenance describe different
bytes from those supplied to Notarius. bytes from those supplied to Notarius.
Validation completes before the single durable rename, so a rejected result Validation completes before the single durable rename, so a rejected result
cannot expose a partial durable bundle. 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 Any executed extraction outcome that replaces a different effective outcome
marks succeeded analysis and delivery dependents stale. Render is an independent marks succeeded analysis and delivery dependents stale. Render is an independent
sibling and remains current. Repeating the same disabled self-skip with no 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 contracts. The same reference helper and transcript identity are resolved again
for artifact evidence, so changing current transcript bytes, reference bytes, for artifact evidence, so changing current transcript bytes, reference bytes,
or producer identity makes the prior extraction obsolete. Operational runner 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, A valid prepared-reference change makes extraction non-resumable. Missing,
unsafe, or checksum-inconsistent prepared evidence is a hard validation error unsafe, or checksum-inconsistent prepared evidence is a hard validation error
@@ -115,9 +128,13 @@ recovery.
`internal/stage/extract_resume.go`, `internal/stage/extract_resume.go`,
`internal/stage/extract_test.go`, `internal/stage/extract_test.go`,
`internal/stage/semantic_contracts_delivery.go` `internal/stage/semantic_contracts_delivery.go`
- Stable snapshot materialization and verification:
`internal/stage/extract_references.go`
- Subprocess boundary: `internal/adapters/notarius/subprocess.go`, - Subprocess boundary: `internal/adapters/notarius/subprocess.go`,
`internal/adapters/notarius/subprocess_test.go` `internal/adapters/notarius/subprocess_test.go`
- Catalog hydration: `internal/artifacts/extraction_catalog.go`, - Catalog hydration: `internal/artifacts/extraction_catalog.go`,
`internal/artifacts/extraction_catalog_test.go` `internal/artifacts/extraction_catalog_test.go`
- Composition and downstream behavior: `internal/app/runner_test.go`, - Composition and downstream behavior: `internal/app/runner_test.go`,
`internal/stage/analyze_test.go`, `internal/stage/publish_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`

View File

@@ -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 bundle path supplied by `internal/artifacts`. Exact physical paths belong in
[Operations](../operations.md#extraction-workflow). [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 ## Locking
`artifacts.LocalStore` enforces the single-writer session lock via an `artifacts.LocalStore` enforces the single-writer session lock via an

View File

@@ -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 Scriptorium, extract does not require Scriptorium, and analyze does not require
the transcription, Seriatim, Audita, or Notarius adapters. 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 ```bash
narratio regenerate-artifacts 2026-04-04 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 Configured Notarius references resolve only from the current manifest-backed
prepared inputs. Their canonical locations are `inputs/party.yml`, prepared inputs. Their canonical locations are `inputs/party.yml`,
`inputs/players.yml`, `inputs/glossary.yml`, and, when configured, `inputs/players.yml`, `inputs/glossary.yml`, and, when configured,
`inputs/spell_catalog.json`. Extraction supplies Notarius with verified copies `inputs/spell_catalog.json`. Extraction supplies Notarius with verified,
under `runs/<run_id>/extract/references/` so a concurrent refresh of canonical content-addressed copies under
prepared files cannot change the bytes consumed by an in-flight invocation. `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 For a canonical party, preparation retains the validated authored party bytes
at `inputs/party.yml` and generates `inputs/players.yml` from that roster. at `inputs/party.yml` and generates `inputs/players.yml` from that roster.
The manifest records their checksums separately, with the players input marked 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; unique durable bundle path. A pre-existing durable destination is a collision;
no duplicate promotion tree is created. 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 ## External Command Lifecycle
When an external command is cancelled or times out, Narratio terminates its When an external command is cancelled or times out, Narratio terminates its
@@ -509,6 +549,7 @@ Durable session paths:
- `logs/**` - `logs/**`
- `config/**` - `config/**`
- `runs/**` - `runs/**`
- `cache/notarius/references/<sha256>/<prepared-basename>`
`manifest.json` is the versioned session-progress document. Narratio accepts `manifest.json` is the versioned session-progress document. Narratio accepts
only `narratio.session-manifest.v1`. Do not edit unsupported session state to 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, Validated Notarius bundles live below `artifacts/notarius/{run_id}/`; receipt,
stderr, and pre-promotion output remain in the producing run's `extract` stderr, and pre-promotion output remain in the producing run's `extract`
directory as described in [Extraction Workflow](#extraction-workflow). 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: Run-local layout:

70
docs/releases/v2.1.0.md Normal file
View File

@@ -0,0 +1,70 @@
# Narratio v2.1.0
This release adds explicit Notarius credential forwarding and cache reuse,
with predictable forced regeneration and clearer artifact-development guidance.
## Summary
Narratio can pass one typed API-key reference to Notarius and opt into reuse of
its accepted execution checkpoints. Stable, verified reference snapshots allow
unchanged references to retain their cache identity across Narratio invocations.
Configuration loading now uses ConfigKit v0.3.0.
## Compatibility
The new `notarius.llm_api_key`, `notarius.resume`, and `notarius.chunk_cache`
fields are optional. Existing v2 configuration needs no schema migration.
Configured credential references must resolve during configuration loading,
even when extraction will not execute; omitted references add no requirement.
Forced extraction now omits Notarius checkpoint loading and requests chunk-plan
refresh. This includes `regenerate-artifacts`, which remains an alias for a
forced run from extract through analyze. `analyze` continues to force analysis
only, preserving extraction results and transcripts.
Notarius v0.7.0 is the tested cache integration baseline, not an unconditional
minimum-version gate. The executable must support the flags selected for the
invocation, including `--chunk_cache refresh` for forced extraction. Reference
bindings retain the contract introduced in Notarius v0.6.0. Receipt and artifact
schemas are unchanged; reused producers may report zero production attempts.
Supported release platforms remain Linux amd64/arm64 and best-effort macOS
arm64. macOS validation is cross-compilation, not native runtime testing.
## Upgrade
No special migration is required for existing v2 configuration. To use the new
features:
- Declare `notarius.llm_api_key` with the environment-variable name expected by
the deployed Notarius profile, supplying its value through the existing
environment or secrets-directory mechanism.
- Enable checkpoint recording in Notarius configuration before setting
Narratio's `notarius.resume: true`. Prefer explicit persistent Notarius cache
roots for service deployments. An optional `notarius.chunk_cache` overrides
its ordinary chunk-plan policy.
- Expect an initial checkpoint miss when moving from old run-local reference
paths to stable session snapshots. No migration or deletion of old caches is
required. Force refreshes the current input's plan, not the whole cache.
- Use `narratio analyze SESSION` for analysis-only iteration, and
`narratio regenerate-artifacts SESSION` when extraction should also be rebuilt.
See [Configuration](../config.md), the
[Notarius integration contract](../integrations/notarius.md), and
[Artifact Development Workflows](../operations.md#artifact-development-workflows)
for current details.
## Changes
- Upgraded ConfigKit from v0.1.2 to v0.3.0 and isolated pipeline-discovery tests
from machine configuration.
- Added optional typed Notarius credentials, exact-name environment delivery,
safe configuration projections, and captured-stream redaction for custom names.
- Added Notarius resume and chunk-cache controls with force taking precedence.
- Added verified, content-addressed session reference snapshots that survive
ordinary run cleanup without becoming published artifacts.
- Preserved relative-workspace support for snapshot delivery and confinement.
- Kept credentials and cache-policy controls outside extract's semantic resume
identity, so operational changes alone do not invalidate completed results.
- Added an opt-in real Notarius integration test using a local provider to
verify reuse, partial-failure recovery, and forced refresh without paid calls.

View File

@@ -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) checksum. Relevant references: [Notarius reference configuration](./config.md#notarius-reference-bindings)
and [Operations: Extraction Workflow](./operations.md#extraction-workflow). 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 ## Notarius reference selector or generated-handoff collision
Symptom: Symptom:

View File

@@ -60,7 +60,7 @@ with the sample campaign and a compatible local- or S3-audio session.
[glossary](campaigns/sample-campaign/glossary.yml), and canonical [glossary](campaigns/sample-campaign/glossary.yml), and canonical
[party](campaigns/sample-campaign/party.yml) fixture, plus an optional [party](campaigns/sample-campaign/party.yml) fixture, plus an optional
[spell-catalog overlay](campaigns/sample-campaign/spell_catalog.json) that [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. from this party source.
- [Sample speaker audio](audio/sample-speaker.flac) is a text placeholder that - [Sample speaker audio](audio/sample-speaker.flac) is a text placeholder that
reserves the expected filename and directory shape. Replace it with a real reserves the expected filename and directory shape. Replace it with a real

View File

@@ -128,7 +128,7 @@ trim:
report: false report: false
notarius: notarius:
# Optional structured extraction between trim and render. # Optional structured extraction after render and before analyze.
enabled: true enabled: true
binary: notarius binary: notarius
# Optional environment-variable reference. Match this name to api_key_env in # Optional environment-variable reference. Match this name to api_key_env in
@@ -138,6 +138,12 @@ notarius:
pipeline_id: dnd-session pipeline_id: dnd-session
timeout: 3h timeout: 3h
working_directory: /usr/local/etc/notarius 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 # External campaign references use prepared Narratio source IDs. Omit an
# optional binding when the selected Notarius pipeline does not need it. # optional binding when the selected Notarius pipeline does not need it.
references: references:

View File

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

View File

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

View File

@@ -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) { func TestSubprocessRunnerForwardsLoadedCredentialWithoutRereadingEnvironment(t *testing.T) {
const envName = "NARRATIO_NOTARIUS_CUSTOM_LLM_KEY" const envName = "NARRATIO_NOTARIUS_CUSTOM_LLM_KEY"
const loadedValue = "loaded-notarius-credential" 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) { func writeValidBundleAndReceipt(t *testing.T, req RunRequest, includeUnknown bool) {
t.Helper() t.Helper()
bundle := filepath.Join(req.OutputRoot, "notarius-run-1") bundle := filepath.Join(req.OutputRoot, "notarius-run-1")

View File

@@ -7,6 +7,7 @@ import (
"fmt" "fmt"
"os" "os"
"path/filepath" "path/filepath"
"reflect"
"sort" "sort"
"strings" "strings"
"testing" "testing"
@@ -182,12 +183,16 @@ func TestExtractLifecyclePrepareBindsVerifiedReferenceSnapshots(t *testing.T) {
for index, expected := range want { for index, expected := range want {
binding := request.References[index] binding := request.References[index]
canonical := filepath.Join(paths.InputsDir, expected.filename) canonical := filepath.Join(paths.InputsDir, expected.filename)
snapshot := filepath.Join( descriptor, ok := artifactpolicy.DescribePreparedInputSource(expected.sourceID)
artifacts.SessionRunNotariusReferencesDirForCampaign( if !ok {
cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, loaded.RunID, t.Fatalf("DescribePreparedInputSource(%q) ok = false", expected.sourceID)
), }
expected.filename, 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] { 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) 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) { func TestExtractLifecycleForcedSelfSkipInvalidatesDownstream(t *testing.T) {
cfg, env, _ := extractionLifecycleFixture(t, false) cfg, env, _ := extractionLifecycleFixture(t, false)
markLifecycleStageSucceeded(t, cfg, "analyze") markLifecycleStageSucceeded(t, cfg, "analyze")
@@ -783,6 +922,14 @@ func lifecycleInputChecksum(t *testing.T, m *manifest.Manifest, kind string) str
return "" 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 { func lifecycleDndOutputs() map[string]config.NotariusOutputConfig {
return 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"}, "item_registry": {LaneID: "item-registry", MediaType: "application/json", SchemaID: "notarius.dnd.item_registry", SchemaVersion: "v1", ModuleKey: "dnd/item-registry"},

View File

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

View File

@@ -10,6 +10,8 @@ import (
"gitea.maximumdirect.net/eric/narratio/internal/pathsafe" "gitea.maximumdirect.net/eric/narratio/internal/pathsafe"
) )
const sha256HexLength = 64
// SessionPaths contains canonical local paths for one session work directory. // SessionPaths contains canonical local paths for one session work directory.
type SessionPaths struct { type SessionPaths struct {
WorkspaceRoot string WorkspaceRoot string
@@ -28,6 +30,7 @@ type SessionPaths struct {
CurrentDir string CurrentDir string
RunsDir string RunsDir string
PreviousDir string PreviousDir string
CacheDir string
PreviousManifestPath string PreviousManifestPath string
PreviousArtifactsDir string PreviousArtifactsDir string
ManifestPath 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") 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. // SessionRunNotariusOutputRootForCampaign returns the invocation-local Notarius output root.
func SessionRunNotariusOutputRootForCampaign(rootDir, campaign, sessionID, runID string) string { func SessionRunNotariusOutputRootForCampaign(rootDir, campaign, sessionID, runID string) string {
return filepath.Join(SessionRunExtractDirForCampaign(rootDir, campaign, sessionID, runID), "notarius-output") 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. // SessionNotariusBundleDirForCampaign returns one immutable durable bundle destination.
func SessionNotariusBundleDirForCampaign(rootDir, campaign, sessionID, runID string) string { func SessionNotariusBundleDirForCampaign(rootDir, campaign, sessionID, runID string) string {
return filepath.Join( return filepath.Join(
@@ -268,6 +291,7 @@ func buildSessionPathsFromRoot(workspaceRoot, campaign, sessionID, root string)
CurrentDir: filepath.Join(root, config.PathCurrentDirSegment), CurrentDir: filepath.Join(root, config.PathCurrentDirSegment),
RunsDir: filepath.Join(root, config.PathRunsDirSegment), RunsDir: filepath.Join(root, config.PathRunsDirSegment),
PreviousDir: filepath.Join(root, config.PathPreviousDirSegment), PreviousDir: filepath.Join(root, config.PathPreviousDirSegment),
CacheDir: filepath.Join(root, config.PathCacheDirSegment),
PreviousManifestPath: filepath.Join(root, config.PathPreviousDirSegment, config.PathManifestFile), PreviousManifestPath: filepath.Join(root, config.PathPreviousDirSegment, config.PathManifestFile),
PreviousArtifactsDir: filepath.Join(root, config.PathPreviousDirSegment, config.PathArtifactsDirSegment), PreviousArtifactsDir: filepath.Join(root, config.PathPreviousDirSegment, config.PathArtifactsDirSegment),
ManifestPath: filepath.Join(root, config.PathManifestFile), ManifestPath: filepath.Join(root, config.PathManifestFile),

View File

@@ -2,6 +2,7 @@ package artifacts
import ( import (
"path/filepath" "path/filepath"
"strings"
"testing" "testing"
) )
@@ -54,7 +55,6 @@ func TestSessionNotariusPathsForCampaign(t *testing.T) {
{name: "extract directory", got: SessionRunExtractDirForCampaign(root, campaign, sessionID, runID), want: extractDir}, {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: "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: "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: "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)}, {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) { func TestSessionPreviousPathsForCampaign(t *testing.T) {
root := "/tmp/workspace" root := "/tmp/workspace"
previousDir := SessionPreviousDirForCampaign(root, "forsaken", "2026-04-19") previousDir := SessionPreviousDirForCampaign(root, "forsaken", "2026-04-19")

View File

@@ -310,6 +310,8 @@ type NotariusConfig struct {
Enabled bool `yaml:"enabled" json:"enabled"` Enabled bool `yaml:"enabled" json:"enabled"`
Binary string `yaml:"binary" json:"binary"` Binary string `yaml:"binary" json:"binary"`
LLMAPIKey *configkit.Secret `yaml:"llm_api_key,omitempty" json:"llm_api_key,omitempty"` 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"` ConfigPath string `yaml:"config_path" json:"config_path"`
PipelineID string `yaml:"pipeline_id" json:"pipeline_id"` PipelineID string `yaml:"pipeline_id" json:"pipeline_id"`
Timeout string `yaml:"timeout" json:"timeout"` Timeout string `yaml:"timeout" json:"timeout"`

View File

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

View File

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

View File

@@ -1,12 +1,15 @@
package config package config
import ( import (
"encoding/json"
"fmt" "fmt"
"os" "os"
"path/filepath" "path/filepath"
"reflect" "reflect"
"strings" "strings"
"testing" "testing"
"gopkg.in/yaml.v3"
) )
func TestNotariusOmittedAndDisabledBehavior(t *testing.T) { func TestNotariusOmittedAndDisabledBehavior(t *testing.T) {
@@ -45,6 +48,9 @@ notarius:
if disabled.Notarius.Binary != DefaultNotariusBinary || disabled.Notarius.Timeout != DefaultNotariusTimeout { if disabled.Notarius.Binary != DefaultNotariusBinary || disabled.Notarius.Timeout != DefaultNotariusTimeout {
t.Fatalf("disabled defaults = %#v", disabled.Notarius) 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 != "" { if disabled.Notarius.ConfigPath != "relative/notarius.yml" || disabled.Notarius.WorkingDirectory != "" {
t.Fatalf("disabled paths were resolved unexpectedly: %#v", disabled.Notarius) 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) { func TestNotariusRelativePathsDoNotMakeEffectiveDigestLocationDependent(t *testing.T) {
pipelineYAML := testPipelineBaseYAML + ` pipelineYAML := testPipelineBaseYAML + `
notarius: notarius:

View File

@@ -182,6 +182,57 @@ func TestNotariusCredentialRootProvenance(t *testing.T) {
t.Fatal("Notarius credential provenance did not identify the root pipeline") 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) { func TestPipelineConfigKitAllowsOneBaseDeclarationWithProfileOverride(t *testing.T) {
dir := t.TempDir() dir := t.TempDir()
writePipelineFixture(t, dir, "pipeline.yml", `configkit: writePipelineFixture(t, dir, "pipeline.yml", `configkit:

View File

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

View File

@@ -39,10 +39,12 @@ func validateNotarius(cfg *NotariusConfig, scriptorium *ScriptoriumConfig) error
if len(cfg.References) > MaxNotariusReferenceBindings { if len(cfg.References) > MaxNotariusReferenceBindings {
return fmt.Errorf("pipeline.notarius.references must contain at most %d bindings", MaxNotariusReferenceBindings) return fmt.Errorf("pipeline.notarius.references must contain at most %d bindings", MaxNotariusReferenceBindings)
} }
if err := validateNotariusCanonicalScalars(cfg); err != nil { if err := validateNotariusCanonicalScalars(cfg); err != nil {
return err 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 { if err := validateNotariusReferences(cfg.References); err != nil {
return err return err
} }
@@ -56,6 +58,7 @@ func validateNotariusCanonicalScalars(cfg *NotariusConfig) error {
want string want string
}{ }{
{name: "binary", got: cfg.Binary, want: strings.TrimSpace(cfg.Binary)}, {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: "config_path", got: cfg.ConfigPath, want: cleanNotariusPath(cfg.ConfigPath)},
{name: "pipeline_id", got: cfg.PipelineID, want: strings.TrimSpace(cfg.PipelineID)}, {name: "pipeline_id", got: cfg.PipelineID, want: strings.TrimSpace(cfg.PipelineID)},
{name: "timeout", got: cfg.Timeout, want: strings.TrimSpace(cfg.Timeout)}, {name: "timeout", got: cfg.Timeout, want: strings.TrimSpace(cfg.Timeout)},
@@ -69,6 +72,15 @@ func validateNotariusCanonicalScalars(cfg *NotariusConfig) error {
return nil 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) { func normalizeNotariusReferences(references map[string]string) (map[string]string, error) {
selectors := make([]string, 0, len(references)) selectors := make([]string, 0, len(references))
for selector := range references { for selector := range references {

View File

@@ -58,9 +58,12 @@ func TestCleanupStageRemovesOnlySelectedPublishedRunTargets(t *testing.T) {
otherSpool := filepath.Join(env.Config.Pipeline.Spool.Root, m.Campaign, m.SessionID, "other-run", "audio") 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") 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") 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(otherSpool, "audio.wav"), "other spool")
writeStageTestFile(t, filepath.Join(otherWorkspace, "result.txt"), "other workspace") writeStageTestFile(t, filepath.Join(otherWorkspace, "result.txt"), "other workspace")
writeStageTestFile(t, cacheFile, "cache") writeStageTestFile(t, cacheFile, "cache")
writeStageTestFile(t, referenceCacheFile, "reference cache")
result, err := (cleanupStage{}).Run(context.Background(), env, m) result, err := (cleanupStage{}).Run(context.Background(), env, m)
if err != nil { if err != nil {
@@ -81,7 +84,7 @@ func TestCleanupStageRemovesOnlySelectedPublishedRunTargets(t *testing.T) {
t.Fatalf("unselected workspace target changed: %v", err) 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 { if _, err := os.Stat(preserved); err != nil {
t.Fatalf("preserved path %q: %v", preserved, err) t.Fatalf("preserved path %q: %v", preserved, err)
} }

View File

@@ -124,20 +124,13 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (re
if err != nil { if err != nil {
return nil, fmt.Errorf("extract: resolve durable bundle path: %w", err) 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)} directories := []string{filepath.Dir(receiptPath), outputRoot, filepath.Dir(durableBundle)}
if len(references.Bindings) > 0 {
directories = append(directories, referenceSnapshotRoot)
}
for _, directory := range directories { for _, directory := range directories {
if err := fileops.EnsureWorkspaceDirectory(directory); err != nil { if err := fileops.EnsureWorkspaceDirectory(directory); err != nil {
return nil, fmt.Errorf("extract: create directory %q: %w", directory, err) 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 { if err != nil {
return nil, fmt.Errorf("extract: materialize Notarius reference snapshots: %w", err) 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 { if err != nil {
return nil, fmt.Errorf("extract: build configuration fingerprint: %w", err) 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{ request := notarius.RunRequest{
Binary: resolvedBinary, LLMAPIKey: notariusConfig.LLMAPIKey, ConfigPath: configPath, PipelineID: notariusConfig.PipelineID, Binary: resolvedBinary, LLMAPIKey: notariusConfig.LLMAPIKey, ConfigPath: configPath, PipelineID: notariusConfig.PipelineID,
InputPath: inputPath, OutputRoot: outputRoot, WorkingDirectory: workingDirectory, 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 stageSucceeded := false
bundleMoved := 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 { if adapterResult.Receipt.RunID == "" || adapterResult.Receipt.PipelineID != notariusConfig.PipelineID {
return nil, fmt.Errorf("extract: notarius receipt identity is missing or incompatible") 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) return nil, fmt.Errorf("extract: verify Notarius reference snapshots: %w", err)
} }

View File

@@ -3,6 +3,7 @@ package stage
import ( import (
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"errors"
"fmt" "fmt"
"io" "io"
"os" "os"
@@ -16,6 +17,7 @@ import (
"gitea.maximumdirect.net/eric/narratio/internal/fileops" "gitea.maximumdirect.net/eric/narratio/internal/fileops"
"gitea.maximumdirect.net/eric/narratio/internal/manifest" "gitea.maximumdirect.net/eric/narratio/internal/manifest"
"gitea.maximumdirect.net/eric/narratio/internal/notariusref" "gitea.maximumdirect.net/eric/narratio/internal/notariusref"
"gitea.maximumdirect.net/eric/narratio/internal/pathsafe"
) )
type extractReferenceIdentity struct { type extractReferenceIdentity struct {
@@ -29,83 +31,204 @@ type extractReferenceIdentity struct {
func materializeExtractReferenceSnapshots( func materializeExtractReferenceSnapshots(
paths artifacts.SessionPaths, paths artifacts.SessionPaths,
references resolvedExtractReferences, references resolvedExtractReferences,
snapshotRoot string,
) ([]notarius.ReferenceBinding, error) { ) ([]notarius.ReferenceBinding, error) {
var err error
paths, err = absoluteExtractSnapshotPaths(paths)
if err != nil {
return nil, err
}
if len(references.Bindings) != len(references.Identities) { if len(references.Bindings) != len(references.Identities) {
return nil, fmt.Errorf("reference bindings and identities have different lengths") return nil, fmt.Errorf("reference bindings and identities have different lengths")
} }
bindings := make([]notarius.ReferenceBinding, len(references.Bindings)) bindings := make([]notarius.ReferenceBinding, len(references.Bindings))
copied := make(map[string]string, len(references.Bindings)) materialized := make(map[string]extractReferenceIdentity, len(references.Bindings))
destinationOwners := make(map[string]string, len(references.Bindings))
for index, binding := range references.Bindings { for index, binding := range references.Bindings {
identity := references.Identities[index] identity := references.Identities[index]
destination, ok := copied[identity.SourceID] destination, err := extractReferenceSnapshotDestination(paths, identity)
if !ok { if err != nil {
filename := filepath.Base(filepath.FromSlash(identity.Path)) return nil, fmt.Errorf("reference %q source %q snapshot path: %w", identity.Selector, identity.SourceID, err)
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
} }
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} bindings[index] = notarius.ReferenceBinding{Selector: binding.Selector, Path: destination}
} }
return bindings, nil return bindings, nil
} }
func verifyExtractReferenceSnapshots(snapshotRoot string, references resolvedExtractReferences) error { func materializeExtractReferenceSnapshot(paths artifacts.SessionPaths, identity extractReferenceIdentity, destination string) error {
verified := make(map[string]struct{}, len(references.Identities)) if err := checkExtractSnapshotDirectory(paths.Root, filepath.Dir(destination), true); err != nil {
for _, identity := range references.Identities { return fmt.Errorf("reference %q snapshot cache: %w", identity.Selector, err)
if _, ok := verified[identity.SourceID]; ok { }
continue 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 { source, err := pathsafe.JoinSlashRelativeUnderRoot(paths.Root, identity.Path)
return fmt.Errorf("reopen reference %q source %q snapshot: %w", identity.Selector, identity.SourceID, err) if err != nil {
} return fmt.Errorf("resolve reference %q prepared source: %w", identity.Selector, err)
digest := sha256.New() }
size, readErr := io.Copy(digest, file) if err := verifyExtractReferenceFile(source, identity); err != nil {
closeErr := file.Close() return fmt.Errorf("reference %q prepared source %q no longer matches its manifest identity; rerun prepare with --force: %w", identity.Selector, source, err)
if readErr != nil { }
return fmt.Errorf("revalidate reference %q source %q snapshot: %w", identity.Selector, identity.SourceID, readErr) checksum, err := fileops.CopyFileAtomicWithChecksum(source, destination, fileops.WorkspaceFileMode)
} if err != nil {
if closeErr != nil { return fmt.Errorf("snapshot reference %q source %q to %q: %w", identity.Selector, identity.SourceID, destination, err)
return fmt.Errorf("close reference %q source %q snapshot: %w", identity.Selector, identity.SourceID, closeErr) }
} if checksum != identity.Checksum {
checksum := hex.EncodeToString(digest.Sum(nil)) return cleanupInvalidExtractSnapshot(paths, identity, destination,
if !strings.EqualFold(checksum, identity.Checksum) || size != identity.SizeBytes { fmt.Errorf("copied checksum %q does not match manifest checksum %q", checksum, identity.Checksum))
return fmt.Errorf("reference %q source %q snapshot changed while Notarius was running", identity.Selector, identity.SourceID) }
} if err := verifyExtractReferenceFile(destination, identity); err != nil {
verified[identity.SourceID] = struct{}{} return cleanupInvalidExtractSnapshot(paths, identity, destination, err)
} }
return nil 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 { type resolvedExtractReferences struct {
Bindings []notarius.ReferenceBinding Bindings []notarius.ReferenceBinding
Identities []extractReferenceIdentity Identities []extractReferenceIdentity

View File

@@ -96,6 +96,9 @@ notarius:
if len(req.References) != 0 { if len(req.References) != 0 {
t.Fatalf("references = %#v, want none", req.References) 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() { if req.LLMAPIKey != loadedPipeline.Notarius.LLMAPIKey || req.LLMAPIKey.EnvName() != credentialName || !req.LLMAPIKey.IsResolved() {
t.Fatal("request did not retain the loaded Notarius credential reference") 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) { func TestExtractStageResolvesAllPreparedReferencesInSelectorOrder(t *testing.T) {
env, m, fake := setupExtractEnv(t) env, m, fake := setupExtractEnv(t)
configured := []struct { configured := []struct {
@@ -149,12 +187,16 @@ func TestExtractStageResolvesAllPreparedReferencesInSelectorOrder(t *testing.T)
} }
for index, selector := range wantSelectors { for index, selector := range wantSelectors {
binding := fake.Requests[0].References[index] binding := fake.Requests[0].References[index]
wantPath := filepath.Join( checksum, err := artifacts.SHA256File(preparedPaths[selector])
artifacts.SessionRunNotariusReferencesDirForCampaign( if err != nil {
env.Config.Pipeline.Workspace.Root, m.Campaign, m.SessionID, m.RunID, t.Fatalf("SHA256File(%q) error = %v", selector, err)
), }
filepath.Base(preparedPaths[selector]), 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) { 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) 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 { type referenceMutationRunner struct {
delegate notarius.Runner delegate notarius.Runner
preparedPath string preparedPath string
preparedBytes []byte preparedBytes []byte
mutateSnapshot bool mutate func(string) error
snapshotBytes []byte snapshotBytes []byte
} }
func (r *referenceMutationRunner) Run(ctx context.Context, req notarius.RunRequest) (notarius.RunResult, error) { 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 return notarius.RunResult{}, err
} }
r.snapshotBytes = data r.snapshotBytes = data
if r.mutateSnapshot { if r.mutate != nil {
if err := os.WriteFile(req.References[0].Path, []byte("tampered snapshot\n"), 0o664); err != nil { if err := r.mutate(req.References[0].Path); err != nil {
return notarius.RunResult{}, err return notarius.RunResult{}, err
} }
} }
@@ -234,21 +276,292 @@ func TestExtractStageInvokesNotariusWithVerifiedReferenceSnapshot(t *testing.T)
t.Fatalf("Notarius reference bytes = %q, want verified snapshot", runner.snapshotBytes) t.Fatalf("Notarius reference bytes = %q, want verified snapshot", runner.snapshotBytes)
} }
if len(fake.Requests) != 1 || fake.Requests[0].References[0].Path == preparedPath { 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) { 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} env.Config.Pipeline.Notarius.References = map[string]string{"party": artifactpolicy.SourceInputParty}
preparedPath := recordPreparedExtractInput(t, env, m, artifactpolicy.SourceInputParty, "verified party\n") preparedPath := recordPreparedExtractInput(t, env, m, artifactpolicy.SourceInputParty, "stable party\n")
env.Notarius = &referenceMutationRunner{ paths := sessionPathsForEnv(env, m.SessionID)
delegate: fake, preparedPath: preparedPath, preparedBytes: []byte("changed after snapshot\n"), mutateSnapshot: true, 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 := os.Remove(firstPath); err != nil {
if err == nil || !strings.Contains(err.Error(), "snapshot changed while Notarius was running") || result != nil { t.Fatalf("Remove(snapshot) error = %v", err)
t.Fatalf("Run() result = %#v error = %v, want changed-snapshot failure", result, 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)
} }
} }

View File

@@ -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)
}

View File

@@ -324,6 +324,9 @@ func TestPublishRejectsDeclaredDirectoryBeforeUploading(t *testing.T) {
func TestPublishDoesNotUploadUndeclaredRunFile(t *testing.T) { func TestPublishDoesNotUploadUndeclaredRunFile(t *testing.T) {
env, m, runRoot := publishFixture(t) env, m, runRoot := publishFixture(t)
writeStageTestFile(t, filepath.Join(runRoot, "unlisted.txt"), "outside of manifest\n") 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 { if _, err := (publishStage{}).Run(context.Background(), env, m); err != nil {
t.Fatalf("Run() error = %v", err) t.Fatalf("Run() error = %v", err)
@@ -333,8 +336,8 @@ func TestPublishDoesNotUploadUndeclaredRunFile(t *testing.T) {
t.Fatal("undeclared run file was uploaded") t.Fatal("undeclared run file was uploaded")
} }
for _, object := range fake.Objects { for _, object := range fake.Objects {
if strings.Contains(string(object.Data), "outside of manifest") { if strings.Contains(string(object.Data), "outside of manifest") || strings.Contains(string(object.Data), "local Notarius reference cache") {
t.Fatalf("undeclared bytes were uploaded in %q", object.Key) t.Fatalf("undeclared or cache bytes were uploaded in %q", object.Key)
} }
} }
} }

View File

@@ -76,6 +76,26 @@ func TestExtractSemanticConfigSensitivityAndOperationalExclusions(t *testing.T)
if !want.Equal(got) { if !want.Equal(got) {
t.Fatalf("operational settings changed fingerprint: got %#v want %#v", got, want) 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) { 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.ConfigPath = "/not/present/private.yml"
env.Config.Pipeline.Notarius.Timeout = "not-a-duration" env.Config.Pipeline.Notarius.Timeout = "not-a-duration"
env.Config.Pipeline.Notarius.WorkingDirectory = "/not/present/work" 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) validation, err := (extractStage{}).ValidateResume(context.Background(), env, m)
if err != nil { if err != nil {