diff --git a/docs/roadmap/adr0005.md b/docs/roadmap/adr0005.md deleted file mode 100644 index 9933bf6..0000000 --- a/docs/roadmap/adr0005.md +++ /dev/null @@ -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 `. 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. diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md deleted file mode 100644 index 837fb1e..0000000 --- a/docs/roadmap/implementation.md +++ /dev/null @@ -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 - `/<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 `. -- 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 - `//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. diff --git a/internal/framework/chunkplan/store.go b/internal/framework/chunkplan/store.go index 60f9c2a..57861a2 100644 --- a/internal/framework/chunkplan/store.go +++ b/internal/framework/chunkplan/store.go @@ -11,6 +11,7 @@ import ( "os" "path/filepath" "strings" + "syscall" "gitea.maximumdirect.net/eric/notarius/internal/core/source" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" @@ -50,7 +51,7 @@ func (s *filesystemStore) Load(sourceDigest string) (pipeline.ChunkPlanRecord, p } defer root.Close() - state, err := inspectDirectory(root, digestDir) + digestRoot, state, err := openDigestRoot(root, digestDir, false) if err != nil { 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 { return invalidDecision() } + defer digestRoot.Close() - target := planPath(digestDir) - state, err = inspectPlan(root, target) + data, state, err := readPlan(digestRoot) 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 { 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() } - 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 decoder := json.NewDecoder(bytes.NewReader(data)) decoder.DisallowUnknownFields() @@ -116,11 +109,15 @@ func (s *filesystemStore) Save(record pipeline.ChunkPlanRecord) error { return fmt.Errorf("open chunk plan root: %w", err) } 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) } - target := planPath(digestDir) - state, err := inspectPlan(root, target) + if state == entryRejected { + return fmt.Errorf("chunk plan directory has an unsupported type") + } + defer digestRoot.Close() + state, err = inspectPlan(digestRoot, planFileName) if err != nil { return fmt.Errorf("inspect chunk plan file: %w", err) } @@ -131,7 +128,7 @@ func (s *filesystemStore) Save(record pipeline.ChunkPlanRecord) error { if writer == nil { 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 nil @@ -174,52 +171,89 @@ const ( 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) if err != nil { 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() { - 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 { - state, err := inspectDirectory(root, digestDir) + state, before, err := inspectDirectory(root, digestDir) if err != nil { - return err + return nil, entryPresent, err } switch state { case entryRejected: - return fmt.Errorf("chunk plan directory has an unsupported type") + return nil, entryRejected, nil case entryMissing: + if !create { + return nil, entryMissing, nil + } if err := root.Mkdir(digestDir, 0o700); err != nil && !errors.Is(err, os.ErrExist) { - return err + return nil, entryPresent, err } 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 { - return err + return nil, entryPresent, err } - info, statErr := directory.Stat() - if statErr == nil && !info.IsDir() { - statErr = fmt.Errorf("chunk plan directory has an unsupported type") + opened, statErr := digestRoot.Stat(".") + afterState, after, afterErr := inspectDirectory(root, digestDir) + if statErr != nil || afterErr != nil { + _ = digestRoot.Close() + if statErr != nil { + return nil, entryPresent, statErr + } + return nil, entryPresent, afterErr } - if statErr == nil { - statErr = directory.Chmod(0o700) + if afterState != entryPresent || !os.SameFile(before, after) || !os.SameFile(opened, after) { + _ = digestRoot.Close() + if afterState == entryRejected { + return nil, entryRejected, nil + } + continue } - closeErr := directory.Close() - if statErr != nil { - return statErr + if create { + directory, openErr := digestRoot.Open(".") + 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 } -func planPath(digestDir string) string { - return digestDir + "/" + planFileName +type planReadHooks struct { + 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) { diff --git a/internal/framework/chunkplan/store_test.go b/internal/framework/chunkplan/store_test.go index 1d02c77..c80862e 100644 --- a/internal/framework/chunkplan/store_test.go +++ b/internal/framework/chunkplan/store_test.go @@ -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) { tests := []struct { name string diff --git a/internal/framework/pipeline/artifact_codec_registry.go b/internal/framework/pipeline/artifact_codec_registry.go index 7d152f9..ec67381 100644 --- a/internal/framework/pipeline/artifact_codec_registry.go +++ b/internal/framework/pipeline/artifact_codec_registry.go @@ -67,7 +67,7 @@ type artifactCodecEntry struct { valueType reflect.Type encode 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) } @@ -132,10 +132,10 @@ func RegisterArtifactCodec[T any](registry *ArtifactCodecRegistry, codec contrac return append([]byte(nil), content...), nil } 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) if err != nil { - return nil + return nil, err } return cloneMetadata(provider.Metadata(typed)) } diff --git a/internal/framework/pipeline/chunk_annotations_test.go b/internal/framework/pipeline/chunk_annotations_test.go index c57f8e0..414f528 100644 --- a/internal/framework/pipeline/chunk_annotations_test.go +++ b/internal/framework/pipeline/chunk_annotations_test.go @@ -29,7 +29,10 @@ func TestChunkCanonicalizationAndClonePreserveAnnotationScopes(t *testing.T) { 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.PlanAnnotations["shared"][0] = '[' if string(chunks[0].Annotations["shared"]) != `{"range":true}` || string(chunks[0].PlanAnnotations["shared"]) != `{"plan":true}` { diff --git a/internal/framework/pipeline/chunk_validation.go b/internal/framework/pipeline/chunk_validation.go index e88f4cd..3b9b10f 100644 --- a/internal/framework/pipeline/chunk_validation.go +++ b/internal/framework/pipeline/chunk_validation.go @@ -21,12 +21,16 @@ func validateAndMaterializeChunkPlan(doc *source.SourceDocument, plan source.Chu 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{ ID: unit.ID, Kind: unit.Kind, Text: unit.Text, Ref: unit.Ref, - Metadata: cloneMetadata(unit.Metadata), - } + Metadata: metadata, + }, nil } diff --git a/internal/framework/pipeline/debug.go b/internal/framework/pipeline/debug.go index 355b8b5..5fac324 100644 --- a/internal/framework/pipeline/debug.go +++ b/internal/framework/pipeline/debug.go @@ -471,7 +471,7 @@ func debugSourceDocumentEnvelope(doc *source.SourceDocument) *debugSourceDocumen Kind: doc.Kind, Format: doc.Format, Digest: doc.Digest, - Units: cloneSourceUnits(doc.Units), + Units: cloneSourceUnitsForDebug(doc.Units), Metadata: redactSensitiveMap(doc.Metadata), } } @@ -483,13 +483,21 @@ func debugSourceChunkEnvelope(chunk source.Chunk) debugSourceChunk { Index: chunk.Index, Ref: chunk.Ref, Content: debugContentEnvelope(chunk.Content, chunk.MediaType, chunk.Metadata, nil), - Units: cloneSourceUnits(chunk.Units), + Units: cloneSourceUnitsForDebug(chunk.Units), Metadata: redactSensitiveMap(chunk.Metadata), Annotations: source.CloneChunkAnnotations(chunk.Annotations), 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 { if len(chunks) == 0 { return nil diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 6f541f9..b1d4c3a 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -78,6 +78,10 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err if err := validateRunInput(input); err != nil { 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.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 - attachModuleManifestMetadata(&output, "input", adapter) + if err := attachModuleManifestMetadata(&output, "input", adapter); err != nil { + return failOutput(output), err + } sourceCheckpoint, sourceDecision := checkpointLoader.Source(adapter.Key()) recordCheckpointEvent(&output, checkpointLoader, "source", "", adapter.Key(), sourceDecision) 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 { 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{ SourceID: input.SourceID, Path: input.Path, Raw: input.RawInput, LLMProfile: input.pipeline.Input.LLMProfile, - Metadata: input.Metadata, + Metadata: requestMetadata, }) if err != nil { _ = 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) 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} chunker := input.Prepared.chunker - attachModuleManifestMetadata(&output, "chunker", chunker) + if err := attachModuleManifestMetadata(&output, "chunker", chunker); err != nil { + return failOutput(output), err + } chunkStarted := time.Now().UTC() chunkMode := effectiveChunkCacheMode(input.ChunkCacheMode) 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) } 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 { return failOutput(output), err } @@ -233,7 +250,9 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err if chunkResult.accepted { 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 { 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()) encoder := input.Prepared.output - attachModuleManifestMetadata(&output, "output", encoder) + if err := attachModuleManifestMetadata(&output, "output", encoder); err != nil { + return failOutput(output), err + } outputStarted := time.Now().UTC() if err := writeDebugTimed(debugRecorder, "output/input.json", debugTimedEnvelope{ Stage: string(StageOutput), @@ -265,13 +286,17 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err }); err != nil { 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{ Manifest: output.Manifest, NormalizeOutputs: cloneSerializedOutputs(output.NormalizeOutputs), Rejected: cloneRejectedOutputs(output.Rejected), Warnings: output.Warnings, LLMProfile: input.pipeline.Output.LLMProfile, - Metadata: input.Metadata, + Metadata: outputMetadata, }) output.Warnings = append(output.Warnings, encoded.Warnings...) 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)) validatorCtx, llmScope := withIsolatedDebugLLMScope(ctx, attemptPath) 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 { 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: - 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: 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 } -func attachModuleManifestMetadata(output *RunOutput, moduleKey string, module any) { +func attachModuleManifestMetadata(output *RunOutput, moduleKey string, module any) error { 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 { - return + return nil } if output.Manifest.ModuleMetadata == nil { output.Manifest.ModuleMetadata = make(map[string]map[string]any) } 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) if !ok { - return nil, false + return nil, false, nil } - moduleMetadata := cloneMetadata(provider.ManifestMetadata()) - if len(moduleMetadata) == 0 { - return nil, false + moduleMetadata, err := cloneMetadata(provider.ManifestMetadata()) + if err != nil { + 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) { @@ -678,12 +718,8 @@ func validateOutputFileName(name string) error { return nil } -func cloneMetadata(metadata map[string]any) map[string]any { - cloned, err := source.CloneMetadata(metadata) - if err != nil { - return metadata - } - return cloned +func cloneMetadata(metadata map[string]any) (map[string]any, error) { + return source.CloneMetadata(metadata) } func cloneLLMProfiles(profiles []artifacts.LLMProfileManifest) []artifacts.LLMProfileManifest { @@ -785,16 +821,19 @@ func resolvedSessionID(explicit string, sourceDocumentID string) string { return strings.TrimSpace(sourceDocumentID) } -func manifestMetadataWithSessionID(metadata map[string]any, sessionID string) map[string]any { - out := cloneMetadata(metadata) +func manifestMetadataWithSessionID(metadata map[string]any, sessionID string) (map[string]any, error) { + out, err := cloneMetadata(metadata) + if err != nil { + return nil, fmt.Errorf("clone run manifest metadata: %w", err) + } if strings.TrimSpace(sessionID) == "" { - return out + return out, nil } if out == nil { out = make(map[string]any) } out["session_id"] = sessionID - return out + return out, nil } func cloneWarnings(warnings []contracts.Warning) []contracts.Warning { @@ -804,43 +843,61 @@ func cloneWarnings(warnings []contracts.Warning) []contracts.Warning { return append([]contracts.Warning(nil), warnings...) } -func cloneSourceChunkPtr(chunk *source.Chunk) *source.Chunk { +func cloneSourceChunkPtr(chunk *source.Chunk) (*source.Chunk, error) { if chunk == nil { - return nil + return nil, nil } - cloned := cloneSourceChunk(*chunk) - return &cloned + cloned, err := cloneSourceChunk(*chunk) + 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.Units = cloneSourceUnits(chunk.Units) - chunk.Metadata = cloneMetadata(chunk.Metadata) + var err error + 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.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 { - return nil + return nil, nil } out := make([]source.Chunk, 0, len(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 { - return nil + return nil, nil } out := make([]source.SourceUnit, 0, len(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 { diff --git a/internal/framework/pipeline/runner_chunk_plan.go b/internal/framework/pipeline/runner_chunk_plan.go index 54598e7..a53a91b 100644 --- a/internal/framework/pipeline/runner_chunk_plan.go +++ b/internal/framework/pipeline/runner_chunk_plan.go @@ -58,7 +58,9 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S case ChunkPlanHit: plan, chunks, validationErr := validateAndMaterializeChunkPlan(doc, record.Plan) 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) result.plan = &plan 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)) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) 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{ Source: doc, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(input.pipeline.ChunkReferences.ReferenceSet), - LLMProfile: input.pipeline.Chunk.LLMProfile, Metadata: input.Metadata, + LLMProfile: input.pipeline.Chunk.LLMProfile, Metadata: requestMetadata, }) if callErr != nil { 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 { 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 := "" if provider, ok := chunker.(contracts.ChunkExecutionClassProvider); ok && provider.ExecutionClass() == contracts.ExecutionClassLLMBacked { profile = input.pipeline.Chunk.LLMProfile @@ -116,7 +125,7 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S Producer: ChunkPlanProducer{ InputModule: input.Prepared.input.Key(), ChunkModule: chunker.Key(), LLMProfile: profile, References: append([]artifacts.ReferenceProvenance(nil), referenceTargetProvenance(input.pipeline.ChunkReferences)...), - Metadata: cloneMetadata(producerMetadata), + Metadata: producerMetadata, }, 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 { 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) attemptWarnings := append(cloneWarnings(chunkResult.Warnings), validationWarnings...) 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 { - 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) if err := input.ChunkPlans.Save(record); err != nil { 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) { - cloned := cloneChunkPlanRecord(record) +func (result *chunkPlanExecution) setCandidate(record ChunkPlanRecord, action string) error { + cloned, err := cloneChunkPlanRecord(record) + if err != nil { + return err + } result.record = &cloned result.action = action result.summary.Action = action result.summary.SourceDigest = record.SourceDigest result.summary.CandidateDigest = record.PlanDigest + return nil } 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.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) - return record + return record, nil } -func applyChunkPlanExecution(output *RunOutput, result chunkPlanExecution) { +func applyChunkPlanExecution(output *RunOutput, result chunkPlanExecution) error { if output == nil { - return + return nil } summary := result.summary output.ChunkPlan = &summary if output.Manifest.ChunkPlan == nil { - return + return nil } manifest := output.Manifest.ChunkPlan manifest.Action = result.action if result.record == nil { - return + return nil } record := result.record manifest.SourceDigest = record.SourceDigest @@ -249,7 +271,12 @@ func applyChunkPlanExecution(output *RunOutput, result chunkPlanExecution) { manifest.ProducerModule = record.Producer.ChunkModule manifest.ProducerLLMProfile = record.Producer.LLMProfile 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 manifest.CreatedAt = &createdAt + return nil } diff --git a/internal/framework/pipeline/runner_chunk_plan_test.go b/internal/framework/pipeline/runner_chunk_plan_test.go index b83ec0d..e4c0ffb 100644 --- a/internal/framework/pipeline/runner_chunk_plan_test.go +++ b/internal/framework/pipeline/runner_chunk_plan_test.go @@ -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) { prepared, plan := preparedTerminalDebugPipeline(t) prepared.resolved.Chunk.LLMProfile = "configured-but-unused" diff --git a/internal/framework/pipeline/runner_concurrent.go b/internal/framework/pipeline/runner_concurrent.go index ea5f2c9..9498da8 100644 --- a/internal/framework/pipeline/runner_concurrent.go +++ b/internal/framework/pipeline/runner_concurrent.go @@ -79,7 +79,9 @@ func (r *Runner) runLanes(parent context.Context, input RunInput, checkpoints Ch if prepared.typed == nil { 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) if err != nil { return output, err @@ -212,7 +214,9 @@ func (r *Runner) runLanes(parent context.Context, input RunInput, checkpoints Ch close(continuations) continuationWorkers.Wait() 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 { return output, err @@ -244,7 +248,10 @@ func prepareLaneExtract(input RunInput, loader CheckpointLoader, doc *source.Sou if decodeErr != nil { 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} if stored.ChunkIndex >= 0 && stored.ChunkIndex < len(chunks) && artifact.ChunkRef == (source.SourceRef{}) { 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 { - state, chunk := job.lane, cloneSourceChunk(job.chunk) + state := job.lane + chunk, cloneErr := cloneSourceChunk(job.chunk) 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 serialized CheckpointArtifact 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)) 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}) - 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 { 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) @@ -414,9 +430,9 @@ func selectRunError(parent context.Context, values []orderedRunError) error { return filtered[0].err } -func mergeLaneOutput(dst *RunOutput, src RunOutput) { +func mergeLaneOutput(dst *RunOutput, src RunOutput) error { if dst == nil { - return + return nil } dst.NormalizeOutputs = append(dst.NormalizeOutputs, cloneSerializedOutputs(src.NormalizeOutputs)...) 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 j := range src.Manifest.ArtifactLanes { 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 } diff --git a/internal/framework/pipeline/runner_extract_handoff_test.go b/internal/framework/pipeline/runner_extract_handoff_test.go index b6e0839..531ea14 100644 --- a/internal/framework/pipeline/runner_extract_handoff_test.go +++ b/internal/framework/pipeline/runner_extract_handoff_test.go @@ -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) { prepared := preparedAttemptDebugPipeline(t) extractCalls := 0 diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 467a775..476bbba 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -30,14 +30,18 @@ func cloneCheckpointArtifact(output CheckpointArtifact) CheckpointArtifact { 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) 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 { output.Artifact.Metadata = nil } - return output + return output, nil } func artifactCheckpointDigests(outputs []CheckpointArtifact) []CheckpointFingerprint { values := make([]CheckpointFingerprint, 0, len(outputs)) @@ -82,9 +86,12 @@ func serializeArtifact(codec artifactCodecEntry, value any, candidate bool) (con schema := codec.spec.Schema metadata := map[string]any(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) { @@ -139,7 +146,9 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi if typed == nil { 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)) 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) } 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) output.Warnings = append(output.Warnings, mergeWarnings...) } 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)) 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}) - 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 { attemptErr := fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, lane.Merge.Module, callErr) return false, nil, terminal.record(nil, attemptErr) @@ -246,7 +262,11 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi if decodeErr != nil { 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...) } else { 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)) 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}) - 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 { attemptErr := fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, lane.Normalize.Module, callErr) return false, nil, terminal.record(nil, attemptErr) @@ -309,9 +333,9 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi 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 { - return + return nil } for i := range output.Manifest.ArtifactLanes { if output.Manifest.ArtifactLanes[i].ID != laneID { @@ -322,15 +346,20 @@ func setTypedLaneManifestMetadata(output *RunOutput, laneID string, extractor, m name string module any }{{"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 } } if len(metadata) > 0 { 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) { @@ -349,17 +378,32 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE 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)) 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 { case ValidatorTargetTyped: - target.llmProfile = binding.LLMProfile - result, err = item.typedValidate(validatorCtx, item.typed, target) + requestTarget.llmProfile = binding.LLMProfile + result, err = item.typedValidate(validatorCtx, item.typed, requestTarget) case ValidatorTargetSerialized: artifact, encodeErr := validationCandidateArtifact(codec, target) if encodeErr != nil { err = encodeErr 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: return nil, nil, fmt.Errorf("validator %q is incompatible with typed artifact validation", binding.Module) }