diff --git a/docs/internal/stage-publish.md b/docs/internal/stage-publish.md index 53d4f80..8b75337 100644 --- a/docs/internal/stage-publish.md +++ b/docs/internal/stage-publish.md @@ -30,9 +30,10 @@ Exact remote placement and the operator workflow belong in both manifests record success. Enablement and upload policy are fingerprinted, so changing either automatically makes the prior result non-resumable. - validates prerequisite stage success and object-store availability. -- derives a deterministic run-archive allowlist from the validated run - `manifest.json`: declared run-local outputs, logs, generated configs, and the - manifest itself. Unlisted workspace files are not archive candidates. +- derives a deterministic run-archive allowlist from the session manifest's + declared current-run outputs, logs, and generated configurations. References + outside the current run root, including canonical session files and prior-run + diagnostics, are not archive candidates. No run manifest is read or uploaded. - opens each archive candidate beneath its archive root without following symlinked ancestors or leaf entries, verifies that it is a regular file and checks a declared checksum when present, then streams the opened descriptor. @@ -82,7 +83,8 @@ Includes counts/lists for: from verified descriptors; symlinks, special files, replacement races, and undeclared entries are rejected or ignored before uploads begin. - run-local diagnostics, including Notarius receipt and stderr files, are - archived only when recorded by the run manifest. + archived only when recorded by the session manifest. A run with no declared + current-run diagnostics legitimately uploads none. - publish locks are not overridden by `--force`; remote locks are revalidated immediately before current-state selection. - post-commit local cleanup is authorized by the committed publish metadata and diff --git a/docs/operations.md b/docs/operations.md index 07c5695..dee0772 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -308,12 +308,15 @@ Run-local diagnostics are: - `runs/{run_id}/extract/notarius.stderr.log` - `runs/{run_id}/extract/notarius-output/` before durable promotion -The run-record upload is an allowlist derived from the validated run manifest, -not a workspace scan. Each declared source is opened without following -symlinked ancestors or the leaf, verified as a regular file, and streamed from -that verified descriptor. Unlisted files and unsafe entries are never uploaded. -The durable bundle is never scanned for implicit publication; only lanes named -by explicit `pipeline.publish.outputs` rules are uploaded. +The run upload is an allowlist derived from the session manifest's declared +current-run outputs, logs, and generated configurations, not a workspace scan. +References to canonical session files and prior-run paths are excluded. Each +declared source is opened without following symlinked ancestors or the leaf, +verified as a regular file, and streamed from that verified descriptor. +Unlisted files and unsafe entries are never uploaded; a run with no declared +current-run diagnostics uploads no run-local files. No run manifest is +uploaded. The durable bundle is never scanned for implicit publication; only +lanes named by explicit `pipeline.publish.outputs` rules are uploaded. To intentionally replace the current extraction result, run: diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 85e8460..31f9c76 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -114,7 +114,7 @@ During every stage: ## Stage 1 — Decouple Publish From The Run Manifest -**Status: Pending** +**Status: Completed** ### Goal diff --git a/internal/stage/publish.go b/internal/stage/publish.go index 2f06a29..fc57819 100644 --- a/internal/stage/publish.go +++ b/internal/stage/publish.go @@ -75,17 +75,9 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S return nil, fmt.Errorf("publish: remote object store backend is required when publish run upload is enabled") } - runManifestPath, err := resolvePublishRunManifestPath(env, m) + runRoot, err := resolvePublishRunRoot(env, m) if err != nil { - return nil, fmt.Errorf("publish: resolve run manifest path: %w", err) - } - runRoot := filepath.Dir(runManifestPath) - runRootInfo, err := os.Lstat(runRoot) - if err != nil { - return nil, fmt.Errorf("publish: run root %q: %w", runRoot, err) - } - if runRootInfo.Mode()&os.ModeSymlink != 0 || !runRootInfo.IsDir() { - return nil, fmt.Errorf("publish: run root %q is not a directory", runRoot) + return nil, fmt.Errorf("publish: resolve run root: %w", err) } runPrefix, err := artifacts.ResolvePublishRunPrefix(env.Config, m) @@ -105,24 +97,11 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S return nil, fmt.Errorf("publish: run id is required") } - runManifestFile, err := openPublishOutputSource(filepath.Join(runRoot, filepath.Base(runManifestPath))) - if err != nil { - return nil, fmt.Errorf("publish: open run manifest: %w", err) - } - defer runManifestFile.Close() - runManifest, err := (&manifest.LocalStore{}).LoadRunReader(ctx, runManifestFile) - if err != nil { - return nil, fmt.Errorf("publish: load run manifest: %w", err) - } - if _, err := runManifestFile.Seek(0, io.SeekStart); err != nil { - return nil, fmt.Errorf("publish: rewind run manifest: %w", err) - } - - runFiles, err := collectPublishRunFiles(runRoot, runManifest) + runFiles, err := collectPublishRunFiles(runRoot, m) if err != nil { return nil, fmt.Errorf("publish: collect run files: %w", err) } - runSources, err := openPublishRunFiles(runRoot, runFiles, runManifestFile) + runSources, err := openPublishRunFiles(runRoot, runFiles) if err != nil { return nil, fmt.Errorf("publish: verify run files: %w", err) } @@ -139,7 +118,7 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S if previousManifestFile != nil { defer previousManifestFile.Close() } - previousSources, err := openPublishRunFiles(sessionPaths.PreviousDir, previousFiles, previousManifestFile) + previousSources, err := openPublishRunFiles(sessionPaths.PreviousDir, previousFiles) if err != nil { return nil, fmt.Errorf("publish: verify previous files: %w", err) } @@ -333,15 +312,9 @@ func buildPublishTransferPlan( return nil } for index, file := range runFiles { - artifactType := artifacts.RemoteArtifactTypeRunFile - source := "run:" + file.RelativePath - if file.RelativePath == config.S3ManifestFile { - artifactType = artifacts.RemoteArtifactTypeRunManifest - source = "run.manifest" - } if err := add(artifacts.RemoteArtifact{ - Type: artifactType, - Source: source, + Type: artifacts.RemoteArtifactTypeRunFile, + Source: "run:" + file.RelativePath, DestinationKey: artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath), }, runSources[index]); err != nil { closePublishTransfers(transfers) @@ -638,7 +611,7 @@ func validatePublishPrerequisites(m *manifest.Manifest) error { return nil } -func resolvePublishRunManifestPath(env *Env, m *manifest.Manifest) (string, error) { +func resolvePublishRunRoot(env *Env, m *manifest.Manifest) (string, error) { sessionID := strings.TrimSpace(env.Config.Session.SessionID) if sessionID == "" && m != nil { sessionID = strings.TrimSpace(m.SessionID) @@ -658,13 +631,13 @@ func resolvePublishRunManifestPath(env *Env, m *manifest.Manifest) (string, erro return "", fmt.Errorf("run id is required") } - canonical := filepath.Clean(artifacts.SessionRunManifestPathForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID)) - canonicalExists, err := directoryExists(filepath.Dir(canonical)) + canonical := filepath.Clean(artifacts.SessionRunRootForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID)) + info, err := os.Lstat(canonical) if err != nil { - return "", fmt.Errorf("check canonical run manifest directory %q: %w", filepath.Dir(canonical), err) + return "", fmt.Errorf("inspect canonical run root %q: %w", canonical, err) } - if !canonicalExists { - return "", fmt.Errorf("run manifest directory not found for campaign %q session %q run %q at canonical path %q", campaign, sessionID, runID, filepath.Dir(canonical)) + if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() { + return "", fmt.Errorf("canonical run root %q is not a directory", canonical) } return canonical, nil } @@ -850,13 +823,11 @@ func buildPublishRuntimeArtifactCatalog( return catalog, nil } -func collectPublishRunFiles(runRoot string, runManifest *manifest.RunManifest) ([]publishUploadFile, error) { - if runManifest == nil { - return nil, fmt.Errorf("run manifest is required") - } - files := map[string]publishUploadFile{ - "manifest.json": {RelativePath: "manifest.json", SourceRelativePath: "manifest.json"}, +func collectPublishRunFiles(runRoot string, sessionManifest *manifest.Manifest) ([]publishUploadFile, error) { + if sessionManifest == nil { + return nil, fmt.Errorf("session manifest is required") } + files := map[string]publishUploadFile{} add := func(path, checksum string) error { if strings.TrimSpace(path) == "" { return nil @@ -875,7 +846,7 @@ func collectPublishRunFiles(runRoot string, runManifest *manifest.RunManifest) ( files[rel] = candidate return nil } - for _, record := range runManifest.Stages { + for _, record := range sessionManifest.Stages { if record == nil { continue } @@ -1001,26 +972,16 @@ func collectPublishPreviousFiles(ctx context.Context, previousDir string) ([]pub return out, manifestFile, nil } -func openPublishRunFiles(runRoot string, files []publishUploadFile, manifestFile *os.File) ([]*os.File, error) { +func openPublishRunFiles(runRoot string, files []publishUploadFile) ([]*os.File, error) { sources := make([]*os.File, 0, len(files)) for _, candidate := range files { - if manifestFile == nil { + file, err := openPublishOutputSource(filepath.Join(runRoot, filepath.FromSlash(candidate.SourceRelativePath))) + if err != nil { closePublishSources(sources) - return nil, fmt.Errorf("archive manifest is required") - } - file := manifestFile - if candidate.SourceRelativePath != "manifest.json" { - var err error - file, err = openPublishOutputSource(filepath.Join(runRoot, filepath.FromSlash(candidate.SourceRelativePath))) - if err != nil { - closePublishSources(sources) - return nil, err - } + return nil, err } if err := verifyPublishSourceChecksum(file, candidate.RelativePath, candidate.Checksum); err != nil { - if file != manifestFile { - _ = file.Close() - } + _ = file.Close() closePublishSources(sources) return nil, err } @@ -1052,17 +1013,6 @@ func closePublishSources(sources []*os.File) { } } -func directoryExists(path string) (bool, error) { - info, err := os.Stat(path) - if err == nil { - return info.IsDir(), nil - } - if os.IsNotExist(err) { - return false, nil - } - return false, err -} - func writeCommittedSessionManifest(m *manifest.Manifest, publishMetadata map[string]any) (string, error) { if m == nil { return "", fmt.Errorf("manifest is required") diff --git a/internal/stage/publish_test.go b/internal/stage/publish_test.go index e7879b3..6b5556e 100644 --- a/internal/stage/publish_test.go +++ b/internal/stage/publish_test.go @@ -101,7 +101,6 @@ func TestPublishUploadsRunRecordPublishedOutputsAndCurrentPointer(t *testing.T) "prepare/inputs/session.yml", "prepare/outputs/audio/speaker.flac", "logs/audita.stderr.log", - "manifest.json", "polish/reports/audita.report.json", "transcribe/outputs/transcripts/raw/speaker.json", "trim/outputs/transcripts/final.trimmed.json", @@ -244,26 +243,24 @@ func TestPublishRejectsDeclaredLeafSymlinkBeforeUploading(t *testing.T) { assertNoPublishUploads(t, env.ObjectStore.(*storage.FakeBackend), "outside secret") } -func TestPublishRejectsSymlinkedRunManifestBeforeUploading(t *testing.T) { +func TestPublishRejectsSymlinkedRunRootBeforeUploading(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("symlink creation requires privileges on Windows") } env, m, runRoot := publishFixture(t) - sentinel := filepath.Join(t.TempDir(), "run-manifest.json") - writeStageTestFile(t, sentinel, `{"session_id":"outside"}`) - manifestPath := filepath.Join(runRoot, "manifest.json") - if err := os.Remove(manifestPath); err != nil { + actualRoot := runRoot + "-actual" + if err := os.Rename(runRoot, actualRoot); err != nil { t.Fatal(err) } - if err := os.Symlink(sentinel, manifestPath); err != nil { + if err := os.Symlink(actualRoot, runRoot); err != nil { t.Fatal(err) } _, err := (publishStage{}).Run(context.Background(), env, m) - if err == nil || !strings.Contains(err.Error(), "not a regular file") { - t.Fatalf("Run() error = %v, want rejected run manifest symlink", err) + if err == nil || !strings.Contains(err.Error(), "not a directory") { + t.Fatalf("Run() error = %v, want rejected run root symlink", err) } - assertNoPublishUploads(t, env.ObjectStore.(*storage.FakeBackend), "outside") + assertNoPublishUploads(t, env.ObjectStore.(*storage.FakeBackend), "") } func TestPublishUsesOrdinarySourceAncestorPathSemantics(t *testing.T) { @@ -293,21 +290,14 @@ func TestPublishRejectsDeclaredSourceReplacement(t *testing.T) { if err != nil { t.Fatal(err) } - now := time.Date(2026, 5, 16, 1, 2, 3, 0, time.UTC) - runManifest := manifest.NewRun(m.SessionID, m.Campaign, m.RunID, false, nil, now) - runManifest.MarkStageSucceeded("archive", now, []manifest.ArtifactRecord{{LocalPath: target, Checksum: checksum}}) - files, err := collectPublishRunFiles(runRoot, runManifest) + m.Stages["polish"].Outputs = []manifest.ArtifactRecord{{LocalPath: target, Checksum: checksum}} + files, err := collectPublishRunFiles(runRoot, m) if err != nil { t.Fatal(err) } writeStageTestFile(t, target, "replacement\n") - manifestFile, err := os.Open(filepath.Join(runRoot, "manifest.json")) - if err != nil { - t.Fatal(err) - } - defer manifestFile.Close() - _, err = openPublishRunFiles(runRoot, files, manifestFile) + _, err = openPublishRunFiles(runRoot, files) if err == nil || !strings.Contains(err.Error(), `source "logs/audita.stderr.log" does not match its declared checksum`) { t.Fatalf("openPublishRunFiles() error = %v, want replacement rejection", err) } @@ -348,6 +338,62 @@ func TestPublishDoesNotUploadUndeclaredRunFile(t *testing.T) { } } +func TestPublishIgnoresPriorRunAndSessionDiagnosticReferences(t *testing.T) { + env, m, _ := publishFixture(t) + priorRunRoot := artifacts.SessionRunRootForCampaign( + env.Config.Pipeline.Workspace.Root, + env.Config.Session.Campaign, + env.Config.Session.SessionID, + "20260515T010203Z-1a2b3c4d", + ) + priorPath := filepath.Join(priorRunRoot, "logs", "prior.log") + sessionPath := filepath.Join( + artifacts.SessionWorkDirForCampaign(env.Config.Pipeline.Workspace.Root, env.Config.Session.Campaign, env.Config.Session.SessionID), + "reports", "session.log", + ) + writeStageTestFile(t, priorPath, "prior diagnostic\n") + writeStageTestFile(t, sessionPath, "session diagnostic\n") + m.Stages["polish"].Logs = append(m.Stages["polish"].Logs, priorPath, sessionPath) + + if _, err := (publishStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("Run() error = %v", err) + } + fake := env.ObjectStore.(*storage.FakeBackend) + for _, path := range []string{"logs/prior.log", "reports/session.log"} { + if _, exists := fake.Objects[m.S3RunPrefix+path]; exists { + t.Fatalf("non-current diagnostic %q was uploaded", path) + } + } +} + +func TestPublishAllowsEmptyCurrentRunDiagnostics(t *testing.T) { + env, m, _ := publishFixture(t) + trimOutputs := append([]manifest.ArtifactRecord(nil), m.Stages["trim"].Outputs[:1]...) + for _, record := range m.Stages { + record.Outputs = nil + record.Logs = nil + record.GeneratedConfigs = nil + } + m.Stages["trim"].Outputs = trimOutputs + + result, err := (publishStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if result.Metadata["run_files_uploaded"] != 0 { + t.Fatalf("run_files_uploaded = %#v, paths = %#v, want 0", result.Metadata["run_files_uploaded"], result.Metadata["run_uploaded_paths"]) + } + commit, err := artifacts.DecodeRemoteCommitManifest(env.ObjectStore.(*storage.FakeBackend).Objects[artifacts.S3RunCommitKey(m.S3SessionPrefix, m.RunID)].Data) + if err != nil { + t.Fatalf("DecodeRemoteCommitManifest() error = %v", err) + } + for _, artifact := range commit.Artifacts { + if artifact.Type == artifacts.RemoteArtifactTypeRunManifest { + t.Fatal("publish emitted a run manifest artifact") + } + } +} + func TestPublishRejectsDeclaredPreviousSymlinkBeforeUploading(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("symlink creation requires privileges on Windows") @@ -845,8 +891,8 @@ func TestPublishFailsWhenCanonicalRunRootMissing(t *testing.T) { if err == nil { t.Fatal("expected missing run-root error, got nil") } - if !strings.Contains(err.Error(), "run manifest directory not found") { - t.Fatalf("error = %v, want missing run-manifest directory error", err) + if !strings.Contains(err.Error(), "canonical run root") { + t.Fatalf("error = %v, want missing canonical run-root error", err) } } @@ -1033,16 +1079,6 @@ func publishFixture(t *testing.T) (*Env, *manifest.Manifest, string) { writeStageTestFile(t, filepath.Join(sessionRoot, "transcripts", "final.trimmed.json"), "{\"segments\":[]}\n") writeStageTestFile(t, filepath.Join(sessionRoot, "artifacts", "session_recap.md"), "# recap\n") - runArtifacts := []string{ - "prepare/inputs/session.yml", - "prepare/outputs/audio/speaker.flac", - "transcribe/outputs/transcripts/raw/speaker.json", - "trim/outputs/transcripts/final.trimmed.json", - "analyze/outputs/artifacts/session_recap.md", - "polish/reports/audita.report.json", - "merge/config/seriatim.generated.yml", - "logs/audita.stderr.log", - } for rel, contents := range map[string]string{ "prepare/inputs/session.yml": "session_id: 2026-04-19\n", "prepare/outputs/audio/speaker.flac": "flac\n", @@ -1056,7 +1092,6 @@ func publishFixture(t *testing.T) (*Env, *manifest.Manifest, string) { writeStageTestFile(t, filepath.Join(runRoot, filepath.FromSlash(rel)), contents) } writeStageTestFile(t, filepath.Join(runRoot, "audio", "speaker.flac"), "flac") - writePublishRunManifest(t, runRoot, sessionID, campaign, runID, runArtifacts) m := manifest.New(sessionID, time.Date(2026, 5, 16, 1, 2, 3, 0, time.UTC)) m.Campaign = campaign @@ -1070,6 +1105,7 @@ func publishFixture(t *testing.T) (*Env, *manifest.Manifest, string) { } setCurrentBuiltInEvidence(t, m, "trim", artifacts.TranscriptOutputKindFinalTrimmed, artifacts.ArtifactTranscriptFinalTrimmed, filepath.Join(sessionRoot, "transcripts", "final.trimmed.json")) setCurrentAnalyzeEvidence(t, m, "session_recap", "artifacts/session_recap.md", filepath.Join(sessionRoot, "artifacts", "session_recap.md")) + recordPublishRunDiagnostics(m, runRoot) env := &Env{ Config: &config.Config{ @@ -1109,21 +1145,17 @@ func publishFixture(t *testing.T) (*Env, *manifest.Manifest, string) { return env, m, runRoot } -func writePublishRunManifest(t *testing.T, runRoot, sessionID, campaign, runID string, archivePaths []string) { - t.Helper() - now := time.Date(2026, 5, 16, 1, 2, 3, 0, time.UTC) - runManifest := manifest.NewRun(sessionID, campaign, runID, false, nil, now) - outputs := make([]manifest.ArtifactRecord, 0, len(archivePaths)) - for _, rel := range archivePaths { - outputs = append(outputs, manifest.ArtifactRecord{LocalPath: filepath.Join(runRoot, filepath.FromSlash(rel))}) +func recordPublishRunDiagnostics(m *manifest.Manifest, runRoot string) { + m.Stages["prepare"].Outputs = []manifest.ArtifactRecord{ + {LocalPath: filepath.Join(runRoot, "prepare", "inputs", "session.yml")}, + {ArchivePath: filepath.Join(runRoot, "prepare", "outputs", "audio", "speaker.flac")}, } - runManifest.MarkStageSucceeded("archive", now, outputs) - runManifest.MarkSucceeded(now) - data, err := json.Marshal(runManifest) - if err != nil { - t.Fatal(err) - } - writeStageTestFile(t, filepath.Join(runRoot, "manifest.json"), string(data)) + m.Stages["transcribe"].Outputs = []manifest.ArtifactRecord{{LocalPath: filepath.Join(runRoot, "transcribe", "outputs", "transcripts", "raw", "speaker.json")}} + m.Stages["trim"].Outputs = append(m.Stages["trim"].Outputs, manifest.ArtifactRecord{LocalPath: filepath.Join(runRoot, "trim", "outputs", "transcripts", "final.trimmed.json")}) + m.Stages["analyze"].Outputs = append(m.Stages["analyze"].Outputs, manifest.ArtifactRecord{LocalPath: filepath.Join(runRoot, "analyze", "outputs", "artifacts", "session_recap.md")}) + m.Stages["polish"].Outputs = []manifest.ArtifactRecord{{LocalPath: filepath.Join(runRoot, "polish", "reports", "audita.report.json")}} + m.Stages["polish"].Logs = []string{filepath.Join(runRoot, "logs", "audita.stderr.log")} + m.Stages["merge"].GeneratedConfigs = []string{filepath.Join(runRoot, "merge", "config", "seriatim.generated.yml")} } func writePublishPreviousManifest(t *testing.T, previousDir, sessionID string, archivePaths ...string) {