From f1d4e3841482458e722f48c21f4c0e820b01bb0d Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Wed, 17 Jun 2026 20:39:22 +0000 Subject: [PATCH] Add batch distributor notification state artifacts --- internal/state/filesystem.go | 56 +++++++++ internal/state/filesystem_test.go | 196 ++++++++++++++++++++++++++++++ internal/state/store.go | 34 ++++++ 3 files changed, 286 insertions(+) diff --git a/internal/state/filesystem.go b/internal/state/filesystem.go index bb41d65..452c038 100644 --- a/internal/state/filesystem.go +++ b/internal/state/filesystem.go @@ -141,6 +141,33 @@ func (s *FilesystemStore) SaveDistributorNotification(_ context.Context, resolve }, artifact) } +func (s *FilesystemStore) BatchDistributorNotificationPath(ref BatchDistributorNotificationRef) (string, error) { + if s == nil { + return "", fmt.Errorf("state store is required") + } + if err := validateBatchNotificationRef(ref); err != nil { + return "", err + } + localDate := ref.StartedAt.In(ref.Location).Format("2006-01-02") + return s.join(s.notificationsDir, "batches", ref.Batch, localDate, ref.BatchRunID+".distributor.json"), nil +} + +func (s *FilesystemStore) SaveBatchDistributorNotification(_ context.Context, ref BatchDistributorNotificationRef, artifact BatchDistributorNotificationArtifact) (string, error) { + path, err := s.BatchDistributorNotificationPath(ref) + if err != nil { + return "", err + } + if artifact.SchemaVersion == "" { + artifact.SchemaVersion = BatchDistributorNotificationSchemaVersion + } + artifact.Batch = ref.Batch + artifact.BatchRunID = ref.BatchRunID + if err := fileutil.WriteJSONAtomic(path, artifact); err != nil { + return "", err + } + return path, nil +} + func (s *FilesystemStore) SaveGeneratedTextRaw(_ context.Context, resolved report.Resolved, data []byte) (string, error) { return s.saveResolvedBytes(resolved, func(paths ArtifactPaths) string { return paths.GeneratedTextRaw @@ -461,6 +488,35 @@ func validateRelativeDir(name string, value string) error { return nil } +func validateBatchNotificationRef(ref BatchDistributorNotificationRef) error { + if err := validatePathSegment("batch kind", ref.Batch); err != nil { + return err + } + if err := validatePathSegment("batch run id", ref.BatchRunID); err != nil { + return err + } + if ref.StartedAt.IsZero() { + return fmt.Errorf("batch started time is required") + } + if ref.Location == nil { + return fmt.Errorf("batch location is required") + } + return nil +} + +func validatePathSegment(name string, value string) error { + if strings.TrimSpace(value) == "" { + return fmt.Errorf("%s is required", name) + } + if strings.ContainsAny(value, `/\`) { + return fmt.Errorf("%s must not contain path separators", name) + } + if value == "." || value == ".." { + return fmt.Errorf("%s must be a safe path segment", name) + } + return nil +} + func readJSON(path string, target any) error { data, err := os.ReadFile(path) if err != nil { diff --git a/internal/state/filesystem_test.go b/internal/state/filesystem_test.go index a963541..1dc706b 100644 --- a/internal/state/filesystem_test.go +++ b/internal/state/filesystem_test.go @@ -69,6 +69,193 @@ func TestDailyPathsUseRunIDValidDateDisambiguator(t *testing.T) { } } +func TestBatchDistributorNotificationPathUsesWorkspaceBatchDateAndRunID(t *testing.T) { + store := newTestStore(t) + location := mustLoadStateLocation(t, "America/Chicago") + startedAt := time.Date(2026, 6, 18, 3, 30, 0, 123456789, time.UTC) + + path, err := store.BatchDistributorNotificationPath(BatchDistributorNotificationRef{ + Batch: "evening", + BatchRunID: "20260618T033000.123456789Z_evening", + StartedAt: startedAt, + Location: location, + }) + if err != nil { + t.Fatalf("BatchDistributorNotificationPath() error = %v", err) + } + want := filepath.Join("notifications", "batches", "evening", "2026-06-17", "20260618T033000.123456789Z_evening.distributor.json") + if !strings.Contains(path, want) { + t.Fatalf("path = %q, want component %q", path, want) + } + if !strings.HasPrefix(path, store.root) { + t.Fatalf("path = %q, want workspace root prefix %q", path, store.root) + } +} + +func TestSaveBatchDistributorNotificationRoundTrip(t *testing.T) { + store := newTestStore(t) + location := mustLoadStateLocation(t, "America/Chicago") + startedAt := time.Date(2026, 6, 17, 12, 0, 0, 0, time.UTC) + bundleCreated := startedAt.Add(2 * time.Second) + attemptedAt := startedAt.Add(3 * time.Second) + acceptedAt := startedAt.Add(4 * time.Second) + finishedAt := startedAt.Add(5 * time.Second) + ref := BatchDistributorNotificationRef{ + Batch: "morning", + BatchRunID: "20260617T120000.000000000Z_morning", + StartedAt: startedAt, + Location: location, + } + + path, err := store.SaveBatchDistributorNotification(context.Background(), ref, BatchDistributorNotificationArtifact{ + AttemptedAt: attemptedAt, + Endpoint: "https://distributor.example.test", + PipelineID: "weatherreporter", + BundleID: "weatherreporter.home.morning", + IdempotencyKey: "weatherreporter.home.morning.20260617T120000.000000000Z_morning", + BundleCreated: bundleCreated, + Reports: []BatchDistributorNotificationReportArtifact{ + { + ReportID: report.Today, + RunID: "20260617T120000.000000000Z_today", + SourcePath: "/workspace/reports/today/20260617T120000.000000000Z_today.md", + BundlePaths: []string{"2026-06-17/today/report.md"}, + }, + { + ReportID: report.Daily, + RunID: "20260617T120000.000000000Z_daily_2026-06-19", + SourcePath: "/workspace/reports/daily/20260617T120000.000000000Z_daily_2026-06-19.md", + BundlePaths: []string{"2026-06-19/daily/report.md"}, + }, + }, + Status: "failed", + Upload: &DistributorUploadResult{ + RunID: "distributor-run", + Status: "accepted", + }, + RunStatus: &DistributorRunStatus{ + RunID: "distributor-run", + PipelineID: "weatherreporter", + Status: "failed", + AcceptedAt: acceptedAt, + FinishedAt: &finishedAt, + Report: json.RawMessage(`{"actions":[{"action":"failed"}]}`), + Error: "destination conflict", + }, + StatusError: "status lookup failed", + Error: "batch upload failed", + }) + if err != nil { + t.Fatalf("SaveBatchDistributorNotification() error = %v", err) + } + wantPath := filepath.Join("notifications", "batches", "morning", "2026-06-17", "20260617T120000.000000000Z_morning.distributor.json") + if !strings.Contains(path, wantPath) { + t.Fatalf("path = %q, want component %q", path, wantPath) + } + + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read batch notification: %v", err) + } + var artifact BatchDistributorNotificationArtifact + if err := json.Unmarshal(data, &artifact); err != nil { + t.Fatalf("decode batch notification: %v", err) + } + if artifact.SchemaVersion != BatchDistributorNotificationSchemaVersion { + t.Fatalf("SchemaVersion = %q, want %q", artifact.SchemaVersion, BatchDistributorNotificationSchemaVersion) + } + if artifact.Batch != "morning" || artifact.BatchRunID != ref.BatchRunID { + t.Fatalf("artifact batch identity = %q/%q, want ref values", artifact.Batch, artifact.BatchRunID) + } + if artifact.Endpoint != "https://distributor.example.test" || artifact.PipelineID != "weatherreporter" || artifact.BundleID != "weatherreporter.home.morning" || artifact.IdempotencyKey == "" { + t.Fatalf("artifact identity = %#v, want distributor identity", artifact) + } + if len(artifact.Reports) != 2 || artifact.Reports[0].ReportID != report.Today || strings.Join(artifact.Reports[1].BundlePaths, ",") != "2026-06-19/daily/report.md" { + t.Fatalf("Reports = %#v, want included report records", artifact.Reports) + } + if artifact.Upload == nil || artifact.Upload.RunID != "distributor-run" { + t.Fatalf("Upload = %#v, want accepted upload result", artifact.Upload) + } + if artifact.RunStatus == nil || artifact.RunStatus.Status != "failed" || !strings.Contains(string(artifact.RunStatus.Report), "failed") || artifact.RunStatus.FinishedAt == nil { + t.Fatalf("RunStatus = %#v, want failed run status with raw report", artifact.RunStatus) + } + if artifact.StatusError != "status lookup failed" || artifact.Error != "batch upload failed" { + t.Fatalf("errors = %q/%q, want persisted error fields", artifact.StatusError, artifact.Error) + } +} + +func TestBatchDistributorNotificationPathRejectsInvalidIdentity(t *testing.T) { + store := newTestStore(t) + location := mustLoadStateLocation(t, "America/Chicago") + valid := BatchDistributorNotificationRef{ + Batch: "morning", + BatchRunID: "20260617T120000.000000000Z_morning", + StartedAt: time.Date(2026, 6, 17, 12, 0, 0, 0, time.UTC), + Location: location, + } + tests := []struct { + name string + mutate func(*BatchDistributorNotificationRef) + wantErr string + }{ + { + name: "Batch", + mutate: func(ref *BatchDistributorNotificationRef) { + ref.Batch = "" + }, + wantErr: "batch kind is required", + }, + { + name: "BatchSeparator", + mutate: func(ref *BatchDistributorNotificationRef) { + ref.Batch = "../morning" + }, + wantErr: "batch kind must not contain path separators", + }, + { + name: "BatchRunID", + mutate: func(ref *BatchDistributorNotificationRef) { + ref.BatchRunID = "" + }, + wantErr: "batch run id is required", + }, + { + name: "BatchRunIDSeparator", + mutate: func(ref *BatchDistributorNotificationRef) { + ref.BatchRunID = "nested/run" + }, + wantErr: "batch run id must not contain path separators", + }, + { + name: "StartedAt", + mutate: func(ref *BatchDistributorNotificationRef) { + ref.StartedAt = time.Time{} + }, + wantErr: "batch started time is required", + }, + { + name: "Location", + mutate: func(ref *BatchDistributorNotificationRef) { + ref.Location = nil + }, + wantErr: "batch location is required", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ref := valid + tt.mutate(&ref) + _, err := store.BatchDistributorNotificationPath(ref) + if err == nil { + t.Fatal("BatchDistributorNotificationPath() error = nil, want error") + } + if !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("error = %q, want %q", err.Error(), tt.wantErr) + } + }) + } +} + func TestGeneratedTextArtifactPathsUseSnapshotTree(t *testing.T) { store := newTestStore(t) tests := []struct { @@ -595,6 +782,15 @@ func newTestStore(t *testing.T) *FilesystemStore { return store } +func mustLoadStateLocation(t *testing.T, name string) *time.Location { + t.Helper() + location, err := time.LoadLocation(name) + if err != nil { + t.Fatalf("LoadLocation(%q) error = %v", name, err) + } + return location +} + func resolveDailyAt(t *testing.T, value string) report.Resolved { t.Helper() return resolveDailyForDateAt(t, value, value) diff --git a/internal/state/store.go b/internal/state/store.go index 8ebe0fe..403f239 100644 --- a/internal/state/store.go +++ b/internal/state/store.go @@ -17,6 +17,7 @@ type Store interface { SaveDataPackage(context.Context, report.Resolved, promptinput.Package) (string, error) SavePreflight(context.Context, report.Resolved, PreflightArtifact) (string, error) SaveDistributorNotification(context.Context, report.Resolved, DistributorNotificationArtifact) (string, error) + SaveBatchDistributorNotification(context.Context, BatchDistributorNotificationRef, BatchDistributorNotificationArtifact) (string, error) SaveGeneratedTextRaw(context.Context, report.Resolved, []byte) (string, error) SaveGeneratedTextResult(context.Context, report.Resolved, any) (string, error) SaveGeneratedText(context.Context, report.Resolved, []byte) (string, error) @@ -45,6 +46,14 @@ type PreflightArtifact struct { } const DistributorNotificationSchemaVersion = "weatherreporter.distributor_notification.v1" +const BatchDistributorNotificationSchemaVersion = "weatherreporter.batch_distributor_notification.v1" + +type BatchDistributorNotificationRef struct { + Batch string + BatchRunID string + StartedAt time.Time + Location *time.Location +} type DistributorNotificationArtifact struct { SchemaVersion string `json:"schemaVersion"` @@ -80,3 +89,28 @@ type DistributorRunStatus struct { Report json.RawMessage `json:"report,omitempty"` Error string `json:"error,omitempty"` } + +type BatchDistributorNotificationArtifact struct { + SchemaVersion string `json:"schemaVersion"` + Batch string `json:"batch"` + BatchRunID string `json:"batchRunId"` + AttemptedAt time.Time `json:"attemptedAt"` + Endpoint string `json:"endpoint"` + PipelineID string `json:"pipelineId,omitempty"` + BundleID string `json:"bundleId,omitempty"` + IdempotencyKey string `json:"idempotencyKey,omitempty"` + BundleCreated time.Time `json:"bundleCreated,omitempty"` + Reports []BatchDistributorNotificationReportArtifact `json:"includedReports,omitempty"` + Status string `json:"status"` + Upload *DistributorUploadResult `json:"upload,omitempty"` + RunStatus *DistributorRunStatus `json:"runStatus,omitempty"` + StatusError string `json:"statusError,omitempty"` + Error string `json:"error,omitempty"` +} + +type BatchDistributorNotificationReportArtifact struct { + ReportID report.ID `json:"reportId"` + RunID string `json:"runId"` + SourcePath string `json:"sourcePath"` + BundlePaths []string `json:"bundlePaths"` +}