eric/notarius
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement final fixes and close out the implemetation roadmap

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-17 23:16:16 -05:00
parent 604c7a7945
commit 9fa9154dda
14 changed files with 463 additions and 879 deletions
Download Patch File Download Diff File Expand all files Collapse all files

165
docs/roadmap/adr0005.md
View File

@@ -1,165 +0,0 @@
# ADR-0005 Feature Roadmap
This roadmap records the implemented target state for
[ADR-0005](../adr/0005-cache-canonical-chunk-plans-by-source.md). The
[implementation record](implementation.md) preserves the completed work.
## Intent
Notarius may run multiple extraction passes over the same source. Chunking,
especially LLM-backed chunking over a large transcript, can be substantially
more expensive than later per-chunk operations. Stable chunk material also
improves the opportunity for provider-side prompt-cache reads across passes.
Notarius therefore prioritizes stable, aggressively reused source partitions
over automatically applying later chunk-module, reference, prompt, model, or
pipeline changes. An operator must explicitly request repartitioning.
## Target State
### Canonical source plan
Each validated generic source document has at most one stored chunk plan. The
source document's canonical digest selects that plan. Pipeline identity,
selected lanes, configured chunk module and options, references, validators,
prompts, schemas, and LLM settings do not participate in lookup.
In the default mode, the first accepted plan for a source is reused by later
pipelines and invocations. A configured chunk module runs only when no valid
stored plan exists, when reuse is bypassed for one invocation, or when the
operator requests refresh.
### Plan and annotation contract
A plan contains an ordered, non-empty collection of inclusive source-unit
ranges. The framework validates these boundaries and deterministically
materializes runtime chunks from the current source document.
Chunk modules may attach optional domain annotations at plan or range scope.
Annotations are namespaced, validated JSON values. They are preserved through
storage and materialization but are never a hard cross-domain capability.
Downstream modules may rely on the generic chunk contract only.
The store contains boundaries, annotations, warnings, and provenance. It does
not contain fully materialized chunks or duplicated source-unit content.
### Persistence
Chunk plans use a dedicated cache root that is independent of
`workspace.directory` and the locations of checkpoints, diagnostics, debug
artifacts, and durable output. The default is the platform-appropriate per-user
cache directory: on Linux, `$XDG_CACHE_HOME/notarius/chunk-plans` when
`XDG_CACHE_HOME` is set to a valid absolute path, otherwise
`$HOME/.cache/notarius/chunk-plans` when it is unset, through the platform
cache-directory resolver. An invalid relative `XDG_CACHE_HOME` is an error, not
a fallback.
Operators may set `workspace.chunk_cache.directory` to replace that root. The
recommended system-wide setting for a dedicated Linux service account is
`/var/cache/notarius/chunk-plans`. This is an operational recommendation, not
the application default; the service account must own the directory and it must
not be shared across mutually untrusted users.
One mutable, versioned plan file is stored beneath the chunk-plan root under the
full canonical source digest. Normal publication and refresh replace that file
atomically; history, rollback, multiple variants, and content deduplication are
outside this feature.
Plan state is potentially sensitive. Paths are confined, directories and files
use restrictive permissions, default diagnostics omit annotation and source
payloads, and debug output remains explicitly opt-in.
### Runtime policy
The effective chunk-cache mode is one of:
- `auto`: load a valid stored plan; otherwise generate, validate, and publish
one;
- `bypass`: do not read or write plan state for this invocation; or
- `refresh`: generate and validate a plan, then atomically replace stored state.
`auto` is the default. Invalid or incompatible stored state is a reported miss
in `auto`; it is replaced only after a newly generated plan is accepted. A
failed generation never overwrites prior state. Concurrent writes must never
expose partial data and use last-successful-atomic-write semantics.
The public CLI override is
`--chunk_cache <auto|bypass|refresh>`. Configuration and environment values use
the same three modes, with CLI taking highest precedence.
### Validation and execution
Framework plan validation and deterministic materialization run for generated
and reused plans. The current pipeline's configured chunk-validator chain then
validates the materialized chunks.
On a cache hit, the configured chunk module is constructed during normal
pipeline preparation but its operation is not invoked and it makes no LLM call.
A validator rejection of a structurally valid reused plan is a run outcome; it
does not implicitly authorize rechunking.
Only the generic `chunks` capability is hard. Domain annotations such as D&D
scene information are opportunistic and cannot be required solely because the
current pipeline selected the module that normally produces them.
Canonical plans are the sole owner of chunk reuse. Invocation-scoped
checkpoints no longer load or record chunk outputs. Downstream checkpoint
identity continues to depend on the digest of the effective materialized
chunks, so refresh invalidates affected downstream work.
### Provenance
Every run distinguishes the chunk module requested by the resolved pipeline
from the producer of the effective stored plan. Durable provenance records the
source and plan digests, plan schema version, effective cache mode and action,
producer module, producer references and LLM profile where applicable, and
non-sensitive producer metadata.
Default diagnostics record lookup, validation, generation, and publication
decisions without source or annotation payloads. Opt-in debug artifacts may
contain complete plan and annotation material and are treated as sensitive.
### Deployment documentation
The implemented configuration reference identifies the dedicated directory
field, its environment override, and the per-user default. The operations guide
documents the resulting filesystem layout and permissions, and recommends
`/var/cache/notarius/chunk-plans` for a system-wide Linux deployment running as
a dedicated service account. It also explains that cache deletion is
recoverable but may repeat expensive chunk generation. Neither document
presents the system-wide path as the default for an ordinary unprivileged
invocation.
## Compatibility Policy
Existing fully materialized chunk checkpoint files are not migrated or promoted
to canonical plans. They are ignored for chunk reuse after this feature lands.
The first `auto` run generates the source's plan through the configured chunk
module.
The existing source digest is the canonical lookup identity. Any source change
that alters that digest creates a separate plan. Earlier plan schema versions,
malformed files, and digest mismatches are incompatible and follow the invalid
`auto`-miss behavior.
## Non-Goals
This feature does not provide:
- plan editing, comparison, history, rollback, or garbage collection;
- remote or shared plan storage;
- multiple active or automatically selected plan variants for one source;
- mandatory domain annotation contracts;
- automatic rechunking because configuration or model inputs changed; or
- a guarantee that an LLM provider will report cache hits.
## Completion Outcomes
**Achieved.** Independent runs over the same canonical source
reuse byte-stable materialized chunks across pipeline, lane, reference,
chunk-module, and LLM configuration changes; bypass and refresh obey their
documented state semantics; provenance identifies the effective producer;
legacy chunk checkpoints cannot compete with plan reuse; and all focused,
integration, compatibility, CLI, and repository-wide validation pass. The
configuration and operations references must also document both the per-user
default and the recommended system-wide Linux setting.

578
docs/roadmap/implementation.md
View File

