5 Commits

45 changed files with 2752 additions and 139 deletions

3
.codebase-memory/.gitattributes vendored Normal file
View File

@@ -0,0 +1,3 @@
# Auto-generated by codebase-memory-mcp
# Prevent merge conflicts on compressed artifact
graph.db.zst merge=ours binary

View File

@@ -0,0 +1,11 @@
{
"schema_version": 2,
"commit": "9e4b989e53d65efa614b5eedd8530caed12f60b5",
"indexed_at": "2026-07-27T18:27:55Z",
"project": "home-eric-Workspace-notarius",
"nodes": 6356,
"edges": 35132,
"original_size": 26017792,
"compressed_size": 4386397,
"compression_level": 3
}

Binary file not shown.

View File

@@ -211,19 +211,41 @@ Validator bindings accept only **module**, **llm_profile**, and **options**.
They reject **references**, **retries**, and nested **validators**. Deterministic
validators reject an explicit **llm_profile**.
The **json** output module accepts one option:
The **json** output module accepts optional **include_chunk_map** and
**evidence_context** settings:
~~~yaml
output:
module: json
options:
include_chunk_map: true
evidence_context:
enabled: true
window_units: 3
lanes:
- npcs
- spells
~~~
**include_chunk_map** is a boolean and defaults to false. It adds the accepted
chunk map when one exists; its wire format is defined in the
[chunk-map contract](integrations/chunk-map.md).
Omitting **evidence_context** disables evidence publication. When present, it
is an object with these strict fields:
| Field | Type | Rules |
| --- | --- | --- |
| **enabled** | boolean | Required. `false` permits no other evidence fields. |
| **lanes** | array of strings | Required and non-empty when enabled. Each value is trimmed and must be unique; every value must name a configured pipeline lane. |
| **window_units** | non-negative integer | Optional when enabled; defaults to 3. Zero retains only directly cited units. |
Unknown outer or nested option fields are rejected, as are incompatible YAML
types. The allowlist remains valid when a run uses lane filtering: a configured
lane that is not active for that invocation simply contributes no evidence.
Evidence publication is opt-in because it can persist source text and metadata.
Its payload contract is [Published Evidence Context](integrations/evidence-context.md).
## References And Ordered Handoffs
Reference maps bind named slots that the selected target declares. A scalar is

View File

@@ -49,6 +49,14 @@ guessed filename. Before decoding a selected payload, verify its descriptor's
media type and schema identity against the relevant published artifact
contract. The JSON bundle contract links to the available lane contracts.
If `index.json` has an `evidence_context` descriptor, treat it as a
pipeline-wide artifact rather than a lane entry. Verify its six descriptor
fields before decoding the linked file according to the [Published Evidence
Context contract](../integrations/evidence-context.md). Use each
`evidence_refs` entry as the citation to source material. Its surrounding
context range and included units explain the citation, but do not widen or
replace the cited source reference.
A zero exit status may still report rejected outputs, warnings, or absent
lanes. The caller decides which lane IDs are required for its own work and
which are optional; it should make that decision explicitly rather than infer
@@ -61,4 +69,6 @@ Keep the receipt with the published `manifest.json`, and retain
them. Treat the input, output bundle, cache, debug bundle, and captured process
logs as potentially sensitive data. Apply the caller's access controls and
retention policy, and avoid copying secrets into arguments, logs, or
provenance records.
provenance records. An evidence-context artifact contains source-unit text and
metadata, and selected lanes can cover most of an input; preserve and share it
only when that source content is authorized for the recipient.

View File

