diff --git a/docs/policy/architecture.md b/docs/policy/architecture.md index 33c7af7..33920a8 100644 --- a/docs/policy/architecture.md +++ b/docs/policy/architecture.md @@ -211,6 +211,13 @@ envelope with `json.RawMessage` payloads; extract modules should own typed Go structs at their boundaries and encode into that generic envelope before returning to framework code. +Output-stage warnings are intentionally out-of-band. An `OutputEncoder` may +return warnings about serialization, formatting, truncation, or destination +concerns, but those warnings are appended to the runner result after encoding +and are not expected to appear inside the encoded artifact bytes. CLI, +diagnostic, or reporting layers should surface output-stage warnings from the +runner result. + Schemas should be versioned per extractor, with a separate envelope/manifest format version. diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md deleted file mode 100644 index 5038306..0000000 --- a/docs/roadmap/implementation.md +++ /dev/null @@ -1,891 +0,0 @@ -# Implementation Plan: Checkpoint 3 Pipeline Stages - -## Status - -This is a staged implementation plan for -[`3-pipeline-stages-chunking-merge-normalize.md`](3-pipeline-stages-chunking-merge-normalize.md). -It is intended for an LLM coding agent to follow stage by stage. - -This plan implements only checkpoint 3. Do not add real Seriatim parsing, real -D&D extraction, production config file loading, CLI command behavior, LLM -provider clients, prompt assets, response schema assets, diagnostics run -directories, or durable output writing. - -## Policy Context - -Follow: - -- [`docs/policy/architecture.md`](../policy/architecture.md) -- [`docs/policy/documentation.md`](../policy/documentation.md) - -Required boundaries: - -- framework code must stay source-agnostic and domain-agnostic; -- concrete business logic remains under `internal/modules//...`, but this - checkpoint should use fake modules only; -- framework plumbing should remain consolidated under `internal/framework/pipeline` - unless a boundary proves itself; -- capabilities are flat strings, not a type system; -- pipeline profiles are fixed-shape templates, not arbitrary step lists; -- no structural module wiring through ad hoc CLI flags. - -## Global Implementation Decisions - -- Add no third-party dependencies. -- Keep new checkpoint 3 framework code in `internal/framework/pipeline`. -- Keep shared request/result contracts in `internal/framework/contracts` when - stage modules will implement or consume them. -- Do not create new framework packages for chunking, merge, normalize, output, - warnings, response schemas, or structured output. -- Use in-memory pipeline profile structs only. Do not implement YAML/TOML - parsing in this checkpoint. -- Keep one shared chunker per resolved pipeline. Per-lane chunker overrides are - deferred. -- Execute chunks serially in checkpoint 3. Contracts must not prevent later - parallel execution. -- Sort artifact lane IDs during resolution for deterministic output and digest - behavior. -- Use `sha256:` for resolved pipeline digests. -- Preserve existing validator decision-cardinality behavior. -- Run `gofmt` on all touched Go files before validation. - -## Stage 1: Stage Contracts And Source Chunks - -### Goal - -Define the source chunk model and contracts for chunk, merge, normalize, and -output stages. - -### Files To Update - -- `internal/framework/contracts/contracts.go` -- `internal/framework/contracts/contracts_test.go` -- `internal/framework/contracts/composition_test.go` - -### Required API - -Add: - -```go -type SourceChunk struct { - ID string `json:"id"` - SourceID string `json:"source_id"` - Index int `json:"index"` - Units []source.SourceUnit `json:"units"` - Metadata map[string]any `json:"metadata,omitempty"` -} - -type ChunkRequest struct { - Source *source.SourceDocument `json:"-"` - Metadata map[string]any `json:"metadata,omitempty"` -} - -type ChunkResult struct { - Chunks []SourceChunk `json:"chunks"` - Warnings []Warning `json:"warnings,omitempty"` -} - -type Chunker interface { - Key() string - Chunk(ctx context.Context, req ChunkRequest) (ChunkResult, error) -} -``` - -Extend `ExtractionRequest`: - -```go -type ExtractionRequest struct { - Source *source.SourceDocument `json:"-"` - Chunk *SourceChunk `json:"chunk,omitempty"` - AmbientContext map[string]any `json:"ambient_context,omitempty"` - LLMClient StructuredLLMClient `json:"-"` - Metadata map[string]any `json:"metadata,omitempty"` -} -``` - -Add: - -```go -type ChunkArtifacts struct { - Chunk SourceChunk `json:"chunk"` - Candidates []artifacts.ArtifactCandidate `json:"candidates"` -} - -type MergeRequest struct { - Source *source.SourceDocument `json:"-"` - LaneID string `json:"lane_id"` - ChunkArtifacts []ChunkArtifacts `json:"chunk_artifacts"` - Metadata map[string]any `json:"metadata,omitempty"` -} - -type MergeResult struct { - Candidates []artifacts.ArtifactCandidate `json:"candidates"` - Warnings []Warning `json:"warnings,omitempty"` -} - -type Merger interface { - Key() string - Merge(ctx context.Context, req MergeRequest) (MergeResult, error) -} - -type NormalizeRequest struct { - Source *source.SourceDocument `json:"-"` - LaneID string `json:"lane_id"` - Candidates []artifacts.ArtifactCandidate `json:"candidates"` - Metadata map[string]any `json:"metadata,omitempty"` -} - -type NormalizeResult struct { - Candidates []artifacts.ArtifactCandidate `json:"candidates"` - Warnings []Warning `json:"warnings,omitempty"` -} - -type Normalizer interface { - Key() string - Normalize(ctx context.Context, req NormalizeRequest) (NormalizeResult, error) -} - -type OutputRequest struct { - Manifest artifacts.RunManifest `json:"manifest"` - Approved []artifacts.Artifact `json:"approved,omitempty"` - Rejected []artifacts.RejectedArtifact `json:"rejected,omitempty"` - Warnings []Warning `json:"warnings,omitempty"` - Metadata map[string]any `json:"metadata,omitempty"` -} - -type OutputResult struct { - Bytes []byte `json:"-"` - ContentType string `json:"content_type,omitempty"` - Warnings []Warning `json:"warnings,omitempty"` -} - -type OutputEncoder interface { - Key() string - Encode(ctx context.Context, req OutputRequest) (OutputResult, error) -} -``` - -If `gofmt` aligns fields differently, keep the formatted output. - -### Required Behavior - -- Existing fake extractor tests must still compile with `ExtractionRequest{Source: doc}`. -- `SourceChunk` is a framework contract type, not a core source type. -- `SourceChunk.Units` should be copied by implementations where mutation risk - exists; this stage only defines the contract. - -### Required Tests - -Update fake implementations in existing contract tests to prove: - -- fake chunker satisfies `contracts.Chunker`; -- fake merger satisfies `contracts.Merger`; -- fake normalizer satisfies `contracts.Normalizer`; -- fake output encoder satisfies `contracts.OutputEncoder`; -- fake extractor can receive a `SourceChunk` and ambient context. - -### Validation - -Run: - -```sh -gofmt -w internal/framework/contracts -go test ./internal/framework/contracts -go test ./... -``` - -## Stage 2: Module Metadata And Stage Registries - -### Goal - -Teach pipeline registries about module metadata and add registries for chunk, -merge, normalize, configured validators, and output modules. - -### Files To Update - -- `internal/framework/pipeline/input_registry.go` -- `internal/framework/pipeline/input_registry_test.go` -- `internal/framework/pipeline/extractor_registry.go` -- `internal/framework/pipeline/extractor_registry_test.go` - -### Files To Add - -- `internal/framework/pipeline/module.go` -- `internal/framework/pipeline/chunker_registry.go` -- `internal/framework/pipeline/chunker_registry_test.go` -- `internal/framework/pipeline/merger_registry.go` -- `internal/framework/pipeline/merger_registry_test.go` -- `internal/framework/pipeline/normalizer_registry.go` -- `internal/framework/pipeline/normalizer_registry_test.go` -- `internal/framework/pipeline/validator_registry.go` -- `internal/framework/pipeline/validator_registry_test.go` -- `internal/framework/pipeline/output_registry.go` -- `internal/framework/pipeline/output_registry_test.go` - -### Required API - -Add: - -```go -type ModuleStage string - -const ( - StageInput ModuleStage = "input" - StageChunk ModuleStage = "chunk" - StageExtract ModuleStage = "extract" - StageMerge ModuleStage = "merge" - StageNormalize ModuleStage = "normalize" - StageValidate ModuleStage = "validate" - StageOutput ModuleStage = "output" -) - -type ModuleSpec struct { - Key string - Stage ModuleStage - Provides []string - Requires []string -} -``` - -Add metadata registration methods to existing registries: - -```go -func (r *InputAdapterRegistry) RegisterWithSpec(spec ModuleSpec, constructor InputAdapterConstructor) error -func (r *InputAdapterRegistry) Spec(key string) (ModuleSpec, bool) - -func (r *ExtractorRegistry) RegisterWithSpec(spec ModuleSpec, constructor ExtractorConstructor) error -func (r *ExtractorRegistry) Spec(key string) (ModuleSpec, bool) -``` - -Keep existing `Register` methods. They should call `RegisterWithSpec` with a -default spec using the provided key and correct stage. - -Add typed registries for other stages: - -```go -type ChunkerConstructor func() (contracts.Chunker, error) -type ChunkerRegistry struct { /* unexported fields */ } -func NewChunkerRegistry() *ChunkerRegistry -func (r *ChunkerRegistry) Register(key string, constructor ChunkerConstructor) error -func (r *ChunkerRegistry) RegisterWithSpec(spec ModuleSpec, constructor ChunkerConstructor) error -func (r *ChunkerRegistry) Build(key string) (contracts.Chunker, error) -func (r *ChunkerRegistry) Spec(key string) (ModuleSpec, bool) -func (r *ChunkerRegistry) RegisteredKeys() []string -``` - -Repeat the same pattern for `MergerRegistry`, `NormalizerRegistry`, -`ValidatorRegistry`, and `OutputEncoderRegistry`. - -`ValidatorRegistry` should use: - -```go -type ValidatorConstructor func() (contracts.Validator, error) -``` - -Because `contracts.Validator` uses `Name()` rather than `Key()`, -`ValidatorRegistry.Build` must require `validator.Name()` to match the -registered key. - -### Required Behavior - -- Registry keys are trimmed. -- Specs are normalized by trimming `Key`, `Provides`, and `Requires`. -- Empty capability strings are ignored. -- `Provides` and `Requires` are deduplicated and sorted for deterministic - behavior. -- `RegisterWithSpec` rejects an empty key, wrong stage, nil constructor, and - duplicate key. -- `Spec` returns false for unknown keys and nil registries. -- `Build` rejects unknown keys, constructor errors, nil modules, and key - mismatches. -- Existing registry tests continue to pass. - -### Required Tests - -For each new registry, cover the same behavior as existing input/extractor -registries: - -- successful registration and build; -- metadata registration and lookup; -- capability normalization; -- wrong stage rejection; -- key trimming; -- empty key rejection; -- duplicate key rejection; -- nil constructor rejection; -- unknown key build error; -- constructor error wrapping; -- nil module rejection; -- key mismatch rejection; -- sorted `RegisteredKeys`; -- nil registry behavior. - -For `ValidatorRegistry`, key mismatch checks should compare the registry key to -`Validator.Name()`. - -For existing input/extractor registries, add tests for metadata registration, -default specs from `Register`, capability normalization, and wrong stage -rejection. - -### Validation - -Run: - -```sh -gofmt -w internal/framework/pipeline -go test ./internal/framework/pipeline -go test ./... -``` - -## Stage 3: Resolved Pipeline Model - -### Goal - -Add fixed-shape pipeline profile structs, default resolution, lane selection, -capability validation, and resolved pipeline digesting. - -### Files To Add - -- `internal/framework/pipeline/profile.go` -- `internal/framework/pipeline/profile_test.go` - -### Files To Update - -- `internal/core/artifacts/artifacts.go` -- `internal/core/artifacts/artifacts_test.go` - -### Required API - -Update `artifacts.RunManifest` to include: - -```go -PipelineID string `json:"pipeline_id,omitempty"` -PipelineDigest string `json:"pipeline_digest,omitempty"` -``` - -Add lane-level manifest details rather than relying on the existing shared -`Extractors`, `Merger`, or `Normalizer` fields for multi-lane runs: - -```go -type ArtifactLaneManifest struct { - ID string `json:"id"` - Extractor string `json:"extractor"` - Merger string `json:"merger"` - Normalizer string `json:"normalizer"` - Validators []string `json:"validators,omitempty"` -} -``` - -Then add: - -```go -ArtifactLanes []ArtifactLaneManifest `json:"artifact_lanes,omitempty"` -``` - -Do not remove the existing singular fields in this checkpoint unless doing so is -required to make the package compile. Treat `ArtifactLanes` as the authoritative -manifest representation for checkpoint 3's multi-lane pipeline model. - -Add pipeline model types: - -```go -type ModuleBinding struct { - Module string `json:"module"` - LLMProfile string `json:"llm_profile,omitempty"` - Options map[string]any `json:"options,omitempty"` -} - -type ArtifactLaneProfile struct { - Extract ModuleBinding `json:"extract"` - Merge ModuleBinding `json:"merge,omitempty"` - Normalize ModuleBinding `json:"normalize,omitempty"` - Validators []ModuleBinding `json:"validators,omitempty"` -} - -type PipelineProfile struct { - ID string `json:"id"` - Input ModuleBinding `json:"input"` - Chunk ModuleBinding `json:"chunk,omitempty"` - Artifacts map[string]ArtifactLaneProfile `json:"artifacts"` - Output ModuleBinding `json:"output,omitempty"` -} - -type ResolveOptions struct { - Only []string -} - -type ResolvedArtifactLane struct { - ID string - Extract ModuleBinding - Merge ModuleBinding - Normalize ModuleBinding - Validators []ModuleBinding -} - -type ResolvedPipeline struct { - ID string - Digest string - Input ModuleBinding - Chunk ModuleBinding - ArtifactLanes []ResolvedArtifactLane - Output ModuleBinding -} -``` - -Add defaults: - -```go -const ( - DefaultChunkModule = "generic" - DefaultMergeModule = "appendorder" - DefaultNormalizeModule = "noop" - DefaultOutputModule = "json" - DefaultLLMProfile = "default" -) -``` - -Add: - -```go -type ModuleCatalog struct { - Inputs *InputAdapterRegistry - Chunkers *ChunkerRegistry - Extractors *ExtractorRegistry - Mergers *MergerRegistry - Normalizers *NormalizerRegistry - Validators *ValidatorRegistry - Outputs *OutputEncoderRegistry -} - -func Binding(module string) ModuleBinding -func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog ModuleCatalog) (ResolvedPipeline, error) -``` - -### Required Behavior - -`ResolvePipeline` must: - -- trim `PipelineProfile.ID`; -- reject empty pipeline IDs; -- require non-empty input module; -- require at least one artifact lane before and after `Only` filtering; -- apply defaults for missing chunk, merge, normalize, output, and binding - `llm_profile` values; -- trim all module keys; -- trim lane names in `ResolveOptions.Only`; -- reject empty or unknown lane names in `ResolveOptions.Only`; -- deduplicate repeated `Only` lane names; -- sort selected lane IDs alphabetically; -- reject unknown module keys using registry metadata, not constructors; -- validate flat capability requirements before execution; -- compute `ResolvedPipeline.Digest` from canonical JSON of the resolved - pipeline excluding the digest field itself. - -Capability validation algorithm: - -1. Start with capabilities provided by the input module. -2. Validate chunker requirements against current capabilities, then add chunker - provided capabilities. -3. For each selected lane, start with input plus chunk capabilities. -4. Validate extractor requirements, then add extractor provided capabilities. -5. Validate merger requirements, then add merger provided capabilities. -6. Validate normalizer requirements, then add normalizer provided capabilities. -7. Validate configured validator requirements in lane order, then add validator - provided capabilities. If no validators are configured for a lane, do not - attempt capability validation for extractor-owned default validators. -8. Validate output requirements against the union of pipeline-level capabilities - and all selected lane capabilities. - -Error messages must name the pipeline ID, lane ID when applicable, stage, module -key, and missing capability when applicable. - -`Binding(module)` should trim the module key and leave `LLMProfile` empty and -`Options` nil before resolution. Resolution should fill empty `LLMProfile` -values with `DefaultLLMProfile` on all resolved bindings. - -### Required Tests - -Cover: - -- successful resolution with explicit modules; -- default chunk, merge, normalize, output, and `llm_profile`; -- lane selection with `Only`; -- `Only` trimming and deduplication; -- unknown `Only` lane rejection; -- empty `Only` lane rejection; -- empty artifact set rejection; -- empty pipeline ID rejection; -- missing input rejection; -- unknown module key rejection for each stage; -- missing capability rejection for input/chunk/extract/merge/normalize/validate/output - interactions; -- deterministic lane ordering; -- deterministic digest for equivalent input maps; -- digest changes when a module binding changes; -- manifest JSON still omits empty optional fields. - -Use fake registries and fake specs only. Do not add config file parsing. - -### Validation - -Run: - -```sh -gofmt -w internal/core/artifacts internal/framework/pipeline -go test ./internal/core/artifacts -go test ./internal/framework/pipeline -go test ./... -``` - -## Stage 4: Generic Merge And Normalize Helpers - -### Goal - -Add generic append-in-chunk-order merge and no-op normalization behavior inside -`internal/framework/pipeline`. - -### Files To Add - -- `internal/framework/pipeline/generic_stages.go` -- `internal/framework/pipeline/generic_stages_test.go` - -### Required API - -Add: - -```go -type AppendOrderMerger struct{} -func (m AppendOrderMerger) Key() string -func (m AppendOrderMerger) Merge(ctx context.Context, req contracts.MergeRequest) (contracts.MergeResult, error) - -type NoopNormalizer struct{} -func (n NoopNormalizer) Key() string -func (n NoopNormalizer) Normalize(ctx context.Context, req contracts.NormalizeRequest) (contracts.NormalizeResult, error) -``` - -### Required Behavior - -`AppendOrderMerger` must: - -- return key `appendorder`; -- concatenate candidates by `ChunkArtifacts` slice order and candidate order; -- copy candidate slices so caller mutation does not affect the result; -- preserve source refs and metadata exactly; -- append no warnings. - -`NoopNormalizer` must: - -- return key `noop`; -- copy candidates so caller mutation does not affect the result; -- preserve ordering and source refs exactly; -- append no warnings. - -### Required Tests - -Cover: - -- key values; -- append order across multiple chunks; -- stable candidate order within chunks; -- no-op normalize preserves order; -- result slices are caller-mutation safe; -- nil/empty input returns empty candidates and no error. - -### Validation - -Run: - -```sh -gofmt -w internal/framework/pipeline -go test ./internal/framework/pipeline -go test ./... -``` - -## Stage 5: Pipeline Runner Refactor - -### Goal - -Refactor the existing runner from extractor-only execution to resolved pipeline -execution across input, chunk, extract, merge, normalize, validate, and output. - -### Files To Update - -- `internal/framework/pipeline/runner.go` -- `internal/framework/pipeline/runner_test.go` -- `internal/framework/pipeline/registry_integration_test.go` - -### Required API - -Replace the extractor-only runner constructor with: - -```go -type Registries struct { - Inputs *InputAdapterRegistry - Chunkers *ChunkerRegistry - Extractors *ExtractorRegistry - Mergers *MergerRegistry - Normalizers *NormalizerRegistry - Validators *ValidatorRegistry - Outputs *OutputEncoderRegistry -} - -type Runner struct { - // unexported fields -} - -func New(registries Registries) *Runner -``` - -Replace `RunInput` with: - -```go -type RunInput struct { - Pipeline ResolvedPipeline - SourceID string - Path string - RawInput []byte - LLMClient contracts.StructuredLLMClient - Metadata map[string]any -} -``` - -Replace or extend `RunOutput` with: - -```go -type RunOutput struct { - Manifest artifacts.RunManifest `json:"manifest"` - Approved []artifacts.Artifact `json:"approved,omitempty"` - Rejected []artifacts.RejectedArtifact `json:"rejected,omitempty"` - Warnings []contracts.Warning `json:"warnings,omitempty"` - EncodedOutput []byte `json:"-"` - ContentType string `json:"content_type,omitempty"` -} -``` - -Keep the existing candidate metadata normalization behavior, but it should now -run per extractor and across all selected lanes. - -### Required Behavior - -`Run` must: - -1. reject nil runner; -2. reject missing registries for stages used by the pipeline; - `Validators` is required only when at least one selected lane declares - validator bindings; -3. reject empty resolved pipeline ID or digest; -4. build the input adapter and parse `RawInput`; -5. validate the resulting `SourceDocument`; -6. build the chunker and produce chunks; -7. reject empty chunk results; -8. for each selected artifact lane in resolved order: - - build extractor, merger, and normalizer; - - extract from every chunk serially; - - pass `Source`, active `Chunk`, `LLMClient`, and metadata to the extractor; - - normalize candidate metadata with global monotonically increasing indices; - - merge per-chunk candidates; - - normalize merged candidates; - - run configured lane validators against normalized candidates when the lane - declares validators; - - otherwise run the extractor-owned default validator chain; - - append approved artifacts and rejected artifacts to output; -9. build the output encoder and encode the final output bundle; -10. populate `RunManifest` with pipeline ID, pipeline digest, input module, - chunker, lane-level extractor/merger/normalizer/validator keys, output - encoder, source digest, and validation status; -11. return partial output plus wrapped errors on failure. - -Validation status: - -- `approved` when all candidates from all lanes survive validation and no lane - returns rejected artifacts; -- `rejected` when at least one candidate is rejected and execution otherwise - succeeds; -- `failed` when returning an error after a manifest has been initialized. - -Warnings: - -- collect warnings from input, chunk, extract, merge, normalize, validator, and - output stages where those stages expose warnings; -- existing input adapters do not expose warnings, so no input warnings are - required in this checkpoint. - -Do not validate source references inside the runner in checkpoint 3. Source -reference validation remains a later deterministic validator. - -### Required Tests - -Refactor existing runner tests and add coverage for: - -- nil runner error; -- missing registry errors; -- invalid resolved pipeline error; -- input adapter build and parse errors; -- invalid source document error; -- chunker build and chunk errors; -- empty chunk result error; -- extractor runs once per chunk; -- extractor receives active chunk and LLM client; -- merge receives per-chunk candidates; -- normalize receives merged candidates; -- validator approvals and rejections still work; -- configured validators are built and run in lane order; -- extractor-owned default validators run when a lane does not configure - validators; -- configured validators replace extractor-owned default validators when present; -- global candidate indices across lanes and chunks; -- warnings from chunk, extract, merge, normalize, validator, and output stages; -- output encoder receives manifest and artifacts; -- manifest pipeline ID, digest, and artifact lane details are populated; -- partial output is returned when a later lane fails. - -Use fake modules only. - -### Validation - -Run: - -```sh -gofmt -w internal/framework/pipeline -go test ./internal/framework/pipeline -go test ./... -``` - -## Stage 6: Fixture-Driven Walking Skeleton - -### Goal - -Add an end-to-end fixture-driven integration test that exercises the full staged -path with fake modules and a fake LLM client. - -### Files To Add - -- `internal/framework/pipeline/testdata/walking_skeleton_input.json` -- `internal/framework/pipeline/testdata/walking_skeleton_output.json` -- `internal/framework/pipeline/walking_skeleton_test.go` - -### Required Fixture Shape - -Use a small input fixture with two or three source units in a simple fake JSON -format. Do not use Seriatim. - -Example input shape: - -```json -{ - "id": "fixture-source", - "units": [ - {"id": "u1", "text": "First event."}, - {"id": "u2", "text": "Second event."}, - {"id": "u3", "text": "Third event."} - ] -} -``` - -The expected output fixture should contain: - -- manifest with `pipeline_id` and `pipeline_digest`; -- approved artifacts emitted from multiple chunks; -- no rejected artifacts for the happy path; -- warnings only if fake stages deliberately emit them. - -Compare expected output by unmarshaled structural equality, not raw bytes, so -formatting differences do not fail the test. - -### Required Test Setup - -Register fake modules with metadata: - -- fake input adapter key `fake/input`, provides `plain_text`; -- fake chunker key `fake/chunk`, requires `plain_text`, provides `chunks`; -- fake extractor key `fake/extract`, requires `chunks`, provides - `fake_artifacts`; -- append-order merger key `appendorder`, requires `fake_artifacts`; -- no-op normalizer key `noop`; -- fake output encoder key `json`. - -Use an in-memory `PipelineProfile`: - -```go -PipelineProfile{ - ID: "walking-skeleton", - Input: Binding("fake/input"), - Chunk: Binding("fake/chunk"), - Artifacts: map[string]ArtifactLaneProfile{ - "events": {Extract: Binding("fake/extract")}, - }, - Output: Binding("json"), -} -``` - -Resolve the profile, run the runner with fixture bytes and fake LLM client, then -assert the encoded output matches the expected fixture structurally. - -### Required Behavior - -- The fake input adapter parses raw fixture bytes into `SourceDocument`. -- The fake chunker produces at least two chunks. -- The fake extractor calls the fake structured LLM client once per chunk. -- The fake extractor returns one artifact candidate per chunk. -- The append-order merger preserves chunk order. -- The no-op normalizer preserves candidates. -- The fake output encoder produces deterministic JSON bytes. -- The test fails before execution if required capabilities are missing. - -### Additional Tests - -Add focused tests for the walking skeleton failure path: - -- missing required capability fails during resolution; -- `Only` with an unknown lane fails during resolution; -- fake LLM call count equals chunk count in the happy path. - -### Validation - -Run: - -```sh -gofmt -w internal/framework/pipeline -go test ./internal/framework/pipeline -go test ./... -``` - -## Stage 7: Final Checkpoint 3 Review Pass - -### Goal - -Clean up naming, formatting, and accidental scope creep before checkpoint 3 is -considered complete. - -### Required Review - -Check: - -- no Seriatim parser exists; -- no D&D module exists; -- no production config file loader exists; -- no CLI behavior was added; -- no LLM provider code exists; -- no prompt, response schema, diagnostics, or durable output package exists; -- no new framework package was created beyond `contracts`, `pipeline`, and - `validate`; -- stage contracts remain source/domain agnostic; -- capabilities remain flat strings; -- resolved pipeline profiles remain fixed-shape templates; -- walking skeleton uses fake modules only. - -### Required Validation - -Run: - -```sh -gofmt -w internal/core internal/framework -go test ./... -go build ./cmd/notarius -``` - -## Open Questions - -None. The deferred choices identified while drafting this plan are intentionally -out of scope for checkpoint 3 and should remain deferred until later roadmap -items. diff --git a/internal/framework/contracts/contracts.go b/internal/framework/contracts/contracts.go index 421c263..c2c405e 100644 --- a/internal/framework/contracts/contracts.go +++ b/internal/framework/contracts/contracts.go @@ -35,10 +35,12 @@ type StructuredLLMClient interface { } type ParseRequest struct { - SourceID string `json:"source_id,omitempty"` - Path string `json:"path,omitempty"` - Raw []byte `json:"-"` - Metadata map[string]any `json:"metadata,omitempty"` + SourceID string `json:"source_id,omitempty"` + Path string `json:"path,omitempty"` + Raw []byte `json:"-"` + LLMProfile string `json:"llm_profile,omitempty"` + Options map[string]any `json:"options,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` } type InputAdapter interface { @@ -55,8 +57,10 @@ type SourceChunk struct { } type ChunkRequest struct { - Source *source.SourceDocument `json:"-"` - Metadata map[string]any `json:"metadata,omitempty"` + Source *source.SourceDocument `json:"-"` + LLMProfile string `json:"llm_profile,omitempty"` + Options map[string]any `json:"options,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` } type ChunkResult struct { @@ -74,6 +78,8 @@ type ExtractionRequest struct { Chunk *SourceChunk `json:"chunk,omitempty"` AmbientContext map[string]any `json:"ambient_context,omitempty"` LLMClient StructuredLLMClient `json:"-"` + LLMProfile string `json:"llm_profile,omitempty"` + Options map[string]any `json:"options,omitempty"` Metadata map[string]any `json:"metadata,omitempty"` } @@ -99,6 +105,8 @@ type MergeRequest struct { Source *source.SourceDocument `json:"-"` LaneID string `json:"lane_id"` ChunkArtifacts []ChunkArtifacts `json:"chunk_artifacts"` + LLMProfile string `json:"llm_profile,omitempty"` + Options map[string]any `json:"options,omitempty"` Metadata map[string]any `json:"metadata,omitempty"` } @@ -116,6 +124,8 @@ type NormalizeRequest struct { Source *source.SourceDocument `json:"-"` LaneID string `json:"lane_id"` Candidates []artifacts.ArtifactCandidate `json:"candidates"` + LLMProfile string `json:"llm_profile,omitempty"` + Options map[string]any `json:"options,omitempty"` Metadata map[string]any `json:"metadata,omitempty"` } @@ -132,6 +142,8 @@ type Normalizer interface { type ValidationRequest struct { Source *source.SourceDocument `json:"-"` Candidates []artifacts.ArtifactCandidate `json:"candidates"` + LLMProfile string `json:"llm_profile,omitempty"` + Options map[string]any `json:"options,omitempty"` Metadata map[string]any `json:"metadata,omitempty"` } @@ -161,11 +173,13 @@ type Warning struct { } type OutputRequest struct { - Manifest artifacts.RunManifest `json:"manifest"` - Approved []artifacts.Artifact `json:"approved,omitempty"` - Rejected []artifacts.RejectedArtifact `json:"rejected,omitempty"` - Warnings []Warning `json:"warnings,omitempty"` - Metadata map[string]any `json:"metadata,omitempty"` + Manifest artifacts.RunManifest `json:"manifest"` + Approved []artifacts.Artifact `json:"approved,omitempty"` + Rejected []artifacts.RejectedArtifact `json:"rejected,omitempty"` + Warnings []Warning `json:"warnings,omitempty"` + LLMProfile string `json:"llm_profile,omitempty"` + Options map[string]any `json:"options,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` } type OutputResult struct { diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 7e51dd7..d030e8f 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -65,10 +65,12 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) { return failOutput(output), fmt.Errorf("build input adapter %q: %w", input.Pipeline.Input.Module, err) } doc, err := adapter.Parse(ctx, contracts.ParseRequest{ - SourceID: input.SourceID, - Path: input.Path, - Raw: input.RawInput, - Metadata: input.Metadata, + SourceID: input.SourceID, + Path: input.Path, + Raw: input.RawInput, + LLMProfile: input.Pipeline.Input.LLMProfile, + Options: cloneOptions(input.Pipeline.Input.Options), + Metadata: input.Metadata, }) if err != nil { return failOutput(output), fmt.Errorf("parse input with adapter %q: %w", adapter.Key(), err) @@ -83,8 +85,10 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) { return failOutput(output), fmt.Errorf("build chunker %q: %w", input.Pipeline.Chunk.Module, err) } chunkResult, err := chunker.Chunk(ctx, contracts.ChunkRequest{ - Source: doc, - Metadata: input.Metadata, + Source: doc, + LLMProfile: input.Pipeline.Chunk.LLMProfile, + Options: cloneOptions(input.Pipeline.Chunk.Options), + Metadata: input.Metadata, }) output.Warnings = append(output.Warnings, chunkResult.Warnings...) if err != nil { @@ -112,11 +116,13 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) { return failOutput(output), fmt.Errorf("build output encoder %q: %w", input.Pipeline.Output.Module, err) } encoded, err := encoder.Encode(ctx, contracts.OutputRequest{ - Manifest: output.Manifest, - Approved: output.Approved, - Rejected: output.Rejected, - Warnings: output.Warnings, - Metadata: input.Metadata, + Manifest: output.Manifest, + Approved: output.Approved, + Rejected: output.Rejected, + Warnings: output.Warnings, + LLMProfile: input.Pipeline.Output.LLMProfile, + Options: cloneOptions(input.Pipeline.Output.Options), + Metadata: input.Metadata, }) output.Warnings = append(output.Warnings, encoded.Warnings...) if err != nil { @@ -142,24 +148,28 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source return fmt.Errorf("build normalizer %q for lane %q: %w", lane.Normalize.Module, lane.ID, err) } - var validators []contracts.Validator + var validators []validatorExecution if len(lane.Validators) > 0 { validators, err = r.buildConfiguredValidators(lane) if err != nil { return err } } else { - validators = extractor.Validators() + for _, validator := range extractor.Validators() { + validators = append(validators, validatorExecution{validator: validator}) + } } chunkArtifacts := make([]contracts.ChunkArtifacts, 0, len(chunks)) for index := range chunks { chunk := chunks[index] result, err := extractor.Extract(ctx, contracts.ExtractionRequest{ - Source: doc, - Chunk: &chunk, - LLMClient: input.LLMClient, - Metadata: input.Metadata, + Source: doc, + Chunk: &chunk, + LLMClient: input.LLMClient, + LLMProfile: lane.Extract.LLMProfile, + Options: cloneOptions(lane.Extract.Options), + Metadata: input.Metadata, }) output.Warnings = append(output.Warnings, result.Warnings...) if err != nil { @@ -180,6 +190,8 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source Source: doc, LaneID: lane.ID, ChunkArtifacts: chunkArtifacts, + LLMProfile: lane.Merge.LLMProfile, + Options: cloneOptions(lane.Merge.Options), Metadata: input.Metadata, }) output.Warnings = append(output.Warnings, mergeResult.Warnings...) @@ -191,6 +203,8 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source Source: doc, LaneID: lane.ID, Candidates: mergeResult.Candidates, + LLMProfile: lane.Normalize.LLMProfile, + Options: cloneOptions(lane.Normalize.Options), Metadata: input.Metadata, }) output.Warnings = append(output.Warnings, normalizeResult.Warnings...) @@ -198,6 +212,10 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source return fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, normalizer.Key(), err) } + if err := validateCandidateEnvelope(extractor, normalizeResult.Candidates); err != nil { + return fmt.Errorf("validate normalized candidates for lane %q: %w", lane.ID, err) + } + approved, rejected, warnings, err := runValidators(ctx, extractor.Key(), validators, doc, normalizeResult.Candidates, input.Metadata) output.Warnings = append(output.Warnings, warnings...) output.Rejected = append(output.Rejected, rejected...) @@ -211,14 +229,22 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source return nil } -func (r *Runner) buildConfiguredValidators(lane ResolvedArtifactLane) ([]contracts.Validator, error) { - validators := make([]contracts.Validator, 0, len(lane.Validators)) +type validatorExecution struct { + validator contracts.Validator + binding ModuleBinding +} + +func (r *Runner) buildConfiguredValidators(lane ResolvedArtifactLane) ([]validatorExecution, error) { + validators := make([]validatorExecution, 0, len(lane.Validators)) for _, binding := range lane.Validators { validator, err := r.registries.Validators.Build(binding.Module) if err != nil { return nil, fmt.Errorf("build validator %q for lane %q: %w", binding.Module, lane.ID, err) } - validators = append(validators, validator) + validators = append(validators, validatorExecution{ + validator: validator, + binding: binding, + }) } return validators, nil } @@ -359,18 +385,51 @@ func normalizeCandidates(extractor contracts.Extractor, candidates []artifacts.A return normalized, nil } -func runValidators(ctx context.Context, extractorKey string, validators []contracts.Validator, doc *source.SourceDocument, candidates []artifacts.ArtifactCandidate, metadata map[string]any) ([]artifacts.ArtifactCandidate, []artifacts.RejectedArtifact, []contracts.Warning, error) { +func validateCandidateEnvelope(extractor contracts.Extractor, candidates []artifacts.ArtifactCandidate) error { + seen := make(map[int]struct{}, len(candidates)) + for _, candidate := range candidates { + if _, ok := seen[candidate.Index]; ok { + return fmt.Errorf("candidate index %d is duplicated", candidate.Index) + } + seen[candidate.Index] = struct{}{} + + if candidate.ExtractorKey == "" { + return fmt.Errorf("candidate index %d extractor_key must not be empty", candidate.Index) + } + if candidate.ExtractorKey != extractor.Key() { + return fmt.Errorf("candidate index %d extractor_key %q does not match extractor %q", candidate.Index, candidate.ExtractorKey, extractor.Key()) + } + if candidate.ArtifactType == "" { + return fmt.Errorf("candidate index %d artifact_type must not be empty", candidate.Index) + } + if candidate.ArtifactType != extractor.ArtifactType() { + return fmt.Errorf("candidate index %d artifact_type %q does not match extractor %q artifact type %q", candidate.Index, candidate.ArtifactType, extractor.Key(), extractor.ArtifactType()) + } + if candidate.SchemaVersion == "" { + return fmt.Errorf("candidate index %d schema_version must not be empty", candidate.Index) + } + if candidate.SchemaVersion != extractor.SchemaVersion() { + return fmt.Errorf("candidate index %d schema_version %q does not match extractor %q schema version %q", candidate.Index, candidate.SchemaVersion, extractor.Key(), extractor.SchemaVersion()) + } + } + return nil +} + +func runValidators(ctx context.Context, extractorKey string, validators []validatorExecution, doc *source.SourceDocument, candidates []artifacts.ArtifactCandidate, metadata map[string]any) ([]artifacts.ArtifactCandidate, []artifacts.RejectedArtifact, []contracts.Warning, error) { eligible := candidates var rejected []artifacts.RejectedArtifact var warnings []contracts.Warning - for validatorIndex, validator := range validators { + for validatorIndex, execution := range validators { + validator := execution.validator if validator == nil { return nil, rejected, warnings, fmt.Errorf("extractor %q validator[%d] must not be nil", extractorKey, validatorIndex) } result, err := validator.Validate(ctx, contracts.ValidationRequest{ Source: doc, Candidates: eligible, + LLMProfile: execution.binding.LLMProfile, + Options: cloneOptions(execution.binding.Options), Metadata: metadata, }) warnings = append(warnings, result.Warnings...) diff --git a/internal/framework/pipeline/runner_test.go b/internal/framework/pipeline/runner_test.go index 69d67ac..50961c1 100644 --- a/internal/framework/pipeline/runner_test.go +++ b/internal/framework/pipeline/runner_test.go @@ -311,6 +311,66 @@ func TestRunPassesInputRequestFields(t *testing.T) { } } +func TestRunPassesModuleBindingConfigToStageRequests(t *testing.T) { + modules := defaultRunnerModules() + pipeline := resolvedPipelineWithValidators("configured") + pipeline.Input = ModuleBinding{Module: "input", LLMProfile: "input-profile", Options: map[string]any{"input_option": "input-value"}} + pipeline.Chunk = ModuleBinding{Module: "chunk", LLMProfile: "chunk-profile", Options: map[string]any{"chunk_option": "chunk-value"}} + pipeline.Output = ModuleBinding{Module: "output", LLMProfile: "output-profile", Options: map[string]any{"output_option": "output-value"}} + pipeline.ArtifactLanes[0].Extract = ModuleBinding{Module: "extract-alpha", LLMProfile: "extract-profile", Options: map[string]any{"extract_option": "extract-value"}} + pipeline.ArtifactLanes[0].Merge = ModuleBinding{Module: "merge", LLMProfile: "merge-profile", Options: map[string]any{"merge_option": "merge-value"}} + pipeline.ArtifactLanes[0].Normalize = ModuleBinding{Module: "normalize", LLMProfile: "normalize-profile", Options: map[string]any{"normalize_option": "normalize-value"}} + pipeline.ArtifactLanes[0].Validators[0] = ModuleBinding{Module: "configured", LLMProfile: "validator-profile", Options: map[string]any{"validator_option": "validator-value"}} + + _, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline}) + if err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + + if got := modules.input.requests[0].LLMProfile; got != "input-profile" { + t.Fatalf("input LLMProfile = %q, want input-profile", got) + } + if got := modules.input.requests[0].Options["input_option"]; got != "input-value" { + t.Fatalf("input Options = %#v, want input option", modules.input.requests[0].Options) + } + if got := modules.chunker.requests[0].LLMProfile; got != "chunk-profile" { + t.Fatalf("chunk LLMProfile = %q, want chunk-profile", got) + } + if got := modules.chunker.requests[0].Options["chunk_option"]; got != "chunk-value" { + t.Fatalf("chunk Options = %#v, want chunk option", modules.chunker.requests[0].Options) + } + if got := modules.extractors["extract-alpha"].requests[0].LLMProfile; got != "extract-profile" { + t.Fatalf("extract LLMProfile = %q, want extract-profile", got) + } + if got := modules.extractors["extract-alpha"].requests[0].Options["extract_option"]; got != "extract-value" { + t.Fatalf("extract Options = %#v, want extract option", modules.extractors["extract-alpha"].requests[0].Options) + } + if got := modules.mergers["merge"].requests[0].LLMProfile; got != "merge-profile" { + t.Fatalf("merge LLMProfile = %q, want merge-profile", got) + } + if got := modules.mergers["merge"].requests[0].Options["merge_option"]; got != "merge-value" { + t.Fatalf("merge Options = %#v, want merge option", modules.mergers["merge"].requests[0].Options) + } + if got := modules.normalizers["normalize"].requests[0].LLMProfile; got != "normalize-profile" { + t.Fatalf("normalize LLMProfile = %q, want normalize-profile", got) + } + if got := modules.normalizers["normalize"].requests[0].Options["normalize_option"]; got != "normalize-value" { + t.Fatalf("normalize Options = %#v, want normalize option", modules.normalizers["normalize"].requests[0].Options) + } + if got := modules.validators["configured"].requests[0].LLMProfile; got != "validator-profile" { + t.Fatalf("validator LLMProfile = %q, want validator-profile", got) + } + if got := modules.validators["configured"].requests[0].Options["validator_option"]; got != "validator-value" { + t.Fatalf("validator Options = %#v, want validator option", modules.validators["configured"].requests[0].Options) + } + if got := modules.output.requests[0].LLMProfile; got != "output-profile" { + t.Fatalf("output LLMProfile = %q, want output-profile", got) + } + if got := modules.output.requests[0].Options["output_option"]; got != "output-value" { + t.Fatalf("output Options = %#v, want output option", modules.output.requests[0].Options) + } +} + func TestRunPassesPerChunkCandidatesToMergeAndNormalize(t *testing.T) { modules := defaultRunnerModules() @@ -346,6 +406,54 @@ func TestRunPassesPerChunkCandidatesToMergeAndNormalize(t *testing.T) { } } +func TestRunRejectsInvalidPostNormalizeCandidateEnvelope(t *testing.T) { + tests := []struct { + name string + candidates []artifacts.ArtifactCandidate + want string + }{ + { + name: "duplicate index", + candidates: []artifacts.ArtifactCandidate{ + runnerCandidate(0), + runnerCandidate(0), + }, + want: "duplicated", + }, + { + name: "missing extractor key", + candidates: []artifacts.ArtifactCandidate{ + {Index: 0, ArtifactType: "artifact", SchemaVersion: "v1", Payload: []byte(`{"value":true}`)}, + }, + want: "extractor_key", + }, + { + name: "mismatched schema version", + candidates: []artifacts.ArtifactCandidate{ + {Index: 0, ExtractorKey: "extract-alpha", ArtifactType: "artifact", SchemaVersion: "other", Payload: []byte(`{"value":true}`)}, + }, + want: "schema_version", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + modules := defaultRunnerModules() + modules.normalizers["normalize"].result = test.candidates + + output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + + assertRunError(t, err, test.want) + if output.Manifest.ValidationStatus != "failed" { + t.Fatalf("ValidationStatus = %q, want failed", output.Manifest.ValidationStatus) + } + if len(output.Approved) != 0 { + t.Fatalf("len(Approved) = %d, want no approved artifacts", len(output.Approved)) + } + }) + } +} + func TestRunValidatorApprovalAndRejection(t *testing.T) { modules := defaultRunnerModules() rejectFirst := &runnerValidator{ @@ -784,6 +892,7 @@ type runnerExtractor struct { validators []contracts.Validator warnings []contracts.Warning err error + requests []contracts.ExtractionRequest seenChunkIDs []string seenLLMClients []contracts.StructuredLLMClient seenMetadata []map[string]any @@ -806,6 +915,7 @@ func (extractor *runnerExtractor) Validators() []contracts.Validator { } func (extractor *runnerExtractor) Extract(ctx context.Context, req contracts.ExtractionRequest) (contracts.ExtractionResult, error) { + extractor.requests = append(extractor.requests, req) if req.Chunk != nil { extractor.seenChunkIDs = append(extractor.seenChunkIDs, req.Chunk.ID) } @@ -880,6 +990,7 @@ type runnerValidator struct { err error order *[]string calls int + requests []contracts.ValidationRequest } func (validator *runnerValidator) Name() string { @@ -888,6 +999,7 @@ func (validator *runnerValidator) Name() string { func (validator *runnerValidator) Validate(ctx context.Context, req contracts.ValidationRequest) (contracts.ValidationResult, error) { validator.calls++ + validator.requests = append(validator.requests, req) if validator.order != nil { *validator.order = append(*validator.order, validator.name) } @@ -981,6 +1093,16 @@ func candidateIndices(candidates []artifacts.ArtifactCandidate) []int { return indices } +func runnerCandidate(index int) artifacts.ArtifactCandidate { + return artifacts.ArtifactCandidate{ + Index: index, + ExtractorKey: "extract-alpha", + ArtifactType: "artifact", + SchemaVersion: "v1", + Payload: []byte(`{"value":true}`), + } +} + func assertRunError(t *testing.T, err error, want string) { t.Helper()