@@ -1,578 +0,0 @@
# ADR-0005 Implementation Record
This document records the implementation of the target state in
[ADR-0005 Feature Roadmap](adr0005.md), governed by
[ADR-0005](../adr/0005-cache-canonical-chunk-plans-by-source.md).
All stages are complete. This document remains a concise historical
implementation record. Current behavior is documented in the canonical
references linked from [Development](../development.md).
## Execution Rules
Before every stage:
1. read [Development](../development.md), both documents under `docs/policy/`,
ADR-0005, this plan, and every stage-specific document named below;
2. inspect `git status` and preserve user changes;
3. inspect the focused package contracts and tests before editing; and
4. confirm that all earlier stages are complete.
During every stage:
- implement only that stage and prerequisites discovered to be inseparable;
- use existing package boundaries and helpers before introducing abstractions;
- keep framework and core packages independent of concrete D&D modules;
- preserve cancellation, deterministic ordering, redaction, path confinement,
and atomic-write invariants;
- add or update focused tests with every behavior change;
- do not update current-behavior documentation before the final documentation
stage, except for ADR acceptance in Stage 1; and
- do not begin the next stage after meeting the current exit criteria.
At the end of every stage:
1. run the focused tests listed for that stage;
2. run `go test ./...`;
3. run `go vet ./...`;
4. run `go build ./cmd/notarius`;
5. run `git diff --check`; and
6. mark the stage complete in this document only after all checks pass.
If repository reality conflicts with this plan, stop and update the plan in the
same change before implementing a materially different design. Do not silently
invent a new contract.
## Fixed Decisions
These choices are settled for this implementation:
- `source.SourceDocument.Digest` is the cache lookup identity.
- The chunk-plan root is independent of `workspace.directory` and every other
state root.
- Its default joins the directory returned by `os.UserCacheDir` with
`notarius/chunk-plans`.
- `workspace.chunk_cache.directory` overrides that default and names the
chunk-plan root itself. The recommended value for a system-wide Linux service
is `/var/cache/notarius/chunk-plans`.
- The plan file is
`<chunk-plan-root>/<64-character-lowercase-source-sha256-hex>/plan.json`. The
validated `sha256:` prefix remains part of the logical digest but is removed
from the path segment.
- The plan envelope schema is `notarius.chunk-plan.v1`.
- There is one mutable plan file per source, atomically replaced. There are no
immutable plan objects, pointers, history, locks, or rollback.
- Concurrent writers use last-successful-atomic-write semantics; they must
never expose a partial file.
- Chunk modules retain the `contracts.Chunker` name but replace `Chunk` with a
`Plan` operation. No legacy interface remains after Stage 2.
- Plans and ranges use `map[string]json.RawMessage` annotations. Annotation
values are canonical valid JSON and namespaces are non-empty trimmed strings.
- Materialized chunks expose range annotations through
`source.Chunk.Annotations` and plan annotations through
`source.Chunk.PlanAnnotations`, never through `Metadata`.
- Framework materialization owns chunk IDs, indexes, references, content,
media type, units, and generic metadata.
- Only the generic `chunks` capability is hard; `chunks.scenes` is removed.
- Cached-plan hits still construct the configured chunker during preparation
but never call its operation or the chunk-stage LLM.
- The current pipeline's configured chunk validators run against materialized
chunks on every path.
- Canonical plan storage is the only chunk-reuse mechanism; chunk checkpoint
APIs and payloads are removed.
- The mode values are `auto`, `bypass`, and `refresh`. They are valid in config,
environment, and CLI, including persistent `refresh` configuration.
- The YAML field is `workspace.chunk_cache.mode`.
- The YAML directory field is `workspace.chunk_cache.directory`.
- The environment variable is `NOTARIUS_WORKSPACE_CHUNK_CACHE_MODE`.
- The root environment variable is `NOTARIUS_WORKSPACE_CHUNK_CACHE_DIR`.
- The CLI flag is `--chunk_cache <auto|bypass|refresh>`.
- Precedence is CLI, environment, file, then default `auto`.
- Chunk-plan-root precedence is environment, file, then the per-user default;
there is no CLI root override.
- Every supplied file, environment, and CLI value is parsed strictly even when
a higher-precedence source would override it; precedence selects among valid
values and does not mask malformed configuration.
- Existing chunk checkpoint files are ignored and never migrated.
- A structurally invalid stored plan is an `auto` miss. A configured-validator
rejection of a structurally valid hit does not trigger regeneration.
## Planned Contracts
Use these names unless an existing collision requires the smallest obvious
adjustment.
### Source-zone types
Add to `internal/core/source`:
```go
type ChunkAnnotations map[string]json.RawMessage
type ChunkPlan struct {
SourceDigest string `json:"source_digest"`
Ranges []ChunkRange `json:"ranges"`
Annotations ChunkAnnotations `json:"annotations,omitempty"`
}
type ChunkRange struct {
StartUnitID int `json:"start_unit_id"`
EndUnitID int `json:"end_unit_id"`
Annotations ChunkAnnotations `json:"annotations,omitempty"`
}
```
Add `Annotations ChunkAnnotations` and
`PlanAnnotations ChunkAnnotations` to `source.Chunk`, with JSON names
`annotations` and `plan_annotations` and `omitempty` on both.
The plan digest covers `SourceDigest`, ordered ranges, and canonical annotation
bytes. It excludes producer provenance, warnings, timestamps, and storage
schema. Materialization produces:
- IDs `chunk-%06d`, numbered from one;
- zero-based indexes in plan order;
- references spanning the first and last units in each range;
- cloned contiguous source units;
- media type `application/json`;
- canonical JSON content shaped as `{"units":[...]}`;
- generic metadata containing `start_unit_id`, `end_unit_id`, and `unit_count`;
- cloned range annotations in `Chunk.Annotations`; and
- cloned plan annotations in `Chunk.PlanAnnotations`.
The plan and range scopes remain distinct even when both use the same namespace.
Copying plan annotations into each runtime chunk is intentional: it makes them
available to downstream chunk-scoped operations without duplicating source
content in durable storage.
Framework validation permits gaps and overlap between ranges, but requires
strictly increasing start positions and rejects duplicate or backward ranges.
Producing modules may impose stricter policy before returning a plan.
### Module contract
Replace the current chunk result and operation with:
```go
type ChunkPlanResult struct {
Plan source.ChunkPlan
Warnings []Warning
}
type Chunker interface {
Key() string
ReferenceSlots() []ReferenceSlot
Plan(context.Context, ChunkRequest) (ChunkPlanResult, error)
}
```
Keep `ChunkRequest`, chunk module keys, stage name `chunk`, configuration
bindings, registries, and preparation ownership unchanged.
### Durable plan record
Define framework-facing record and store contracts beside other pipeline
collaborator contracts:
```go
type ChunkPlanProducer struct {
InputModule string
ChunkModule string
LLMProfile string
References []artifacts.ReferenceProvenance
Metadata map[string]any
}
type ChunkPlanRecord struct {
SchemaVersion string
SourceDigest string
PlanDigest string
Plan source.ChunkPlan
Producer ChunkPlanProducer
Warnings []contracts.Warning
CreatedAt time.Time
}
type ChunkPlanStore interface {
Load(sourceDigest string) (ChunkPlanRecord, ChunkPlanDecision, error)
Save(ChunkPlanRecord) error
}
type ChunkPlanStoreFactory func(root string) (ChunkPlanStore, error)
type ChunkPlanDecision struct {
Status ChunkPlanStatus
Reason string
}
type ChunkPlanStatus string
const (
ChunkPlanHit ChunkPlanStatus = "hit"
ChunkPlanMissing ChunkPlanStatus = "missing"
ChunkPlanInvalid ChunkPlanStatus = "invalid"
)
```
`ChunkPlanDecision` reports `hit`, `missing`, or `invalid` plus a redacted
reason. A miss or invalid result returns a zero record and nil error. Missing
and structurally invalid records are recoverable in `auto`. Filesystem access
failures other than absence return an error and are framework failures.
The concrete constructor is:
```go
// Package chunkplan
func NewFilesystemStore(root string) (pipeline.ChunkPlanStore, error)
```
It treats `root` as the exact chunk-plan root. CLI composition injects this
constructor through `cli.Options.ChunkPlanStoreFactory`; tests may inject a
recording factory. `cli.Options.UserCacheDir` has signature
`func() (string, error)` and defaults to `os.UserCacheDir`.
The JSON envelope uses snake-case equivalents of the record fields. It stores
references as provenance only, never reference content. Before returning a hit,
the loader validates schema version, source digest, plan digest, plan structure,
annotation JSON, and materializability against the current source in the runner.
### Cache policy and provenance
Add a `ChunkCacheMode` enum with `auto`, `bypass`, and `refresh`, plus strict
parsing and validation.
Add to `RunInput`:
```go
ChunkCacheMode ChunkCacheMode
ChunkPlans ChunkPlanStore
```
An empty mode used by direct framework callers means `bypass`; the production
CLI always supplies the effective non-empty mode.
Add a run-manifest summary:
```go
type ChunkPlanManifest struct {
Mode string `json:"mode"`
Action string `json:"action,omitempty"`
SourceDigest string `json:"source_digest,omitempty"`
PlanDigest string `json:"plan_digest,omitempty"`
PlanSchemaVersion string `json:"plan_schema_version,omitempty"`
RequestedModule string `json:"requested_module"`
ProducerInputModule string `json:"producer_input_module,omitempty"`
ProducerModule string `json:"producer_module,omitempty"`
ProducerLLMProfile string `json:"producer_llm_profile,omitempty"`
ProducerReferences []ReferenceProvenance `json:"producer_references,omitempty"`
ProducerMetadata map[string]any `json:"producer_metadata,omitempty"`
CreatedAt *time.Time `json:"created_at,omitempty"`
}
```
`Action` is one of `reused`, `generated`, `refreshed`, or `bypassed`. Retain
the existing top-level `input_module` and `chunker` fields as the requested
pipeline modules. Add this field to `artifacts.RunManifest`:
```go
ChunkPlan *ChunkPlanManifest `json:"chunk_plan,omitempty"`
```
The runner initializes it with the effective mode and requested chunk module,
then fills candidate and producer fields when a plan reaches materialization.
## Completed Stage Summaries
### Stage 1: Finalize ADR and add the source-zone plan model
**Status:** Complete
Accepted ADR-0005 and added the domain-neutral chunk-plan, range, annotation,
validation, digest, cloning, and deterministic materialization primitives in
`internal/core/source`.
### Stage 2: Replace the chunk operation with plan generation
**Status:** Complete
Replaced chunk-returning module operations with `Chunker.Plan`, converted the
generic and D&D scene chunkers to produce ranges and optional annotations, and
made the runner materialize and validate generated plans.
### Stage 3: Add dedicated cache configuration and the plan store
**Status:** Complete
Added strict cache modes, independent per-user or explicitly configured
chunk-plan roots, the source-addressed filesystem store, restrictive
permissions, envelope validation, and atomic replacement.
### Stage 4: Integrate cache policy and remove chunk checkpoints
**Status:** Complete
Implemented runner semantics for `auto`, `bypass`, and `refresh`; retained
current chunk-validator behavior; made accepted canonical plans the sole chunk
reuse mechanism; and removed chunk checkpoint loading and recording.
### Stage 5: Wire persistent policy into the CLI
**Status:** Complete
Added CLI, environment, and file configuration precedence; per-user root
resolution; persistent store construction; bypass isolation; and cross-run
reuse through normal CLI execution.
### Stage 6: Expose producer provenance and safe diagnostics
**Status:** Complete
Added requested-versus-producer chunk-plan provenance to manifests and a
payload-free chunk-plan decision summary to normal diagnostics, with complete
plan material confined to opt-in debug artifacts.
### Stage 7: Add cross-run, corruption, and concurrency hardening
**Status:** Complete
Added integration, corruption, interruption, concurrency, compatibility,
cross-domain annotation, import-boundary, and race coverage for the implemented
cache design.
### Stage 8: Publish current-behavior documentation
**Status:** Complete
Updated architecture, pipeline, module, configuration, CLI, operations,
diagnostics, integration, and overview documentation for the implemented
feature and its per-user and system-wide deployment guidance.
## Stage 9: Redact invalid cache-record diagnostics
**Status:** Complete
### Objective
Ensure malformed or incompatible cache contents cannot enter normal errors,
warnings, manifests, or diagnostics while retaining useful closed lookup
decisions.
### Read first
- [Architecture Policy](../policy/architecture.md), especially secret handling
- [Diagnostics Internals](../internal/diagnostics.md)
- `internal/framework/chunkplan/store.go` and its focused tests
- `internal/framework/pipeline/runner_chunk_plan.go` and its focused tests
- `internal/core/artifacts/artifacts.go`
- the CLI path that writes `chunk-plan.json`
### Implement
1. Stop interpolating decoder errors, unknown field names, schema values,
annotation namespaces, timestamps, paths, or any other stored content into
recoverable `ChunkPlanDecision.Reason` values.
2. At the normal-diagnostic boundary, ignore the reason supplied by a
`ChunkPlanStore` and derive `ChunkPlanSummary.LookupReason` solely from
status using exactly:
- `hit`: `stored chunk plan is valid`;
- `missing`: `chunk plan not found`;
- `invalid`: `stored chunk plan is invalid`; and
- `skipped`: `chunk plan lookup skipped`.
3. Keep operational read failures as framework errors rather than recoverable
invalid-record decisions. Their normal error text must identify the failed
operation without including cache-file content.
4. Do not add raw invalid-record details to another default surface. Direct
operator inspection or a future explicitly sensitive debug facility may
expose them, but this stage does not need to add such a facility.
### Tests
Add focused tests proving that sentinel secrets placed in unknown field names,
schema values, annotation namespaces, timestamps, and malformed trailing JSON
appear in neither the store decision nor `chunk-plan.json`. Add a runner test
with a custom store whose reason contains a sentinel and prove that the summary
uses the fixed status-derived reason. Preserve useful missing, hit, invalid, and
operational-error classification.
Run focused tests for `internal/framework/chunkplan`,
`internal/framework/pipeline`, and `internal/cli`, then perform the
repository-wide validation in the execution rules.
### Exit criteria
Every normal lookup reason belongs to the fixed safe vocabulary, hostile
invalid-file content cannot reach default diagnostics, operational failures
remain correctly classified, and all focused and repository-wide checks pass.
## Stage 10: Enforce rooted chunk-plan filesystem access
**Status:** Complete
### Objective
Make the cache's path-confinement guarantee resistant to symlinked cache entries
while preserving atomicity, permissions, and concurrent access semantics.
### Read first
- [Architecture Policy](../policy/architecture.md), especially state and path
invariants
- [Operations](../operations.md)
- `internal/framework/chunkplan/store.go` and all store tests
- Go's `os.Root` API for the repository's declared Go version
### Implement
1. Treat the configured cache root as the filesystem trust boundary. A symlink
in the operator-supplied root path itself may be followed when the root is
opened, but every store-owned operation after that point must remain beneath
the opened root.
2. Perform digest-directory, plan-file, temporary-file, and rename operations
through `os.Root` or an equivalently race-resistant directory-relative
mechanism. Do not rely on `filepath.Join` plus a preflight `Lstat` as the
confinement boundary.
3. Reject symlinks and unexpected file types at the store-owned digest-directory
and `plan.json` positions. A missing cache root on `Load` remains a normal
miss and is not created; `Save` creates the root and digest directory as
needed.
4. Preserve the exact
`<root>/<full-source-sha256-hex>/plan.json` layout, `0700` directories,
`0600` plan files, strict digest validation, and no implicit path suffix.
5. Publish using an exclusively created random temporary file inside the digest
directory, sync and close it, then atomically rename it through the same
rooted handle. Remove temporary files on every pre-rename failure. Preserve
last-successful-write and complete-reader semantics.
6. Keep store access serial at the runner boundary and safe for concurrent
independent store callers. Do not introduce a lock, history, or rollback
protocol.
### Tests
Add load and save cases for digest-directory and plan-file symlinks that resolve
outside the configured root. Prove that they are rejected and that no outside
file is read, created, chmodded, or replaced. Include unexpected file types and
failed-publication preservation.
Retain and run tests for absent-root behavior, exact paths, permissions, round
trip, malformed digests, atomic replacement, interrupted writes, temporary-file
cleanup, concurrent readers, and concurrent writers. Run the race detector for
`internal/framework/chunkplan`, then perform the repository-wide validation in
the execution rules.
### Exit criteria
No store-owned operation can escape the opened root through a cache entry,
normal layout and permissions remain unchanged, publication remains atomic,
concurrency tests and the race detector pass, and the repository is green.
## Stage 11: Deep-clone source metadata during materialization
**Status:** Complete
### Objective
Ensure source documents, materialized chunks, and separate materializations
never share mutable metadata storage, regardless of the concrete JSON-shaped
map, slice, array, interface, raw-message, or byte-slice types used by an input
adapter.
### Read first
- `internal/core/source/source.go`
- `internal/core/source/chunk_plan.go` and its focused tests
- `internal/core/source/digest.go` and validation behavior
- every metadata clone helper and ownership boundary found with `rg`
- focused runner tests that hand chunks to validators and extractors
### Implement
1. Replace the concrete-type allowlist used by plan materialization with one
reusable source-owned deep-clone implementation for supported
JSON-serializable metadata.
2. Recursively clone string-keyed maps, slices, arrays, interfaces,
`json.RawMessage`, and byte slices without sharing mutable backing storage.
Preserve scalar values and the logical JSON representation. Typed composite
containers must not fall through as shared values.
3. Detect cyclic or unsupported values and return a contextual error from
materialization rather than retaining an alias, panicking, or recursing
indefinitely.
4. Use the source-owned helper at pipeline ownership boundaries where doing so
removes a duplicate incomplete implementation. Do not introduce a dependency
from `internal/core/source` to a framework package and do not expand this
stage into unrelated metadata-schema redesign.
5. Preserve deterministic materialized content, chunk digests, annotations, IDs,
references, and metadata fields for all currently supported production
inputs.
### Tests
Cover typed nested maps, slices of typed maps, arrays, interfaces,
`json.RawMessage`, and byte slices. Mutate each materialized value and prove
that neither the source document nor a separately materialized chunk changes.
Add cyclic and unsupported-value cases that fail deterministically with
contextual errors. Retain the existing exact-materialization and digest tests.
Run focused tests for `internal/core/source`,
`internal/framework/pipeline`, and affected module packages, then perform the
repository-wide validation in the execution rules.
### Exit criteria
Every supported mutable metadata value has independent ownership across source,
chunk, validation, and extraction boundaries; invalid metadata fails safely;
current materialized bytes remain stable; and all checks pass.
## Stage 12: Finalize remediation documentation and roadmap status
**Status:** Complete
### Objective
Make the roadmap and canonical documentation accurately describe the completed
feature after Stages 9 through 11 land.
### Read first
- [Documentation Policy](../policy/documentation.md)
- [ADR-0005 Feature Roadmap](adr0005.md)
- [Architecture Policy](../policy/architecture.md)
- [Diagnostics Internals](../internal/diagnostics.md)
- [Operations](../operations.md)
- the final code and tests from Stages 9 through 11
### Implement
1. Remove the feature roadmap's statement that ADR-0005 is unimplemented and
describe it as the implemented target-state record. Mark its completion
outcomes as achieved without adding staged implementation detail.
2. Recheck architecture, diagnostics, operations, pipeline, and module
documentation against the remediated behavior. Update a canonical document
only where Stages 9 through 11 changed a current operator or developer
contract; do not duplicate implementation detail across documents.
3. Mark Stages 9 through 12 complete only after their respective code,
documentation, focused tests, and repository-wide validation have passed.
Update this document's introduction to state that all stages are complete and
that it remains as a concise historical implementation record.
4. Validate every changed link, heading, symbol, diagnostic field, permission,
and filesystem claim against the implementation.
### Tests and validation
Run:
```sh
go test ./...
go vet ./...
go build ./cmd/notarius
git diff --check
```
Run any repository documentation or link checker if present, and manually
inspect the reading map in [Development](../development.md) and component links
in [Internal Overview](../internal/overview.md).
### Exit criteria
The feature roadmap and canonical references describe implemented behavior,
this plan accurately records completion, documentation ownership remains clear,
all links and names are current, and the repository-wide validation is green.