@@ -0,0 +1,116 @@
# Published Evidence Context
This contract defines the optional `source/evidence-context` artifact emitted
by the production JSON output. Its configuration is owned by
[Configuration](../config.md#module-bindings-and-validators); its logical-file
discovery is owned by [Published JSON Output](json-output.md).
## Identity And Discovery
When enabled, the JSON bundle contains `evidence-context.json` and an
`index.json` `evidence_context` descriptor with the same six fields as other
pipeline-wide artifact descriptors.
| Property | Value |
| --- | --- |
| Artifact kind | `source/evidence-context` |
| Media type | `application/json` |
| Schema ID | `notarius.source.evidence_context` |
| Schema name | `notarius_source_evidence_context_v1` |
| Schema version | `v1` |
| Logical file | `evidence-context.json` |
Consumers must discover the file from the descriptor, verify all six descriptor
fields, and decode only a supported schema version. The descriptor is optional:
its absence means evidence publication was not enabled for that bundle.
## Payload
The v1 payload is a JSON object with required `source_id`, `source_digest`,
`window_units`, `selected_lanes`, and `contexts` fields. `selected_lanes` and
`contexts` are always arrays; an enabled configuration with no accepted direct
evidence publishes `contexts: []`.
```json
{
"source_id": "session-alpha",
"source_digest": "sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
"window_units": 1,
"selected_lanes": ["npcs", "spells"],
"contexts": [
{
"context_ref": {
"source_id": "session-alpha",
"start_unit_id": 10,
"end_unit_id": 20
},
"evidence_refs": [
{
"lane_id": "spells",
"source_ref": {
"source_id": "session-alpha",
"start_unit_id": 10,
"end_unit_id": 10
}
}
],
"units": [
{
"id": 10,
"kind": "transcript_segment",
"text": "Aria casts Cure Wounds.",
"ref": {
"source_id": "session-alpha",
"start_unit_id": 10,
"end_unit_id": 10
}
},
{
"id": 20,
"kind": "transcript_segment",
"text": "The party regroups.",
"ref": {
"source_id": "session-alpha",
"start_unit_id": 20,
"end_unit_id": 20
}
}
]
}
]
}
```
Each context requires `context_ref`, `evidence_refs`, and `units` arrays.
`context_ref` identifies the first and last included unit. Each evidence entry
contains a selected `lane_id` and an original `source_ref`. A unit uses the
existing source-unit shape: required `id`, `kind`, `text`, and self `ref`, plus
optional JSON-object `metadata`. Fixed payload objects reject unknown fields;
unit metadata may contain application-defined JSON values.
## Citations And Context
`evidence_refs` are the authoritative citations. They identify the direct
references emitted by accepted normalized artifacts. `context_ref` and the
units collection include those cited units plus nearby source units selected by
the configured window. They are explanatory context, not widened citations.
Only accepted outputs from the configured lane allowlist contribute. Rejected,
failed, absent, and lane-filtered outputs do not contribute. The artifact never
contains raw input bytes, prompts, model responses, auxiliary reference
content, credentials, or filesystem paths.
## Ordering And Compatibility
The selected lane allowlist is lexical. Contexts and units are in source
document position order, not numeric unit-ID order. Direct evidence entries
are deterministically ordered by lane and source reference. Overlapping or
contiguous windows merge, and each source unit appears at most once in the
resulting contexts.
The artifact is additive to the JSON bundle and is not a lane payload,
normalized-output count, checkpoint, or generated reference. Consumers that
do not need it must tolerate the absent optional descriptor. Consumers that do
use it should preserve the artifact and its schema identity with the run
provenance, and should treat its source text and metadata as sensitive durable
content.

View File

@@ -3,14 +3,14 @@
This document defines the logical JSON bundle emitted by the production JSON
output encoder. The bundles physical destination, atomic publication, and
retention are operational concerns; see [Operations](../operations.md#output-bundles).
Output configuration, including chunk-map export, belongs in
Output configuration, including chunk-map and evidence-context publication, belongs in
[Configuration](../config.md#module-bindings-and-validators).
## Bundle Layout
All paths below are logical, relative, slash-separated bundle paths. The
encoder always emits the first four JSON files below and adds lane or chunk-map
files when their corresponding artifacts are available:
encoder always emits the first four JSON files below and adds lane or
pipeline-wide artifact files when their corresponding artifacts are available:
A subprocess caller first obtains the physical bundle root from the
[run-result receipt](run-result.md), then resolves `index.json` beneath that
@@ -24,6 +24,7 @@ root for the logical discovery described here.
| `warnings.json` | Accepted-output and run warnings. |
| `lanes/<safe-lane-id>.json` | One normalized artifact payload for each lane. |
| `chunk-map.json` | Optional accepted chunk map, when its export is enabled and available. |
| `evidence-context.json` | Optional source-context artifact, when evidence publication is enabled. |
JSON files are pretty-printed with a trailing newline. Lane payloads are
accepted only when their media type is `application/json`.
@@ -49,13 +50,15 @@ normalized lanes has this valid minimal index:
| `rejected_file` | Yes | Always `rejected.json`. |
| `warnings_file` | Yes | Always `warnings.json`. |
| `chunk_map` | No | Descriptor for the pipeline-wide `chunk-map.json`; never a lane descriptor. |
| `evidence_context` | No | Descriptor for the pipeline-wide `evidence-context.json`; never a lane descriptor. |
Each lane descriptor has required `lane_id` and `file`. It may also include
`media_type`, `module_key`, `schema_id`, `schema_name`, and `schema_version`
when supplied by the normalized artifact. A `chunk_map` descriptor contains
`artifact_kind`, `file`, `media_type`, `schema_id`, `schema_name`, and
`schema_version`; its payload is defined by the
[Accepted Chunk Map contract](chunk-map.md).
when supplied by the normalized artifact. Each pipeline-wide artifact
descriptor (`chunk_map` or `evidence_context`) contains `artifact_kind`,
`file`, `media_type`, `schema_id`, `schema_name`, and `schema_version`. Their
payloads are defined by the [Accepted Chunk Map contract](chunk-map.md) and
[Published Evidence Context](evidence-context.md), respectively.
The lane path is derived from its lane ID. Characters outside letters, digits,
periods, underscores, and hyphens become underscores; `..` sequences are

View File

@@ -30,6 +30,14 @@ they need, register each leaf implementation, and add any family-owned assets
or default validator chains. They return contextual errors so production
composition fails at startup rather than at the first run.
An artifact family can register an optional typed evidence projector alongside
its codec. The projector returns defensive copies of the artifact's direct
generic source references and must use the codec's exact Go type. It does not
interpret surrounding context or publish files; the pipeline validates the
capability during preparation and the output boundary owns publication. See
the [Published Evidence Context contract](../integrations/evidence-context.md)
for the durable result.
## Production Composition
Production composition is intentionally split by family:

View File

@@ -52,6 +52,13 @@ checkpoint fingerprints. Missing registrations, incompatible typed entries,
nil implementations, and constructor failures are reported before source
parsing or any stage operation begins.
An output encoder can opt into source-evidence publication through its output
policy. Preparation keeps the configured lane allowlist and active lanes
separate, then verifies an exact typed evidence projector and registered codec
for each active lane. The resulting private plan is immutable; lanes excluded
by invocation filtering remain configured but do not acquire a projector for
that run.
## Typed Lanes And References
Each resolved lane has one artifact kind, codec, and exact Go type. The
@@ -110,11 +117,16 @@ directives consume this same budget and validate any final safe fallback through
the normalizer chain.
After terminal lane work, the runner assembles manifest provenance, normalized
artifacts, rejections, warnings, and an optional accepted chunk map. The output
encoder returns logical files; it does not choose a physical directory. The CLI
publishes those files only after the runner returns without a framework error.
Logical file names and schemas are defined by the
[output integration contracts](../integrations/).
artifacts, rejections, warnings, and an optional accepted chunk map. When an
output policy selected evidence lanes, it decodes accepted serialized normalize
outputs through their registered codecs and invokes the prepared typed
projectors. Rejected or absent lanes contribute nothing. This reconstruction is
also used after normalized-checkpoint reuse, so no second typed output channel
is retained. The runner passes the resulting owned artifact to the output
encoder, which returns logical files and does not choose a physical directory.
The CLI publishes those files only after the runner returns without a framework
error. Logical file names and schemas are defined by the [output integration
contracts](../integrations/).
## Checkpoint And Debug Hooks

View File

@@ -36,7 +36,11 @@ On supported Unix systems, output directories and files are created with
requested modes **0755** and **0644**. Chunk-plan, checkpoint, and debug
directories and files use **0700** and **0600**. The operating system's umask
may impose stricter output modes. Cache and debug roots may contain sensitive
source-derived data, so provision them for one trusted account or service.
source-derived data, so provision them for one trusted account or service. An
output bundle can also contain source content when its JSON output enables
evidence publication. Apply an appropriate umask and output-root access policy
before enabling that option; the requested output modes alone may not be
suitable for transcript-bearing bundles.
## Run Lifecycle
@@ -66,7 +70,13 @@ run directory remains for inspection and is never removed automatically.
Treat an output bundle as durable user data. Do not use cache-cleanup policy to
remove it. An optional accepted chunk map is also durable output and can carry
source- or model-derived annotations; its content and compatibility contract
are defined in [Accepted Chunk Map](integrations/chunk-map.md).
are defined in [Accepted Chunk Map](integrations/chunk-map.md). An optional
[evidence context](integrations/evidence-context.md) contains source-unit text
and metadata. It is not a cache or debug artifact: retain it with the output
bundle only for as long as consumers need it, and apply source-content access
controls to the entire bundle. Selected lanes may collectively cite most of a
transcript, so a broad allowlist can make the evidence artifact nearly as
sensitive and large as the source itself.
## Chunk-Plan Cache

View File

@@ -79,6 +79,14 @@ Pipeline resolution requires a compatible codec and matching kind-specific
variants before a typed lane can be accepted. Framework-owned erasure remains
private and must report type incompatibility as an error rather than a panic.
An artifact kind may additionally provide a typed evidence projection that
copies its direct generic source references. Preparation proves that projection
matches the artifact codec's exact Go type before retaining it for an output
policy. The runner reconstructs evidence only from accepted serialized
normalized artifacts, and the output boundary owns any resulting publication.
Generic framework code never infers evidence by inspecting domain JSON or
depends on domain artifact types.
Auxiliary references provide context or disambiguation. They are not source
evidence and must not be converted into source references.

View File

@@ -2,7 +2,7 @@
## Status
Accepted for implementation.
Implemented.
## Purpose

View File

@@ -2,7 +2,7 @@
## Status
Ready for implementation.
Completed.
## Objective

View File

@@ -28,6 +28,15 @@ pipelines:
module: json
options:
include_chunk_map: true
evidence_context:
enabled: true
window_units: 3
lanes:
- item-events
- npcs
- spells
- combat-turns
- npc-interactions
steps:
# Establish session-wide reference artifacts alongside independent item events.
- id: describe-session

View File

@@ -24,6 +24,7 @@ func newProductionComponents() (productionComponents, error) {
Inputs: pipeline.NewInputAdapterRegistry(),
Chunkers: pipeline.NewChunkerRegistry(),
ArtifactCodecs: pipeline.NewArtifactCodecRegistry(),
ArtifactEvidence: pipeline.NewArtifactEvidenceRegistry(),
Extractors: pipeline.NewExtractorRegistry(),
Mergers: pipeline.NewMergerRegistry(),
Normalizers: pipeline.NewNormalizerRegistry(),
@@ -91,6 +92,7 @@ func catalogFromRegistries(registries pipeline.Registries) pipeline.ModuleCatalo
Inputs: registries.Inputs,
Chunkers: registries.Chunkers,
ArtifactCodecs: registries.ArtifactCodecs,
ArtifactEvidence: registries.ArtifactEvidence,
Extractors: registries.Extractors,
Mergers: registries.Mergers,
Normalizers: registries.Normalizers,
@@ -105,6 +107,7 @@ func registriesFromCatalog(catalog pipeline.ModuleCatalog) pipeline.Registries {
Inputs: catalog.Inputs,
Chunkers: catalog.Chunkers,
ArtifactCodecs: catalog.ArtifactCodecs,
ArtifactEvidence: catalog.ArtifactEvidence,
Extractors: catalog.Extractors,
Mergers: catalog.Mergers,
Normalizers: catalog.Normalizers,
@@ -118,6 +121,7 @@ func isEmptyCatalog(catalog pipeline.ModuleCatalog) bool {
return catalog.Inputs == nil &&
catalog.Chunkers == nil &&
catalog.ArtifactCodecs == nil &&
catalog.ArtifactEvidence == nil &&
catalog.Extractors == nil &&
catalog.Mergers == nil &&
catalog.Normalizers == nil &&
@@ -130,6 +134,7 @@ func isEmptyRegistries(registries pipeline.Registries) bool {
return registries.Inputs == nil &&
registries.Chunkers == nil &&
registries.ArtifactCodecs == nil &&
registries.ArtifactEvidence == nil &&
registries.Extractors == nil &&
registries.Mergers == nil &&
registries.Normalizers == nil &&

View File

@@ -161,8 +161,8 @@ func TestProductionCatalogCoversMaintainedConfigurations(t *testing.T) {
catalog := catalogFromRegistries(registries)
converted := registriesFromCatalog(catalog)
if converted.ArtifactCodecs != registries.ArtifactCodecs || converted.ValidatorChains != registries.ValidatorChains {
t.Fatal("catalog/registry conversion did not preserve codec and validator-chain registries")
if converted.ArtifactCodecs != registries.ArtifactCodecs || converted.ArtifactEvidence != registries.ArtifactEvidence || converted.ValidatorChains != registries.ValidatorChains {
t.Fatal("catalog/registry conversion did not preserve artifact and validator registries")
}
codecSpec, ok := catalog.ArtifactCodecs.Spec(dnd.SpellListKind)
if !ok || codecSpec.Kind != dnd.SpellListKind || codecSpec.Schema.ID != spellcodec.SchemaID {

View File

@@ -292,6 +292,7 @@ type OutputRequest struct {
LLMProfile string `json:"llm_profile,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
ChunkMap *SerializedArtifact `json:"chunk_map,omitempty"`
EvidenceContext *SerializedArtifact `json:"evidence_context,omitempty"`
}
type OutputFile struct {

View File

@@ -0,0 +1,66 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "notarius.source.evidence_context",
"title": "notarius_source_evidence_context_v1",
"type": "object",
"additionalProperties": false,
"required": ["source_id", "source_digest", "window_units", "selected_lanes", "contexts"],
"properties": {
"source_id": {"type": "string", "minLength": 1},
"source_digest": {"type": "string", "pattern": "^sha256:[0-9a-f]{64}$"},
"window_units": {"type": "integer", "minimum": 0},
"selected_lanes": {
"type": "array",
"minItems": 1,
"uniqueItems": true,
"items": {"type": "string", "minLength": 1}
},
"contexts": {
"type": "array",
"items": {"$ref": "#/$defs/context"}
}
},
"$defs": {
"source_ref": {
"type": "object",
"additionalProperties": false,
"required": ["source_id", "start_unit_id", "end_unit_id"],
"properties": {
"source_id": {"type": "string", "minLength": 1},
"start_unit_id": {"type": "integer", "minimum": 1},
"end_unit_id": {"type": "integer", "minimum": 1}
}
},
"unit": {
"type": "object",
"additionalProperties": false,
"required": ["id", "kind", "text", "ref"],
"properties": {
"id": {"type": "integer", "minimum": 1},
"kind": {"type": "string", "minLength": 1},
"text": {"type": "string", "minLength": 1},
"ref": {"$ref": "#/$defs/source_ref"},
"metadata": {"type": "object", "additionalProperties": true}
}
},
"evidence_ref": {
"type": "object",
"additionalProperties": false,
"required": ["lane_id", "source_ref"],
"properties": {
"lane_id": {"type": "string", "minLength": 1},
"source_ref": {"$ref": "#/$defs/source_ref"}
}
},
"context": {
"type": "object",
"additionalProperties": false,
"required": ["context_ref", "evidence_refs", "units"],
"properties": {
"context_ref": {"$ref": "#/$defs/source_ref"},
"evidence_refs": {"type": "array", "minItems": 1, "items": {"$ref": "#/$defs/evidence_ref"}},
"units": {"type": "array", "minItems": 1, "items": {"$ref": "#/$defs/unit"}}
}
}
}
}

View File

@@ -0,0 +1,196 @@
package evidencecontext
import (
"fmt"
"sort"
"strings"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
)
type contribution struct {
laneID string
ref source.SourceRef
startPos int
endPos int
}
type expandedRange struct {
startPos int
endPos int
contributions []contribution
}
// Build validates accepted direct references, expands them by source-document
// position, and returns their deterministic context union.
func Build(request BuildRequest) (Document, error) {
if request.WindowUnits < 0 {
return Document{}, fmt.Errorf("window_units must not be negative")
}
lanes, err := normalizeSelectedLanes(request.SelectedLanes)
if err != nil {
return Document{}, err
}
if err := source.ValidateDocument(request.Source); err != nil {
return Document{}, fmt.Errorf("validate source document: %w", err)
}
digest, err := source.DigestDocument(request.Source)
if err != nil {
return Document{}, fmt.Errorf("digest source document: %w", err)
}
if digest != request.Source.Digest {
return Document{}, fmt.Errorf("source digest does not match source document digest")
}
selected := make(map[string]struct{}, len(lanes))
for _, laneID := range lanes {
selected[laneID] = struct{}{}
}
index := source.NewDocumentIndex(request.Source)
seen := make(map[evidenceKey]struct{})
contributions := make([]contribution, 0)
for laneIndex, laneEvidence := range request.LaneEvidence {
laneID := strings.TrimSpace(laneEvidence.LaneID)
if _, ok := selected[laneID]; !ok {
return Document{}, fmt.Errorf("lane evidence[%d] lane %q is not selected", laneIndex, laneID)
}
for refIndex, ref := range laneEvidence.SourceRefs {
if err := index.ValidateRef(ref); err != nil {
return Document{}, fmt.Errorf("lane %q source reference[%d]: %w", laneID, refIndex, err)
}
key := evidenceKey{laneID: laneID, ref: ref}
if _, exists := seen[key]; exists {
continue
}
seen[key] = struct{}{}
startPos, _ := index.Position(ref.StartUnitID)
endPos, _ := index.Position(ref.EndUnitID)
contributions = append(contributions, contribution{laneID: laneID, ref: ref, startPos: expandStart(startPos, request.WindowUnits), endPos: expandEnd(endPos, len(request.Source.Units), request.WindowUnits)})
}
}
sort.Slice(contributions, func(i, j int) bool { return lessContribution(contributions[i], contributions[j]) })
document := Document{
SourceID: request.Source.ID,
SourceDigest: digest,
WindowUnits: request.WindowUnits,
SelectedLanes: lanes,
Contexts: make([]Context, 0),
}
for _, rangeValue := range mergeRanges(contributions) {
context, err := buildContext(request.Source, rangeValue)
if err != nil {
return Document{}, err
}
document.Contexts = append(document.Contexts, context)
}
canonical, err := canonicalize(document)
if err != nil {
return Document{}, fmt.Errorf("validate evidence context: %w", err)
}
return clone(canonical)
}
type evidenceKey struct {
laneID string
ref source.SourceRef
}
func normalizeSelectedLanes(values []string) ([]string, error) {
if len(values) == 0 {
return nil, fmt.Errorf("selected_lanes must not be empty")
}
seen := make(map[string]struct{}, len(values))
lanes := make([]string, 0, len(values))
for index, raw := range values {
laneID := strings.TrimSpace(raw)
if laneID == "" {
return nil, fmt.Errorf("selected_lanes[%d] must not be empty", index)
}
if _, exists := seen[laneID]; exists {
return nil, fmt.Errorf("selected_lanes lane %q is duplicated", laneID)
}
seen[laneID] = struct{}{}
lanes = append(lanes, laneID)
}
sort.Strings(lanes)
return lanes, nil
}
func expandStart(position, window int) int {
if window > position {
return 0
}
return position - window
}
func expandEnd(position, length, window int) int {
last := length - 1
if window > last-position {
return last
}
return position + window
}
func lessContribution(left, right contribution) bool {
if left.startPos != right.startPos {
return left.startPos < right.startPos
}
if left.endPos != right.endPos {
return left.endPos < right.endPos
}
return lessEvidenceRef(EvidenceRef{LaneID: left.laneID, SourceRef: left.ref}, EvidenceRef{LaneID: right.laneID, SourceRef: right.ref})
}
func mergeRanges(values []contribution) []expandedRange {
if len(values) == 0 {
return nil
}
ranges := make([]expandedRange, 0, len(values))
for _, value := range values {
if len(ranges) == 0 || value.startPos > ranges[len(ranges)-1].endPos+1 {
ranges = append(ranges, expandedRange{startPos: value.startPos, endPos: value.endPos, contributions: []contribution{value}})
continue
}
current := &ranges[len(ranges)-1]
if value.endPos > current.endPos {
current.endPos = value.endPos
}
current.contributions = append(current.contributions, value)
}
return ranges
}
func buildContext(document *source.SourceDocument, value expandedRange) (Context, error) {
evidenceRefs := make([]EvidenceRef, 0, len(value.contributions))
for _, contribution := range value.contributions {
evidenceRefs = append(evidenceRefs, EvidenceRef{LaneID: contribution.laneID, SourceRef: contribution.ref})
}
sort.Slice(evidenceRefs, func(i, j int) bool { return lessEvidenceRef(evidenceRefs[i], evidenceRefs[j]) })
units := make([]source.SourceUnit, 0, value.endPos-value.startPos+1)
for position := value.startPos; position <= value.endPos; position++ {
unit, err := cloneSourceUnit(document.Units[position])
if err != nil {
return Context{}, fmt.Errorf("clone source unit at position %d: %w", position, err)
}
units = append(units, unit)
}
return Context{
ContextRef: source.SourceRef{SourceID: document.ID, StartUnitID: units[0].ID, EndUnitID: units[len(units)-1].ID},
EvidenceRefs: evidenceRefs,
Units: units,
}, nil
}
func lessEvidenceRef(left, right EvidenceRef) bool {
if left.LaneID != right.LaneID {
return left.LaneID < right.LaneID
}
if left.SourceRef.SourceID != right.SourceRef.SourceID {
return left.SourceRef.SourceID < right.SourceRef.SourceID
}
if left.SourceRef.StartUnitID != right.SourceRef.StartUnitID {
return left.SourceRef.StartUnitID < right.SourceRef.StartUnitID
}
return left.SourceRef.EndUnitID < right.SourceRef.EndUnitID
}

View File

@@ -0,0 +1,329 @@
package evidencecontext
import (
"bytes"
"embed"
"encoding/json"
"fmt"
"io"
"regexp"
"strings"
"sync"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"github.com/santhosh-tekuri/jsonschema/v6"
)
//go:embed assets/schemas/source_evidence_context.v1.json
var schemaAssets embed.FS
var digestPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
var (
loadSchemaOnce sync.Once
loadedSchema []byte
compiledSchema *jsonschema.Schema
loadSchemaErr error
)
// Codec owns strict serialization for the durable evidence-context contract.
type Codec struct{}
func New() *Codec { return &Codec{} }
func (c *Codec) Kind() contracts.ArtifactKind { return ArtifactKind }
func (c *Codec) Schema() contracts.ArtifactSchema {
raw, err := c.schemaBytes()
if err != nil {
return contracts.ArtifactSchema{}
}
return contracts.ArtifactSchema{ID: SchemaID, Name: SchemaName, Version: SchemaVersion, JSONSchema: raw}
}
func (c *Codec) MediaType() string { return MediaType }
// Serialize builds and encodes the framework-owned serialized artifact.
func Serialize(request BuildRequest) (contracts.SerializedArtifact, error) {
value, err := Build(request)
if err != nil {
return contracts.SerializedArtifact{}, err
}
codec := New()
content, err := codec.Encode(value)
if err != nil {
return contracts.SerializedArtifact{}, err
}
return contracts.SerializedArtifact{Kind: ArtifactKind, Schema: codec.Schema(), MediaType: MediaType, Content: content}, nil
}
func (c *Codec) Encode(value Document) ([]byte, error) {
if _, err := c.schemaBytes(); err != nil {
return nil, err
}
canonical, err := canonicalize(value)
if err != nil {
return nil, fmt.Errorf("encode evidence context: %w", err)
}
content, err := json.Marshal(canonical)
if err != nil {
return nil, fmt.Errorf("encode evidence context: %w", err)
}
if err := validateSchemaInstance(content); err != nil {
return nil, fmt.Errorf("encode evidence context: %w", err)
}
return content, nil
}
func (c *Codec) Decode(content []byte) (Document, error) {
if _, err := c.schemaBytes(); err != nil {
return Document{}, err
}
if err := validateSchemaInstance(content); err != nil {
return Document{}, fmt.Errorf("decode evidence context: %w", err)
}
decoder := json.NewDecoder(bytes.NewReader(content))
decoder.DisallowUnknownFields()
var value Document
if err := decoder.Decode(&value); err != nil {
return Document{}, fmt.Errorf("decode evidence context: %w", err)
}
var trailing any
if err := decoder.Decode(&trailing); err != io.EOF {
return Document{}, fmt.Errorf("decode evidence context: multiple JSON values")
}
canonical, err := canonicalize(value)
if err != nil {
return Document{}, fmt.Errorf("decode evidence context: %w", err)
}
return clone(canonical)
}
func (c *Codec) schemaBytes() ([]byte, error) {
loadSchemaOnce.Do(loadAndCompileSchema)
if loadSchemaErr != nil {
return nil, loadSchemaErr
}
return append([]byte(nil), loadedSchema...), nil
}
func loadAndCompileSchema() {
raw, err := schemaAssets.ReadFile("assets/schemas/source_evidence_context.v1.json")
if err != nil {
loadSchemaErr = fmt.Errorf("read source evidence context schema: %w", err)
return
}
var identity struct {
ID string `json:"$id"`
Title string `json:"title"`
Type string `json:"type"`
Required []string `json:"required"`
}
if err := json.Unmarshal(raw, &identity); err != nil {
loadSchemaErr = fmt.Errorf("decode source evidence context schema: %w", err)
return
}
if identity.ID != SchemaID || identity.Title != SchemaName || identity.Type != "object" || !hasRequiredFields(identity.Required) {
loadSchemaErr = fmt.Errorf("source evidence context schema identity or required fields are invalid")
return
}
schemaDocument, err := jsonschema.UnmarshalJSON(bytes.NewReader(raw))
if err != nil {
loadSchemaErr = fmt.Errorf("parse source evidence context schema: %w", err)
return
}
compiler := jsonschema.NewCompiler()
if err := compiler.AddResource("source-evidence-context-schema.json", schemaDocument); err != nil {
loadSchemaErr = fmt.Errorf("load source evidence context schema: %w", err)
return
}
compiled, err := compiler.Compile("source-evidence-context-schema.json")
if err != nil {
loadSchemaErr = fmt.Errorf("compile source evidence context schema: %w", err)
return
}
loadedSchema = append([]byte(nil), raw...)
compiledSchema = compiled
}
func validateSchemaInstance(content []byte) error {
instance, err := jsonschema.UnmarshalJSON(bytes.NewReader(content))
if err != nil {
return fmt.Errorf("payload is not valid JSON: %w", err)
}
if err := compiledSchema.Validate(instance); err != nil {
return fmt.Errorf("payload does not conform to source evidence context schema: %w", err)
}
return nil
}
func hasRequiredFields(required []string) bool {
want := map[string]bool{"source_id": true, "source_digest": true, "window_units": true, "selected_lanes": true, "contexts": true}
for _, field := range required {
delete(want, field)
}
return len(want) == 0
}
func canonicalize(value Document) (Document, error) {
owned, err := clone(value)
if err != nil {
return Document{}, err
}
value = owned
if err := requireIdentity("source_id", value.SourceID); err != nil {
return Document{}, err
}
if !digestPattern.MatchString(value.SourceDigest) {
return Document{}, fmt.Errorf("source_digest must be a sha256 digest")
}
if value.WindowUnits < 0 {
return Document{}, fmt.Errorf("window_units must not be negative")
}
if err := validateSelectedLanes(value.SelectedLanes); err != nil {
return Document{}, err
}
if value.Contexts == nil {
value.Contexts = make([]Context, 0)
}
selected := make(map[string]struct{}, len(value.SelectedLanes))
for _, laneID := range value.SelectedLanes {
selected[laneID] = struct{}{}
}
seenUnits := make(map[int]struct{})
for contextIndex := range value.Contexts {
context, err := canonicalizeContext(value.SourceID, selected, seenUnits, value.Contexts[contextIndex], contextIndex)
if err != nil {
return Document{}, err
}
value.Contexts[contextIndex] = context
}
return value, nil
}
func validateSelectedLanes(lanes []string) error {
if len(lanes) == 0 {
return fmt.Errorf("selected_lanes must not be empty")
}
for index, laneID := range lanes {
if err := requireIdentity(fmt.Sprintf("selected_lanes[%d]", index), laneID); err != nil {
return err
}
if index > 0 && lanes[index-1] >= laneID {
return fmt.Errorf("selected_lanes must be unique and in lexical order")
}
}
return nil
}
func canonicalizeContext(sourceID string, selected map[string]struct{}, seenUnits map[int]struct{}, value Context, contextIndex int) (Context, error) {
prefix := fmt.Sprintf("contexts[%d]", contextIndex)
if len(value.EvidenceRefs) == 0 {
return Context{}, fmt.Errorf("%s.evidence_refs must not be empty", prefix)
}
if len(value.Units) == 0 {
return Context{}, fmt.Errorf("%s.units must not be empty", prefix)
}
if err := validateRefIdentity(sourceID, value.ContextRef, prefix+".context_ref"); err != nil {
return Context{}, err
}
positions := make(map[int]int, len(value.Units))
for unitIndex := range value.Units {
unit, err := cloneSourceUnit(value.Units[unitIndex])
if err != nil {
return Context{}, fmt.Errorf("%s.units[%d]: %w", prefix, unitIndex, err)
}
if unit.ID <= 0 || strings.TrimSpace(unit.Kind) == "" || strings.TrimSpace(unit.Text) == "" {
return Context{}, fmt.Errorf("%s.units[%d] has invalid required fields", prefix, unitIndex)
}
if err := validateRefIdentity(sourceID, unit.Ref, fmt.Sprintf("%s.units[%d].ref", prefix, unitIndex)); err != nil {
return Context{}, err
}
if unit.Ref.StartUnitID != unit.ID || unit.Ref.EndUnitID != unit.ID {
return Context{}, fmt.Errorf("%s.units[%d].ref must identify unit id %d", prefix, unitIndex, unit.ID)
}
if _, exists := positions[unit.ID]; exists {
return Context{}, fmt.Errorf("%s.units contains duplicate unit id %d", prefix, unit.ID)
}
if _, exists := seenUnits[unit.ID]; exists {
return Context{}, fmt.Errorf("contexts contain duplicate unit id %d", unit.ID)
}
positions[unit.ID] = unitIndex
seenUnits[unit.ID] = struct{}{}
value.Units[unitIndex] = unit
}
if value.ContextRef.StartUnitID != value.Units[0].ID || value.ContextRef.EndUnitID != value.Units[len(value.Units)-1].ID {
return Context{}, fmt.Errorf("%s.context_ref must identify the first and last units", prefix)
}
for evidenceIndex := range value.EvidenceRefs {
evidence := value.EvidenceRefs[evidenceIndex]
if _, ok := selected[evidence.LaneID]; !ok {
return Context{}, fmt.Errorf("%s.evidence_refs[%d].lane_id is not selected", prefix, evidenceIndex)
}
if err := requireIdentity(fmt.Sprintf("%s.evidence_refs[%d].lane_id", prefix, evidenceIndex), evidence.LaneID); err != nil {
return Context{}, err
}
if err := validateRefIdentity(sourceID, evidence.SourceRef, fmt.Sprintf("%s.evidence_refs[%d].source_ref", prefix, evidenceIndex)); err != nil {
return Context{}, err
}
start, startOK := positions[evidence.SourceRef.StartUnitID]
end, endOK := positions[evidence.SourceRef.EndUnitID]
if !startOK || !endOK || start > end {
return Context{}, fmt.Errorf("%s.evidence_refs[%d].source_ref is outside context units", prefix, evidenceIndex)
}
if evidenceIndex > 0 && !lessEvidenceRef(value.EvidenceRefs[evidenceIndex-1], evidence) {
return Context{}, fmt.Errorf("%s.evidence_refs must be unique and in deterministic order", prefix)
}
}
return value, nil
}
func validateRefIdentity(sourceID string, ref source.SourceRef, field string) error {
if ref.SourceID != sourceID {
return fmt.Errorf("%s.source_id does not match source_id", field)
}
if ref.StartUnitID <= 0 || ref.EndUnitID <= 0 {
return fmt.Errorf("%s endpoints must be positive", field)
}
return nil
}
func requireIdentity(field, value string) error {
if strings.TrimSpace(value) == "" || strings.TrimSpace(value) != value {
return fmt.Errorf("%s must be a non-empty trimmed string", field)
}
return nil
}
func clone(value Document) (Document, error) {
value.SelectedLanes = append([]string(nil), value.SelectedLanes...)
if value.Contexts == nil {
value.Contexts = make([]Context, 0)
} else {
contexts := make([]Context, len(value.Contexts))
for contextIndex, context := range value.Contexts {
contexts[contextIndex].ContextRef = context.ContextRef
contexts[contextIndex].EvidenceRefs = append([]EvidenceRef(nil), context.EvidenceRefs...)
contexts[contextIndex].Units = make([]source.SourceUnit, len(context.Units))
for unitIndex, unit := range context.Units {
cloned, err := cloneSourceUnit(unit)
if err != nil {
return Document{}, fmt.Errorf("clone contexts[%d].units[%d]: %w", contextIndex, unitIndex, err)
}
contexts[contextIndex].Units[unitIndex] = cloned
}
}
value.Contexts = contexts
}
return value, nil
}
func cloneSourceUnit(unit source.SourceUnit) (source.SourceUnit, error) {
metadata, err := source.CloneMetadata(unit.Metadata)
if err != nil {
return source.SourceUnit{}, fmt.Errorf("clone metadata: %w", err)
}
unit.Metadata = metadata
return unit, nil
}

View File

@@ -0,0 +1,350 @@
package evidencecontext
import (
"bytes"
"encoding/json"
"math"
"os"
"reflect"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
)
func TestBuildExpandsAndMergesEvidenceByDocumentPosition(t *testing.T) {
for _, test := range []struct {
name string
window int
evidence []LaneEvidence
wantUnits [][]int
wantRefs [][]EvidenceRef
}{
{
name: "zero window",
evidence: []LaneEvidence{{LaneID: "npcs", SourceRefs: []source.SourceRef{ref(3, 3)}}},
wantUnits: [][]int{{3}},
wantRefs: [][]EvidenceRef{{{LaneID: "npcs", SourceRef: ref(3, 3)}}},
},
{
name: "non monotonic ids use positions and clip boundaries",
window: 1,
evidence: []LaneEvidence{{LaneID: "npcs", SourceRefs: []source.SourceRef{ref(3, 3)}}},
wantUnits: [][]int{{10, 3, 30}},
wantRefs: [][]EvidenceRef{{{LaneID: "npcs", SourceRef: ref(3, 3)}}},
},
{
name: "separate gaps stay separate",
evidence: []LaneEvidence{{LaneID: "npcs", SourceRefs: []source.SourceRef{ref(10, 10), ref(50, 50)}}},
wantUnits: [][]int{{10}, {50}},
wantRefs: [][]EvidenceRef{{{LaneID: "npcs", SourceRef: ref(10, 10)}}, {{LaneID: "npcs", SourceRef: ref(50, 50)}}},
},
{
name: "overlapping windows merge",
window: 1,
evidence: []LaneEvidence{{LaneID: "npcs", SourceRefs: []source.SourceRef{ref(3, 3), ref(30, 30)}}},
wantUnits: [][]int{{10, 3, 30, 7}},
wantRefs: [][]EvidenceRef{{{LaneID: "npcs", SourceRef: ref(3, 3)}, {LaneID: "npcs", SourceRef: ref(30, 30)}}},
},
{
name: "contiguous windows merge",
window: 1,
evidence: []LaneEvidence{{LaneID: "npcs", SourceRefs: []source.SourceRef{ref(10, 10), ref(7, 7)}}},
wantUnits: [][]int{{10, 3, 30, 7, 50}},
wantRefs: [][]EvidenceRef{{{LaneID: "npcs", SourceRef: ref(7, 7)}, {LaneID: "npcs", SourceRef: ref(10, 10)}}},
},
{
name: "duplicate contributions retain unique lane attribution",
evidence: []LaneEvidence{
{LaneID: "spells", SourceRefs: []source.SourceRef{ref(30, 30), ref(30, 30)}},
{LaneID: "npcs", SourceRefs: []source.SourceRef{ref(30, 30)}},
},
wantUnits: [][]int{{30}},
wantRefs: [][]EvidenceRef{{{LaneID: "npcs", SourceRef: ref(30, 30)}, {LaneID: "spells", SourceRef: ref(30, 30)}}},
},
{
name: "empty contributions retain explicit empty contexts",
evidence: []LaneEvidence{{LaneID: "npcs"}},
wantUnits: [][]int{},
wantRefs: [][]EvidenceRef{},
},
{
name: "largest window clips without overflow",
window: math.MaxInt,
evidence: []LaneEvidence{{LaneID: "npcs", SourceRefs: []source.SourceRef{ref(30, 30)}}},
wantUnits: [][]int{{10, 3, 30, 7, 50}},
wantRefs: [][]EvidenceRef{{{LaneID: "npcs", SourceRef: ref(30, 30)}}},
},
} {
t.Run(test.name, func(t *testing.T) {
document := testDocument(t)
got, err := Build(BuildRequest{Source: document, WindowUnits: test.window, SelectedLanes: []string{"spells", "npcs"}, LaneEvidence: test.evidence})
if err != nil {
t.Fatalf("Build() error = %v", err)
}
if want := []string{"npcs", "spells"}; !reflect.DeepEqual(got.SelectedLanes, want) {
t.Fatalf("SelectedLanes = %#v, want %#v", got.SelectedLanes, want)
}
if got.WindowUnits != test.window || got.SourceID != document.ID || got.SourceDigest != document.Digest {
t.Fatalf("Build() identity = %#v, want source and window identity", got)
}
if actual := contextUnitIDs(got.Contexts); !reflect.DeepEqual(actual, test.wantUnits) {
t.Fatalf("context unit ids = %#v, want %#v", actual, test.wantUnits)
}
if actual := contextEvidenceRefs(got.Contexts); !reflect.DeepEqual(actual, test.wantRefs) {
t.Fatalf("context evidence refs = %#v, want %#v", actual, test.wantRefs)
}
})
}
}
func TestBuildIsStableAndOwnsSourceAndInputs(t *testing.T) {
document := testDocument(t)
refs := []source.SourceRef{ref(30, 30), ref(3, 3)}
request := BuildRequest{
Source: document,
WindowUnits: 1,
SelectedLanes: []string{"spells", "npcs"},
LaneEvidence: []LaneEvidence{{LaneID: "spells", SourceRefs: refs}, {LaneID: "npcs", SourceRefs: []source.SourceRef{ref(3, 3)}}},
}
first, err := Build(request)
if err != nil {
t.Fatal(err)
}
secondRequest := request
secondRequest.LaneEvidence = []LaneEvidence{{LaneID: "npcs", SourceRefs: []source.SourceRef{ref(3, 3)}}, {LaneID: "spells", SourceRefs: []source.SourceRef{ref(3, 3), ref(30, 30)}}}
second, err := Build(secondRequest)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(first, second) {
t.Fatalf("Build() order differs:\nfirst: %#v\nsecond: %#v", first, second)
}
first.SelectedLanes[0] = "changed"
first.Contexts[0].Units[0].Metadata["nested"].(map[string]any)["value"] = "changed"
if document.Units[0].Metadata["nested"].(map[string]any)["value"] != "original" {
t.Fatal("Build() returned metadata aliases to source document")
}
document.Units[0].Metadata["nested"].(map[string]any)["value"] = "later"
if second.Contexts[0].Units[0].Metadata["nested"].(map[string]any)["value"] != "original" {
t.Fatal("Build() retained metadata aliases to source document")
}
refs[0].StartUnitID = 999
if !containsEvidenceRef(second.Contexts[0].EvidenceRefs, ref(30, 30)) {
t.Fatal("Build() retained source-reference input aliases")
}
}
func TestBuildRejectsInvalidInputs(t *testing.T) {
for _, test := range []struct {
name string
mutate func(*BuildRequest)
want string
}{
{name: "negative window", mutate: func(request *BuildRequest) { request.WindowUnits = -1 }, want: "window_units"},
{name: "blank selected lane", mutate: func(request *BuildRequest) { request.SelectedLanes = []string{" "} }, want: "selected_lanes"},
{name: "duplicate selected lane", mutate: func(request *BuildRequest) { request.SelectedLanes = []string{"npcs", " npcs "} }, want: "duplicated"},
{name: "unselected contribution", mutate: func(request *BuildRequest) {
request.LaneEvidence = []LaneEvidence{{LaneID: "other", SourceRefs: []source.SourceRef{ref(3, 3)}}}
}, want: "not selected"},
{name: "source digest mismatch", mutate: func(request *BuildRequest) { request.Source.Digest = "sha256:" + strings.Repeat("0", 64) }, want: "does not match"},
{name: "invalid reference", mutate: func(request *BuildRequest) {
request.LaneEvidence = []LaneEvidence{{LaneID: "npcs", SourceRefs: []source.SourceRef{ref(99, 99)}}}
}, want: "source reference[0]"},
} {
t.Run(test.name, func(t *testing.T) {
request := BuildRequest{Source: testDocument(t), SelectedLanes: []string{"npcs"}, LaneEvidence: []LaneEvidence{{LaneID: "npcs", SourceRefs: []source.SourceRef{ref(3, 3)}}}}
test.mutate(&request)
if _, err := Build(request); err == nil || !strings.Contains(err.Error(), test.want) {
t.Fatalf("Build() error = %v, want %q", err, test.want)
}
})
}
}
func TestCodecRoundTripsCompactFixtureAndOwnsDecodedValues(t *testing.T) {
fixture, err := os.ReadFile("testdata/source_evidence_context.v1.json")
if err != nil {
t.Fatal(err)
}
codec := New()
value, err := codec.Decode(fixture)
if err != nil {
t.Fatalf("Decode(fixture) error = %v", err)
}
encoded, err := codec.Encode(value)
if err != nil {
t.Fatal(err)
}
if !bytes.Equal(encoded, bytes.TrimSpace(fixture)) {
t.Fatalf("fixture does not use canonical encoding\nwant: %s\n got: %s", fixture, encoded)
}
value.Contexts[0].Units[0].Text = "changed"
decoded, err := codec.Decode(fixture)
if err != nil {
t.Fatal(err)
}
if decoded.Contexts[0].Units[0].Text != "The party meets Rowan." {
t.Fatal("Decode() reused mutable document storage")
}
built, err := Build(BuildRequest{Source: testDocument(t), WindowUnits: 1, SelectedLanes: []string{"npcs"}, LaneEvidence: []LaneEvidence{{LaneID: "npcs", SourceRefs: []source.SourceRef{ref(3, 3)}}}})
if err != nil {
t.Fatal(err)
}
content, err := codec.Encode(built)
if err != nil {
t.Fatal(err)
}
first, err := codec.Decode(content)
if err != nil {
t.Fatal(err)
}
second, err := codec.Decode(content)
if err != nil {
t.Fatal(err)
}
first.Contexts[0].Units[0].Metadata["nested"].(map[string]any)["value"] = "changed"
if second.Contexts[0].Units[0].Metadata["nested"].(map[string]any)["value"] != "original" {
t.Fatal("Decode() returned metadata aliases")
}
}
func TestCodecRejectsInvalidDurableBoundaries(t *testing.T) {
value, err := Build(BuildRequest{Source: testDocument(t), SelectedLanes: []string{"npcs"}, LaneEvidence: []LaneEvidence{{LaneID: "npcs", SourceRefs: []source.SourceRef{ref(3, 3)}}}})
if err != nil {
t.Fatal(err)
}
for _, test := range []struct {
name string
mutate func(*Document)
}{
{name: "unsorted lanes", mutate: func(value *Document) { value.SelectedLanes = []string{"z", "a"} }},
{name: "context range mismatch", mutate: func(value *Document) { value.Contexts[0].ContextRef.EndUnitID = 999 }},
{name: "mismatched evidence source", mutate: func(value *Document) { value.Contexts[0].EvidenceRefs[0].SourceRef.SourceID = "other" }},
{name: "invalid evidence range", mutate: func(value *Document) {
value.Contexts[0].EvidenceRefs[0].SourceRef.StartUnitID = 10
}},
{name: "duplicate context unit", mutate: func(value *Document) { value.Contexts = append(value.Contexts, value.Contexts[0]) }},
} {
t.Run(test.name, func(t *testing.T) {
candidate, err := clone(value)
if err != nil {
t.Fatal(err)
}
test.mutate(&candidate)
if _, err := New().Encode(candidate); err == nil {
t.Fatal("Encode() error = nil, want durable model rejection")
}
})
}
content, err := New().Encode(value)
if err != nil {
t.Fatal(err)
}
for _, test := range []struct {
name string
mutate func(map[string]any)
}{
{name: "missing contexts", mutate: func(value map[string]any) { delete(value, "contexts") }},
{name: "null contexts", mutate: func(value map[string]any) { value["contexts"] = nil }},
{name: "unknown fixed field", mutate: func(value map[string]any) { value["unknown"] = true }},
{name: "missing units", mutate: func(value map[string]any) { delete(contextObject(value, 0), "units") }},
{name: "null evidence refs", mutate: func(value map[string]any) { contextObject(value, 0)["evidence_refs"] = nil }},
} {
t.Run(test.name, func(t *testing.T) {
raw := decodeJSON(t, content)
test.mutate(raw)
mutated, err := json.Marshal(raw)
if err != nil {
t.Fatal(err)
}
if _, err := New().Decode(mutated); err == nil {
t.Fatal("Decode() error = nil, want strict payload rejection")
}
})
}
if _, err := New().Decode(append(content, []byte(" {}")...)); err == nil {
t.Fatal("Decode() error = nil, want trailing JSON rejection")
}
}
func TestSerializeUsesFixedArtifactIdentity(t *testing.T) {
artifact, err := Serialize(BuildRequest{Source: testDocument(t), SelectedLanes: []string{"npcs"}})
if err != nil {
t.Fatal(err)
}
if artifact.Kind != ArtifactKind || artifact.MediaType != MediaType || artifact.Schema.ID != SchemaID || artifact.Schema.Name != SchemaName || artifact.Schema.Version != SchemaVersion {
t.Fatalf("Serialize() = %#v, want fixed artifact identity", artifact)
}
decoded, err := New().Decode(artifact.Content)
if err != nil || len(decoded.Contexts) != 0 || decoded.Contexts == nil {
t.Fatalf("Decode(Serialize()) = %#v, %v; want explicit empty contexts", decoded, err)
}
}
func testDocument(t *testing.T) *source.SourceDocument {
t.Helper()
document := &source.SourceDocument{
ID: "session", Kind: "transcript", Format: "application/json",
Units: []source.SourceUnit{
{ID: 10, Kind: "segment", Text: "first", Ref: ref(10, 10), Metadata: map[string]any{"nested": map[string]any{"value": "original"}}},
{ID: 3, Kind: "segment", Text: "second", Ref: ref(3, 3)},
{ID: 30, Kind: "segment", Text: "third", Ref: ref(30, 30)},
{ID: 7, Kind: "segment", Text: "fourth", Ref: ref(7, 7)},
{ID: 50, Kind: "segment", Text: "fifth", Ref: ref(50, 50)},
},
}
digest, err := source.DigestDocument(document)
if err != nil {
t.Fatal(err)
}
document.Digest = digest
return document
}
func ref(start, end int) source.SourceRef {
return source.SourceRef{SourceID: "session", StartUnitID: start, EndUnitID: end}
}
func contextUnitIDs(contexts []Context) [][]int {
values := make([][]int, len(contexts))
for index, context := range contexts {
values[index] = make([]int, len(context.Units))
for unitIndex, unit := range context.Units {
values[index][unitIndex] = unit.ID
}
}
return values
}
func contextEvidenceRefs(contexts []Context) [][]EvidenceRef {
values := make([][]EvidenceRef, len(contexts))
for index, context := range contexts {
values[index] = append([]EvidenceRef(nil), context.EvidenceRefs...)
}
return values
}
func containsEvidenceRef(values []EvidenceRef, want source.SourceRef) bool {
for _, value := range values {
if value.SourceRef == want {
return true
}
}
return false
}
func decodeJSON(t *testing.T, content []byte) map[string]any {
t.Helper()
decoder := json.NewDecoder(bytes.NewReader(content))
decoder.UseNumber()
var value map[string]any
if err := decoder.Decode(&value); err != nil {
t.Fatal(err)
}
return value
}
func contextObject(value map[string]any, index int) map[string]any {
return value["contexts"].([]any)[index].(map[string]any)
}

View File

@@ -0,0 +1,50 @@
// Package evidencecontext owns the durable source evidence-context contract.
package evidencecontext
import (
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
const (
ArtifactKind contracts.ArtifactKind = "source/evidence-context"
SchemaID = "notarius.source.evidence_context"
SchemaName = "notarius_source_evidence_context_v1"
SchemaVersion = "v1"
MediaType = "application/json"
)
// Document is the durable union of direct evidence and surrounding source
// context selected for one accepted source document.
type Document struct {
SourceID string `json:"source_id"`
SourceDigest string `json:"source_digest"`
WindowUnits int `json:"window_units"`
SelectedLanes []string `json:"selected_lanes"`
Contexts []Context `json:"contexts"`
}
type Context struct {
ContextRef source.SourceRef `json:"context_ref"`
EvidenceRefs []EvidenceRef `json:"evidence_refs"`
Units []source.SourceUnit `json:"units"`
}
type EvidenceRef struct {
LaneID string `json:"lane_id"`
SourceRef source.SourceRef `json:"source_ref"`
}
// LaneEvidence attributes direct source references to one selected lane.
type LaneEvidence struct {
LaneID string `json:"lane_id"`
SourceRefs []source.SourceRef `json:"source_refs"`
}
// BuildRequest supplies accepted source material and direct lane evidence.
type BuildRequest struct {
Source *source.SourceDocument
WindowUnits int
SelectedLanes []string
LaneEvidence []LaneEvidence
}

View File

@@ -0,0 +1 @@
{"source_id":"session-alpha","source_digest":"sha256:0000000000000000000000000000000000000000000000000000000000000000","window_units":0,"selected_lanes":["npcs"],"contexts":[{"context_ref":{"source_id":"session-alpha","start_unit_id":13,"end_unit_id":13},"evidence_refs":[{"lane_id":"npcs","source_ref":{"source_id":"session-alpha","start_unit_id":13,"end_unit_id":13}}],"units":[{"id":13,"kind":"transcript_segment","text":"The party meets Rowan.","ref":{"source_id":"session-alpha","start_unit_id":13,"end_unit_id":13}}]}]}

View File

@@ -0,0 +1,148 @@
package pipeline
import (
"fmt"
"reflect"
"sort"
"strings"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
// ArtifactEvidenceProjector returns the direct source references represented
// by one normalized artifact value.
type ArtifactEvidenceProjector[T any] func(T) []source.SourceRef
// ArtifactEvidenceRegistry keeps the typed projection boundary private while
// allowing preparation and execution to discover registered capabilities.
type ArtifactEvidenceRegistry struct {
entries map[contracts.ArtifactKind]artifactEvidenceEntry
}
type artifactEvidenceEntry struct {
valueType reflect.Type
project func(any) ([]source.SourceRef, error)
}
func NewArtifactEvidenceRegistry() *ArtifactEvidenceRegistry {
return &ArtifactEvidenceRegistry{entries: make(map[contracts.ArtifactKind]artifactEvidenceEntry)}
}
// RegisterArtifactEvidence registers one evidence projector for an artifact
// kind. Projectors are invoked only after an exact Go-type check.
func RegisterArtifactEvidence[T any](registry *ArtifactEvidenceRegistry, kind contracts.ArtifactKind, projector ArtifactEvidenceProjector[T]) error {
if registry == nil {
return fmt.Errorf("artifact evidence registry must not be nil")
}
if projector == nil {
return fmt.Errorf("artifact evidence projector must not be nil")
}
kind = normalizeArtifactKind(kind)
if kind == "" {
return fmt.Errorf("artifact evidence kind must not be empty")
}
if _, ok := registry.entries[kind]; ok {
return fmt.Errorf("artifact evidence %q is already registered", kind)
}
valueType := reflect.TypeFor[T]()
if registry.entries == nil {
registry.entries = make(map[contracts.ArtifactKind]artifactEvidenceEntry)
}
registry.entries[kind] = artifactEvidenceEntry{
valueType: valueType,
project: func(value any) ([]source.SourceRef, error) {
actualType := reflect.TypeOf(value)
if actualType != valueType {
return nil, newArtifactEvidenceTypeError(kind, valueType, actualType)
}
typed, ok := value.(T)
if !ok {
return nil, newArtifactEvidenceTypeError(kind, valueType, actualType)
}
return append([]source.SourceRef(nil), projector(typed)...), nil
},
}
return nil
}
func (r *ArtifactEvidenceRegistry) RegisteredKinds() []contracts.ArtifactKind {
if r == nil || len(r.entries) == 0 {
return nil
}
kinds := make([]contracts.ArtifactKind, 0, len(r.entries))
for kind := range r.entries {
kinds = append(kinds, kind)
}
sort.Slice(kinds, func(i, j int) bool { return kinds[i] < kinds[j] })
return kinds
}
// Project returns independently owned direct references for a registered
// artifact value.
func (r *ArtifactEvidenceRegistry) Project(kind contracts.ArtifactKind, value any) ([]source.SourceRef, error) {
entry, normalizedKind, err := r.entry(kind)
if err != nil {
return nil, err
}
refs, err := entry.project(value)
if err != nil {
return nil, fmt.Errorf("project artifact evidence %q: %w", normalizedKind, err)
}
return append([]source.SourceRef(nil), refs...), nil
}
func (r *ArtifactEvidenceRegistry) entry(kind contracts.ArtifactKind) (artifactEvidenceEntry, contracts.ArtifactKind, error) {
if r == nil {
return artifactEvidenceEntry{}, "", fmt.Errorf("artifact evidence registry must not be nil")
}
kind = normalizeArtifactKind(kind)
if kind == "" {
return artifactEvidenceEntry{}, "", fmt.Errorf("artifact evidence kind must not be empty")
}
entry, ok := r.entries[kind]
if !ok {
return artifactEvidenceEntry{}, kind, fmt.Errorf("artifact evidence %q is not registered", kind)
}
return entry, kind, nil
}
func (r *ArtifactEvidenceRegistry) valueType(kind contracts.ArtifactKind) (reflect.Type, bool) {
if r == nil {
return nil, false
}
entry, ok := r.entries[normalizeArtifactKind(kind)]
if !ok {
return nil, false
}
return entry.valueType, true
}
func newArtifactEvidenceTypeError(kind contracts.ArtifactKind, expected, actual reflect.Type) error {
actualName := "<nil>"
if actual != nil {
actualName = actual.String()
}
return fmt.Errorf("project artifact evidence %q: expected exact Go type %s, got %s", kind, expected, actualName)
}
func normalizeEvidenceLaneIDs(values []string) ([]string, error) {
if len(values) == 0 {
return nil, fmt.Errorf("evidence lane ids must not be empty")
}
seen := make(map[string]struct{}, len(values))
lanes := make([]string, 0, len(values))
for _, raw := range values {
lane := strings.TrimSpace(raw)
if lane == "" {
return nil, fmt.Errorf("evidence lane id must not be empty")
}
if _, ok := seen[lane]; ok {
return nil, fmt.Errorf("evidence lane id %q is duplicated", lane)
}
seen[lane] = struct{}{}
lanes = append(lanes, lane)
}
sort.Strings(lanes)
return lanes, nil
}

View File

@@ -0,0 +1,55 @@
package pipeline
import (
"reflect"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
func TestArtifactEvidenceRegistryProjectsExactRegisteredTypesWithOwnedReferences(t *testing.T) {
registry := NewArtifactEvidenceRegistry()
input := codecNotes{Items: []string{"one"}}
refs := []source.SourceRef{{SourceID: "source", StartUnitID: 2, EndUnitID: 2}}
if err := RegisterArtifactEvidence(registry, "test/notes", func(codecNotes) []source.SourceRef { return refs }); err != nil {
t.Fatalf("RegisterArtifactEvidence() error = %v, want nil", err)
}
if err := RegisterArtifactEvidence(registry, "test/score", func(codecScore) []source.SourceRef { return nil }); err != nil {
t.Fatalf("RegisterArtifactEvidence() second kind error = %v, want nil", err)
}
if got, want := registry.RegisteredKinds(), []contracts.ArtifactKind{"test/notes", "test/score"}; !reflect.DeepEqual(got, want) {
t.Fatalf("RegisteredKinds() = %#v, want %#v", got, want)
}
projected, err := registry.Project(" test/notes ", input)
if err != nil {
t.Fatalf("Project() error = %v, want nil", err)
}
projected[0].StartUnitID = 99
if refs[0].StartUnitID != 2 {
t.Fatal("Project() retained projector reference storage")
}
if _, err := registry.Project("test/notes", codecNotesAlias(input)); err == nil || !strings.Contains(err.Error(), "expected exact Go type") {
t.Fatalf("Project() exact type error = %v, want exact type failure", err)
}
}
func TestArtifactEvidenceRegistryRejectsInvalidRegistration(t *testing.T) {
registry := NewArtifactEvidenceRegistry()
if err := RegisterArtifactEvidence[codecNotes](nil, "test/notes", func(codecNotes) []source.SourceRef { return nil }); err == nil || !strings.Contains(err.Error(), "must not be nil") {
t.Fatalf("nil registry error = %v, want failure", err)
}
if err := RegisterArtifactEvidence(registry, " ", func(codecNotes) []source.SourceRef { return nil }); err == nil || !strings.Contains(err.Error(), "must not be empty") {
t.Fatalf("blank kind error = %v, want failure", err)
}
if err := RegisterArtifactEvidence(registry, "test/notes", ArtifactEvidenceProjector[codecNotes](nil)); err == nil || !strings.Contains(err.Error(), "must not be nil") {
t.Fatalf("nil projector error = %v, want failure", err)
}
if err := RegisterArtifactEvidence(registry, "test/notes", func(codecNotes) []source.SourceRef { return nil }); err != nil {
t.Fatal(err)
}
if err := RegisterArtifactEvidence(registry, "test/notes", func(codecNotes) []source.SourceRef { return nil }); err == nil || !strings.Contains(err.Error(), "already registered") {
t.Fatalf("duplicate kind error = %v, want failure", err)
}
}

View File

@@ -0,0 +1,111 @@
package pipeline
import (
"fmt"
"strings"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/evidencecontext"
)
// debugEvidenceContextSummary intentionally contains only publication-safe
// identifiers and aggregate counts. The evidence document itself can include
// source text and must never be written to this debug envelope.
type debugEvidenceContextSummary struct {
ArtifactKind contracts.ArtifactKind `json:"artifact_kind"`
MediaType string `json:"media_type"`
SchemaID string `json:"schema_id"`
SchemaName string `json:"schema_name"`
SchemaVersion string `json:"schema_version"`
SelectedLanes []string `json:"selected_lanes"`
WindowUnits int `json:"window_units"`
ContextCount int `json:"context_count"`
UnitCount int `json:"unit_count"`
SourceDigest string `json:"source_digest"`
}
// buildOutputEvidenceContext projects the prepared output policy from accepted
// normalized artifacts. It is intentionally separate from lane execution so
// checkpointed normalized outputs use the same reconstruction path.
func buildOutputEvidenceContext(prepared *PreparedPipeline, doc *source.SourceDocument, outputs []contracts.SerializedOutput) (*contracts.SerializedArtifact, *debugEvidenceContextSummary, error) {
if prepared == nil || prepared.evidencePlan == nil {
return nil, nil, nil
}
if doc == nil {
return nil, nil, fmt.Errorf("evidence context output: source document is unavailable")
}
if prepared.artifactCodecs == nil {
return nil, nil, fmt.Errorf("evidence context output: artifact codecs are unavailable")
}
byLane := make(map[string]contracts.SerializedOutput, len(outputs))
for _, output := range outputs {
laneID := strings.TrimSpace(output.LaneID)
if _, exists := byLane[laneID]; exists {
return nil, nil, fmt.Errorf("evidence context output: accepted normalized outputs contain duplicate lane %q", laneID)
}
byLane[laneID] = contracts.CloneSerializedOutput(output)
}
request := evidencecontext.BuildRequest{
Source: doc,
WindowUnits: prepared.evidencePlan.policy.WindowUnits,
SelectedLanes: append([]string(nil), prepared.evidencePlan.policy.LaneIDs...),
LaneEvidence: make([]evidencecontext.LaneEvidence, 0, len(prepared.evidencePlan.lanes)),
}
for _, lane := range prepared.evidencePlan.lanes {
output, ok := byLane[lane.laneID]
if !ok {
continue
}
if output.SourceID != doc.ID {
return nil, nil, fmt.Errorf("evidence context output lane %q: accepted normalized output source is incompatible", lane.laneID)
}
if output.Artifact.Kind != lane.kind {
return nil, nil, fmt.Errorf("evidence context output lane %q: accepted normalized output artifact kind is incompatible", lane.laneID)
}
value, err := prepared.artifactCodecs.Decode(contracts.CloneSerializedArtifact(output.Artifact))
if err != nil {
return nil, nil, fmt.Errorf("evidence context output lane %q: accepted normalized artifact cannot be decoded", lane.laneID)
}
references, err := lane.project(value)
if err != nil {
return nil, nil, fmt.Errorf("evidence context output lane %q: accepted normalized artifact cannot be projected", lane.laneID)
}
request.LaneEvidence = append(request.LaneEvidence, evidencecontext.LaneEvidence{
LaneID: lane.laneID,
SourceRefs: append([]source.SourceRef(nil), references...),
})
}
document, err := evidencecontext.Build(request)
if err != nil {
return nil, nil, fmt.Errorf("evidence context output: accepted evidence references are invalid")
}
content, err := evidencecontext.New().Encode(document)
if err != nil {
return nil, nil, fmt.Errorf("evidence context output: evidence context serialization failed")
}
artifact := &contracts.SerializedArtifact{
Kind: evidencecontext.ArtifactKind,
Schema: evidencecontext.New().Schema(),
MediaType: evidencecontext.MediaType,
Content: content,
}
summary := debugEvidenceContextSummary{
ArtifactKind: artifact.Kind,
MediaType: artifact.MediaType,
SchemaID: artifact.Schema.ID,
SchemaName: artifact.Schema.Name,
SchemaVersion: artifact.Schema.Version,
SelectedLanes: append([]string(nil), document.SelectedLanes...),
WindowUnits: document.WindowUnits,
ContextCount: len(document.Contexts),
SourceDigest: document.SourceDigest,
}
for _, context := range document.Contexts {
summary.UnitCount += len(context.Units)
}
return contracts.CloneSerializedArtifactPointer(artifact), &summary, nil
}

View File

@@ -0,0 +1,252 @@
package pipeline
import (
"context"
"errors"
"reflect"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/evidencecontext"
)
type capturingEvidenceOutput struct {
requests []contracts.OutputRequest
}
func (*capturingEvidenceOutput) Key() string { return "capture/evidence-context" }
func (output *capturingEvidenceOutput) Encode(_ context.Context, request contracts.OutputRequest) (contracts.OutputResult, error) {
request.EvidenceContext = contracts.CloneSerializedArtifactPointer(request.EvidenceContext)
output.requests = append(output.requests, request)
return contracts.OutputResult{}, nil
}
func installEvidencePlan(prepared *PreparedPipeline, window int, selected []string, project func(codecNotes) ([]source.SourceRef, error)) {
lanes := make([]preparedEvidenceLane, 0, len(selected))
for _, laneID := range selected {
for _, step := range prepared.Steps {
for _, lane := range step.lanes {
if lane.resolved.ID != laneID {
continue
}
lanes = append(lanes, preparedEvidenceLane{
laneID: laneID,
kind: lane.resolved.ArtifactKind,
project: func(value any) ([]source.SourceRef, error) {
notes, ok := value.(codecNotes)
if !ok {
return nil, errors.New("unexpected artifact type")
}
return project(notes)
},
})
}
}
}
prepared.evidencePlan = &preparedEvidencePlan{policy: EvidenceContextPolicy{Enabled: true, WindowUnits: window, LaneIDs: append([]string(nil), selected...)}, lanes: lanes}
}
func setNormalizedNotes(prepared *PreparedPipeline, values map[string]codecNotes) {
for stepIndex := range prepared.Steps {
for laneIndex := range prepared.Steps[stepIndex].lanes {
lane := &prepared.Steps[stepIndex].lanes[laneIndex]
value, ok := values[lane.resolved.ID]
if !ok {
continue
}
lane.typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
return erasedTypedResult{Value: value}, nil
}
}
}
}
func decodeCapturedEvidence(t *testing.T, output *capturingEvidenceOutput) evidencecontext.Document {
t.Helper()
if len(output.requests) != 1 || output.requests[0].EvidenceContext == nil {
t.Fatalf("output requests = %#v, want one evidence context", output.requests)
}
value, err := evidencecontext.New().Decode(output.requests[0].EvidenceContext.Content)
if err != nil {
t.Fatalf("Decode(evidence context): %v", err)
}
return value
}
func TestRunnerBuildsEvidenceContextFromSelectedNormalizedOutputs(t *testing.T) {
prepared := preparedOrderedPipeline(t, 3, orderedLaneSpec{id: "alpha", profile: "notes"}, orderedLaneSpec{id: "beta", profile: "notes"})
encoder := &capturingEvidenceOutput{}
prepared.output = encoder
setNormalizedNotes(prepared, map[string]codecNotes{"alpha": {Items: []string{"one"}}, "beta": {Items: []string{"two"}}})
installEvidencePlan(prepared, 1, []string{"alpha", "beta", "inactive"}, func(notes codecNotes) ([]source.SourceRef, error) {
switch notes.Items[0] {
case "one":
return []source.SourceRef{{SourceID: "source", StartUnitID: 1, EndUnitID: 1}}, nil
case "two":
return []source.SourceRef{{SourceID: "source", StartUnitID: 2, EndUnitID: 2}}, nil
default:
return nil, nil
}
})
debug := newCapturedDebugRecorder()
if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug}); err != nil {
t.Fatalf("Run() error = %v", err)
}
value := decodeCapturedEvidence(t, encoder)
if !reflect.DeepEqual(value.SelectedLanes, []string{"alpha", "beta", "inactive"}) || len(value.Contexts) != 1 || len(value.Contexts[0].Units) != 3 {
t.Fatalf("evidence context = %#v, want selected union", value)
}
if got := value.Contexts[0].EvidenceRefs; len(got) != 2 || got[0].LaneID != "alpha" || got[1].LaneID != "beta" {
t.Fatalf("evidence refs = %#v, want both selected lanes", got)
}
debugJSON := string(debug.json["output/evidence-context.json"])
if strings.Contains(debugJSON, "text-1") || strings.Contains(debugJSON, "metadata") || !strings.Contains(debugJSON, `"artifact_kind":"source/evidence-context"`) || !strings.Contains(debugJSON, `"schema_id":"notarius.source.evidence_context"`) || !strings.Contains(debugJSON, `"context_count":1`) || !strings.Contains(debugJSON, `"unit_count":3`) {
t.Fatalf("evidence debug envelope = %s, want only allowlisted summary", debugJSON)
}
}
func TestRunnerEvidenceContextOmitsAbsentAndRejectedLanes(t *testing.T) {
prepared := preparedOrderedPipeline(t, 2, orderedLaneSpec{id: "present", profile: "notes"}, orderedLaneSpec{id: "rejected", profile: "notes"})
encoder := &capturingEvidenceOutput{}
prepared.output = encoder
setNormalizedNotes(prepared, map[string]codecNotes{"present": {Items: []string{"present"}}, "rejected": {Items: []string{"rejected"}}})
prepared.Steps[1].lanes[0].mergeValidators.validators = []preparedValidator{{
resolved: ResolvedValidator{Binding: Binding("reject"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"},
typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {
return contracts.ValidationResult{Approved: false, ReasonCode: "rejected", Message: "not accepted"}, nil
},
}}
installEvidencePlan(prepared, 0, []string{"absent", "present", "rejected"}, func(notes codecNotes) ([]source.SourceRef, error) {
if len(notes.Items) > 0 && notes.Items[0] == "present" {
return []source.SourceRef{{SourceID: "source", StartUnitID: 1, EndUnitID: 1}}, nil
}
return []source.SourceRef{{SourceID: "source", StartUnitID: 2, EndUnitID: 2}}, nil
})
result, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")})
if err != nil {
t.Fatalf("Run() error = %v", err)
}
if len(result.Rejected) != 1 || result.Rejected[0].LaneID != "rejected" {
t.Fatalf("rejections = %#v, want rejected lane unchanged", result.Rejected)
}
value := decodeCapturedEvidence(t, encoder)
if len(value.Contexts) != 1 || len(value.Contexts[0].EvidenceRefs) != 1 || value.Contexts[0].EvidenceRefs[0].LaneID != "present" {
t.Fatalf("evidence context = %#v, want present lane only", value)
}
}
func TestRunnerEvidenceContextFailurePreventsOutputEncoding(t *testing.T) {
prepared := preparedConcurrentPipeline(t, 1)
encoder := &capturingEvidenceOutput{}
prepared.output = encoder
setNormalizedNotes(prepared, map[string]codecNotes{"notes": {Items: []string{"invalid"}}})
installEvidencePlan(prepared, 0, []string{"notes"}, func(codecNotes) ([]source.SourceRef, error) {
return []source.SourceRef{{SourceID: "source", StartUnitID: 99, EndUnitID: 99}}, nil
})
result, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")})
if err == nil || !strings.Contains(err.Error(), "accepted evidence references are invalid") {
t.Fatalf("Run() error = %v, want evidence context failure", err)
}
if len(encoder.requests) != 0 || result.Manifest.ValidationStatus != "failed" || len(result.Rejected) != 0 {
t.Fatalf("output requests = %#v manifest = %#v rejected = %#v, want failed run before output encoding", encoder.requests, result.Manifest, result.Rejected)
}
}
func TestBuildOutputEvidenceContextRejectsIncompatibleAcceptedOutputs(t *testing.T) {
prepared := preparedConcurrentPipeline(t, 1)
doc := prepared.input.(*typedTestInput).doc
installEvidencePlan(prepared, 0, []string{"notes"}, func(codecNotes) ([]source.SourceRef, error) {
return []source.SourceRef{{SourceID: doc.ID, StartUnitID: 1, EndUnitID: 1}}, nil
})
artifact, err := prepared.artifactCodecs.Encode("test/notes", codecNotes{Items: []string{"valid"}})
if err != nil {
t.Fatal(err)
}
valid := contracts.SerializedOutput{LaneID: "notes", SourceID: doc.ID, Artifact: artifact}
for _, test := range []struct {
name string
outputs []contracts.SerializedOutput
mutate func(*contracts.SerializedOutput)
want string
}{
{name: "duplicate lane", outputs: []contracts.SerializedOutput{valid, valid}, want: "duplicate lane"},
{name: "foreign source", outputs: []contracts.SerializedOutput{valid}, mutate: func(output *contracts.SerializedOutput) { output.SourceID = "other" }, want: "source is incompatible"},
{name: "wrong artifact kind", outputs: []contracts.SerializedOutput{valid}, mutate: func(output *contracts.SerializedOutput) { output.Artifact.Kind = "test/score" }, want: "artifact kind is incompatible"},
{name: "invalid artifact payload", outputs: []contracts.SerializedOutput{valid}, mutate: func(output *contracts.SerializedOutput) { output.Artifact.Content = []byte("not JSON") }, want: "cannot be decoded"},
} {
t.Run(test.name, func(t *testing.T) {
outputs := append([]contracts.SerializedOutput(nil), test.outputs...)
for index := range outputs {
outputs[index] = contracts.CloneSerializedOutput(outputs[index])
}
if test.mutate != nil {
test.mutate(&outputs[0])
}
_, _, err := buildOutputEvidenceContext(prepared, doc, outputs)
if err == nil || !strings.Contains(err.Error(), test.want) {
t.Fatalf("buildOutputEvidenceContext() error = %v, want %q", err, test.want)
}
if strings.Contains(err.Error(), "not JSON") {
t.Fatalf("buildOutputEvidenceContext() exposed artifact payload: %v", err)
}
})
}
}
func TestEvidenceContextOutputRequestOwnsArtifactBytes(t *testing.T) {
artifact := &contracts.SerializedArtifact{Content: []byte("original"), Metadata: map[string]any{"source": "original"}}
request := contracts.OutputRequest{EvidenceContext: contracts.CloneSerializedArtifactPointer(artifact)}
request.EvidenceContext.Content[0] = 'X'
request.EvidenceContext.Metadata["source"] = "changed"
if string(artifact.Content) != "original" || artifact.Metadata["source"] != "original" {
t.Fatalf("output request evidence context aliases source artifact: %#v", artifact)
}
}
func TestRunnerEvidenceContextRebuildsFromAcceptedCheckpoint(t *testing.T) {
prepared := preparedConcurrentPipeline(t, 1)
encoder := &capturingEvidenceOutput{}
prepared.output = encoder
doc := prepared.input.(*typedTestInput).doc
lane := prepared.Steps[0].lanes[0]
stored, err := checkpointArtifact(lane.typed.codec, lane.resolved.ID, lane.resolved.Normalize.Module, doc.ID, codecNotes{Items: []string{"stored"}})
if err != nil {
t.Fatal(err)
}
loader := newAcceptedCheckpointLoader()
key := CheckpointLaneKey(lane.resolved.StepID, lane.resolved.ID)
loader.accepted[key] = NormalizeCheckpoint{Output: stored}
loader.acceptedDecision[key] = NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused)
installEvidencePlan(prepared, 0, []string{"notes"}, func(notes codecNotes) ([]source.SourceRef, error) {
if !reflect.DeepEqual(notes.Items, []string{"stored"}) {
return nil, errors.New("checkpoint artifact was not projected")
}
return []source.SourceRef{{SourceID: doc.ID, StartUnitID: 1, EndUnitID: 1}}, nil
})
policy := CheckpointExecutionPolicy{RequireReusableLanes: map[string]struct{}{key: {}}}
if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoint: loader, CheckpointPolicy: policy}); err != nil {
t.Fatalf("Run() error = %v", err)
}
value := decodeCapturedEvidence(t, encoder)
if len(value.Contexts) != 1 || value.Contexts[0].EvidenceRefs[0].LaneID != "notes" {
t.Fatalf("evidence context = %#v, want checkpointed normalized output", value)
}
}
func TestRunnerSkipsEvidenceContextWhenOutputDoesNotOptIn(t *testing.T) {
prepared := preparedConcurrentPipeline(t, 1)
encoder := &capturingEvidenceOutput{}
prepared.output = encoder
projected := 0
prepared.evidencePlan = nil
if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")}); err != nil {
t.Fatalf("Run() error = %v", err)
}
if projected != 0 || len(encoder.requests) != 1 || encoder.requests[0].EvidenceContext != nil {
t.Fatalf("projected = %d requests = %#v, want no evidence work", projected, encoder.requests)
}
}

View File

@@ -0,0 +1,20 @@
package pipeline
// EvidenceContextPolicy controls optional source-context publication by an
// output encoder.
type EvidenceContextPolicy struct {
Enabled bool
WindowUnits int
LaneIDs []string
}
// EvidenceContextPolicyProvider is implemented by output encoders that opt in
// to evidence-context publication.
type EvidenceContextPolicyProvider interface {
EvidenceContextPolicy() EvidenceContextPolicy
}
func cloneEvidenceContextPolicy(policy EvidenceContextPolicy) EvidenceContextPolicy {
policy.LaneIDs = append([]string(nil), policy.LaneIDs...)
return policy
}

View File

@@ -0,0 +1,98 @@
package pipeline
import (
"context"
"reflect"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
type testEvidenceOutput struct {
policy EvidenceContextPolicy
}
func (output testEvidenceOutput) Key() string { return "output" }
func (output testEvidenceOutput) Encode(context.Context, contracts.OutputRequest) (contracts.OutputResult, error) {
return contracts.OutputResult{}, nil
}
func (output testEvidenceOutput) EvidenceContextPolicy() EvidenceContextPolicy {
return cloneEvidenceContextPolicy(output.policy)
}
func TestPrepareEvidencePlanSelectsActiveLanesAndOwnsPolicy(t *testing.T) {
registries, _ := constructionRegistries(t, nil, nil)
registries.ArtifactEvidence = NewArtifactEvidenceRegistry()
registerTestEvidenceOutput(t, &registries, EvidenceContextPolicy{Enabled: true, WindowUnits: 2, LaneIDs: []string{"artifact", "inactive"}})
if err := RegisterArtifactEvidence(registries.ArtifactEvidence, "test/notes", func(codecNotes) []source.SourceRef { return nil }); err != nil {
t.Fatal(err)
}
profile := constructionProfile()
profile.Output.Options = map[string]any{"known": true}
resolved, err := ResolvePipeline(profile, ResolveOptions{}, registries.catalog())
if err != nil {
t.Fatal(err)
}
prepared, err := Prepare(resolved, registries, ModuleDependencies{})
if err != nil {
t.Fatalf("Prepare() error = %v, want nil", err)
}
if prepared.evidencePlan == nil || !reflect.DeepEqual(prepared.evidencePlan.policy.LaneIDs, []string{"artifact", "inactive"}) {
t.Fatalf("evidence plan = %#v, want complete configured policy", prepared.evidencePlan)
}
if got := prepared.evidencePlan.lanes; len(got) != 1 || got[0].laneID != "artifact" {
t.Fatalf("active evidence lanes = %#v, want artifact only", got)
}
prepared.evidencePlan.policy.LaneIDs[0] = "mutated"
if policy := prepared.output.(EvidenceContextPolicyProvider).EvidenceContextPolicy(); policy.LaneIDs[0] != "artifact" {
t.Fatalf("prepared plan mutated provider policy: %#v", policy)
}
}
func TestPrepareEvidencePlanRejectsMissingAndMismatchedCapabilities(t *testing.T) {
for _, test := range []struct {
name string
configure func(*Registries)
want string
}{
{name: "missing registry", configure: func(registries *Registries) { registries.ArtifactEvidence = nil }, want: "artifact evidence registry"},
{name: "unsupported kind", configure: func(registries *Registries) {}, want: "artifact evidence \"test/notes\" is not registered"},
{name: "mismatched type", configure: func(registries *Registries) {
if err := RegisterArtifactEvidence(registries.ArtifactEvidence, "test/notes", func(codecScore) []source.SourceRef { return nil }); err != nil {
t.Fatal(err)
}
}, want: "requires Go type"},
} {
t.Run(test.name, func(t *testing.T) {
registries, _ := constructionRegistries(t, nil, nil)
registries.ArtifactEvidence = NewArtifactEvidenceRegistry()
registerTestEvidenceOutput(t, &registries, EvidenceContextPolicy{Enabled: true, LaneIDs: []string{"artifact"}})
test.configure(&registries)
profile := constructionProfile()
profile.Output.Options = map[string]any{"known": true}
resolved, err := ResolvePipeline(profile, ResolveOptions{}, registries.catalog())
if err != nil {
t.Fatal(err)
}
_, err = Prepare(resolved, registries, ModuleDependencies{})
if err == nil || !strings.Contains(err.Error(), test.want) {
t.Fatalf("Prepare() error = %v, want %q", err, test.want)
}
})
}
}
func registerTestEvidenceOutput(t *testing.T, registries *Registries, policy EvidenceContextPolicy) {
t.Helper()
registry := NewOutputEncoderRegistry()
if err := registry.RegisterBuilderWithSpec(defaultModuleSpec("output", StageOutput), func(options map[string]any) error {
return RejectUnknownOptions(options, "known")
}, func(BuildRequest) (contracts.OutputEncoder, error) {
return testEvidenceOutput{policy: cloneEvidenceContextPolicy(policy)}, nil
}); err != nil {
t.Fatal(err)
}
registries.Outputs = registry
}

View File

@@ -2,7 +2,7 @@ package pipeline
import "fmt"
func validateResolvedOptions(resolved ResolvedPipeline, catalog ModuleCatalog) error {
func validateResolvedOptions(resolved ResolvedPipeline, catalog ModuleCatalog, configuredLaneIDs []string) error {
if err := catalog.Inputs.ValidateOptions(resolved.Input.Module, resolved.Input.Options); err != nil {
return moduleOptionsError(resolved.ID, "", StageInput, resolved.Input.Module, err)
}
@@ -42,6 +42,9 @@ func validateResolvedOptions(resolved ResolvedPipeline, catalog ModuleCatalog) e
if err := catalog.Outputs.ValidateOptions(resolved.Output.Module, resolved.Output.Options); err != nil {
return moduleOptionsError(resolved.ID, "", StageOutput, resolved.Output.Module, err)
}
if err := catalog.Outputs.ValidateProfileOptions(resolved.Output.Module, OutputProfileOptionContext{LaneIDs: configuredLaneIDs}, resolved.Output.Options); err != nil {
return moduleOptionsError(resolved.ID, "", StageOutput, resolved.Output.Module, err)
}
return nil
}

View File

@@ -10,9 +10,16 @@ import (
type OutputEncoderConstructor func() (contracts.OutputEncoder, error)
type OutputEncoderBuilder func(BuildRequest) (contracts.OutputEncoder, error)
type OutputProfileOptionContext struct {
LaneIDs []string
}
type OutputProfileOptionValidator func(OutputProfileOptionContext, map[string]any) error
type OutputEncoderRegistry struct {
builders map[string]OutputEncoderBuilder
optionValidators map[string]OptionValidator
profileValidators map[string]OutputProfileOptionValidator
specs map[string]ModuleSpec
}
@@ -20,6 +27,7 @@ func NewOutputEncoderRegistry() *OutputEncoderRegistry {
return &OutputEncoderRegistry{
builders: make(map[string]OutputEncoderBuilder),
optionValidators: make(map[string]OptionValidator),
profileValidators: make(map[string]OutputProfileOptionValidator),
specs: make(map[string]ModuleSpec),
}
}
@@ -38,6 +46,12 @@ func (r *OutputEncoderRegistry) RegisterWithSpec(spec ModuleSpec, constructor Ou
}
func (r *OutputEncoderRegistry) RegisterBuilderWithSpec(spec ModuleSpec, validateOptions OptionValidator, builder OutputEncoderBuilder) error {
return r.RegisterBuilderWithProfileValidation(spec, validateOptions, nil, builder)
}
// RegisterBuilderWithProfileValidation registers an output builder with an
// optional validator that can inspect all configured lane identities.
func (r *OutputEncoderRegistry) RegisterBuilderWithProfileValidation(spec ModuleSpec, validateOptions OptionValidator, validateProfile OutputProfileOptionValidator, builder OutputEncoderBuilder) error {
if r == nil {
return fmt.Errorf("output encoder registry must not be nil")
}
@@ -62,11 +76,15 @@ func (r *OutputEncoderRegistry) RegisterBuilderWithSpec(spec ModuleSpec, validat
if r.optionValidators == nil {
r.optionValidators = make(map[string]OptionValidator)
}
if r.profileValidators == nil {
r.profileValidators = make(map[string]OutputProfileOptionValidator)
}
if r.specs == nil {
r.specs = make(map[string]ModuleSpec)
}
r.builders[normalizedSpec.Key] = builder
r.optionValidators[normalizedSpec.Key] = validateOptions
r.profileValidators[normalizedSpec.Key] = validateProfile
r.specs[normalizedSpec.Key] = cloneModuleSpec(normalizedSpec)
return nil
}
@@ -116,6 +134,21 @@ func (r *OutputEncoderRegistry) ValidateOptions(key string, options map[string]a
return validateRegisteredOptions(validator, options)
}
func (r *OutputEncoderRegistry) ValidateProfileOptions(key string, context OutputProfileOptionContext, options map[string]any) error {
if r == nil {
return fmt.Errorf("output encoder registry must not be nil")
}
normalizedKey := strings.TrimSpace(key)
validator, ok := r.profileValidators[normalizedKey]
if !ok {
return fmt.Errorf("output encoder %q is not registered", normalizedKey)
}
if validator == nil {
return nil
}
return validator(OutputProfileOptionContext{LaneIDs: append([]string(nil), context.LaneIDs...)}, cloneOptions(options))
}
func (r *OutputEncoderRegistry) Spec(key string) (ModuleSpec, bool) {
if r == nil {
return ModuleSpec{}, false

View File

@@ -1,6 +1,7 @@
package pipeline
import (
"reflect"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
@@ -56,3 +57,33 @@ func TestOutputEncoderRegistryBehavior(t *testing.T) {
},
})
}
func TestOutputProfileValidationReceivesOwnedOptionsAndLaneIDs(t *testing.T) {
registry := NewOutputEncoderRegistry()
if err := registry.RegisterBuilderWithProfileValidation(defaultModuleSpec("profile-output", StageOutput), func(options map[string]any) error {
options["nested"].(map[string]any)["value"] = "changed"
return nil
}, func(context OutputProfileOptionContext, options map[string]any) error {
context.LaneIDs[0] = "changed"
options["nested"].(map[string]any)["value"] = "changed-again"
return nil
}, func(BuildRequest) (contracts.OutputEncoder, error) {
return registryOutputEncoder{key: "profile-output"}, nil
}); err != nil {
t.Fatal(err)
}
options := map[string]any{"nested": map[string]any{"value": "original"}}
if err := registry.ValidateOptions("profile-output", options); err != nil {
t.Fatal(err)
}
context := OutputProfileOptionContext{LaneIDs: []string{"artifact"}}
if err := registry.ValidateProfileOptions("profile-output", context, options); err != nil {
t.Fatal(err)
}
if got := options["nested"].(map[string]any)["value"]; got != "original" {
t.Fatalf("options mutated by validator: %q", got)
}
if want := []string{"artifact"}; !reflect.DeepEqual(context.LaneIDs, want) {
t.Fatalf("lane ids mutated by validator: %#v, want %#v", context.LaneIDs, want)
}
}

View File

@@ -5,6 +5,7 @@ import (
"reflect"
"strings"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
@@ -24,6 +25,7 @@ type PreparedPipeline struct {
chunkValidators preparedValidatorChain
output contracts.OutputEncoder
artifactCodecs *ArtifactCodecRegistry
evidencePlan *preparedEvidencePlan
checkpointFingerprints []CheckpointFingerprint
}
@@ -55,6 +57,17 @@ type preparedTypedLane struct {
codec artifactCodecEntry
}
type preparedEvidencePlan struct {
policy EvidenceContextPolicy
lanes []preparedEvidenceLane
}
type preparedEvidenceLane struct {
laneID string
kind contracts.ArtifactKind
project func(any) ([]source.SourceRef, error)
}
type preparedValidatorChain struct {
resolved ResolvedValidatorChain
validators []preparedValidator
@@ -129,6 +142,10 @@ func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDepend
return nil, constructionError(stable.ID, "", StageOutput, stable.Output.Module, "", err)
}
prepared.output = output
prepared.evidencePlan, err = prepareEvidencePlan(stable, registries, output)
if err != nil {
return nil, err
}
prepared.checkpointFingerprints, err = collectPreparedCheckpointFingerprints(prepared)
if err != nil {
return nil, err
@@ -136,6 +153,52 @@ func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDepend
return prepared, nil
}
func prepareEvidencePlan(resolved ResolvedPipeline, registries Registries, output contracts.OutputEncoder) (*preparedEvidencePlan, error) {
provider, ok := output.(EvidenceContextPolicyProvider)
if !ok {
return nil, nil
}
policy := cloneEvidenceContextPolicy(provider.EvidenceContextPolicy())
if !policy.Enabled {
return nil, nil
}
if policy.WindowUnits < 0 {
return nil, constructionError(resolved.ID, "", StageOutput, resolved.Output.Module, "", fmt.Errorf("evidence window units must not be negative"))
}
lanes, err := normalizeEvidenceLaneIDs(policy.LaneIDs)
if err != nil {
return nil, constructionError(resolved.ID, "", StageOutput, resolved.Output.Module, "", err)
}
policy.LaneIDs = lanes
active := make(map[string]ResolvedArtifactLane)
for _, lane := range resolved.AllArtifactLanes() {
active[lane.ID] = lane
}
plan := &preparedEvidencePlan{policy: cloneEvidenceContextPolicy(policy)}
for _, laneID := range policy.LaneIDs {
lane, ok := active[laneID]
if !ok {
continue
}
if registries.ArtifactEvidence == nil {
return nil, constructionError(resolved.ID, laneID, StageOutput, resolved.Output.Module, "", fmt.Errorf("artifact evidence registry must not be nil for active evidence lane"))
}
evidence, _, evidenceErr := registries.ArtifactEvidence.entry(lane.ArtifactKind)
if evidenceErr != nil {
return nil, constructionError(resolved.ID, laneID, StageOutput, resolved.Output.Module, "", evidenceErr)
}
codecType, ok := registries.ArtifactCodecs.valueType(lane.ArtifactKind)
if !ok {
return nil, constructionError(resolved.ID, laneID, StageOutput, resolved.Output.Module, "", fmt.Errorf("artifact codec %q has no Go type", lane.ArtifactKind))
}
if evidence.valueType != codecType {
return nil, constructionError(resolved.ID, laneID, StageOutput, resolved.Output.Module, "", fmt.Errorf("artifact evidence kind %q requires Go type %s, but active artifact codec uses %s", lane.ArtifactKind, typeName(evidence.valueType), typeName(codecType)))
}
plan.lanes = append(plan.lanes, preparedEvidenceLane{laneID: laneID, kind: lane.ArtifactKind, project: evidence.project})
}
return plan, nil
}
func prepareLane(pipeline ResolvedPipeline, lane ResolvedArtifactLane, registries Registries, deps ModuleDependencies) (preparedLaneExecutor, error) {
executor := preparedLaneExecutor{resolved: cloneResolvedArtifactLane(lane)}
request := func(binding ModuleBinding, references contracts.ReferenceSet) BuildRequest {
@@ -303,7 +366,7 @@ func constructionError(pipelineID, laneID string, stage ModuleStage, moduleKey,
func (registries Registries) catalog() ModuleCatalog {
return ModuleCatalog{
Inputs: registries.Inputs, Chunkers: registries.Chunkers, ArtifactCodecs: registries.ArtifactCodecs,
Inputs: registries.Inputs, Chunkers: registries.Chunkers, ArtifactCodecs: registries.ArtifactCodecs, ArtifactEvidence: registries.ArtifactEvidence,
Extractors: registries.Extractors, Mergers: registries.Mergers, Normalizers: registries.Normalizers,
Validators: registries.Validators, ValidatorChains: registries.ValidatorChains, Outputs: registries.Outputs,
}

View File

@@ -216,6 +216,7 @@ type ModuleCatalog struct {
Inputs *InputAdapterRegistry
Chunkers *ChunkerRegistry
ArtifactCodecs *ArtifactCodecRegistry
ArtifactEvidence *ArtifactEvidenceRegistry
Extractors *ExtractorRegistry
Mergers *MergerRegistry
Normalizers *NormalizerRegistry
@@ -251,6 +252,10 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo
if len(steps) == 0 {
return ResolvedPipeline{}, fmt.Errorf("pipeline %q must declare at least one artifact lane", pipelineID)
}
configuredLaneIDs, err := configuredProfileLaneIDs(pipelineID, steps)
if err != nil {
return ResolvedPipeline{}, err
}
input := resolveBinding(profile.Input, "")
if input.Module == "" {
@@ -381,7 +386,7 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo
if missing, ok := outputCapabilities.missing(outputSpec.Requires); ok {
return ResolvedPipeline{}, capabilityError(pipelineID, "", StageOutput, resolved.Output.Module, missing)
}
if err := validateResolvedOptions(resolved, catalog); err != nil {
if err := validateResolvedOptions(resolved, catalog, configuredLaneIDs); err != nil {
return ResolvedPipeline{}, err
}
@@ -393,6 +398,29 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo
return resolved, nil
}
// configuredProfileLaneIDs validates the complete configured lane identity
// set before invocation-level selection removes legacy-profile lanes.
func configuredProfileLaneIDs(pipelineID string, steps []PipelineStepProfile) ([]string, error) {
seen := make(map[string]string)
var laneIDs []string
for _, step := range steps {
_, ids, err := selectedArtifactLanes(pipelineID, step.Artifacts, ResolveOptions{})
if err != nil {
return nil, err
}
stepID := strings.TrimSpace(step.ID)
for _, laneID := range ids {
if previous, ok := seen[laneID]; ok {
return nil, fmt.Errorf("pipeline %q artifact lane id %q is duplicated across steps %q and %q", pipelineID, laneID, previous, stepID)
}
seen[laneID] = stepID
laneIDs = append(laneIDs, laneID)
}
}
sort.Strings(laneIDs)
return laneIDs, nil
}
func resolveArtifactLane(
pipelineID string,
stepID string,
@@ -1190,11 +1218,32 @@ func cloneOptions(options map[string]any) map[string]any {
copied := make(map[string]any, len(options))
for key, value := range options {
copied[key] = value
copied[key] = cloneOptionValue(value)
}
return copied
}
func cloneOptionValue(value any) any {
switch typed := value.(type) {
case map[string]any:
return cloneOptions(typed)
case []any:
out := make([]any, len(typed))
for i := range typed {
out[i] = cloneOptionValue(typed[i])
}
return out
case []string:
return append([]string(nil), typed...)
case []byte:
return append([]byte(nil), typed...)
case json.RawMessage:
return append(json.RawMessage(nil), typed...)
default:
return value
}
}
func normalizeReferenceMap(values map[string]ReferenceSource) map[string]ReferenceSource {
if len(values) == 0 {
return nil

View File

@@ -24,6 +24,7 @@ type Registries struct {
Inputs *InputAdapterRegistry
Chunkers *ChunkerRegistry
ArtifactCodecs *ArtifactCodecRegistry
ArtifactEvidence *ArtifactEvidenceRegistry
Extractors *ExtractorRegistry
Mergers *MergerRegistry
Normalizers *NormalizerRegistry
@@ -293,6 +294,20 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
return failOutput(output), err
}
outputStarted := time.Now().UTC()
evidenceArtifact, evidenceSummary, err := buildOutputEvidenceContext(input.Prepared, doc, output.NormalizeOutputs)
if err != nil {
return failOutput(output), err
}
if evidenceSummary != nil {
if err := writeDebugTimed(debugRecorder, "output/evidence-context.json", debugTimedEnvelope{
Stage: string(StageOutput),
ModuleKey: encoder.Key(),
StartedAt: outputStarted,
Payload: *evidenceSummary,
}); err != nil {
return failOutput(output), fmt.Errorf("write evidence context debug artifact: %w", err)
}
}
outputDebugPayload := map[string]any{
"manifest": output.Manifest,
"normalize_outputs": debugSerializedOutputEnvelopes(output.NormalizeOutputs),
@@ -324,6 +339,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
LLMProfile: input.pipeline.Output.LLMProfile,
Metadata: outputMetadata,
ChunkMap: contracts.CloneSerializedArtifactPointer(acceptedChunkMap),
EvidenceContext: contracts.CloneSerializedArtifactPointer(evidenceArtifact),
})
output.Warnings = append(output.Warnings, encoded.Warnings...)
if err != nil {

View File

@@ -0,0 +1,74 @@
package register
import (
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd"
)
func registerEvidence(registry *pipeline.ArtifactEvidenceRegistry) error {
return runRegistrations([]registration{
{name: "spells evidence", register: func() error { return pipeline.RegisterArtifactEvidence(registry, dnd.SpellListKind, spellEvidence) }},
{name: "npcs evidence", register: func() error { return pipeline.RegisterArtifactEvidence(registry, dnd.NPCListKind, npcEvidence) }},
{name: "combat turns evidence", register: func() error {
return pipeline.RegisterArtifactEvidence(registry, dnd.CombatTurnListKind, combatTurnEvidence)
}},
{name: "item events evidence", register: func() error {
return pipeline.RegisterArtifactEvidence(registry, dnd.ItemEventListKind, itemEventEvidence)
}},
{name: "npc interactions evidence", register: func() error {
return pipeline.RegisterArtifactEvidence(registry, dnd.NPCInteractionListKind, npcInteractionEvidence)
}},
{name: "scene descriptions evidence", register: func() error {
return pipeline.RegisterArtifactEvidence(registry, dnd.SceneDescriptionListKind, sceneDescriptionEvidence)
}},
})
}
func spellEvidence(value dnd.SpellList) []source.SourceRef {
var refs []source.SourceRef
for _, record := range value.SpellCasts {
refs = append(refs, record.SourceRefs...)
}
return append([]source.SourceRef(nil), refs...)
}
func npcEvidence(value dnd.NPCList) []source.SourceRef {
var refs []source.SourceRef
for _, record := range value.NPCs {
refs = append(refs, record.SourceRefs...)
}
return append([]source.SourceRef(nil), refs...)
}
func combatTurnEvidence(value dnd.CombatTurnList) []source.SourceRef {
var refs []source.SourceRef
for _, record := range value.CombatTurns {
refs = append(refs, record.SourceRefs...)
}
return append([]source.SourceRef(nil), refs...)
}
func itemEventEvidence(value dnd.ItemEventList) []source.SourceRef {
var refs []source.SourceRef
for _, record := range value.Events {
refs = append(refs, record.SourceRefs...)
}
return append([]source.SourceRef(nil), refs...)
}
func npcInteractionEvidence(value dnd.NPCInteractionList) []source.SourceRef {
var refs []source.SourceRef
for _, record := range value.Interactions {
refs = append(refs, record.SourceRefs...)
}
return append([]source.SourceRef(nil), refs...)
}
func sceneDescriptionEvidence(value dnd.SceneDescriptionList) []source.SourceRef {
refs := make([]source.SourceRef, 0, len(value.Scenes))
for _, record := range value.Scenes {
refs = append(refs, record.SourceRef)
}
return refs
}

View File

@@ -21,6 +21,9 @@ func Register(registries pipeline.Registries, assets *llm.AssetRegistry) error {
if err := registerModules(registries); err != nil {
return err
}
if err := registerEvidence(registries.ArtifactEvidence); err != nil {
return err
}
if err := registerValidators(registries); err != nil {
return err
}
@@ -45,6 +48,8 @@ func validateRegistries(registries pipeline.Registries, assets *llm.AssetRegistr
return fmt.Errorf("dnd registrar: chunker registry must not be nil")
case registries.ArtifactCodecs == nil:
return fmt.Errorf("dnd registrar: artifact codec registry must not be nil")
case registries.ArtifactEvidence == nil:
return fmt.Errorf("dnd registrar: artifact evidence registry must not be nil")
case registries.Extractors == nil:
return fmt.Errorf("dnd registrar: extractor registry must not be nil")
case registries.Mergers == nil:

View File

@@ -50,6 +50,7 @@ func TestRegisterAddsDNDFamily(t *testing.T) {
assertContainsKeys(t, "extractors", registries.Extractors.RegisteredKeys(), []string{"dnd/spells", npcextract.Key, combatextract.Key, itemeventextract.Key, interactionextract.Key, scenedescriptionextract.Key})
assertContainsKeys(t, "normalizers", registries.Normalizers.RegisteredKeys(), []string{spellnormalize.Key, npcnormalize.Key, combatnormalize.Key, itemeventnormalize.Key, interactionnormalize.Key, scenedescriptionnormalize.Key, pipeline.DefaultNormalizeModule})
assertContainsArtifactKinds(t, registries.ArtifactCodecs.RegisteredKinds(), []contracts.ArtifactKind{dnd.SpellListKind, dnd.NPCListKind, dnd.CombatTurnListKind, dnd.ItemEventListKind, dnd.NPCInteractionListKind, dnd.SceneDescriptionListKind})
assertContainsArtifactKinds(t, registries.ArtifactEvidence.RegisteredKinds(), []contracts.ArtifactKind{dnd.SpellListKind, dnd.NPCListKind, dnd.CombatTurnListKind, dnd.ItemEventListKind, dnd.NPCInteractionListKind, dnd.SceneDescriptionListKind})
assertContainsArtifactKinds(t, registries.Mergers.RegisteredArtifactKinds(pipeline.DefaultMergeModule), []contracts.ArtifactKind{dnd.SpellListKind, dnd.NPCListKind, dnd.CombatTurnListKind, dnd.ItemEventListKind, dnd.NPCInteractionListKind, dnd.SceneDescriptionListKind})
assertContainsArtifactKinds(t, registries.Normalizers.RegisteredArtifactKinds(pipeline.DefaultNormalizeModule), []contracts.ArtifactKind{dnd.SpellListKind, dnd.NPCListKind, dnd.CombatTurnListKind, dnd.ItemEventListKind, dnd.NPCInteractionListKind, dnd.SceneDescriptionListKind})
assertContainsArtifactKinds(t, registries.Normalizers.RegisteredArtifactKinds(npcnormalize.Key), []contracts.ArtifactKind{dnd.NPCListKind})
@@ -318,6 +319,46 @@ func TestRegisterAddsDNDFamily(t *testing.T) {
}
}
func TestEvidenceProjectorsPreserveDirectReferencesWithIndependentStorage(t *testing.T) {
first := source.SourceRef{SourceID: "session", StartUnitID: 1, EndUnitID: 1}
second := source.SourceRef{SourceID: "session", StartUnitID: 2, EndUnitID: 2}
for _, test := range []struct {
name string
project func() []source.SourceRef
want []source.SourceRef
}{
{name: "spells", project: func() []source.SourceRef {
return spellEvidence(dnd.SpellList{SpellCasts: []dnd.SpellCast{{SourceRefs: []source.SourceRef{first, second}}}})
}, want: []source.SourceRef{first, second}},
{name: "npcs", project: func() []source.SourceRef {
return npcEvidence(dnd.NPCList{NPCs: []dnd.NPC{{SourceRefs: []source.SourceRef{first, second}}}})
}, want: []source.SourceRef{first, second}},
{name: "combat turns", project: func() []source.SourceRef {
return combatTurnEvidence(dnd.CombatTurnList{CombatTurns: []dnd.CombatTurn{{SourceRefs: []source.SourceRef{first, second}}}})
}, want: []source.SourceRef{first, second}},
{name: "item events", project: func() []source.SourceRef {
return itemEventEvidence(dnd.ItemEventList{Events: []dnd.ItemEvent{{SourceRefs: []source.SourceRef{first, second}}}})
}, want: []source.SourceRef{first, second}},
{name: "npc interactions", project: func() []source.SourceRef {
return npcInteractionEvidence(dnd.NPCInteractionList{Interactions: []dnd.NPCInteraction{{SourceRefs: []source.SourceRef{first, second}}}})
}, want: []source.SourceRef{first, second}},
{name: "scene descriptions", project: func() []source.SourceRef {
return sceneDescriptionEvidence(dnd.SceneDescriptionList{Scenes: []dnd.SceneDescription{{SourceRef: first}, {SourceRef: second}}})
}, want: []source.SourceRef{first, second}},
} {
t.Run(test.name, func(t *testing.T) {
got := test.project()
if !reflect.DeepEqual(got, test.want) {
t.Fatalf("projected references = %#v, want %#v", got, test.want)
}
got[0].StartUnitID = 99
if first.StartUnitID != 1 {
t.Fatal("projector returned aliased reference storage")
}
})
}
}
func referenceSlot(slots []contracts.ReferenceSlot, name string) contracts.ReferenceSlot {
for _, slot := range slots {
if slot.Name == name {
@@ -531,6 +572,7 @@ func TestRegisterRejectsMissingDNDDependenciesBeforeMutation(t *testing.T) {
}{
{name: "chunkers", remove: func(r *pipeline.Registries, _ **llm.AssetRegistry) { r.Chunkers = nil }, wantErr: "chunker registry"},
{name: "artifact codecs", remove: func(r *pipeline.Registries, _ **llm.AssetRegistry) { r.ArtifactCodecs = nil }, wantErr: "artifact codec registry"},
{name: "artifact evidence", remove: func(r *pipeline.Registries, _ **llm.AssetRegistry) { r.ArtifactEvidence = nil }, wantErr: "artifact evidence registry"},
{name: "extractors", remove: func(r *pipeline.Registries, _ **llm.AssetRegistry) { r.Extractors = nil }, wantErr: "extractor registry"},
{name: "mergers", remove: func(r *pipeline.Registries, _ **llm.AssetRegistry) { r.Mergers = nil }, wantErr: "merger registry"},
{name: "normalizers", remove: func(r *pipeline.Registries, _ **llm.AssetRegistry) { r.Normalizers = nil }, wantErr: "normalizer registry"},
@@ -571,6 +613,7 @@ func completeRegistries() pipeline.Registries {
Inputs: pipeline.NewInputAdapterRegistry(),
Chunkers: pipeline.NewChunkerRegistry(),
ArtifactCodecs: pipeline.NewArtifactCodecRegistry(),
ArtifactEvidence: pipeline.NewArtifactEvidenceRegistry(),
Extractors: pipeline.NewExtractorRegistry(),
Mergers: pipeline.NewMergerRegistry(),
Normalizers: pipeline.NewNormalizerRegistry(),

View File

@@ -13,6 +13,7 @@ import (
"gitea.maximumdirect.net/eric/notarius/internal/framework/chunkmap"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/evidencecontext"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
)
@@ -22,12 +23,15 @@ const contentTypeJSON = "application/json"
const chunkMapFileName = "chunk-map.json"
const evidenceContextFileName = "evidence-context.json"
var safeOutputFileChar = regexp.MustCompile(`[^A-Za-z0-9._-]`)
var _ contracts.OutputEncoder = (*Encoder)(nil)
type Options struct {
IncludeChunkMap bool
EvidenceContext pipeline.EvidenceContextPolicy
}
type Encoder struct {
@@ -39,6 +43,7 @@ func New() *Encoder {
}
func NewWithOptions(options Options) *Encoder {
options.EvidenceContext.LaneIDs = append([]string(nil), options.EvidenceContext.LaneIDs...)
return &Encoder{options: options}
}
@@ -46,6 +51,15 @@ func (e *Encoder) Key() string {
return Key
}
func (e *Encoder) EvidenceContextPolicy() pipeline.EvidenceContextPolicy {
if e == nil {
return pipeline.EvidenceContextPolicy{}
}
policy := e.options.EvidenceContext
policy.LaneIDs = append([]string(nil), policy.LaneIDs...)
return policy
}
func (e *Encoder) Encode(ctx context.Context, req contracts.OutputRequest) (contracts.OutputResult, error) {
if e == nil {
return contracts.OutputResult{}, encoderErrorf("encoder must not be nil")
@@ -74,7 +88,7 @@ func ModuleSpec() pipeline.ModuleSpec {
}
func Register(registry *pipeline.OutputEncoderRegistry) error {
return registry.RegisterBuilderWithSpec(ModuleSpec(), validateOptions, func(request pipeline.BuildRequest) (contracts.OutputEncoder, error) {
return registry.RegisterBuilderWithProfileValidation(ModuleSpec(), validateOptions, validateProfileOptions, func(request pipeline.BuildRequest) (contracts.OutputEncoder, error) {
options, err := DecodeOptions(request.Options)
if err != nil {
return nil, err
@@ -89,17 +103,126 @@ func validateOptions(options map[string]any) error {
}
func DecodeOptions(options map[string]any) (Options, error) {
if err := pipeline.RejectUnknownOptions(options, "include_chunk_map"); err != nil {
if err := pipeline.RejectUnknownOptions(options, "include_chunk_map", "evidence_context"); err != nil {
return Options{}, encoderErrorf("%w", err)
}
decoded := Options{}
if value, ok := options["include_chunk_map"]; ok {
enabled, ok := value.(bool)
if !ok {
return Options{}, encoderErrorf("option %q must be a boolean", "include_chunk_map")
}
return Options{IncludeChunkMap: enabled}, nil
decoded.IncludeChunkMap = enabled
}
return Options{}, nil
if value, ok := options["evidence_context"]; ok {
policy, err := decodeEvidenceContextPolicy(value)
if err != nil {
return Options{}, err
}
decoded.EvidenceContext = policy
}
return decoded, nil
}
func validateProfileOptions(context pipeline.OutputProfileOptionContext, options map[string]any) error {
decoded, err := DecodeOptions(options)
if err != nil || !decoded.EvidenceContext.Enabled {
return err
}
configured := make(map[string]struct{}, len(context.LaneIDs))
for _, laneID := range context.LaneIDs {
configured[laneID] = struct{}{}
}
for _, laneID := range decoded.EvidenceContext.LaneIDs {
if _, ok := configured[laneID]; !ok {
return encoderErrorf("evidence_context lane %q is not configured", laneID)
}
}
return nil
}
func decodeEvidenceContextPolicy(value any) (pipeline.EvidenceContextPolicy, error) {
object, ok := value.(map[string]any)
if !ok {
return pipeline.EvidenceContextPolicy{}, encoderErrorf("option %q must be an object", "evidence_context")
}
if err := pipeline.RejectUnknownOptions(object, "enabled", "lanes", "window_units"); err != nil {
return pipeline.EvidenceContextPolicy{}, encoderErrorf("evidence_context: %w", err)
}
enabledValue, ok := object["enabled"]
if !ok {
return pipeline.EvidenceContextPolicy{}, encoderErrorf("evidence_context option %q is required", "enabled")
}
enabled, ok := enabledValue.(bool)
if !ok {
return pipeline.EvidenceContextPolicy{}, encoderErrorf("evidence_context option %q must be a boolean", "enabled")
}
if !enabled {
if _, ok := object["lanes"]; ok {
return pipeline.EvidenceContextPolicy{}, encoderErrorf("evidence_context option %q is not allowed when disabled", "lanes")
}
if _, ok := object["window_units"]; ok {
return pipeline.EvidenceContextPolicy{}, encoderErrorf("evidence_context option %q is not allowed when disabled", "window_units")
}
return pipeline.EvidenceContextPolicy{}, nil
}
rawLanes, ok := object["lanes"]
if !ok {
return pipeline.EvidenceContextPolicy{}, encoderErrorf("evidence_context option %q is required when enabled", "lanes")
}
lanes, err := decodeEvidenceLaneIDs(rawLanes)
if err != nil {
return pipeline.EvidenceContextPolicy{}, err
}
windowUnits := 3
if rawWindow, ok := object["window_units"]; ok {
value, ok := rawWindow.(int)
if !ok {
return pipeline.EvidenceContextPolicy{}, encoderErrorf("evidence_context option %q must be an integer", "window_units")
}
if value < 0 {
return pipeline.EvidenceContextPolicy{}, encoderErrorf("evidence_context option %q must not be negative", "window_units")
}
windowUnits = value
}
return pipeline.EvidenceContextPolicy{Enabled: true, WindowUnits: windowUnits, LaneIDs: lanes}, nil
}
func decodeEvidenceLaneIDs(value any) ([]string, error) {
var raw []any
switch typed := value.(type) {
case []any:
raw = typed
case []string:
raw = make([]any, len(typed))
for i := range typed {
raw[i] = typed[i]
}
default:
return nil, encoderErrorf("evidence_context option %q must be an array", "lanes")
}
if len(raw) == 0 {
return nil, encoderErrorf("evidence_context option %q must not be empty", "lanes")
}
seen := make(map[string]struct{}, len(raw))
lanes := make([]string, 0, len(raw))
for _, value := range raw {
lane, ok := value.(string)
if !ok {
return nil, encoderErrorf("evidence_context lane values must be strings")
}
lane = strings.TrimSpace(lane)
if lane == "" {
return nil, encoderErrorf("evidence_context lane values must not be empty")
}
if _, ok := seen[lane]; ok {
return nil, encoderErrorf("evidence_context lane %q is duplicated", lane)
}
seen[lane] = struct{}{}
lanes = append(lanes, lane)
}
sort.Strings(lanes)
return lanes, nil
}
type indexFile struct {
@@ -107,10 +230,11 @@ type indexFile struct {
OutputFiles []outputFileIndex `json:"output_files"`
RejectedFile string `json:"rejected_file"`
WarningsFile string `json:"warnings_file"`
ChunkMap *chunkMapIndex `json:"chunk_map,omitempty"`
ChunkMap *artifactIndex `json:"chunk_map,omitempty"`
EvidenceContext *artifactIndex `json:"evidence_context,omitempty"`
}
type chunkMapIndex struct {
type artifactIndex struct {
ArtifactKind contracts.ArtifactKind `json:"artifact_kind"`
File string `json:"file"`
MediaType string `json:"media_type"`
@@ -144,7 +268,7 @@ func logicalFiles(req contracts.OutputRequest, options Options) ([]contracts.Out
})
outputIndexes := make([]outputFileIndex, 0, len(outputs))
files := make([]contracts.OutputFile, 0, len(outputs)+4)
files := make([]contracts.OutputFile, 0, len(outputs)+5)
manifestFile, err := jsonFile("manifest.json", req.Manifest)
if err != nil {
return nil, err
@@ -191,6 +315,14 @@ func logicalFiles(req contracts.OutputRequest, options Options) ([]contracts.Out
files = append(files, chunkMapOutput)
index.ChunkMap = &chunkMapDescriptor
}
if req.EvidenceContext != nil {
evidenceOutput, evidenceDescriptor, err := serializedEvidenceContextFile(*req.EvidenceContext)
if err != nil {
return nil, err
}
files = append(files, evidenceOutput)
index.EvidenceContext = &evidenceDescriptor
}
indexOutput, err := jsonFile("index.json", index)
if err != nil {
return nil, err
@@ -210,24 +342,24 @@ func logicalFiles(req contracts.OutputRequest, options Options) ([]contracts.Out
return files, nil
}
func serializedChunkMapFile(artifact contracts.SerializedArtifact) (contracts.OutputFile, chunkMapIndex, error) {
func serializedChunkMapFile(artifact contracts.SerializedArtifact) (contracts.OutputFile, artifactIndex, error) {
if artifact.Kind != chunkmap.ArtifactKind {
return contracts.OutputFile{}, chunkMapIndex{}, encoderErrorf("chunk map has unexpected artifact kind %q", artifact.Kind)
return contracts.OutputFile{}, artifactIndex{}, encoderErrorf("chunk map has unexpected artifact kind %q", artifact.Kind)
}
if artifact.Schema.ID != chunkmap.SchemaID || artifact.Schema.Name != chunkmap.SchemaName || artifact.Schema.Version != chunkmap.SchemaVersion {
return contracts.OutputFile{}, chunkMapIndex{}, encoderErrorf("chunk map has unexpected schema identity")
return contracts.OutputFile{}, artifactIndex{}, encoderErrorf("chunk map has unexpected schema identity")
}
if strings.TrimSpace(artifact.MediaType) != chunkmap.MediaType {
return contracts.OutputFile{}, chunkMapIndex{}, encoderErrorf("chunk map has unsupported media type %q", artifact.MediaType)
return contracts.OutputFile{}, artifactIndex{}, encoderErrorf("chunk map has unsupported media type %q", artifact.MediaType)
}
if _, err := chunkmap.New().Decode(artifact.Content); err != nil {
return contracts.OutputFile{}, chunkMapIndex{}, encoderErrorf("decode chunk map: %w", err)
return contracts.OutputFile{}, artifactIndex{}, encoderErrorf("decode chunk map: %w", err)
}
file, err := serializedOutputFile(chunkMapFileName, artifact)
if err != nil {
return contracts.OutputFile{}, chunkMapIndex{}, err
return contracts.OutputFile{}, artifactIndex{}, err
}
return file, chunkMapIndex{
return file, artifactIndex{
ArtifactKind: artifact.Kind,
File: chunkMapFileName,
MediaType: chunkmap.MediaType,
@@ -237,6 +369,32 @@ func serializedChunkMapFile(artifact contracts.SerializedArtifact) (contracts.Ou
}, nil
}
func serializedEvidenceContextFile(artifact contracts.SerializedArtifact) (contracts.OutputFile, artifactIndex, error) {
codec := evidencecontext.New()
expected := codec.Schema()
if artifact.Kind != evidencecontext.ArtifactKind ||
artifact.Schema.ID != expected.ID || artifact.Schema.Name != expected.Name || artifact.Schema.Version != expected.Version ||
contracts.DigestArtifactSchema(artifact.Schema) != contracts.DigestArtifactSchema(expected) ||
strings.TrimSpace(artifact.MediaType) != evidencecontext.MediaType {
return contracts.OutputFile{}, artifactIndex{}, encoderErrorf("evidence context artifact is invalid")
}
if _, err := codec.Decode(artifact.Content); err != nil {
return contracts.OutputFile{}, artifactIndex{}, encoderErrorf("evidence context artifact is invalid")
}
file, err := serializedOutputFile(evidenceContextFileName, artifact)
if err != nil {
return contracts.OutputFile{}, artifactIndex{}, encoderErrorf("evidence context artifact is invalid")
}
return file, artifactIndex{
ArtifactKind: evidencecontext.ArtifactKind,
File: evidenceContextFileName,
MediaType: evidencecontext.MediaType,
SchemaID: expected.ID,
SchemaName: expected.Name,
SchemaVersion: expected.Version,
}, nil
}
func serializedOutputFile(name string, artifact contracts.SerializedArtifact) (contracts.OutputFile, error) {
content := append([]byte(nil), artifact.Content...)
if len(content) == 0 {

View File

@@ -11,6 +11,7 @@ import (
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/chunkmap"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/evidencecontext"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
)
@@ -68,13 +69,62 @@ func TestDecodeOptions(t *testing.T) {
if err != nil {
t.Fatalf("DecodeOptions() error = %v, want nil", err)
}
if got != test.want {
if !reflect.DeepEqual(got, test.want) {
t.Fatalf("DecodeOptions() = %#v, want %#v", got, test.want)
}
})
}
}
func TestDecodeEvidenceContextOptions(t *testing.T) {
for _, test := range []struct {
name string
options map[string]any
want pipeline.EvidenceContextPolicy
wantErr string
}{
{name: "disabled", options: map[string]any{"evidence_context": map[string]any{"enabled": false}}},
{name: "enabled default window", options: map[string]any{"evidence_context": map[string]any{"enabled": true, "lanes": []any{"npcs"}}}, want: pipeline.EvidenceContextPolicy{Enabled: true, WindowUnits: 3, LaneIDs: []string{"npcs"}}},
{name: "explicit zero window and normalized lanes", options: map[string]any{"evidence_context": map[string]any{"enabled": true, "lanes": []any{" spells ", "npcs"}, "window_units": 0}}, want: pipeline.EvidenceContextPolicy{Enabled: true, WindowUnits: 0, LaneIDs: []string{"npcs", "spells"}}},
{name: "duplicate lanes", options: map[string]any{"evidence_context": map[string]any{"enabled": true, "lanes": []any{"npcs", " npcs "}}}, wantErr: "duplicated"},
{name: "unknown nested option", options: map[string]any{"evidence_context": map[string]any{"enabled": false, "extra": true}}, wantErr: "unknown option"},
{name: "disabled nested fields", options: map[string]any{"evidence_context": map[string]any{"enabled": false, "lanes": []any{"npcs"}}}, wantErr: "not allowed"},
{name: "invalid object", options: map[string]any{"evidence_context": true}, wantErr: "must be an object"},
{name: "invalid lane type", options: map[string]any{"evidence_context": map[string]any{"enabled": true, "lanes": "npcs"}}, wantErr: "must be an array"},
{name: "invalid window type", options: map[string]any{"evidence_context": map[string]any{"enabled": true, "lanes": []any{"npcs"}, "window_units": "3"}}, wantErr: "must be an integer"},
} {
t.Run(test.name, func(t *testing.T) {
got, err := DecodeOptions(test.options)
if test.wantErr != "" {
if err == nil || !strings.Contains(err.Error(), test.wantErr) {
t.Fatalf("DecodeOptions() error = %v, want %q", err, test.wantErr)
}
return
}
if err != nil {
t.Fatalf("DecodeOptions() error = %v, want nil", err)
}
if !reflect.DeepEqual(got.EvidenceContext, test.want) {
t.Fatalf("EvidenceContext = %#v, want %#v", got.EvidenceContext, test.want)
}
})
}
}
func TestProfileValidationRejectsUnknownEvidenceLaneAndCopiesInputs(t *testing.T) {
registry := pipeline.NewOutputEncoderRegistry()
if err := Register(registry); err != nil {
t.Fatal(err)
}
options := map[string]any{"evidence_context": map[string]any{"enabled": true, "lanes": []any{"npcs"}}}
if err := registry.ValidateProfileOptions(Key, pipeline.OutputProfileOptionContext{LaneIDs: []string{"npcs", "spells"}}, options); err != nil {
t.Fatalf("ValidateProfileOptions() error = %v, want nil", err)
}
if err := registry.ValidateProfileOptions(Key, pipeline.OutputProfileOptionContext{LaneIDs: []string{"spells"}}, options); err == nil || !strings.Contains(err.Error(), "not configured") {
t.Fatalf("ValidateProfileOptions() error = %v, want unknown lane failure", err)
}
}
func TestEncodeReturnsLogicalFilesForNormalizedOutputs(t *testing.T) {
req := contracts.OutputRequest{
Manifest: artifacts.RunManifest{RunID: "run-1", PipelineID: "pipeline-1"},
@@ -289,6 +339,82 @@ func TestEncodeRejectsInvalidChunkMapArtifact(t *testing.T) {
}
}
func TestEncodeIncludesValidatedEvidenceContext(t *testing.T) {
artifact := acceptedEvidenceContextArtifact(t)
result, err := New().Encode(context.Background(), contracts.OutputRequest{EvidenceContext: &artifact})
if err != nil {
t.Fatalf("Encode() error = %v", err)
}
if got := string(fileBytes(t, result.Files, evidenceContextFileName)); !strings.HasSuffix(got, "\n") || !stdjson.Valid([]byte(got)) {
t.Fatalf("evidence context file = %q, want pretty valid newline-terminated JSON", got)
}
value, err := evidencecontext.New().Decode(fileBytes(t, result.Files, evidenceContextFileName))
if err != nil {
t.Fatalf("Decode(evidence context file) error = %v", err)
}
if len(value.Contexts) != 0 {
t.Fatalf("evidence context = %#v, want explicit empty contexts", value)
}
index := decodeObject(t, fileBytes(t, result.Files, "index.json"))
if got, want := index["evidence_context"], map[string]any{
"artifact_kind": string(evidencecontext.ArtifactKind),
"file": evidenceContextFileName,
"media_type": evidencecontext.MediaType,
"schema_id": evidencecontext.SchemaID,
"schema_name": evidencecontext.SchemaName,
"schema_version": evidencecontext.SchemaVersion,
}; !reflect.DeepEqual(got, want) {
t.Fatalf("evidence context descriptor = %#v, want %#v", got, want)
}
for _, entry := range index["output_files"].([]any) {
if entry.(map[string]any)["file"] == evidenceContextFileName {
t.Fatalf("output_files = %#v, want no evidence context lane entry", index["output_files"])
}
}
}
func TestEncodeRejectsInvalidEvidenceContextArtifactWithoutContentLeakage(t *testing.T) {
artifact := acceptedEvidenceContextArtifact(t)
for _, test := range []struct {
name string
mutate func(*contracts.SerializedArtifact)
}{
{name: "kind", mutate: func(artifact *contracts.SerializedArtifact) { artifact.Kind = "other/evidence" }},
{name: "schema identity", mutate: func(artifact *contracts.SerializedArtifact) { artifact.Schema.Version = "v2" }},
{name: "schema digest", mutate: func(artifact *contracts.SerializedArtifact) { artifact.Schema.JSONSchema[0] = '[' }},
{name: "media type", mutate: func(artifact *contracts.SerializedArtifact) { artifact.MediaType = "text/plain" }},
{name: "payload", mutate: func(artifact *contracts.SerializedArtifact) {
artifact.Content = []byte(`{"source_id":"secret transcript text"}`)
}},
} {
t.Run(test.name, func(t *testing.T) {
candidate := contracts.CloneSerializedArtifact(artifact)
test.mutate(&candidate)
_, err := New().Encode(context.Background(), contracts.OutputRequest{EvidenceContext: &candidate})
if err == nil || err.Error() != "json output encoder: evidence context artifact is invalid" {
t.Fatalf("Encode() error = %v, want fixed evidence artifact error", err)
}
if strings.Contains(err.Error(), "secret transcript text") {
t.Fatalf("Encode() leaked artifact content: %v", err)
}
})
}
}
func TestEncodeOmitsEvidenceContextWhenArtifactIsAbsent(t *testing.T) {
result, err := New().Encode(context.Background(), contracts.OutputRequest{Manifest: artifacts.RunManifest{RunID: "run-1"}})
if err != nil {
t.Fatalf("Encode() error = %v", err)
}
if got := outputFileNames(result.Files); containsString(got, evidenceContextFileName) {
t.Fatalf("file names = %#v, want no evidence context", got)
}
index := decodeObject(t, fileBytes(t, result.Files, "index.json"))
if _, ok := index["evidence_context"]; ok {
t.Fatalf("index = %#v, want no evidence context descriptor", index)
}
}
func TestEncodeChunkMapDoesNotMutateRequest(t *testing.T) {
artifact := acceptedChunkMapArtifact(t)
artifact.Metadata = map[string]any{"owner": "caller"}
@@ -600,6 +726,33 @@ func acceptedChunkMapArtifactWithPlanAnnotation(t *testing.T, annotation stdjson
return artifact
}
func acceptedEvidenceContextArtifact(t *testing.T) contracts.SerializedArtifact {
t.Helper()
document := &source.SourceDocument{
ID: "source-1",
Kind: "text",
Format: "text/plain",
Units: []source.SourceUnit{{
ID: 7,
Kind: "text",
Text: "Source content retained only in the evidence artifact.",
Ref: source.SourceRef{SourceID: "source-1", StartUnitID: 7, EndUnitID: 7},
}},
}
digest, err := source.DigestDocument(document)
if err != nil {
t.Fatal(err)
}
document.Digest = digest
artifact, err := evidencecontext.Serialize(evidencecontext.BuildRequest{
Source: document, WindowUnits: 3, SelectedLanes: []string{"spells"},
})
if err != nil {
t.Fatal(err)
}
return artifact
}
func outputFileNames(files []contracts.OutputFile) []string {
names := make([]string, 0, len(files))
for _, file := range files {

View File

@@ -153,7 +153,7 @@ func TestRunnerIndependentlyBoundsWorkersAndProviderCallsAcrossRegisteredModules
if err != nil {
t.Fatalf("Resolve() error = %v", err)
}
registries := pipeline.Registries{Inputs: catalog.Inputs, Chunkers: catalog.Chunkers, ArtifactCodecs: catalog.ArtifactCodecs, Extractors: catalog.Extractors, Mergers: catalog.Mergers, Normalizers: catalog.Normalizers, Validators: catalog.Validators, ValidatorChains: catalog.ValidatorChains, Outputs: catalog.Outputs}
registries := pipeline.Registries{Inputs: catalog.Inputs, Chunkers: catalog.Chunkers, ArtifactCodecs: catalog.ArtifactCodecs, ArtifactEvidence: catalog.ArtifactEvidence, Extractors: catalog.Extractors, Mergers: catalog.Mergers, Normalizers: catalog.Normalizers, Validators: catalog.Validators, ValidatorChains: catalog.ValidatorChains, Outputs: catalog.Outputs}
output, err := runPreparedPipeline(t, registries, resolved.ResolvedPipeline, client, pipeline.RunInput{RawInput: readDNDSpellsFixture(t), ExtractWorkers: 3})
if err != nil {
t.Fatalf("Run() error = %v", err)

View File

@@ -7,6 +7,7 @@ import (
"fmt"
"os"
"path/filepath"
"reflect"
"strings"
"sync"
"testing"
@@ -14,6 +15,7 @@ import (
"gitea.maximumdirect.net/eric/notarius/internal/core/config"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/evidencecontext"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd"
combatcodec "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/codec/combatturns"
@@ -254,6 +256,65 @@ func TestNPCOutputGroundsSpellAndCombatConsumersThroughOneOperation(t *testing.T
assertCurrentEvidence(t, combatValue.CombatTurns[0].SourceRefs)
}
func TestProductionDNDOutputPublishesSelectedEvidenceContext(t *testing.T) {
registries := productionNPCRegistries(t)
configValue := loadGroundedPipelineConfig(t)
profile := configValue.Pipelines["dnd-npc-grounded"]
profile.Output.Options = map[string]any{"evidence_context": map[string]any{
"enabled": true, "window_units": 0, "lanes": []any{"npcs", "spells", "combat"},
}}
configValue.Pipelines["dnd-npc-grounded"] = profile
raw := strings.NewReplacer(
`"id": 1`, `"id": 10`,
`"id": 2`, `"id": 30`,
`"id": 3`, `"id": 20`,
`"id": 4`, `"id": 50`,
`"id": 5`, `"id": 40`,
).Replace(string(readNPCFixture(t)))
output := runGroundedPipelineWithRaw(t, configValue, registries, &groundedDNDLLMClient{firstUnitID: 10, thirdUnitID: 20}, nil, []byte(raw))
value, err := evidencecontext.New().Decode(outputFileContent(t, output.OutputFiles, "evidence-context.json"))
if err != nil {
t.Fatalf("Decode(evidence context) error = %v", err)
}
if !reflect.DeepEqual(value.SelectedLanes, []string{"combat", "npcs", "spells"}) {
t.Fatalf("selected lanes = %#v, want configured production lanes without scene descriptions", value.SelectedLanes)
}
if len(value.Contexts) != 2 || len(value.Contexts[0].Units) != 1 || len(value.Contexts[1].Units) != 1 || value.Contexts[0].Units[0].ID != 10 || value.Contexts[1].Units[0].ID != 20 {
t.Fatalf("evidence contexts = %#v, want source-position union with non-monotonic unit IDs", value.Contexts)
}
firstRefs := value.Contexts[0].EvidenceRefs
if len(firstRefs) != 3 || firstRefs[0].LaneID != "combat" || firstRefs[1].LaneID != "npcs" || firstRefs[2].LaneID != "spells" {
t.Fatalf("first context evidence = %#v, want overlapping selected lane references", firstRefs)
}
for _, context := range value.Contexts {
for _, reference := range context.EvidenceRefs {
if reference.LaneID == "scene-descriptions" {
t.Fatalf("evidence refs = %#v, want scene descriptions excluded by allowlist", value.Contexts)
}
}
}
}
func TestProductionDNDOutputCanExplicitlySelectSceneDescriptionEvidence(t *testing.T) {
registries := productionNPCRegistries(t)
configValue := loadGroundedPipelineConfig(t)
profile := configValue.Pipelines["dnd-npc-grounded"]
profile.Output.Options = map[string]any{"evidence_context": map[string]any{
"enabled": true, "lanes": []any{"scene-descriptions"},
}}
configValue.Pipelines["dnd-npc-grounded"] = profile
output := runGroundedPipeline(t, configValue, registries, &groundedDNDLLMClient{}, nil)
value, err := evidencecontext.New().Decode(outputFileContent(t, output.OutputFiles, "evidence-context.json"))
if err != nil {
t.Fatalf("Decode(evidence context) error = %v", err)
}
if !reflect.DeepEqual(value.SelectedLanes, []string{"scene-descriptions"}) || len(value.Contexts) == 0 || len(value.Contexts[0].EvidenceRefs) == 0 || value.Contexts[0].EvidenceRefs[0].LaneID != "scene-descriptions" {
t.Fatalf("evidence context = %#v, want explicitly selected scene-description evidence", value)
}
}
func TestGroundedPipelineSkipsCombatForExactNarrativeScene(t *testing.T) {
registries := productionNPCRegistries(t)
configValue := loadGroundedPipelineConfig(t)
@@ -338,6 +399,10 @@ func configureGroundedCampaignReferences(t *testing.T, cfg *config.Config) map[s
}
func runGroundedPipeline(t *testing.T, configValue config.Config, registries pipeline.Registries, client *groundedDNDLLMClient, checkpoint pipeline.CheckpointLoader) pipeline.RunOutput {
return runGroundedPipelineWithRaw(t, configValue, registries, client, checkpoint, readNPCFixture(t))
}
func runGroundedPipelineWithRaw(t *testing.T, configValue config.Config, registries pipeline.Registries, client *groundedDNDLLMClient, checkpoint pipeline.CheckpointLoader, raw []byte) pipeline.RunOutput {
t.Helper()
catalog := moduleCatalog(registries)
effective, err := configValue.Resolve(config.ResolveInput{PipelineID: "dnd-npc-grounded", Catalog: catalog})
@@ -354,7 +419,7 @@ func runGroundedPipeline(t *testing.T, configValue config.Config, registries pip
}
output, err := pipeline.New().Run(context.Background(), pipeline.RunInput{
Prepared: prepared,
RawInput: readNPCFixture(t),
RawInput: append([]byte(nil), raw...),
ExtractWorkers: 1,
Checkpoint: checkpoint,
})
@@ -364,6 +429,17 @@ func runGroundedPipeline(t *testing.T, configValue config.Config, registries pip
return output
}
func outputFileContent(t *testing.T, files []contracts.OutputFile, name string) []byte {
t.Helper()
for _, file := range files {
if file.Name == name {
return append([]byte(nil), file.Bytes...)
}
}
t.Fatalf("output files = %#v, missing %q", files, name)
return nil
}
func normalizedCombatOutput(t *testing.T, output pipeline.RunOutput) dnd.CombatTurnList {
t.Helper()
for _, serialized := range output.NormalizeOutputs {
@@ -440,6 +516,8 @@ type groundedDNDLLMClient struct {
requests []contracts.StructuredCompletionRequest
sceneKind dnd.SceneKind
sceneTitle string
firstUnitID int
thirdUnitID int
}
func (client *groundedDNDLLMClient) CompleteStructured(ctx context.Context, request contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) {
@@ -451,14 +529,22 @@ func (client *groundedDNDLLMClient) CompleteStructured(ctx context.Context, requ
client.mu.Unlock()
var payload any
firstUnitID := client.firstUnitID
if firstUnitID == 0 {
firstUnitID = 1
}
thirdUnitID := client.thirdUnitID
if thirdUnitID == 0 {
thirdUnitID = 3
}
switch request.PromptID {
case npcs.PromptID:
payload = map[string]any{"npcs": []any{
map[string]any{
"name": "Mira Thorn", "source_refs": []any{map[string]int{"start_unit_id": 1, "end_unit_id": 1}},
"name": "Mira Thorn", "source_refs": []any{map[string]int{"start_unit_id": firstUnitID, "end_unit_id": firstUnitID}},
},
map[string]any{
"name": "Hooded Guard", "source_refs": []any{map[string]int{"start_unit_id": 3, "end_unit_id": 3}},
"name": "Hooded Guard", "source_refs": []any{map[string]int{"start_unit_id": thirdUnitID, "end_unit_id": thirdUnitID}},
},
}}
case npcnormalize.PromptID:
@@ -476,13 +562,13 @@ func (client *groundedDNDLLMClient) CompleteStructured(ctx context.Context, requ
case spells.PromptID:
payload = map[string]any{"spell_casts": []any{map[string]any{
"caster": "Mira Thorn", "spell": "Cure Wounds",
"source_refs": []any{map[string]int{"start_unit_id": 1, "end_unit_id": 1}},
"source_refs": []any{map[string]int{"start_unit_id": firstUnitID, "end_unit_id": firstUnitID}},
}}}
case combatextract.PromptID:
payload = map[string]any{"combat_turns": []any{map[string]any{
"actor": "Mira Thorn",
"turn_kind": "turn",
"source_refs": []any{map[string]int{"start_unit_id": 1, "end_unit_id": 1}},
"source_refs": []any{map[string]int{"start_unit_id": firstUnitID, "end_unit_id": firstUnitID}},
}}}
default:
return contracts.StructuredCompletionResponse{}, fmt.Errorf("unexpected grounded prompt %q", request.PromptID)

View File

@@ -290,6 +290,7 @@ func productionNPCRegistries(t *testing.T) pipeline.Registries {
Inputs: pipeline.NewInputAdapterRegistry(),
Chunkers: pipeline.NewChunkerRegistry(),
ArtifactCodecs: pipeline.NewArtifactCodecRegistry(),
ArtifactEvidence: pipeline.NewArtifactEvidenceRegistry(),
Extractors: pipeline.NewExtractorRegistry(),
Mergers: pipeline.NewMergerRegistry(),
Normalizers: pipeline.NewNormalizerRegistry(),
@@ -315,7 +316,7 @@ func productionNPCRegistries(t *testing.T) pipeline.Registries {
func moduleCatalog(registries pipeline.Registries) pipeline.ModuleCatalog {
return pipeline.ModuleCatalog{
Inputs: registries.Inputs, Chunkers: registries.Chunkers, ArtifactCodecs: registries.ArtifactCodecs,
Inputs: registries.Inputs, Chunkers: registries.Chunkers, ArtifactCodecs: registries.ArtifactCodecs, ArtifactEvidence: registries.ArtifactEvidence,
Extractors: registries.Extractors, Mergers: registries.Mergers, Normalizers: registries.Normalizers,
Validators: registries.Validators, ValidatorChains: registries.ValidatorChains, Outputs: registries.Outputs,
}

View File

@@ -88,6 +88,7 @@ func dndCapabilityCatalog(t *testing.T, inputSpec, extractorSpec pipeline.Module
}
codecs := pipeline.NewArtifactCodecRegistry()
evidence := pipeline.NewArtifactEvidenceRegistry()
if err := pipeline.RegisterArtifactCodec(codecs, spellcodec.New()); err != nil {
t.Fatalf("register capability codec: %v", err)
}
@@ -112,7 +113,7 @@ func dndCapabilityCatalog(t *testing.T, inputSpec, extractorSpec pipeline.Module
}
return pipeline.ModuleCatalog{
Inputs: inputs, Chunkers: chunkers, ArtifactCodecs: codecs, Extractors: extractors,
Inputs: inputs, Chunkers: chunkers, ArtifactCodecs: codecs, ArtifactEvidence: evidence, Extractors: extractors,
Mergers: mergers, Normalizers: normalizers, ValidatorChains: pipeline.NewValidatorChainRegistry(), Outputs: outputs,
}
}
@@ -148,6 +149,7 @@ func dndSpellsTestCatalog(t *testing.T, specs dndSpellsCatalogSpecs) pipeline.Mo
chunkers := pipeline.NewChunkerRegistry()
extractors := pipeline.NewExtractorRegistry()
codecs := pipeline.NewArtifactCodecRegistry()
evidence := pipeline.NewArtifactEvidenceRegistry()
mergers := pipeline.NewMergerRegistry()
normalizers := pipeline.NewNormalizerRegistry()
outputs := pipeline.NewOutputEncoderRegistry()
@@ -221,6 +223,7 @@ func dndSpellsTestCatalog(t *testing.T, specs dndSpellsCatalogSpecs) pipeline.Mo
Inputs: inputs,
Chunkers: chunkers,
ArtifactCodecs: codecs,
ArtifactEvidence: evidence,
Extractors: extractors,
Mergers: mergers,
Normalizers: normalizers,

View File

@@ -287,6 +287,7 @@ func dndSpellsRunnerRegistries(t *testing.T) pipeline.Registries {
Inputs: catalog.Inputs,
Chunkers: catalog.Chunkers,
ArtifactCodecs: catalog.ArtifactCodecs,
ArtifactEvidence: catalog.ArtifactEvidence,
Extractors: catalog.Extractors,
Mergers: catalog.Mergers,
Normalizers: catalog.Normalizers,