From 49d94cc2e91ddcc9216bdd75bd5ad34af638a6c2 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sun, 5 Jul 2026 17:51:36 +0000 Subject: [PATCH] Add prompt input materials and session IDs --- internal/cli/run.go | 45 ++++++- internal/cli/run_test.go | 100 ++++++++++++++++ internal/core/artifacts/artifacts.go | 1 + internal/framework/contracts/contracts.go | 78 +++++++++--- .../framework/contracts/contracts_test.go | 47 ++++++++ internal/framework/pipeline/runner.go | 113 ++++++++++++++---- internal/framework/pipeline/runner_test.go | 78 ++++++++++++ 7 files changed, 423 insertions(+), 39 deletions(-) diff --git a/internal/cli/run.go b/internal/cli/run.go index 0d0b8e2..93296b9 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -25,7 +25,7 @@ const defaultOutputRoot = "./notarius-output" const usage = `Usage: notarius help - notarius run --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] [--reference selector=path] [--without-reference selector] + notarius run --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] [--session-id id] [--reference selector=path] [--without-reference selector] notarius config validate --config path/to/config.yml [--pipeline pipeline-id] [--only lane-a,lane-b] notarius pipelines list --config path/to/config.yml [--json] ` @@ -95,10 +95,16 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i outputDir := fs.String("output-dir", "", "output directory") diagnosticsDir := fs.String("diagnostics-dir", "", "diagnostics directory") llmProfile := fs.String("llm-profile", "", "LLM profile override") + sessionID := sessionIDFlag{} referenceFlags := stringListFlag{} withoutReferenceFlags := stringListFlag{} + fs.Var(&sessionID, "session-id", "prompt session identifier") fs.Var(&referenceFlags, "reference", "reference binding, as slot=path, chunk.slot=path, lane.slot=path, lane.extract.slot=path, or lane.normalize.slot=path") fs.Var(&withoutReferenceFlags, "without-reference", "unbind a reference, using the same selector forms as --reference") + if err := validateRunFlagValues(args); err != nil { + fmt.Fprintf(stderr, "notarius: %v\n", err) + return 2 + } if err := fs.Parse(reorderRunArgs(args)); err != nil { fmt.Fprintf(stderr, "notarius: %v\n", err) return 2 @@ -120,6 +126,10 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i fmt.Fprintln(stderr, "notarius: run requires --input") return 2 } + if sessionID.set && strings.TrimSpace(sessionID.value) == "" { + fmt.Fprintln(stderr, "notarius: --session-id must not be empty") + return 2 + } only, err := parseOnly(*onlyRaw) if err != nil { fmt.Fprintf(stderr, "notarius: %v\n", err) @@ -236,6 +246,7 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i Path: strings.TrimSpace(*inputPath), RawInput: rawInput, LLMClient: llmClient, + SessionID: strings.TrimSpace(sessionID.value), RunID: runDir.RunID(), StartedAt: startedAt, LLMProfiles: llmProfiles, @@ -448,13 +459,25 @@ func reorderRunArgs(args []string) []string { func runFlagTakesValue(arg string) bool { switch arg { - case "--config", "--input", "--only", "--output-dir", "--diagnostics-dir", "--llm-profile", "--reference", "--without-reference": + case "--config", "--input", "--only", "--output-dir", "--diagnostics-dir", "--llm-profile", "--session-id", "--reference", "--without-reference": return true default: return false } } +func validateRunFlagValues(args []string) error { + for i, arg := range args { + if arg != "--session-id" { + continue + } + if i+1 >= len(args) || strings.HasPrefix(args[i+1], "-") { + return fmt.Errorf("flag needs an argument: --session-id") + } + } + return nil +} + func effectiveLLMProfileIDs(resolved pipeline.ResolvedPipeline) []string { seen := make(map[string]struct{}) add := func(binding pipeline.ModuleBinding) { @@ -711,6 +734,24 @@ func (flag *stringListFlag) Set(value string) error { return nil } +type sessionIDFlag struct { + value string + set bool +} + +func (flag *sessionIDFlag) String() string { + if flag == nil { + return "" + } + return flag.value +} + +func (flag *sessionIDFlag) Set(value string) error { + flag.value = value + flag.set = true + return nil +} + type cliReferenceRequest struct { Selector cliReferenceSelector Source string diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index 3f3b01e..652272f 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -739,6 +739,102 @@ func TestRunPipelineLLMProfileOverrideSelectsFactoryProfile(t *testing.T) { } } +func TestRunPipelineSessionIDFlagRecordsExplicitTrimmedValue(t *testing.T) { + configPath := writeTestConfig(t, mvpConfigYAML("dnd-session", "dnd/spells")) + inputPath := writeSeriatimInput(t) + outputDir := t.TempDir() + diagnosticsDir := t.TempDir() + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{ + "run", "dnd-session", + "--config", configPath, + "--input", inputPath, + "--session-id", " external-session ", + "--output-dir", outputDir, + "--diagnostics-dir", diagnosticsDir, + }, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(newFakeRunLLMClient(false), nil), + }) + + if code != 0 { + t.Fatalf("RunWithOptions() code = %d, stderr=%q", code, stderr.String()) + } + var manifest artifacts.RunManifest + readJSONFile(t, filepath.Join(onlyChildDir(t, outputDir), "manifest.json"), &manifest) + if got := manifest.Metadata["session_id"]; got != "external-session" { + t.Fatalf("manifest metadata = %#v, want trimmed session ID", manifest.Metadata) + } +} + +func TestRunPipelineSessionIDDefaultsToParsedSourceID(t *testing.T) { + configPath := writeTestConfig(t, mvpConfigYAML("dnd-session", "dnd/spells")) + inputPath := writeSeriatimInput(t) + outputDir := t.TempDir() + diagnosticsDir := t.TempDir() + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{ + "run", "dnd-session", + "--config", configPath, + "--input", inputPath, + "--output-dir", outputDir, + "--diagnostics-dir", diagnosticsDir, + }, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(newFakeRunLLMClient(false), nil), + }) + + if code != 0 { + t.Fatalf("RunWithOptions() code = %d, stderr=%q", code, stderr.String()) + } + var manifest artifacts.RunManifest + readJSONFile(t, filepath.Join(onlyChildDir(t, outputDir), "manifest.json"), &manifest) + if got := manifest.Metadata["session_id"]; got != "session-alpha" { + t.Fatalf("manifest metadata = %#v, want parsed source ID default", manifest.Metadata) + } +} + +func TestRunPipelineSessionIDFlagRejectsMissingOrBlankValue(t *testing.T) { + configPath := writeTestConfig(t, mvpConfigYAML("dnd-session", "dnd/spells")) + inputPath := writeSeriatimInput(t) + tests := []struct { + name string + args []string + want string + }{ + { + name: "missing value", + args: []string{"run", "dnd-session", "--config", configPath, "--input", inputPath, "--session-id"}, + want: "flag needs an argument", + }, + { + name: "blank value", + args: []string{"run", "dnd-session", "--config", configPath, "--input", inputPath, "--session-id", " \t "}, + want: "--session-id must not be empty", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions(test.args, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(newFakeRunLLMClient(false), nil), + }) + + if code != 2 { + t.Fatalf("RunWithOptions() code = %d, want 2", code) + } + if !strings.Contains(stderr.String(), test.want) { + t.Fatalf("stderr = %q, want substring %q", stderr.String(), test.want) + } + }) + } +} + func TestRunPipelineReferenceFlagBindsUnambiguousSlot(t *testing.T) { configPath := writeTestConfig(t, testConfigYAML("example", "events")) inputPath := filepath.Join(t.TempDir(), "missing.json") @@ -1478,6 +1574,10 @@ func TestRunPipelineReferenceBytesProduceDistinctManifests(t *testing.T) { if !reflect.DeepEqual(resolvedReferences, manifest.References) { t.Fatalf("resolved references = %#v, want manifest references %#v", resolvedReferences, manifest.References) } + runManifestJSON := string(readFile(t, filepath.Join(onlyChildDir(t, diagnosticsDir), diagnostics.ArtifactRunManifest))) + if strings.Contains(runManifestJSON, "source text") || strings.Contains(runManifestJSON, referenceText) { + t.Fatalf("run manifest diagnostics contains raw prompt material: %s", runManifestJSON) + } resolvedReferenceJSON := string(readFile(t, filepath.Join(onlyChildDir(t, diagnosticsDir), diagnostics.ArtifactResolvedReferences))) if strings.Contains(resolvedReferenceJSON, referenceText) || strings.Contains(resolvedReferenceJSON, "content") { t.Fatalf("resolved references diagnostics contains content: %s", resolvedReferenceJSON) diff --git a/internal/core/artifacts/artifacts.go b/internal/core/artifacts/artifacts.go index e03981a..46a54ff 100644 --- a/internal/core/artifacts/artifacts.go +++ b/internal/core/artifacts/artifacts.go @@ -75,6 +75,7 @@ type RunManifest struct { ArtifactLanes []ArtifactLaneManifest `json:"artifact_lanes,omitempty"` References []ReferenceProvenance `json:"references,omitempty"` LLMProfiles []LLMProfileManifest `json:"llm_profiles,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` SchemaVersion string `json:"schema_version,omitempty"` ValidationStatus string `json:"validation_status,omitempty"` StartedAt *time.Time `json:"started_at,omitempty"` diff --git a/internal/framework/contracts/contracts.go b/internal/framework/contracts/contracts.go index 12383c9..142b65e 100644 --- a/internal/framework/contracts/contracts.go +++ b/internal/framework/contracts/contracts.go @@ -15,6 +15,12 @@ type LLMMessage struct { type StructuredCompletionRequest struct { StageName string `json:"stage_name"` + PromptID string `json:"prompt_id,omitempty"` + PromptVersion string `json:"prompt_version,omitempty"` + ProfileID string `json:"profile_id,omitempty"` + SessionID string `json:"session_id,omitempty"` + Inputs LLMInputSet `json:"inputs,omitempty"` + Vars map[string]any `json:"vars,omitempty"` Messages []LLMMessage `json:"messages"` Model string `json:"model,omitempty"` ResponseSchemaName string `json:"response_schema_name,omitempty"` @@ -34,6 +40,44 @@ type StructuredLLMClient interface { CompleteStructured(ctx context.Context, req StructuredCompletionRequest, out any) (StructuredCompletionResponse, error) } +type LLMInputMaterial struct { + Name string `json:"name"` + MediaType string `json:"media_type,omitempty"` + Content []byte `json:"-"` + Digest string `json:"digest,omitempty"` + OriginURI string `json:"origin_uri,omitempty"` + SizeBytes int64 `json:"size_bytes,omitempty"` +} + +func NewLLMInputMaterial(name string, mediaType string, content []byte, digest string, originURI string) LLMInputMaterial { + return LLMInputMaterial{ + Name: name, + MediaType: mediaType, + Content: append([]byte(nil), content...), + Digest: digest, + OriginURI: originURI, + SizeBytes: int64(len(content)), + } +} + +func (material LLMInputMaterial) Clone() LLMInputMaterial { + material.Content = append([]byte(nil), material.Content...) + return material +} + +type LLMInputSet map[string]LLMInputMaterial + +func (set LLMInputSet) Clone() LLMInputSet { + if len(set) == 0 { + return nil + } + out := make(LLMInputSet, len(set)) + for key, material := range set { + out[key] = material.Clone() + } + return out +} + type ParseRequest struct { SourceID string `json:"source_id,omitempty"` Path string `json:"path,omitempty"` @@ -57,12 +101,14 @@ type SourceChunk struct { } type ChunkRequest struct { - Source *source.SourceDocument `json:"-"` - References ReferenceSet `json:"references,omitempty"` - LLMClient StructuredLLMClient `json:"-"` - LLMProfile string `json:"llm_profile,omitempty"` - Options map[string]any `json:"options,omitempty"` - Metadata map[string]any `json:"metadata,omitempty"` + Source *source.SourceDocument `json:"-"` + SourceInput LLMInputMaterial `json:"source_input,omitempty"` + SessionID string `json:"session_id,omitempty"` + References ReferenceSet `json:"references,omitempty"` + LLMClient StructuredLLMClient `json:"-"` + LLMProfile string `json:"llm_profile,omitempty"` + Options map[string]any `json:"options,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` } type ChunkResult struct { @@ -118,6 +164,8 @@ type ExtractionRequest struct { Source *source.SourceDocument `json:"-"` Chunk *SourceChunk `json:"chunk,omitempty"` AmbientContext map[string]any `json:"ambient_context,omitempty"` + SourceInput LLMInputMaterial `json:"source_input,omitempty"` + SessionID string `json:"session_id,omitempty"` References ReferenceSet `json:"references,omitempty"` LLMClient StructuredLLMClient `json:"-"` LLMProfile string `json:"llm_profile,omitempty"` @@ -164,14 +212,16 @@ type Merger interface { } type NormalizeRequest struct { - Source *source.SourceDocument `json:"-"` - LaneID string `json:"lane_id"` - Candidates []artifacts.ArtifactCandidate `json:"candidates"` - References ReferenceSet `json:"references,omitempty"` - LLMClient StructuredLLMClient `json:"-"` - LLMProfile string `json:"llm_profile,omitempty"` - Options map[string]any `json:"options,omitempty"` - Metadata map[string]any `json:"metadata,omitempty"` + Source *source.SourceDocument `json:"-"` + LaneID string `json:"lane_id"` + Candidates []artifacts.ArtifactCandidate `json:"candidates"` + SourceInput LLMInputMaterial `json:"source_input,omitempty"` + SessionID string `json:"session_id,omitempty"` + References ReferenceSet `json:"references,omitempty"` + LLMClient StructuredLLMClient `json:"-"` + LLMProfile string `json:"llm_profile,omitempty"` + Options map[string]any `json:"options,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` } type NormalizeResult struct { diff --git a/internal/framework/contracts/contracts_test.go b/internal/framework/contracts/contracts_test.go index 9e8db95..21dfe6e 100644 --- a/internal/framework/contracts/contracts_test.go +++ b/internal/framework/contracts/contracts_test.go @@ -251,6 +251,53 @@ func TestReferenceItemJSONOmitsContent(t *testing.T) { } } +func TestLLMInputMaterialCopiesContentAndOmitsContentFromJSON(t *testing.T) { + content := []byte("raw source bytes") + material := NewLLMInputMaterial("transcript", "application/json", content, "sha256:source", "file:///tmp/source.json") + content[0] = 'R' + if got := string(material.Content); got != "raw source bytes" { + t.Fatalf("material content = %q, want defensive copy", got) + } + if material.SizeBytes != int64(len("raw source bytes")) { + t.Fatalf("SizeBytes = %d, want content length", material.SizeBytes) + } + + clone := material.Clone() + clone.Content[0] = 'X' + if got := string(material.Content); got != "raw source bytes" { + t.Fatalf("cloned material content aliased original: %q", got) + } + + encoded, err := json.Marshal(material) + if err != nil { + t.Fatalf("json.Marshal() error = %v, want nil", err) + } + var got map[string]any + if err := json.Unmarshal(encoded, &got); err != nil { + t.Fatalf("json.Unmarshal() error = %v, want nil", err) + } + if _, ok := got["content"]; ok { + t.Fatalf("encoded material leaked content: %s", encoded) + } + if _, ok := got["Content"]; ok { + t.Fatalf("encoded material leaked Content: %s", encoded) + } + if got["digest"] != "sha256:source" || got["origin_uri"] != "file:///tmp/source.json" { + t.Fatalf("encoded material = %#v, want non-secret provenance", got) + } +} + +func TestLLMInputSetCloneCopiesContent(t *testing.T) { + set := LLMInputSet{ + "transcript": NewLLMInputMaterial("transcript", "application/json", []byte("source"), "sha256:source", "file:///tmp/source.json"), + } + clone := set.Clone() + clone["transcript"].Content[0] = 'S' + if got := string(set["transcript"].Content); got != "source" { + t.Fatalf("input set clone aliased content: %q", got) + } +} + func TestFakeMergeNormalizeAndOutputContracts(t *testing.T) { candidate := artifacts.ArtifactCandidate{ Index: 0, diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 3f169ef..b430391 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -2,8 +2,12 @@ package pipeline import ( "context" + "crypto/sha256" + "encoding/hex" "fmt" + "mime" "path" + "path/filepath" "strings" "time" @@ -37,6 +41,7 @@ type RunInput struct { Path string RawInput []byte LLMClient contracts.StructuredLLMClient + SessionID string RunID string StartedAt time.Time LLMProfiles []artifacts.LLMProfileManifest @@ -86,6 +91,9 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) { if err := source.ValidateDocument(doc); err != nil { return failOutput(output), fmt.Errorf("validate source document: %w", err) } + sourceInput := sourceInputMaterial(input.Path, input.RawInput) + sessionID := resolvedSessionID(input.SessionID, doc.ID) + output.Manifest.Metadata = manifestMetadataWithSessionID(output.Manifest.Metadata, sessionID) output.Manifest.SourceDigests = []string{doc.Digest} chunker, err := r.registries.Chunkers.Build(input.Pipeline.Chunk.Module) @@ -94,12 +102,14 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) { } attachModuleManifestMetadata(&output, "chunker", chunker) chunkResult, err := chunker.Chunk(ctx, contracts.ChunkRequest{ - Source: doc, - References: CloneReferenceSet(input.Pipeline.ChunkReferences.ReferenceSet), - LLMClient: input.LLMClient, - LLMProfile: input.Pipeline.Chunk.LLMProfile, - Options: cloneOptions(input.Pipeline.Chunk.Options), - Metadata: input.Metadata, + Source: doc, + SourceInput: sourceInput.Clone(), + SessionID: sessionID, + References: CloneReferenceSet(input.Pipeline.ChunkReferences.ReferenceSet), + LLMClient: input.LLMClient, + LLMProfile: input.Pipeline.Chunk.LLMProfile, + Options: cloneOptions(input.Pipeline.Chunk.Options), + Metadata: input.Metadata, }) output.Warnings = append(output.Warnings, chunkResult.Warnings...) if err != nil { @@ -115,7 +125,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) { nextCandidateIndex := 0 for _, lane := range input.Pipeline.ArtifactLanes { - if err := r.runLane(ctx, input, doc, canonicalChunks, lane, &output, &nextCandidateIndex); err != nil { + if err := r.runLane(ctx, input, doc, sourceInput, sessionID, canonicalChunks, lane, &output, &nextCandidateIndex); err != nil { return failOutput(output), err } } @@ -154,7 +164,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) { return output, nil } -func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.SourceDocument, chunks []contracts.SourceChunk, lane ResolvedArtifactLane, output *RunOutput, nextCandidateIndex *int) error { +func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []contracts.SourceChunk, lane ResolvedArtifactLane, output *RunOutput, nextCandidateIndex *int) error { extractor, err := r.registries.Extractors.Build(lane.Extract.Module) if err != nil { return fmt.Errorf("build extractor %q for lane %q: %w", lane.Extract.Module, lane.ID, err) @@ -185,13 +195,15 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source for index := range chunks { chunk := chunks[index] result, err := extractor.Extract(ctx, contracts.ExtractionRequest{ - Source: doc, - Chunk: &chunk, - References: CloneReferenceSet(lane.ExtractReferences.ReferenceSet), - LLMClient: input.LLMClient, - LLMProfile: lane.Extract.LLMProfile, - Options: cloneOptions(lane.Extract.Options), - Metadata: input.Metadata, + Source: doc, + Chunk: &chunk, + SourceInput: sourceInput.Clone(), + SessionID: sessionID, + References: CloneReferenceSet(lane.ExtractReferences.ReferenceSet), + LLMClient: input.LLMClient, + LLMProfile: lane.Extract.LLMProfile, + Options: cloneOptions(lane.Extract.Options), + Metadata: input.Metadata, }) output.Warnings = append(output.Warnings, result.Warnings...) if err != nil { @@ -222,14 +234,16 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source } normalizeResult, err := normalizer.Normalize(ctx, contracts.NormalizeRequest{ - Source: doc, - LaneID: lane.ID, - Candidates: mergeResult.Candidates, - References: CloneReferenceSet(lane.NormalizeReferences.ReferenceSet), - LLMClient: input.LLMClient, - LLMProfile: lane.Normalize.LLMProfile, - Options: cloneOptions(lane.Normalize.Options), - Metadata: input.Metadata, + Source: doc, + LaneID: lane.ID, + Candidates: mergeResult.Candidates, + SourceInput: sourceInput.Clone(), + SessionID: sessionID, + References: CloneReferenceSet(lane.NormalizeReferences.ReferenceSet), + LLMClient: input.LLMClient, + LLMProfile: lane.Normalize.LLMProfile, + Options: cloneOptions(lane.Normalize.Options), + Metadata: input.Metadata, }) output.Warnings = append(output.Warnings, normalizeResult.Warnings...) if err != nil { @@ -510,6 +524,59 @@ func cloneLLMProfiles(profiles []artifacts.LLMProfileManifest) []artifacts.LLMPr return append([]artifacts.LLMProfileManifest(nil), profiles...) } +func sourceInputMaterial(inputPath string, content []byte) contracts.LLMInputMaterial { + return contracts.NewLLMInputMaterial( + "source", + sourceInputMediaType(inputPath), + content, + sourceInputDigest(content), + sourceInputOriginURI(inputPath), + ) +} + +func sourceInputMediaType(inputPath string) string { + extension := strings.ToLower(filepath.Ext(strings.TrimSpace(inputPath))) + if extension == ".json" { + return "application/json" + } + mediaType := mime.TypeByExtension(extension) + if strings.TrimSpace(mediaType) == "" { + return unknownMediaType + } + return canonicalMediaType(mediaType) +} + +func sourceInputDigest(content []byte) string { + sum := sha256.Sum256(content) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +func sourceInputOriginURI(inputPath string) string { + if strings.TrimSpace(inputPath) == "" { + return "" + } + return fileURI(inputPath) +} + +func resolvedSessionID(explicit string, sourceDocumentID string) string { + if trimmed := strings.TrimSpace(explicit); trimmed != "" { + return trimmed + } + return strings.TrimSpace(sourceDocumentID) +} + +func manifestMetadataWithSessionID(metadata map[string]any, sessionID string) map[string]any { + out := cloneMetadata(metadata) + if strings.TrimSpace(sessionID) == "" { + return out + } + if out == nil { + out = make(map[string]any) + } + out["session_id"] = sessionID + return out +} + func cloneWarnings(warnings []contracts.Warning) []contracts.Warning { if len(warnings) == 0 { return nil diff --git a/internal/framework/pipeline/runner_test.go b/internal/framework/pipeline/runner_test.go index bc01218..64b099b 100644 --- a/internal/framework/pipeline/runner_test.go +++ b/internal/framework/pipeline/runner_test.go @@ -476,6 +476,84 @@ func TestRunExecutesChunksAndPassesChunkAndLLMClient(t *testing.T) { } } +func TestRunPassesSourceInputAndSessionIDToPromptCapableStages(t *testing.T) { + modules := defaultRunnerModules() + rawInput := []byte("{\"source\":\"exact bytes\"}") + + output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ + Pipeline: resolvedPipeline(), + Path: "session.json", + RawInput: rawInput, + SessionID: " explicit-session ", + }) + if err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + + if got := output.Manifest.Metadata["session_id"]; got != "explicit-session" { + t.Fatalf("manifest metadata = %#v, want session_id", output.Manifest.Metadata) + } + + requests := []struct { + name string + material contracts.LLMInputMaterial + sessionID string + }{ + {name: "chunk", material: modules.chunker.requests[0].SourceInput, sessionID: modules.chunker.requests[0].SessionID}, + {name: "extract first", material: modules.extractors["extract-alpha"].requests[0].SourceInput, sessionID: modules.extractors["extract-alpha"].requests[0].SessionID}, + {name: "extract second", material: modules.extractors["extract-alpha"].requests[1].SourceInput, sessionID: modules.extractors["extract-alpha"].requests[1].SessionID}, + {name: "normalize", material: modules.normalizers["normalize"].requests[0].SourceInput, sessionID: modules.normalizers["normalize"].requests[0].SessionID}, + } + for _, req := range requests { + if req.sessionID != "explicit-session" { + t.Fatalf("%s session ID = %q, want explicit-session", req.name, req.sessionID) + } + if got := string(req.material.Content); got != string(rawInput) { + t.Fatalf("%s source input content = %q, want exact raw input", req.name, got) + } + if req.material.Name != "source" || req.material.MediaType != "application/json" || req.material.SizeBytes != int64(len(rawInput)) { + t.Fatalf("%s source input = %#v, want source metadata", req.name, req.material) + } + if req.material.Digest != sourceInputDigest(rawInput) { + t.Fatalf("%s digest = %q, want %q", req.name, req.material.Digest, sourceInputDigest(rawInput)) + } + if !strings.HasPrefix(req.material.OriginURI, "file://") || !strings.HasSuffix(req.material.OriginURI, "/session.json") { + t.Fatalf("%s origin URI = %q, want file URI ending in session.json", req.name, req.material.OriginURI) + } + } + + modules.chunker.requests[0].SourceInput.Content[0] = 'X' + if got := string(modules.extractors["extract-alpha"].requests[0].SourceInput.Content); got != string(rawInput) { + t.Fatalf("source input content aliased across requests: %q", got) + } + if got := string(rawInput); got != "{\"source\":\"exact bytes\"}" { + t.Fatalf("raw input mutated through request material: %q", got) + } +} + +func TestRunDefaultsSessionIDFromParsedSourceDocumentID(t *testing.T) { + modules := defaultRunnerModules() + + output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ + Pipeline: resolvedPipeline(), + Path: "notes.unknown", + RawInput: []byte("notes"), + }) + if err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + + if got := modules.chunker.requests[0].SessionID; got != "source-1" { + t.Fatalf("chunk session ID = %q, want parsed source document ID", got) + } + if got := output.Manifest.Metadata["session_id"]; got != "source-1" { + t.Fatalf("manifest metadata = %#v, want default session id", output.Manifest.Metadata) + } + if got := modules.chunker.requests[0].SourceInput.MediaType; got != unknownMediaType { + t.Fatalf("source input media type = %q, want fallback %q", got, unknownMediaType) + } +} + func TestRunPassesInputRequestFields(t *testing.T) { modules := defaultRunnerModules() metadata := map[string]any{"request": "test"}