158
internal/framework/chunkplan/store.go
View File

@@ -11,6 +11,7 @@ import (
"os" "os"
"path/filepath" "path/filepath"
"strings" "strings"
"syscall"
"gitea.maximumdirect.net/eric/notarius/internal/core/source" "gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
@@ -50,7 +51,7 @@ func (s *filesystemStore) Load(sourceDigest string) (pipeline.ChunkPlanRecord, p
} }
defer root.Close() defer root.Close()
state, err := inspectDirectory(root, digestDir) digestRoot, state, err := openDigestRoot(root, digestDir, false)
if err != nil { if err != nil {
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, fmt.Errorf("inspect chunk plan directory: %w", err) return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, fmt.Errorf("inspect chunk plan directory: %w", err)
} }
@@ -60,11 +61,11 @@ func (s *filesystemStore) Load(sourceDigest string) (pipeline.ChunkPlanRecord, p
if state == entryRejected { if state == entryRejected {
return invalidDecision() return invalidDecision()
} }
defer digestRoot.Close()
target := planPath(digestDir) data, state, err := readPlan(digestRoot)
state, err = inspectPlan(root, target)
if err != nil { if err != nil {
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, fmt.Errorf("inspect chunk plan file: %w", err) return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, fmt.Errorf("read chunk plan: %w", err)
} }
if state == entryMissing { if state == entryMissing {
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanMissing, Reason: lookupReason(pipeline.ChunkPlanMissing)}, nil return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanMissing, Reason: lookupReason(pipeline.ChunkPlanMissing)}, nil
@@ -73,14 +74,6 @@ func (s *filesystemStore) Load(sourceDigest string) (pipeline.ChunkPlanRecord, p
return invalidDecision() return invalidDecision()
} }
data, err := root.ReadFile(target)
if err != nil {
if os.IsNotExist(err) {
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanMissing, Reason: lookupReason(pipeline.ChunkPlanMissing)}, nil
}
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, fmt.Errorf("read chunk plan: %w", err)
}
var record pipeline.ChunkPlanRecord var record pipeline.ChunkPlanRecord
decoder := json.NewDecoder(bytes.NewReader(data)) decoder := json.NewDecoder(bytes.NewReader(data))
decoder.DisallowUnknownFields() decoder.DisallowUnknownFields()
@@ -116,11 +109,15 @@ func (s *filesystemStore) Save(record pipeline.ChunkPlanRecord) error {
return fmt.Errorf("open chunk plan root: %w", err) return fmt.Errorf("open chunk plan root: %w", err)
} }
defer root.Close() defer root.Close()
if err := ensureDirectory(root, digestDir); err != nil { digestRoot, state, err := openDigestRoot(root, digestDir, true)
if err != nil {
return fmt.Errorf("prepare chunk plan directory: %w", err) return fmt.Errorf("prepare chunk plan directory: %w", err)
} }
target := planPath(digestDir) if state == entryRejected {
state, err := inspectPlan(root, target) return fmt.Errorf("chunk plan directory has an unsupported type")
}
defer digestRoot.Close()
state, err = inspectPlan(digestRoot, planFileName)
if err != nil { if err != nil {
return fmt.Errorf("inspect chunk plan file: %w", err) return fmt.Errorf("inspect chunk plan file: %w", err)
} }
@@ -131,7 +128,7 @@ func (s *filesystemStore) Save(record pipeline.ChunkPlanRecord) error {
if writer == nil { if writer == nil {
writer = writeAtomic writer = writeAtomic
} }
if err := writer(root, target, data); err != nil { if err := writer(digestRoot, planFileName, data); err != nil {
return fmt.Errorf("write chunk plan: %w", err) return fmt.Errorf("write chunk plan: %w", err)
} }
return nil return nil
@@ -174,52 +171,89 @@ const (
entryRejected entryRejected
) )
func inspectDirectory(root *os.Root, digestDir string) (entryState, error) { func inspectDirectory(root *os.Root, digestDir string) (entryState, os.FileInfo, error) {
info, err := root.Lstat(digestDir) info, err := root.Lstat(digestDir)
if err != nil { if err != nil {
if os.IsNotExist(err) { if os.IsNotExist(err) {
return entryMissing, nil return entryMissing, nil, nil
} }
return entryPresent, err return entryPresent, nil, err
} }
if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() { if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
return entryRejected, nil return entryRejected, info, nil
} }
return entryPresent, nil return entryPresent, info, nil
} }
func ensureDirectory(root *os.Root, digestDir string) error { type digestOpenHooks struct {
BeforeOpen func() error
}
func openDigestRoot(root *os.Root, digestDir string, create bool) (*os.Root, entryState, error) {
return openDigestRootWithHooks(root, digestDir, create, digestOpenHooks{})
}
func openDigestRootWithHooks(root *os.Root, digestDir string, create bool, hooks digestOpenHooks) (*os.Root, entryState, error) {
for { for {
state, err := inspectDirectory(root, digestDir) state, before, err := inspectDirectory(root, digestDir)
if err != nil { if err != nil {
return err return nil, entryPresent, err
} }
switch state { switch state {
case entryRejected: case entryRejected:
return fmt.Errorf("chunk plan directory has an unsupported type") return nil, entryRejected, nil
case entryMissing: case entryMissing:
if !create {
return nil, entryMissing, nil
}
if err := root.Mkdir(digestDir, 0o700); err != nil && !errors.Is(err, os.ErrExist) { if err := root.Mkdir(digestDir, 0o700); err != nil && !errors.Is(err, os.ErrExist) {
return err return nil, entryPresent, err
} }
continue continue
} }
directory, err := root.Open(digestDir) if hooks.BeforeOpen != nil {
if err := hooks.BeforeOpen(); err != nil {
return nil, entryPresent, err
}
}
digestRoot, err := root.OpenRoot(digestDir)
if err != nil { if err != nil {
return err return nil, entryPresent, err
} }
info, statErr := directory.Stat() opened, statErr := digestRoot.Stat(".")
if statErr == nil && !info.IsDir() { afterState, after, afterErr := inspectDirectory(root, digestDir)
statErr = fmt.Errorf("chunk plan directory has an unsupported type") if statErr != nil || afterErr != nil {
_ = digestRoot.Close()
if statErr != nil {
return nil, entryPresent, statErr
}
return nil, entryPresent, afterErr
} }
if statErr == nil { if afterState != entryPresent || !os.SameFile(before, after) || !os.SameFile(opened, after) {
statErr = directory.Chmod(0o700) _ = digestRoot.Close()
if afterState == entryRejected {
return nil, entryRejected, nil
}
continue
} }
closeErr := directory.Close() if create {
if statErr != nil { directory, openErr := digestRoot.Open(".")
return statErr if openErr != nil {
_ = digestRoot.Close()
return nil, entryPresent, openErr
}
chmodErr := directory.Chmod(0o700)
closeErr := directory.Close()
if chmodErr != nil || closeErr != nil {
_ = digestRoot.Close()
if chmodErr != nil {
return nil, entryPresent, chmodErr
}
return nil, entryPresent, closeErr
}
} }
return closeErr return digestRoot, entryPresent, nil
} }
} }
@@ -237,8 +271,54 @@ func inspectPlan(root *os.Root, target string) (entryState, error) {
return entryPresent, nil return entryPresent, nil
} }
func planPath(digestDir string) string { type planReadHooks struct {
return digestDir + "/" + planFileName BeforeOpen func() error
}
func readPlan(root *os.Root) ([]byte, entryState, error) {
return readPlanWithHooks(root, planReadHooks{})
}
func readPlanWithHooks(root *os.Root, hooks planReadHooks) ([]byte, entryState, error) {
state, err := inspectPlan(root, planFileName)
if err != nil || state != entryPresent {
return nil, state, err
}
if hooks.BeforeOpen != nil {
if err := hooks.BeforeOpen(); err != nil {
return nil, entryPresent, err
}
}
file, err := root.OpenFile(planFileName, os.O_RDONLY|syscall.O_NOFOLLOW|syscall.O_NONBLOCK, 0)
if err != nil {
if os.IsNotExist(err) {
return nil, entryMissing, nil
}
if errors.Is(err, syscall.ELOOP) {
return nil, entryRejected, nil
}
return nil, entryPresent, err
}
defer file.Close()
opened, err := file.Stat()
if err != nil {
return nil, entryPresent, err
}
if !opened.Mode().IsRegular() {
return nil, entryRejected, nil
}
currentState, currentErr := inspectPlan(root, planFileName)
if currentErr != nil {
return nil, entryPresent, currentErr
}
if currentState == entryRejected {
return nil, entryRejected, nil
}
data, err := io.ReadAll(file)
if err != nil {
return nil, entryPresent, err
}
return data, entryPresent, nil
} }
func digestPathSegment(digest string) (string, error) { func digestPathSegment(digest string) (string, error) {

55
internal/framework/chunkplan/store_test.go
View File

@@ -163,6 +163,61 @@ func TestFilesystemStoreRejectsSymlinkedEntries(t *testing.T) {
} }
} }
func TestOpenDigestRootRejectsEntryReplacedDuringOpen(t *testing.T) {
rootPath := t.TempDir()
digestDir := strings.TrimPrefix(testSourceDigest, "sha256:")
if err := os.Mkdir(filepath.Join(rootPath, digestDir), 0o700); err != nil {
t.Fatal(err)
}
if err := os.Mkdir(filepath.Join(rootPath, "redirect"), 0o700); err != nil {
t.Fatal(err)
}
root, err := os.OpenRoot(rootPath)
if err != nil {
t.Fatal(err)
}
defer root.Close()
traced := false
opened, state, err := openDigestRootWithHooks(root, digestDir, false, digestOpenHooks{BeforeOpen: func() error {
if traced {
return nil
}
traced = true
if err := os.Rename(filepath.Join(rootPath, digestDir), filepath.Join(rootPath, "original")); err != nil {
return err
}
return os.Symlink("redirect", filepath.Join(rootPath, digestDir))
}})
if opened != nil {
_ = opened.Close()
}
if err != nil || state != entryRejected {
t.Fatalf("openDigestRootWithHooks() root=%v state=%v error=%v, want nil/rejected/nil", opened, state, err)
}
}
func TestReadPlanRejectsEntryReplacedDuringOpen(t *testing.T) {
rootPath := t.TempDir()
writeFile(t, filepath.Join(rootPath, planFileName), []byte("original"), 0o600)
writeFile(t, filepath.Join(rootPath, "redirect.json"), []byte("redirect"), 0o600)
root, err := os.OpenRoot(rootPath)
if err != nil {
t.Fatal(err)
}
defer root.Close()
data, state, err := readPlanWithHooks(root, planReadHooks{BeforeOpen: func() error {
if err := os.Remove(filepath.Join(rootPath, planFileName)); err != nil {
return err
}
return os.Symlink("redirect.json", filepath.Join(rootPath, planFileName))
}})
if err != nil || state != entryRejected || data != nil {
t.Fatalf("readPlanWithHooks() data=%q state=%v error=%v, want nil/rejected/nil", data, state, err)
}
}
func TestFilesystemStoreRejectsUnexpectedEntryTypes(t *testing.T) { func TestFilesystemStoreRejectsUnexpectedEntryTypes(t *testing.T) {
tests := []struct { tests := []struct {
name string name string

6
internal/framework/pipeline/artifact_codec_registry.go
View File

@@ -67,7 +67,7 @@ type artifactCodecEntry struct {
valueType reflect.Type valueType reflect.Type
encode func(any) ([]byte, error) encode func(any) ([]byte, error)
encodeCandidate func(any) ([]byte, error) encodeCandidate func(any) ([]byte, error)
metadata func(any) map[string]any metadata func(any) (map[string]any, error)
decode func([]byte) (any, error) decode func([]byte) (any, error)
} }
@@ -132,10 +132,10 @@ func RegisterArtifactCodec[T any](registry *ArtifactCodecRegistry, codec contrac
return append([]byte(nil), content...), nil return append([]byte(nil), content...), nil
} }
if provider, ok := any(codec).(interface{ Metadata(T) map[string]any }); ok { if provider, ok := any(codec).(interface{ Metadata(T) map[string]any }); ok {
entry.metadata = func(value any) map[string]any { entry.metadata = func(value any) (map[string]any, error) {
typed, err := exactTypedValue[T]("artifact metadata", value) typed, err := exactTypedValue[T]("artifact metadata", value)
if err != nil { if err != nil {
return nil return nil, err
} }
return cloneMetadata(provider.Metadata(typed)) return cloneMetadata(provider.Metadata(typed))
} }

5
internal/framework/pipeline/chunk_annotations_test.go
View File

@@ -29,7 +29,10 @@ func TestChunkCanonicalizationAndClonePreserveAnnotationScopes(t *testing.T) {
t.Fatalf("plan annotation = %q", got) t.Fatalf("plan annotation = %q", got)
} }
cloned := cloneSourceChunk(chunks[0]) cloned, err := cloneSourceChunk(chunks[0])
if err != nil {
t.Fatal(err)
}
cloned.Annotations["shared"][0] = '[' cloned.Annotations["shared"][0] = '['
cloned.PlanAnnotations["shared"][0] = '[' cloned.PlanAnnotations["shared"][0] = '['
if string(chunks[0].Annotations["shared"]) != `{"range":true}` || string(chunks[0].PlanAnnotations["shared"]) != `{"plan":true}` { if string(chunks[0].Annotations["shared"]) != `{"range":true}` || string(chunks[0].PlanAnnotations["shared"]) != `{"plan":true}` {

10
internal/framework/pipeline/chunk_validation.go
View File

@@ -21,12 +21,16 @@ func validateAndMaterializeChunkPlan(doc *source.SourceDocument, plan source.Chu
return canonical, chunks, nil return canonical, chunks, nil
} }
func cloneSourceUnit(unit source.SourceUnit) source.SourceUnit { func cloneSourceUnit(unit source.SourceUnit) (source.SourceUnit, error) {
metadata, err := cloneMetadata(unit.Metadata)
if err != nil {
return source.SourceUnit{}, fmt.Errorf("clone source unit %d metadata: %w", unit.ID, err)
}
return source.SourceUnit{ return source.SourceUnit{
ID: unit.ID, ID: unit.ID,
Kind: unit.Kind, Kind: unit.Kind,
Text: unit.Text, Text: unit.Text,
Ref: unit.Ref, Ref: unit.Ref,
Metadata: cloneMetadata(unit.Metadata), Metadata: metadata,
} }, nil
} }

12
internal/framework/pipeline/debug.go
View File

@@ -471,7 +471,7 @@ func debugSourceDocumentEnvelope(doc *source.SourceDocument) *debugSourceDocumen
Kind: doc.Kind, Kind: doc.Kind,
Format: doc.Format, Format: doc.Format,
Digest: doc.Digest, Digest: doc.Digest,
Units: cloneSourceUnits(doc.Units), Units: cloneSourceUnitsForDebug(doc.Units),
Metadata: redactSensitiveMap(doc.Metadata), Metadata: redactSensitiveMap(doc.Metadata),
} }
} }
@@ -483,13 +483,21 @@ func debugSourceChunkEnvelope(chunk source.Chunk) debugSourceChunk {
Index: chunk.Index, Index: chunk.Index,
Ref: chunk.Ref, Ref: chunk.Ref,
Content: debugContentEnvelope(chunk.Content, chunk.MediaType, chunk.Metadata, nil), Content: debugContentEnvelope(chunk.Content, chunk.MediaType, chunk.Metadata, nil),
Units: cloneSourceUnits(chunk.Units), Units: cloneSourceUnitsForDebug(chunk.Units),
Metadata: redactSensitiveMap(chunk.Metadata), Metadata: redactSensitiveMap(chunk.Metadata),
Annotations: source.CloneChunkAnnotations(chunk.Annotations), Annotations: source.CloneChunkAnnotations(chunk.Annotations),
PlanAnnotations: source.CloneChunkAnnotations(chunk.PlanAnnotations), PlanAnnotations: source.CloneChunkAnnotations(chunk.PlanAnnotations),
} }
} }
func cloneSourceUnitsForDebug(units []source.SourceUnit) []source.SourceUnit {
cloned, err := cloneSourceUnits(units)
if err != nil {
return nil
}
return cloned
}
func debugSourceChunkEnvelopes(chunks []source.Chunk) []debugSourceChunk { func debugSourceChunkEnvelopes(chunks []source.Chunk) []debugSourceChunk {
if len(chunks) == 0 { if len(chunks) == 0 {
return nil return nil

149
internal/framework/pipeline/runner.go
View File

@@ -78,6 +78,10 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
if err := validateRunInput(input); err != nil { if err := validateRunInput(input); err != nil {
return output, err return output, err
} }
input.Metadata, err = cloneMetadata(input.Metadata)
if err != nil {
return output, fmt.Errorf("clone run metadata: %w", err)
}
input.pipeline = input.Prepared.resolved input.pipeline = input.Prepared.resolved
input.llmClient = input.Prepared.dependencies.LLM input.llmClient = input.Prepared.dependencies.LLM
@@ -121,7 +125,9 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
} }
adapter := input.Prepared.input adapter := input.Prepared.input
attachModuleManifestMetadata(&output, "input", adapter) if err := attachModuleManifestMetadata(&output, "input", adapter); err != nil {
return failOutput(output), err
}
sourceCheckpoint, sourceDecision := checkpointLoader.Source(adapter.Key()) sourceCheckpoint, sourceDecision := checkpointLoader.Source(adapter.Key())
recordCheckpointEvent(&output, checkpointLoader, "source", "", adapter.Key(), sourceDecision) recordCheckpointEvent(&output, checkpointLoader, "source", "", adapter.Key(), sourceDecision)
doc := sourceCheckpoint.Document doc := sourceCheckpoint.Document
@@ -144,12 +150,16 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
if err := checkpoints.SourceRunning(adapter.Key()); err != nil { if err := checkpoints.SourceRunning(adapter.Key()); err != nil {
return failOutput(output), fmt.Errorf("write source checkpoint: %w", err) return failOutput(output), fmt.Errorf("write source checkpoint: %w", err)
} }
requestMetadata, metadataErr := cloneMetadata(input.Metadata)
if metadataErr != nil {
return failOutput(output), fmt.Errorf("clone input adapter metadata: %w", metadataErr)
}
doc, err = adapter.Parse(ctx, contracts.ParseRequest{ doc, err = adapter.Parse(ctx, contracts.ParseRequest{
SourceID: input.SourceID, SourceID: input.SourceID,
Path: input.Path, Path: input.Path,
Raw: input.RawInput, Raw: input.RawInput,
LLMProfile: input.pipeline.Input.LLMProfile, LLMProfile: input.pipeline.Input.LLMProfile,
Metadata: input.Metadata, Metadata: requestMetadata,
}) })
if err != nil { if err != nil {
_ = checkpoints.SourceFailed(adapter.Key(), err) _ = checkpoints.SourceFailed(adapter.Key(), err)
@@ -177,11 +187,16 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
} }
sourceInput := sourceInputMaterial(input.Path, input.RawInput) sourceInput := sourceInputMaterial(input.Path, input.RawInput)
sessionID := resolvedSessionID(input.SessionID, doc.ID) sessionID := resolvedSessionID(input.SessionID, doc.ID)
output.Manifest.Metadata = manifestMetadataWithSessionID(output.Manifest.Metadata, sessionID) output.Manifest.Metadata, err = manifestMetadataWithSessionID(output.Manifest.Metadata, sessionID)
if err != nil {
return failOutput(output), err
}
output.Manifest.SourceDigests = []string{doc.Digest} output.Manifest.SourceDigests = []string{doc.Digest}
chunker := input.Prepared.chunker chunker := input.Prepared.chunker
attachModuleManifestMetadata(&output, "chunker", chunker) if err := attachModuleManifestMetadata(&output, "chunker", chunker); err != nil {
return failOutput(output), err
}
chunkStarted := time.Now().UTC() chunkStarted := time.Now().UTC()
chunkMode := effectiveChunkCacheMode(input.ChunkCacheMode) chunkMode := effectiveChunkCacheMode(input.ChunkCacheMode)
if err := writeDebugTimed(debugRecorder, "chunk/input.json", debugTimedEnvelope{ if err := writeDebugTimed(debugRecorder, "chunk/input.json", debugTimedEnvelope{
@@ -199,7 +214,9 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
return failOutput(output), fmt.Errorf("write chunk debug artifact: %w", err) return failOutput(output), fmt.Errorf("write chunk debug artifact: %w", err)
} }
chunkResult, err := r.runChunkPlan(ctx, input, doc, sourceInput, sessionID) chunkResult, err := r.runChunkPlan(ctx, input, doc, sourceInput, sessionID)
applyChunkPlanExecution(&output, chunkResult) if applyErr := applyChunkPlanExecution(&output, chunkResult); applyErr != nil {
return failOutput(output), applyErr
}
if err != nil { if err != nil {
return failOutput(output), err return failOutput(output), err
} }
@@ -233,7 +250,9 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
if chunkResult.accepted { if chunkResult.accepted {
laneOutput, laneErr := r.runLanes(ctx, input, checkpoints, checkpointLoader, doc, sourceInput, sessionID, chunkResult.chunks) laneOutput, laneErr := r.runLanes(ctx, input, checkpoints, checkpointLoader, doc, sourceInput, sessionID, chunkResult.chunks)
mergeLaneOutput(&output, laneOutput) if err := mergeLaneOutput(&output, laneOutput); err != nil {
return failOutput(output), err
}
if laneErr != nil { if laneErr != nil {
return failOutput(output), laneErr return failOutput(output), laneErr
} }
@@ -248,7 +267,9 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
output.Manifest.CompletedAt = timePtr(time.Now().UTC()) output.Manifest.CompletedAt = timePtr(time.Now().UTC())
encoder := input.Prepared.output encoder := input.Prepared.output
attachModuleManifestMetadata(&output, "output", encoder) if err := attachModuleManifestMetadata(&output, "output", encoder); err != nil {
return failOutput(output), err
}
outputStarted := time.Now().UTC() outputStarted := time.Now().UTC()
if err := writeDebugTimed(debugRecorder, "output/input.json", debugTimedEnvelope{ if err := writeDebugTimed(debugRecorder, "output/input.json", debugTimedEnvelope{
Stage: string(StageOutput), Stage: string(StageOutput),
@@ -265,13 +286,17 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
}); err != nil { }); err != nil {
return failOutput(output), fmt.Errorf("write output debug artifact: %w", err) return failOutput(output), fmt.Errorf("write output debug artifact: %w", err)
} }
outputMetadata, err := cloneMetadata(input.Metadata)
if err != nil {
return failOutput(output), fmt.Errorf("clone output encoder metadata: %w", err)
}
encoded, err := encoder.Encode(ctx, contracts.OutputRequest{ encoded, err := encoder.Encode(ctx, contracts.OutputRequest{
Manifest: output.Manifest, Manifest: output.Manifest,
NormalizeOutputs: cloneSerializedOutputs(output.NormalizeOutputs), NormalizeOutputs: cloneSerializedOutputs(output.NormalizeOutputs),
Rejected: cloneRejectedOutputs(output.Rejected), Rejected: cloneRejectedOutputs(output.Rejected),
Warnings: output.Warnings, Warnings: output.Warnings,
LLMProfile: input.pipeline.Output.LLMProfile, LLMProfile: input.pipeline.Output.LLMProfile,
Metadata: input.Metadata, Metadata: outputMetadata,
}) })
output.Warnings = append(output.Warnings, encoded.Warnings...) output.Warnings = append(output.Warnings, encoded.Warnings...)
if err != nil { if err != nil {
@@ -351,11 +376,19 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument,
attemptPath := path.Join("validate", debugPathComponent(string(StageChunk)), "", debugPathComponent(moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d", index+1, debugPathComponent(binding.Module), attempt)) attemptPath := path.Join("validate", debugPathComponent(string(StageChunk)), "", debugPathComponent(moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d", index+1, debugPathComponent(binding.Module), attempt))
validatorCtx, llmScope := withIsolatedDebugLLMScope(ctx, attemptPath) validatorCtx, llmScope := withIsolatedDebugLLMScope(ctx, attemptPath)
var result contracts.ValidationResult var result contracts.ValidationResult
requestMetadata, cloneErr := cloneMetadata(metadata)
if cloneErr != nil {
return nil, nil, fmt.Errorf("clone chunk validation metadata: %w", cloneErr)
}
requestChunks, cloneErr := cloneSourceChunks(chunks)
if cloneErr != nil {
return nil, nil, fmt.Errorf("clone chunks for validation: %w", cloneErr)
}
switch item.resolved.Target { switch item.resolved.Target {
case ValidatorTargetChunk: case ValidatorTargetChunk:
result, err = item.chunk.Validate(validatorCtx, contracts.ChunkValidationRequest{ModuleKey: moduleKey, Source: doc, SourceID: doc.ID, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(references), LLMProfile: binding.LLMProfile, Metadata: cloneMetadata(metadata), Chunks: cloneSourceChunks(chunks)}) result, err = item.chunk.Validate(validatorCtx, contracts.ChunkValidationRequest{ModuleKey: moduleKey, Source: doc, SourceID: doc.ID, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(references), LLMProfile: binding.LLMProfile, Metadata: requestMetadata, Chunks: requestChunks})
case ValidatorTargetSerialized: case ValidatorTargetSerialized:
result, err = item.serialized.Validate(validatorCtx, contracts.SerializedValidationRequest{Stage: string(StageChunk), ModuleKey: moduleKey, Source: doc, SourceID: doc.ID, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(references), LLMProfile: binding.LLMProfile, Metadata: cloneMetadata(metadata), Chunks: cloneSourceChunks(chunks), Schema: contracts.CloneArtifactSchema(schema), MediaType: "application/json", Content: append([]byte(nil), content...)}) result, err = item.serialized.Validate(validatorCtx, contracts.SerializedValidationRequest{Stage: string(StageChunk), ModuleKey: moduleKey, Source: doc, SourceID: doc.ID, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(references), LLMProfile: binding.LLMProfile, Metadata: requestMetadata, Chunks: requestChunks, Schema: contracts.CloneArtifactSchema(schema), MediaType: "application/json", Content: append([]byte(nil), content...)})
default: default:
return nil, nil, fmt.Errorf("validator %q is incompatible with chunk validation", binding.Module) return nil, nil, fmt.Errorf("validator %q is incompatible with chunk validation", binding.Module)
} }
@@ -616,31 +649,38 @@ func rejectedOutputManifests(rejected []contracts.RejectedOutput) []artifacts.Re
return manifests return manifests
} }
func attachModuleManifestMetadata(output *RunOutput, moduleKey string, module any) { func attachModuleManifestMetadata(output *RunOutput, moduleKey string, module any) error {
if output == nil { if output == nil {
return return nil
}
metadata, ok, err := moduleManifestMetadata(module)
if err != nil {
return fmt.Errorf("clone manifest metadata for module %q: %w", moduleKey, err)
} }
metadata, ok := moduleManifestMetadata(module)
if !ok { if !ok {
return return nil
} }
if output.Manifest.ModuleMetadata == nil { if output.Manifest.ModuleMetadata == nil {
output.Manifest.ModuleMetadata = make(map[string]map[string]any) output.Manifest.ModuleMetadata = make(map[string]map[string]any)
} }
output.Manifest.ModuleMetadata[moduleKey] = metadata output.Manifest.ModuleMetadata[moduleKey] = metadata
return nil
} }
func moduleManifestMetadata(module any) (map[string]any, bool) { func moduleManifestMetadata(module any) (map[string]any, bool, error) {
provider, ok := module.(contracts.ManifestMetadataProvider) provider, ok := module.(contracts.ManifestMetadataProvider)
if !ok { if !ok {
return nil, false return nil, false, nil
} }
moduleMetadata := cloneMetadata(provider.ManifestMetadata()) moduleMetadata, err := cloneMetadata(provider.ManifestMetadata())
if len(moduleMetadata) == 0 { if err != nil {
return nil, false return nil, false, err
} }
return moduleMetadata, true if len(moduleMetadata) == 0 {
return nil, false, nil
}
return moduleMetadata, true, nil
} }
func outputFilesFromResult(result contracts.OutputResult) ([]contracts.OutputFile, error) { func outputFilesFromResult(result contracts.OutputResult) ([]contracts.OutputFile, error) {
@@ -678,12 +718,8 @@ func validateOutputFileName(name string) error {
return nil return nil
} }
func cloneMetadata(metadata map[string]any) map[string]any { func cloneMetadata(metadata map[string]any) (map[string]any, error) {
cloned, err := source.CloneMetadata(metadata) return source.CloneMetadata(metadata)
if err != nil {
return metadata
}
return cloned
} }
func cloneLLMProfiles(profiles []artifacts.LLMProfileManifest) []artifacts.LLMProfileManifest { func cloneLLMProfiles(profiles []artifacts.LLMProfileManifest) []artifacts.LLMProfileManifest {
@@ -785,16 +821,19 @@ func resolvedSessionID(explicit string, sourceDocumentID string) string {
return strings.TrimSpace(sourceDocumentID) return strings.TrimSpace(sourceDocumentID)
} }
func manifestMetadataWithSessionID(metadata map[string]any, sessionID string) map[string]any { func manifestMetadataWithSessionID(metadata map[string]any, sessionID string) (map[string]any, error) {
out := cloneMetadata(metadata) out, err := cloneMetadata(metadata)
if err != nil {
return nil, fmt.Errorf("clone run manifest metadata: %w", err)
}
if strings.TrimSpace(sessionID) == "" { if strings.TrimSpace(sessionID) == "" {
return out return out, nil
} }
if out == nil { if out == nil {
out = make(map[string]any) out = make(map[string]any)
} }
out["session_id"] = sessionID out["session_id"] = sessionID
return out return out, nil
} }
func cloneWarnings(warnings []contracts.Warning) []contracts.Warning { func cloneWarnings(warnings []contracts.Warning) []contracts.Warning {
@@ -804,43 +843,61 @@ func cloneWarnings(warnings []contracts.Warning) []contracts.Warning {
return append([]contracts.Warning(nil), warnings...) return append([]contracts.Warning(nil), warnings...)
} }
func cloneSourceChunkPtr(chunk *source.Chunk) *source.Chunk { func cloneSourceChunkPtr(chunk *source.Chunk) (*source.Chunk, error) {
if chunk == nil { if chunk == nil {
return nil return nil, nil
} }
cloned := cloneSourceChunk(*chunk) cloned, err := cloneSourceChunk(*chunk)
return &cloned if err != nil {
return nil, err
}
return &cloned, nil
} }
func cloneSourceChunk(chunk source.Chunk) source.Chunk { func cloneSourceChunk(chunk source.Chunk) (source.Chunk, error) {
chunk.Content = append([]byte(nil), chunk.Content...) chunk.Content = append([]byte(nil), chunk.Content...)
chunk.Units = cloneSourceUnits(chunk.Units) var err error
chunk.Metadata = cloneMetadata(chunk.Metadata) chunk.Units, err = cloneSourceUnits(chunk.Units)
if err != nil {
return source.Chunk{}, err
}
chunk.Metadata, err = cloneMetadata(chunk.Metadata)
if err != nil {
return source.Chunk{}, fmt.Errorf("clone chunk metadata: %w", err)
}
chunk.Annotations = source.CloneChunkAnnotations(chunk.Annotations) chunk.Annotations = source.CloneChunkAnnotations(chunk.Annotations)
chunk.PlanAnnotations = source.CloneChunkAnnotations(chunk.PlanAnnotations) chunk.PlanAnnotations = source.CloneChunkAnnotations(chunk.PlanAnnotations)
return chunk return chunk, nil
} }
func cloneSourceChunks(chunks []source.Chunk) []source.Chunk { func cloneSourceChunks(chunks []source.Chunk) ([]source.Chunk, error) {
if len(chunks) == 0 { if len(chunks) == 0 {
return nil return nil, nil
} }
out := make([]source.Chunk, 0, len(chunks)) out := make([]source.Chunk, 0, len(chunks))
for _, chunk := range chunks { for _, chunk := range chunks {
out = append(out, cloneSourceChunk(chunk)) cloned, err := cloneSourceChunk(chunk)
if err != nil {
return nil, err
}
out = append(out, cloned)
} }
return out return out, nil
} }
func cloneSourceUnits(units []source.SourceUnit) []source.SourceUnit { func cloneSourceUnits(units []source.SourceUnit) ([]source.SourceUnit, error) {
if len(units) == 0 { if len(units) == 0 {
return nil return nil, nil
} }
out := make([]source.SourceUnit, 0, len(units)) out := make([]source.SourceUnit, 0, len(units))
for _, unit := range units { for _, unit := range units {
out = append(out, cloneSourceUnit(unit)) cloned, err := cloneSourceUnit(unit)
if err != nil {
return nil, err
}
out = append(out, cloned)
} }
return out return out, nil
} }
func cloneSerializedOutputs(outputs []contracts.SerializedOutput) []contracts.SerializedOutput { func cloneSerializedOutputs(outputs []contracts.SerializedOutput) []contracts.SerializedOutput {

59
internal/framework/pipeline/runner_chunk_plan.go
View File

@@ -58,7 +58,9 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
case ChunkPlanHit: case ChunkPlanHit:
plan, chunks, validationErr := validateAndMaterializeChunkPlan(doc, record.Plan) plan, chunks, validationErr := validateAndMaterializeChunkPlan(doc, record.Plan)
if validationErr == nil { if validationErr == nil {
result.setCandidate(record, "reused") if err := result.setCandidate(record, "reused"); err != nil {
return result, fmt.Errorf("clone reused chunk plan record: %w", err)
}
validationWarnings, rejection, err := r.validateChunks(ctx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.Metadata, input.Prepared.chunkValidators, 1, input.Debug) validationWarnings, rejection, err := r.validateChunks(ctx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.Metadata, input.Prepared.chunkValidators, 1, input.Debug)
result.plan = &plan result.plan = &plan
result.chunks = chunks result.chunks = chunks
@@ -87,10 +89,14 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
attemptPath := path.Join("chunk", fmt.Sprintf("attempt-%02d", attempt)) attemptPath := path.Join("chunk", fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "chunk", llmScope, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, StartedAt: attemptStarted}) terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "chunk", llmScope, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, StartedAt: attemptStarted})
requestMetadata, metadataErr := cloneMetadata(input.Metadata)
if metadataErr != nil {
return false, nil, terminal.record(nil, fmt.Errorf("clone chunk request metadata: %w", metadataErr))
}
chunkResult, callErr := chunker.Plan(attemptCtx, contracts.ChunkRequest{ chunkResult, callErr := chunker.Plan(attemptCtx, contracts.ChunkRequest{
Source: doc, SourceInput: sourceInput.Clone(), SessionID: sessionID, Source: doc, SourceInput: sourceInput.Clone(), SessionID: sessionID,
References: CloneReferenceSet(input.pipeline.ChunkReferences.ReferenceSet), References: CloneReferenceSet(input.pipeline.ChunkReferences.ReferenceSet),
LLMProfile: input.pipeline.Chunk.LLMProfile, Metadata: input.Metadata, LLMProfile: input.pipeline.Chunk.LLMProfile, Metadata: requestMetadata,
}) })
if callErr != nil { if callErr != nil {
return false, nil, terminal.record(nil, fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), callErr)) return false, nil, terminal.record(nil, fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), callErr))
@@ -105,7 +111,10 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
if digestErr != nil { if digestErr != nil {
return false, nil, terminal.record(nil, fmt.Errorf("digest generated chunk plan: %w", digestErr)) return false, nil, terminal.record(nil, fmt.Errorf("digest generated chunk plan: %w", digestErr))
} }
producerMetadata, _ := moduleManifestMetadata(chunker) producerMetadata, _, metadataErr := moduleManifestMetadata(chunker)
if metadataErr != nil {
return false, nil, terminal.record(nil, fmt.Errorf("clone chunker manifest metadata: %w", metadataErr))
}
profile := "" profile := ""
if provider, ok := chunker.(contracts.ChunkExecutionClassProvider); ok && provider.ExecutionClass() == contracts.ExecutionClassLLMBacked { if provider, ok := chunker.(contracts.ChunkExecutionClassProvider); ok && provider.ExecutionClass() == contracts.ExecutionClassLLMBacked {
profile = input.pipeline.Chunk.LLMProfile profile = input.pipeline.Chunk.LLMProfile
@@ -116,7 +125,7 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
Producer: ChunkPlanProducer{ Producer: ChunkPlanProducer{
InputModule: input.Prepared.input.Key(), ChunkModule: chunker.Key(), LLMProfile: profile, InputModule: input.Prepared.input.Key(), ChunkModule: chunker.Key(), LLMProfile: profile,
References: append([]artifacts.ReferenceProvenance(nil), referenceTargetProvenance(input.pipeline.ChunkReferences)...), References: append([]artifacts.ReferenceProvenance(nil), referenceTargetProvenance(input.pipeline.ChunkReferences)...),
Metadata: cloneMetadata(producerMetadata), Metadata: producerMetadata,
}, },
Warnings: cloneWarnings(chunkResult.Warnings), CreatedAt: time.Now().UTC(), Warnings: cloneWarnings(chunkResult.Warnings), CreatedAt: time.Now().UTC(),
} }
@@ -127,7 +136,9 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
if mode == ChunkCacheBypass { if mode == ChunkCacheBypass {
action = "bypassed" action = "bypassed"
} }
result.setCandidate(candidate, action) if candidateErr := result.setCandidate(candidate, action); candidateErr != nil {
return false, nil, terminal.record(nil, fmt.Errorf("clone generated chunk plan record: %w", candidateErr))
}
validationWarnings, rejected, validationErr := r.validateChunks(attemptCtx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.Metadata, input.Prepared.chunkValidators, attempt, input.Debug) validationWarnings, rejected, validationErr := r.validateChunks(attemptCtx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.Metadata, input.Prepared.chunkValidators, attempt, input.Debug)
attemptWarnings := append(cloneWarnings(chunkResult.Warnings), validationWarnings...) attemptWarnings := append(cloneWarnings(chunkResult.Warnings), validationWarnings...)
payload := map[string]any{ payload := map[string]any{
@@ -161,7 +172,10 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
} }
if mode == ChunkCacheAuto || mode == ChunkCacheRefresh { if mode == ChunkCacheAuto || mode == ChunkCacheRefresh {
record := cloneChunkPlanRecord(*result.record) record, cloneErr := cloneChunkPlanRecord(*result.record)
if cloneErr != nil {
return result, fmt.Errorf("clone chunk plan record for publication: %w", cloneErr)
}
record.Warnings = cloneWarnings(producerWarnings) record.Warnings = cloneWarnings(producerWarnings)
if err := input.ChunkPlans.Save(record); err != nil { if err := input.ChunkPlans.Save(record); err != nil {
return result, fmt.Errorf("save chunk plan: %w", err) return result, fmt.Errorf("save chunk plan: %w", err)
@@ -197,13 +211,17 @@ func chunkPlanLookupReason(status ChunkPlanStatus) string {
} }
} }
func (result *chunkPlanExecution) setCandidate(record ChunkPlanRecord, action string) { func (result *chunkPlanExecution) setCandidate(record ChunkPlanRecord, action string) error {
cloned := cloneChunkPlanRecord(record) cloned, err := cloneChunkPlanRecord(record)
if err != nil {
return err
}
result.record = &cloned result.record = &cloned
result.action = action result.action = action
result.summary.Action = action result.summary.Action = action
result.summary.SourceDigest = record.SourceDigest result.summary.SourceDigest = record.SourceDigest
result.summary.CandidateDigest = record.PlanDigest result.summary.CandidateDigest = record.PlanDigest
return nil
} }
func (result *chunkPlanExecution) setValidation(warnings []contracts.Warning, rejection *contracts.RejectedOutput, err error) { func (result *chunkPlanExecution) setValidation(warnings []contracts.Warning, rejection *contracts.RejectedOutput, err error) {
@@ -219,27 +237,31 @@ func (result *chunkPlanExecution) setValidation(warnings []contracts.Warning, re
} }
} }
func cloneChunkPlanRecord(record ChunkPlanRecord) ChunkPlanRecord { func cloneChunkPlanRecord(record ChunkPlanRecord) (ChunkPlanRecord, error) {
record.Plan = source.CloneChunkPlan(record.Plan) record.Plan = source.CloneChunkPlan(record.Plan)
record.Producer.References = append([]artifacts.ReferenceProvenance(nil), record.Producer.References...) record.Producer.References = append([]artifacts.ReferenceProvenance(nil), record.Producer.References...)
record.Producer.Metadata = cloneMetadata(record.Producer.Metadata) metadata, err := cloneMetadata(record.Producer.Metadata)
if err != nil {
return ChunkPlanRecord{}, fmt.Errorf("clone chunk plan producer metadata: %w", err)
}
record.Producer.Metadata = metadata
record.Warnings = cloneWarnings(record.Warnings) record.Warnings = cloneWarnings(record.Warnings)
return record return record, nil
} }
func applyChunkPlanExecution(output *RunOutput, result chunkPlanExecution) { func applyChunkPlanExecution(output *RunOutput, result chunkPlanExecution) error {
if output == nil { if output == nil {
return return nil
} }
summary := result.summary summary := result.summary
output.ChunkPlan = &summary output.ChunkPlan = &summary
if output.Manifest.ChunkPlan == nil { if output.Manifest.ChunkPlan == nil {
return return nil
} }
manifest := output.Manifest.ChunkPlan manifest := output.Manifest.ChunkPlan
manifest.Action = result.action manifest.Action = result.action
if result.record == nil { if result.record == nil {
return return nil
} }
record := result.record record := result.record
manifest.SourceDigest = record.SourceDigest manifest.SourceDigest = record.SourceDigest
@@ -249,7 +271,12 @@ func applyChunkPlanExecution(output *RunOutput, result chunkPlanExecution) {
manifest.ProducerModule = record.Producer.ChunkModule manifest.ProducerModule = record.Producer.ChunkModule
manifest.ProducerLLMProfile = record.Producer.LLMProfile manifest.ProducerLLMProfile = record.Producer.LLMProfile
manifest.ProducerReferences = append([]artifacts.ReferenceProvenance(nil), record.Producer.References...) manifest.ProducerReferences = append([]artifacts.ReferenceProvenance(nil), record.Producer.References...)
manifest.ProducerMetadata = cloneMetadata(record.Producer.Metadata) metadata, err := cloneMetadata(record.Producer.Metadata)
if err != nil {
return fmt.Errorf("clone chunk plan manifest producer metadata: %w", err)
}
manifest.ProducerMetadata = metadata
createdAt := record.CreatedAt createdAt := record.CreatedAt
manifest.CreatedAt = &createdAt manifest.CreatedAt = &createdAt
return nil
} }

14
internal/framework/pipeline/runner_chunk_plan_test.go
View File

@@ -438,6 +438,20 @@ func TestRunnerStoresProducerProvenanceAndProducerWarnings(t *testing.T) {
} }
} }
func TestRunnerRejectsUncloneableModuleManifestMetadata(t *testing.T) {
prepared, plan := preparedTerminalDebugPipeline(t)
cyclic := make(map[string]any)
cyclic["self"] = cyclic
prepared.chunker = manifestChunker{
terminalChunker: terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan},
metadata: map[string]any{"cyclic": cyclic},
}
_, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")})
if err == nil || !strings.Contains(err.Error(), `clone manifest metadata for module "chunker": metadata.cyclic.self contains a cycle`) {
t.Fatalf("Run() error = %v, want contextual module metadata clone failure", err)
}
}
func TestRunnerOmitsProducerProfileForDeterministicChunker(t *testing.T) { func TestRunnerOmitsProducerProfileForDeterministicChunker(t *testing.T) {
prepared, plan := preparedTerminalDebugPipeline(t) prepared, plan := preparedTerminalDebugPipeline(t)
prepared.resolved.Chunk.LLMProfile = "configured-but-unused" prepared.resolved.Chunk.LLMProfile = "configured-but-unused"

39
internal/framework/pipeline/runner_concurrent.go
View File

@@ -79,7 +79,9 @@ func (r *Runner) runLanes(parent context.Context, input RunInput, checkpoints Ch
if prepared.typed == nil { if prepared.typed == nil {
return output, fmt.Errorf("typed lane %q executor is not prepared", prepared.resolved.ID) return output, fmt.Errorf("typed lane %q executor is not prepared", prepared.resolved.ID)
} }
setTypedLaneManifestMetadata(&output, prepared.resolved.ID, prepared.typed.extractor, prepared.typed.merger, prepared.typed.normalizer) if err := setTypedLaneManifestMetadata(&output, prepared.resolved.ID, prepared.typed.extractor, prepared.typed.merger, prepared.typed.normalizer); err != nil {
return output, err
}
state, err := prepareLaneExtract(input, loader, doc, chunks, i, prepared) state, err := prepareLaneExtract(input, loader, doc, chunks, i, prepared)
if err != nil { if err != nil {
return output, err return output, err
@@ -212,7 +214,9 @@ func (r *Runner) runLanes(parent context.Context, input RunInput, checkpoints Ch
close(continuations) close(continuations)
continuationWorkers.Wait() continuationWorkers.Wait()
for i := range completedOutputs { for i := range completedOutputs {
mergeLaneOutput(&output, completedOutputs[i]) if err := mergeLaneOutput(&output, completedOutputs[i]); err != nil {
return output, err
}
} }
if err := selectRunError(parent, runErrors); err != nil { if err := selectRunError(parent, runErrors); err != nil {
return output, err return output, err
@@ -244,7 +248,10 @@ func prepareLaneExtract(input RunInput, loader CheckpointLoader, doc *source.Sou
if decodeErr != nil { if decodeErr != nil {
return nil, fmt.Errorf("decode extract checkpoint for lane %q: %w", lane.ID, decodeErr) return nil, fmt.Errorf("decode extract checkpoint for lane %q: %w", lane.ID, decodeErr)
} }
stored = hydrateCheckpointArtifact(typed.codec, stored, value) stored, decodeErr = hydrateCheckpointArtifact(typed.codec, stored, value)
if decodeErr != nil {
return nil, fmt.Errorf("hydrate extract checkpoint for lane %q: %w", lane.ID, decodeErr)
}
artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: stored.ChunkID, ChunkIndex: stored.ChunkIndex, ChunkRef: stored.ChunkRef, Value: value} artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: stored.ChunkID, ChunkIndex: stored.ChunkIndex, ChunkRef: stored.ChunkRef, Value: value}
if stored.ChunkIndex >= 0 && stored.ChunkIndex < len(chunks) && artifact.ChunkRef == (source.SourceRef{}) { if stored.ChunkIndex >= 0 && stored.ChunkIndex < len(chunks) && artifact.ChunkRef == (source.SourceRef{}) {
artifact.ChunkRef = chunks[stored.ChunkIndex].Ref artifact.ChunkRef = chunks[stored.ChunkIndex].Ref
@@ -258,9 +265,14 @@ func prepareLaneExtract(input RunInput, loader CheckpointLoader, doc *source.Sou
} }
func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, job extractJob) extractJobResult { func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, job extractJob) extractJobResult {
state, chunk := job.lane, cloneSourceChunk(job.chunk) state := job.lane
chunk, cloneErr := cloneSourceChunk(job.chunk)
lane, typed := state.prepared.resolved, state.prepared.typed lane, typed := state.prepared.resolved, state.prepared.typed
result := extractJobResult{laneIndex: state.index, chunkIndex: chunk.Index} result := extractJobResult{laneIndex: state.index, chunkIndex: job.chunk.Index}
if cloneErr != nil {
result.err = fmt.Errorf("clone chunk %q for extraction: %w", job.chunk.ID, cloneErr)
return result
}
var accepted erasedExtractArtifact var accepted erasedExtractArtifact
var serialized CheckpointArtifact var serialized CheckpointArtifact
var acceptedWarnings []contracts.Warning var acceptedWarnings []contracts.Warning
@@ -269,7 +281,11 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source.
attemptPath := path.Join("extract", debugPathComponent(lane.ID), fmt.Sprintf("chunk-%06d", chunk.Index+1), fmt.Sprintf("attempt-%02d", attempt)) attemptPath := path.Join("extract", debugPathComponent(lane.ID), fmt.Sprintf("chunk-%06d", chunk.Index+1), fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "extract", llmScope, debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started}) terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "extract", llmScope, debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started})
extracted, callErr := typed.extract(attemptCtx, typed.extractor, contracts.TypedExtractionRequest{Source: doc, Chunk: &chunk, SourceInput: chunkInputMaterial(sourceInput, chunk), SessionID: sessionID, References: CloneReferenceSet(lane.ExtractReferences.ReferenceSet), LLMProfile: lane.Extract.LLMProfile, Metadata: cloneMetadata(input.Metadata)}) requestMetadata, metadataErr := cloneMetadata(input.Metadata)
if metadataErr != nil {
return false, nil, terminal.record(nil, fmt.Errorf("clone extract request metadata: %w", metadataErr))
}
extracted, callErr := typed.extract(attemptCtx, typed.extractor, contracts.TypedExtractionRequest{Source: doc, Chunk: &chunk, SourceInput: chunkInputMaterial(sourceInput, chunk), SessionID: sessionID, References: CloneReferenceSet(lane.ExtractReferences.ReferenceSet), LLMProfile: lane.Extract.LLMProfile, Metadata: requestMetadata})
if callErr != nil { if callErr != nil {
attemptErr := fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, lane.Extract.Module, callErr) attemptErr := fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, lane.Extract.Module, callErr)
return false, nil, terminal.record(nil, attemptErr) return false, nil, terminal.record(nil, attemptErr)
@@ -414,9 +430,9 @@ func selectRunError(parent context.Context, values []orderedRunError) error {
return filtered[0].err return filtered[0].err
} }
func mergeLaneOutput(dst *RunOutput, src RunOutput) { func mergeLaneOutput(dst *RunOutput, src RunOutput) error {
if dst == nil { if dst == nil {
return return nil
} }
dst.NormalizeOutputs = append(dst.NormalizeOutputs, cloneSerializedOutputs(src.NormalizeOutputs)...) dst.NormalizeOutputs = append(dst.NormalizeOutputs, cloneSerializedOutputs(src.NormalizeOutputs)...)
dst.Rejected = append(dst.Rejected, cloneRejectedOutputs(src.Rejected)...) dst.Rejected = append(dst.Rejected, cloneRejectedOutputs(src.Rejected)...)
@@ -425,8 +441,13 @@ func mergeLaneOutput(dst *RunOutput, src RunOutput) {
for i := range dst.Manifest.ArtifactLanes { for i := range dst.Manifest.ArtifactLanes {
for j := range src.Manifest.ArtifactLanes { for j := range src.Manifest.ArtifactLanes {
if dst.Manifest.ArtifactLanes[i].ID == src.Manifest.ArtifactLanes[j].ID && src.Manifest.ArtifactLanes[j].Metadata != nil { if dst.Manifest.ArtifactLanes[i].ID == src.Manifest.ArtifactLanes[j].ID && src.Manifest.ArtifactLanes[j].Metadata != nil {
dst.Manifest.ArtifactLanes[i].Metadata = cloneMetadata(src.Manifest.ArtifactLanes[j].Metadata) metadata, err := cloneMetadata(src.Manifest.ArtifactLanes[j].Metadata)
if err != nil {
return fmt.Errorf("clone lane %q manifest metadata: %w", dst.Manifest.ArtifactLanes[i].ID, err)
}
dst.Manifest.ArtifactLanes[i].Metadata = metadata
} }
} }
} }
return nil
} }

14
internal/framework/pipeline/runner_extract_handoff_test.go
View File

@@ -135,6 +135,20 @@ func TestRunnerPassesIndependentConcreteMetadataToValidatorsAndExtractors(t *tes
} }
} }
func TestRunnerRejectsMetadataThatCannotBeCloned(t *testing.T) {
prepared := preparedAttemptDebugPipeline(t)
cyclic := make(map[string]any)
cyclic["self"] = cyclic
_, err := New().Run(context.Background(), RunInput{
Prepared: prepared,
RawInput: []byte("input"),
Metadata: map[string]any{"cyclic": cyclic},
})
if err == nil || !strings.Contains(err.Error(), "clone run metadata: metadata.cyclic.self contains a cycle") {
t.Fatalf("Run() error = %v, want contextual metadata clone failure", err)
}
}
func TestRunnerContinuesFromFreshAndReusedExtractResults(t *testing.T) { func TestRunnerContinuesFromFreshAndReusedExtractResults(t *testing.T) {
prepared := preparedAttemptDebugPipeline(t) prepared := preparedAttemptDebugPipeline(t)
extractCalls := 0 extractCalls := 0

78
internal/framework/pipeline/runner_typed.go
View File

@@ -30,14 +30,18 @@ func cloneCheckpointArtifact(output CheckpointArtifact) CheckpointArtifact {
return output return output
} }
func hydrateCheckpointArtifact(codec artifactCodecEntry, output CheckpointArtifact, value any) CheckpointArtifact { func hydrateCheckpointArtifact(codec artifactCodecEntry, output CheckpointArtifact, value any) (CheckpointArtifact, error) {
output.Artifact.Schema = contracts.CloneArtifactSchema(codec.spec.Schema) output.Artifact.Schema = contracts.CloneArtifactSchema(codec.spec.Schema)
if codec.metadata != nil { if codec.metadata != nil {
output.Artifact.Metadata = cloneMetadata(codec.metadata(value)) metadata, err := codec.metadata(value)
if err != nil {
return CheckpointArtifact{}, fmt.Errorf("clone artifact metadata: %w", err)
}
output.Artifact.Metadata = metadata
} else { } else {
output.Artifact.Metadata = nil output.Artifact.Metadata = nil
} }
return output return output, nil
} }
func artifactCheckpointDigests(outputs []CheckpointArtifact) []CheckpointFingerprint { func artifactCheckpointDigests(outputs []CheckpointArtifact) []CheckpointFingerprint {
values := make([]CheckpointFingerprint, 0, len(outputs)) values := make([]CheckpointFingerprint, 0, len(outputs))
@@ -82,9 +86,12 @@ func serializeArtifact(codec artifactCodecEntry, value any, candidate bool) (con
schema := codec.spec.Schema schema := codec.spec.Schema
metadata := map[string]any(nil) metadata := map[string]any(nil)
if codec.metadata != nil { if codec.metadata != nil {
metadata = codec.metadata(value) metadata, err = codec.metadata(value)
if err != nil {
return contracts.SerializedArtifact{}, fmt.Errorf("clone artifact metadata: %w", err)
}
} }
return contracts.SerializedArtifact{Kind: codec.spec.Kind, Schema: contracts.CloneArtifactSchema(schema), MediaType: codec.spec.MediaType, Content: append([]byte(nil), content...), Metadata: cloneMetadata(metadata)}, nil return contracts.SerializedArtifact{Kind: codec.spec.Kind, Schema: contracts.CloneArtifactSchema(schema), MediaType: codec.spec.MediaType, Content: append([]byte(nil), content...), Metadata: metadata}, nil
} }
func decodeCheckpointArtifact(codec artifactCodecEntry, artifact CheckpointArtifact) (any, error) { func decodeCheckpointArtifact(codec artifactCodecEntry, artifact CheckpointArtifact) (any, error) {
@@ -139,7 +146,9 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
if typed == nil { if typed == nil {
return fmt.Errorf("typed lane %q executor is not prepared", lane.ID) return fmt.Errorf("typed lane %q executor is not prepared", lane.ID)
} }
setTypedLaneManifestMetadata(output, lane.ID, typed.extractor, typed.merger, typed.normalizer) if err := setTypedLaneManifestMetadata(output, lane.ID, typed.extractor, typed.merger, typed.normalizer); err != nil {
return err
}
mergeInputs := make([]contracts.ExtractArtifact[any], len(extracts.accepted)) mergeInputs := make([]contracts.ExtractArtifact[any], len(extracts.accepted))
for i, value := range extracts.accepted { for i, value := range extracts.accepted {
@@ -165,7 +174,10 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
return fmt.Errorf("decode merge checkpoint for lane %q: %w", lane.ID, decodeErr) return fmt.Errorf("decode merge checkpoint for lane %q: %w", lane.ID, decodeErr)
} }
merged = erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: value} merged = erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: value}
serializedMerge = hydrateCheckpointArtifact(typed.codec, cloneCheckpointArtifact(mergeCP.Output), value) serializedMerge, decodeErr = hydrateCheckpointArtifact(typed.codec, cloneCheckpointArtifact(mergeCP.Output), value)
if decodeErr != nil {
return fmt.Errorf("hydrate merge checkpoint for lane %q: %w", lane.ID, decodeErr)
}
mergeWarnings = cloneWarnings(mergeCP.Warnings) mergeWarnings = cloneWarnings(mergeCP.Warnings)
output.Warnings = append(output.Warnings, mergeWarnings...) output.Warnings = append(output.Warnings, mergeWarnings...)
} else { } else {
@@ -177,7 +189,11 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
attemptPath := path.Join("merge", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt)) attemptPath := path.Join("merge", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "merge", llmScope, debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: attempt, StartedAt: started}) terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "merge", llmScope, debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: attempt, StartedAt: started})
result, callErr := typed.merge(attemptCtx, typed.merger, contracts.TypedMergeRequest[any]{Source: doc, LaneID: lane.ID, ExtractOutputs: mergeInputs, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(lane.MergeReferences.ReferenceSet), LLMProfile: lane.Merge.LLMProfile, Metadata: cloneMetadata(input.Metadata)}) requestMetadata, metadataErr := cloneMetadata(input.Metadata)
if metadataErr != nil {
return false, nil, terminal.record(nil, fmt.Errorf("clone merge request metadata: %w", metadataErr))
}
result, callErr := typed.merge(attemptCtx, typed.merger, contracts.TypedMergeRequest[any]{Source: doc, LaneID: lane.ID, ExtractOutputs: mergeInputs, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(lane.MergeReferences.ReferenceSet), LLMProfile: lane.Merge.LLMProfile, Metadata: requestMetadata})
if callErr != nil { if callErr != nil {
attemptErr := fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, lane.Merge.Module, callErr) attemptErr := fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, lane.Merge.Module, callErr)
return false, nil, terminal.record(nil, attemptErr) return false, nil, terminal.record(nil, attemptErr)
@@ -246,7 +262,11 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
if decodeErr != nil { if decodeErr != nil {
return fmt.Errorf("decode normalize checkpoint for lane %q: %w", lane.ID, decodeErr) return fmt.Errorf("decode normalize checkpoint for lane %q: %w", lane.ID, decodeErr)
} }
serializedNormalize, normalizeWarnings = hydrateCheckpointArtifact(typed.codec, cloneCheckpointArtifact(normalizeCP.Output), value), cloneWarnings(normalizeCP.Warnings) serializedNormalize, decodeErr = hydrateCheckpointArtifact(typed.codec, cloneCheckpointArtifact(normalizeCP.Output), value)
if decodeErr != nil {
return fmt.Errorf("hydrate normalize checkpoint for lane %q: %w", lane.ID, decodeErr)
}
normalizeWarnings = cloneWarnings(normalizeCP.Warnings)
output.Warnings = append(output.Warnings, normalizeWarnings...) output.Warnings = append(output.Warnings, normalizeWarnings...)
} else { } else {
if err := checkpoints.NormalizeRunning(lane.ID, lane.Normalize.Module, normalizeDeps); err != nil { if err := checkpoints.NormalizeRunning(lane.ID, lane.Normalize.Module, normalizeDeps); err != nil {
@@ -257,7 +277,11 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
attemptPath := path.Join("normalize", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt)) attemptPath := path.Join("normalize", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "normalize", llmScope, debugTimedEnvelope{Stage: string(StageNormalize), LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: attempt, StartedAt: started}) terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "normalize", llmScope, debugTimedEnvelope{Stage: string(StageNormalize), LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: attempt, StartedAt: started})
result, callErr := typed.normalize(attemptCtx, typed.normalizer, contracts.TypedNormalizeRequest[any]{Source: doc, LaneID: lane.ID, MergeOutput: contracts.MergeArtifact[any]{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: merged.Value}, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(lane.NormalizeReferences.ReferenceSet), LLMProfile: lane.Normalize.LLMProfile, Metadata: cloneMetadata(input.Metadata)}) requestMetadata, metadataErr := cloneMetadata(input.Metadata)
if metadataErr != nil {
return false, nil, terminal.record(nil, fmt.Errorf("clone normalize request metadata: %w", metadataErr))
}
result, callErr := typed.normalize(attemptCtx, typed.normalizer, contracts.TypedNormalizeRequest[any]{Source: doc, LaneID: lane.ID, MergeOutput: contracts.MergeArtifact[any]{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: merged.Value}, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(lane.NormalizeReferences.ReferenceSet), LLMProfile: lane.Normalize.LLMProfile, Metadata: requestMetadata})
if callErr != nil { if callErr != nil {
attemptErr := fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, lane.Normalize.Module, callErr) attemptErr := fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, lane.Normalize.Module, callErr)
return false, nil, terminal.record(nil, attemptErr) return false, nil, terminal.record(nil, attemptErr)
@@ -309,9 +333,9 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
return nil return nil
} }
func setTypedLaneManifestMetadata(output *RunOutput, laneID string, extractor, merger, normalizer any) { func setTypedLaneManifestMetadata(output *RunOutput, laneID string, extractor, merger, normalizer any) error {
if output == nil { if output == nil {
return return nil
} }
for i := range output.Manifest.ArtifactLanes { for i := range output.Manifest.ArtifactLanes {
if output.Manifest.ArtifactLanes[i].ID != laneID { if output.Manifest.ArtifactLanes[i].ID != laneID {
@@ -322,15 +346,20 @@ func setTypedLaneManifestMetadata(output *RunOutput, laneID string, extractor, m
name string name string
module any module any
}{{"extractor", extractor}, {"merger", merger}, {"normalizer", normalizer}} { }{{"extractor", extractor}, {"merger", merger}, {"normalizer", normalizer}} {
if value, ok := moduleManifestMetadata(item.module); ok { value, ok, err := moduleManifestMetadata(item.module)
if err != nil {
return fmt.Errorf("clone %s manifest metadata for lane %q: %w", item.name, laneID, err)
}
if ok {
metadata[item.name] = value metadata[item.name] = value
} }
} }
if len(metadata) > 0 { if len(metadata) > 0 {
output.Manifest.ArtifactLanes[i].Metadata = metadata output.Manifest.ArtifactLanes[i].Metadata = metadata
} }
return return nil
} }
return nil
} }
func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecEntry, target typedValidationTarget, chain preparedValidatorChain, attempt int, debug DebugRecorder) ([]contracts.Warning, *contracts.RejectedOutput, error) { func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecEntry, target typedValidationTarget, chain preparedValidatorChain, attempt int, debug DebugRecorder) ([]contracts.Warning, *contracts.RejectedOutput, error) {
@@ -349,17 +378,32 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE
started := time.Now().UTC() started := time.Now().UTC()
attemptPath := path.Join("validate", debugPathComponent(string(target.stage)), debugPathComponent(target.laneID), debugPathComponent(target.moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d", index+1, debugPathComponent(binding.Module), attempt)) attemptPath := path.Join("validate", debugPathComponent(string(target.stage)), debugPathComponent(target.laneID), debugPathComponent(target.moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d", index+1, debugPathComponent(binding.Module), attempt))
validatorCtx, llmScope := withIsolatedDebugLLMScope(ctx, attemptPath) validatorCtx, llmScope := withIsolatedDebugLLMScope(ctx, attemptPath)
requestTarget := target
requestTarget.sourceInput = target.sourceInput.Clone()
requestTarget.references = CloneReferenceSet(target.references)
requestTarget.metadata, err = cloneMetadata(target.metadata)
if err != nil {
return nil, nil, fmt.Errorf("clone typed validation metadata: %w", err)
}
requestTarget.chunk, err = cloneSourceChunkPtr(target.chunk)
if err != nil {
return nil, nil, fmt.Errorf("clone typed validation chunk: %w", err)
}
requestTarget.chunks, err = cloneSourceChunks(target.chunks)
if err != nil {
return nil, nil, fmt.Errorf("clone typed validation chunks: %w", err)
}
switch item.resolved.Target { switch item.resolved.Target {
case ValidatorTargetTyped: case ValidatorTargetTyped:
target.llmProfile = binding.LLMProfile requestTarget.llmProfile = binding.LLMProfile
result, err = item.typedValidate(validatorCtx, item.typed, target) result, err = item.typedValidate(validatorCtx, item.typed, requestTarget)
case ValidatorTargetSerialized: case ValidatorTargetSerialized:
artifact, encodeErr := validationCandidateArtifact(codec, target) artifact, encodeErr := validationCandidateArtifact(codec, target)
if encodeErr != nil { if encodeErr != nil {
err = encodeErr err = encodeErr
break break
} }
result, err = item.serialized.Validate(validatorCtx, contracts.SerializedValidationRequest{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Source: target.source, SourceID: target.sourceID, SourceInput: target.sourceInput.Clone(), SessionID: target.sessionID, References: CloneReferenceSet(target.references), LLMProfile: binding.LLMProfile, Metadata: cloneMetadata(target.metadata), Chunk: cloneSourceChunkPtr(target.chunk), Chunks: cloneSourceChunks(target.chunks), Schema: contracts.CloneArtifactSchema(artifact.Artifact.Schema), MediaType: artifact.Artifact.MediaType, Content: append([]byte(nil), artifact.Artifact.Content...)}) result, err = item.serialized.Validate(validatorCtx, contracts.SerializedValidationRequest{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Source: target.source, SourceID: target.sourceID, SourceInput: requestTarget.sourceInput, SessionID: target.sessionID, References: requestTarget.references, LLMProfile: binding.LLMProfile, Metadata: requestTarget.metadata, Chunk: requestTarget.chunk, Chunks: requestTarget.chunks, Schema: contracts.CloneArtifactSchema(artifact.Artifact.Schema), MediaType: artifact.Artifact.MediaType, Content: append([]byte(nil), artifact.Artifact.Content...)})
default: default:
return nil, nil, fmt.Errorf("validator %q is incompatible with typed artifact validation", binding.Module) return nil, nil, fmt.Errorf("validator %q is incompatible with typed artifact validation", binding.Module)
} }