diff --git a/docs/config.md b/docs/config.md index da79701..19092e9 100644 --- a/docs/config.md +++ b/docs/config.md @@ -134,6 +134,7 @@ stub source slots use the missing-source policy. - `reports_dir`: managed Markdown report directory under `workspace.root`. Default: `reports`. - `data_packages_dir`: prompt input package directory under `workspace.root`. Default: `data-packages`. - `preflight_dir`: Scriptorium render output directory under `workspace.root`. Default: `preflight`. +- `notifications_dir`: distributor notification debug artifact directory under `workspace.root`. Default: `notifications`. Workspace subdirectories must be relative paths that stay inside `workspace.root`. diff --git a/docs/internal/app-orchestration.md b/docs/internal/app-orchestration.md index 689de7b..9168ecf 100644 --- a/docs/internal/app-orchestration.md +++ b/docs/internal/app-orchestration.md @@ -81,12 +81,17 @@ Single-report generation follows this order: 14. Save metadata with the managed report path. 15. If distributor notification is enabled, notify using the managed report path as the source file. +16. Save a distributor notification debug artifact and update metadata with its + path. If render preflight returns both a result and an error, preflight JSON and metadata are persisted before the error is returned. If Scriptorium report generation returns an error after writing output, the managed report and metadata remain inspectable. Notification is not attempted after Weather API, briefing, prompt input, render, Scriptorium run, or metadata-save failures. +When notification is attempted, the debug artifact records request identity, +accepted upload fields, distributor status fields, raw status report JSON when +available, and redacted failure context. `--out` copies are never used as notification source files. ## Batch Workflow diff --git a/docs/internal/distributor-adapter.md b/docs/internal/distributor-adapter.md index df0bfda..d4e983f 100644 --- a/docs/internal/distributor-adapter.md +++ b/docs/internal/distributor-adapter.md @@ -8,7 +8,8 @@ This document describes the distributor upload adapter in The adapter submits generated weatherreporter Markdown reports to a configured distributor HTTP upload endpoint. It isolates distributor package types, token-env lookup, upload client construction, source-bundle file mapping, -timeout handling, and upload error wrapping from app orchestration. +timeout handling, status polling, and upload error wrapping from app +orchestration. ## Inputs And Outputs @@ -21,12 +22,14 @@ Inputs: - idempotency key - source Markdown report path - bundle-relative Markdown path +- bundle created timestamp - context for cancellation Outputs: - accepted distributor run ID -- accepted distributor status +- accepted distributor upload status +- distributor run status, status polling error, and raw run report JSON when available - weatherreporter-owned idempotency conflict error when applicable ## Boundaries @@ -66,11 +69,21 @@ The adapter calls distributor `UploadFiles` with exactly one file: - source path: the managed Markdown report path selected by app orchestration - bundle path: the rendered bundle-relative report path +- created: the report generation timestamp The adapter creates a distributor upload client with the configured endpoint, bearer token, and timeout-backed HTTP client. It also wraps the upload context with the configured timeout when the timeout is greater than zero. +After upload acceptance, the adapter polls distributor `Status` for the accepted +run ID until the run reaches `succeeded` or `failed`, or until the configured +timeout expires. It returns the latest status, error text, and raw report JSON in +weatherreporter-owned types so app orchestration can persist them in the +notification debug artifact. Status lookup failures or timeout before a terminal +state are kept as debug status errors on an otherwise accepted upload. A +terminal distributor run status of `failed` is returned as a notification failure +with the status report preserved. + ## Failure Behavior The adapter validates required endpoint, token env name, token value, bundle ID, diff --git a/docs/internal/state.md b/docs/internal/state.md index 85e613d..7fd8c24 100644 --- a/docs/internal/state.md +++ b/docs/internal/state.md @@ -47,6 +47,7 @@ converts adapter render results into that shape before saving. - `workspace.reports_dir` - `workspace.data_packages_dir` - `workspace.preflight_dir` +- `workspace.notifications_dir` Workspace subdirectories must be relative paths that stay under `workspace.root`. @@ -62,12 +63,14 @@ valid-period start date for JSON artifacts, and the RunID. snapshots///.metadata.json data-packages///.data_package.json preflight///.render.json + notifications///.distributor.json reports//.md ``` Metadata is stored beside briefing snapshots and links the briefing, data -package, preflight, report paths, and configured prompt location. Report -listing walks metadata files under the snapshots directory. +package, preflight, report paths, notification path when attempted, and +configured prompt location. Report listing walks metadata files under the +snapshots directory. ## Prior Lookup @@ -89,6 +92,8 @@ current report definition. Durable JSON writes use shared atomic file helpers. Managed Markdown reports are prepared by creating their parent directory; Scriptorium writes the report body to the prepared path. Extra Markdown copies are handled by app orchestration. +Distributor notification debug artifacts are written atomically when +notification is attempted. Inspection helpers read existing metadata, briefing, and data package files. Missing metadata directories return no inspection records or no prior snapshot diff --git a/docs/operations.md b/docs/operations.md index b43ef66..ed13622 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -91,6 +91,19 @@ workspace/ storm/ YYYY-MM-DD/ .render.json + notifications/ + daily/ + YYYY-MM-DD/ + .distributor.json + three-day/ + YYYY-MM-DD/ + .distributor.json + weekend/ + YYYY-MM-DD/ + .distributor.json + storm/ + YYYY-MM-DD/ + .distributor.json reports/ daily/ .md @@ -122,6 +135,7 @@ Each generated report writes metadata that links: - prompt input data package path - preflight output path - managed Markdown report path +- distributor notification debug artifact path, when notification is attempted Batch summaries include report status, error text when applicable, notification outcome when attempted, valid period, and known artifact paths for each @@ -153,6 +167,16 @@ not trigger notification. A notification failure fails that report. In a batch, other reports continue, the failed report includes notification fields in the JSON summary, and the batch returns nonzero. +Each notification attempt writes a debug artifact under `notifications/`. The +artifact records the rendered bundle ID, idempotency key, managed source path, +bundle-relative path, bundle created timestamp, accepted upload response, and +the latest distributor run status response when available. Weatherreporter polls +status until distributor reports `succeeded` or `failed`, or until the configured +notification timeout expires. The run status includes the distributor status, +error text, and raw run report JSON, which can show actions such as +`replace_older`, `skip_same`, `skip_destination_newer`, or `failed`. Token values +are not written. + Weatherreporter is responsible for selecting the managed Markdown report, constructing a source bundle, and submitting it to the configured distributor HTTP endpoint. Distributor remains responsible for destination routing, diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index c8da3e4..75b9a90 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -248,6 +248,10 @@ Diagnostic: inspect the failed batch JSON or stderr line for bundle and idempotency context. Compare the configured templates with the report RunID and report path. +Also inspect the notification artifact linked from metadata. It records the +rendered bundle ID, idempotency key, upload result, distributor run status, +status error, and raw run report JSON when available. + Safe fix: keep idempotency templates stable for retries of the same generated report, but do not reuse the same rendered key for different generated report content. @@ -268,6 +272,11 @@ Diagnostic: inspect stdout JSON or stderr status lines for `notify.distributor.report_path_template`, and token configuration. Token values are redacted from weatherreporter errors. +If the upload was accepted but destination output did not change, inspect the +notification artifact's `runStatus.report`. Distributor actions such as +`replace_older`, `skip_same`, `skip_destination_newer`, or `failed` explain how +the destination handled the uploaded bundle. + Safe fix: fix the endpoint, token, templates, or distributor-side upload configuration. The weatherreporter upload source is the managed Markdown report, not `--out` or `--out-dir` copies. diff --git a/examples/config.yml b/examples/config.yml index 130bcc0..2eedee1 100644 --- a/examples/config.yml +++ b/examples/config.yml @@ -40,6 +40,7 @@ workspace: reports_dir: reports data_packages_dir: data-packages preflight_dir: preflight + notifications_dir: notifications dayparts: - name: overnight diff --git a/internal/adapters/distributor/client.go b/internal/adapters/distributor/client.go index 9bb3ac0..8947b0b 100644 --- a/internal/adapters/distributor/client.go +++ b/internal/adapters/distributor/client.go @@ -3,6 +3,7 @@ package distributor import ( "context" + "encoding/json" "errors" "fmt" "net/http" @@ -28,11 +29,26 @@ type UploadRequest struct { IdempotencyKey string SourcePath string BundlePath string + CreatedAt time.Time } type UploadResult struct { - RunID string - Status string + RunID string + Status string + UploadStatus string + StatusError string + RunStatus *RunStatus +} + +type RunStatus struct { + RunID string + PipelineID string + Status string + AcceptedAt time.Time + StartedAt *time.Time + FinishedAt *time.Time + Report json.RawMessage + Error string } type IdempotencyConflictError struct { @@ -57,6 +73,7 @@ type uploadClientFactory func(endpoint, token string, timeout time.Duration) (up type uploadClient interface { UploadFiles(ctx context.Context, opts uploadFilesOptions) (uploadFilesResult, error) + Status(ctx context.Context, runID string) (runStatus, error) } type uploadFilesOptions struct { @@ -64,6 +81,7 @@ type uploadFilesOptions struct { IdempotencyKey string SourcePath string BundlePath string + CreatedAt time.Time } type uploadFilesResult struct { @@ -71,6 +89,19 @@ type uploadFilesResult struct { Status string } +type runStatus struct { + RunID string + PipelineID string + Status string + AcceptedAt time.Time + StartedAt *time.Time + FinishedAt *time.Time + Report json.RawMessage + Error string +} + +const statusPollInterval = 250 * time.Millisecond + func New(cfg config.DistributorNotifyConfig) *Client { return newClient(cfg, newDistributorUploadClient) } @@ -138,6 +169,7 @@ func (c *Client) Upload(ctx context.Context, req UploadRequest) (UploadResult, e IdempotencyKey: req.IdempotencyKey, SourcePath: req.SourcePath, BundlePath: req.BundlePath, + CreatedAt: req.CreatedAt, }) if err != nil { return UploadResult{}, wrapUploadError(err, uploadErrorContext{ @@ -150,10 +182,65 @@ func (c *Client) Upload(ctx context.Context, req UploadRequest) (UploadResult, e }) } - return UploadResult{ - RunID: result.RunID, - Status: result.Status, - }, nil + uploadResult := UploadResult{ + RunID: result.RunID, + Status: result.Status, + UploadStatus: result.Status, + } + status, statusErr := waitForRunStatus(runCtx, uploadClient, result.RunID, c.Timeout > 0) + if status.RunID != "" || status.Status != "" { + uploadResult.RunStatus = &RunStatus{ + RunID: status.RunID, + PipelineID: status.PipelineID, + Status: status.Status, + AcceptedAt: status.AcceptedAt, + StartedAt: status.StartedAt, + FinishedAt: status.FinishedAt, + Report: append(json.RawMessage(nil), status.Report...), + Error: redactTokenString(status.Error, token), + } + if status.Status != "" { + uploadResult.Status = status.Status + } + } + if statusErr != nil { + uploadResult.StatusError = redactTokenString(statusErr.Error(), token) + return uploadResult, nil + } + if status.Status == "failed" { + return uploadResult, fmt.Errorf("distributor run %q failed: %s", status.RunID, uploadResult.RunStatus.Error) + } + return uploadResult, nil +} + +func waitForRunStatus(ctx context.Context, client uploadClient, runID string, poll bool) (runStatus, error) { + status, err := client.Status(ctx, runID) + if err != nil || terminalRunStatus(status.Status) || !poll { + return status, err + } + + for { + timer := time.NewTimer(statusPollInterval) + select { + case <-ctx.Done(): + timer.Stop() + return status, fmt.Errorf("distributor run %q did not reach terminal status before timeout: %w", runID, ctx.Err()) + case <-timer.C: + } + + next, err := client.Status(ctx, runID) + if err != nil { + return status, err + } + status = next + if terminalRunStatus(status.Status) { + return status, nil + } + } +} + +func terminalRunStatus(status string) bool { + return status == "succeeded" || status == "failed" } type distributorUploadClient struct { @@ -179,6 +266,7 @@ func newDistributorUploadClient(endpoint, token string, timeout time.Duration) ( func (c distributorUploadClient) UploadFiles(ctx context.Context, opts uploadFilesOptions) (uploadFilesResult, error) { result, err := c.client.UploadFiles(ctx, distributorupload.UploadFilesOptions{ ID: opts.BundleID, + Created: opts.CreatedAt, IdempotencyKey: opts.IdempotencyKey, Files: []distributorbundle.BundleFile{ {SourcePath: opts.SourcePath, Path: opts.BundlePath}, @@ -193,6 +281,23 @@ func (c distributorUploadClient) UploadFiles(ctx context.Context, opts uploadFil }, nil } +func (c distributorUploadClient) Status(ctx context.Context, runID string) (runStatus, error) { + status, err := c.client.Status(ctx, runID) + if err != nil { + return runStatus{}, err + } + return runStatus{ + RunID: status.RunID, + PipelineID: status.PipelineID, + Status: status.Status, + AcceptedAt: status.AcceptedAt, + StartedAt: status.StartedAt, + FinishedAt: status.FinishedAt, + Report: append(json.RawMessage(nil), status.Report...), + Error: status.Error, + }, nil +} + type uploadErrorContext struct { Endpoint string BundleID string diff --git a/internal/adapters/distributor/client_test.go b/internal/adapters/distributor/client_test.go index fa74e8a..6608b24 100644 --- a/internal/adapters/distributor/client_test.go +++ b/internal/adapters/distributor/client_test.go @@ -2,6 +2,7 @@ package distributor import ( "context" + "encoding/json" "errors" "fmt" "strings" @@ -23,6 +24,7 @@ func TestUploadUsesConfiguredClientAndSingleFile(t *testing.T) { factory := &fakeUploadFactory{ client: &fakeUploadClient{ result: uploadFilesResult{RunID: "run-123", Status: "accepted"}, + status: runStatus{RunID: "run-123", PipelineID: "reports", Status: "succeeded", Report: json.RawMessage(`{"actions":[{"action":"replace_older"}]}`)}, }, } client := newClient(cfg, factory.newClient) @@ -32,13 +34,17 @@ func TestUploadUsesConfiguredClientAndSingleFile(t *testing.T) { IdempotencyKey: "weatherreporter.home.daily.run", SourcePath: "/tmp/report.md", BundlePath: "daily.md", + CreatedAt: time.Date(2026, 6, 7, 12, 0, 0, 123, time.UTC), }) if err != nil { t.Fatalf("Upload() error = %v", err) } - if result.RunID != "run-123" || result.Status != "accepted" { + if result.RunID != "run-123" || result.Status != "succeeded" || result.UploadStatus != "accepted" { t.Fatalf("result = %#v, want accepted run", result) } + if result.RunStatus == nil || result.RunStatus.PipelineID != "reports" || !strings.Contains(string(result.RunStatus.Report), "replace_older") { + t.Fatalf("RunStatus = %#v, want parsed run report", result.RunStatus) + } if factory.endpoint != cfg.Endpoint { t.Fatalf("factory endpoint = %q, want %q", factory.endpoint, cfg.Endpoint) } @@ -61,6 +67,12 @@ func TestUploadUsesConfiguredClientAndSingleFile(t *testing.T) { if got.BundlePath != "daily.md" { t.Fatalf("BundlePath = %q, want daily.md", got.BundlePath) } + if got.CreatedAt.IsZero() { + t.Fatal("CreatedAt is zero, want generated report timestamp") + } + if factory.client.statusRunID != "run-123" { + t.Fatalf("Status runID = %q, want run-123", factory.client.statusRunID) + } } func TestUploadRejectsMissingInputs(t *testing.T) { @@ -168,6 +180,109 @@ func TestUploadWrapsUploadFailureWithContextWithoutToken(t *testing.T) { } } +func TestUploadReturnsAcceptedWhenStatusLookupFails(t *testing.T) { + cfg := config.Defaults().Notify.Distributor + cfg.Endpoint = "https://distributor.example.test" + t.Setenv(cfg.TokenEnv, "secret-token") + factory := &fakeUploadFactory{ + client: &fakeUploadClient{ + result: uploadFilesResult{RunID: "run-123", Status: "accepted"}, + statusErr: fmt.Errorf("status rejected secret-token"), + }, + } + client := newClient(cfg, factory.newClient) + + result, err := client.Upload(context.Background(), validUploadRequest()) + if err != nil { + t.Fatalf("Upload() error = %v, want accepted upload despite status lookup failure", err) + } + if result.Status != "accepted" || result.StatusError == "" { + t.Fatalf("result = %#v, want accepted status with status error", result) + } + if strings.Contains(result.StatusError, "secret-token") { + t.Fatalf("StatusError = %q, want token redacted", result.StatusError) + } +} + +func TestUploadPollsUntilTerminalStatus(t *testing.T) { + cfg := config.Defaults().Notify.Distributor + cfg.Endpoint = "https://distributor.example.test" + cfg.Timeout = 2 * time.Second + t.Setenv(cfg.TokenEnv, "secret-token") + factory := &fakeUploadFactory{ + client: &fakeUploadClient{ + result: uploadFilesResult{RunID: "run-123", Status: "accepted"}, + statuses: []runStatus{ + {RunID: "run-123", Status: "accepted"}, + {RunID: "run-123", Status: "succeeded", Report: json.RawMessage(`{"actions":[{"action":"replace_older"}]}`)}, + }, + }, + } + client := newClient(cfg, factory.newClient) + + result, err := client.Upload(context.Background(), validUploadRequest()) + if err != nil { + t.Fatalf("Upload() error = %v", err) + } + if result.Status != "succeeded" || result.RunStatus == nil || !strings.Contains(string(result.RunStatus.Report), "replace_older") { + t.Fatalf("result = %#v, want terminal succeeded status with run report", result) + } + if factory.client.statusCalls != 2 { + t.Fatalf("status calls = %d, want 2", factory.client.statusCalls) + } +} + +func TestUploadReturnsLatestStatusWhenPollingTimesOut(t *testing.T) { + cfg := config.Defaults().Notify.Distributor + cfg.Endpoint = "https://distributor.example.test" + cfg.Timeout = time.Millisecond + t.Setenv(cfg.TokenEnv, "secret-token") + factory := &fakeUploadFactory{ + client: &fakeUploadClient{ + result: uploadFilesResult{RunID: "run-123", Status: "accepted"}, + status: runStatus{RunID: "run-123", Status: "running"}, + }, + } + client := newClient(cfg, factory.newClient) + + result, err := client.Upload(context.Background(), validUploadRequest()) + if err != nil { + t.Fatalf("Upload() error = %v, want accepted upload with status timeout recorded", err) + } + if result.Status != "running" || result.StatusError == "" { + t.Fatalf("result = %#v, want latest status and status timeout", result) + } +} + +func TestUploadFailsWhenDistributorRunFailed(t *testing.T) { + cfg := config.Defaults().Notify.Distributor + cfg.Endpoint = "https://distributor.example.test" + t.Setenv(cfg.TokenEnv, "secret-token") + factory := &fakeUploadFactory{ + client: &fakeUploadClient{ + result: uploadFilesResult{RunID: "run-123", Status: "accepted"}, + status: runStatus{ + RunID: "run-123", + Status: "failed", + Error: "destination rejected secret-token", + Report: json.RawMessage(`{"actions":[{"action":"failed"}]}`), + }, + }, + } + client := newClient(cfg, factory.newClient) + + result, err := client.Upload(context.Background(), validUploadRequest()) + if err == nil { + t.Fatal("Upload() error = nil, want failed distributor run error") + } + if result.RunStatus == nil || result.RunStatus.Status != "failed" || !strings.Contains(string(result.RunStatus.Report), "failed") { + t.Fatalf("result = %#v, want failed run status report", result) + } + if strings.Contains(err.Error(), "secret-token") || strings.Contains(result.RunStatus.Error, "secret-token") { + t.Fatalf("error/result leaked token: err=%q result=%#v", err.Error(), result) + } +} + func TestUploadPreservesIdempotencyConflictDiagnosis(t *testing.T) { cfg := config.Defaults().Notify.Distributor cfg.Endpoint = "https://distributor.example.test" @@ -207,6 +322,7 @@ func validUploadRequest() UploadRequest { IdempotencyKey: "weatherreporter.home.daily.run", SourcePath: "/tmp/report.md", BundlePath: "daily.md", + CreatedAt: time.Date(2026, 6, 7, 12, 0, 0, 123, time.UTC), } } @@ -229,9 +345,14 @@ func (f *fakeUploadFactory) newClient(endpoint, token string, timeout time.Durat } type fakeUploadClient struct { - opts uploadFilesOptions - result uploadFilesResult - err error + opts uploadFilesOptions + statusRunID string + statusCalls int + result uploadFilesResult + status runStatus + statuses []runStatus + err error + statusErr error } func (c *fakeUploadClient) UploadFiles(ctx context.Context, opts uploadFilesOptions) (uploadFilesResult, error) { @@ -241,3 +362,19 @@ func (c *fakeUploadClient) UploadFiles(ctx context.Context, opts uploadFilesOpti } return c.result, nil } + +func (c *fakeUploadClient) Status(ctx context.Context, runID string) (runStatus, error) { + c.statusRunID = runID + c.statusCalls++ + if c.statusErr != nil { + return runStatus{}, c.statusErr + } + if len(c.statuses) > 0 { + index := c.statusCalls - 1 + if index >= len(c.statuses) { + index = len(c.statuses) - 1 + } + return c.statuses[index], nil + } + return c.status, nil +} diff --git a/internal/app/app.go b/internal/app/app.go index 1040738..0334dcd 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -86,20 +86,21 @@ type BriefingResult struct { } type ReportResult struct { - Briefing briefing.Package - BriefingPath string - DataPackage promptinput.Package - DataPackagePath string - PreflightPath string - ReportPath string - OutputPath string - Metadata state.Metadata - MetadataPath string - PriorSnapshot *state.PriorSnapshot - RecentChanges []changes.Change - RenderResult *scriptorium.RenderResult - RunResult *scriptorium.RunResult - Notification *NotificationResult + Briefing briefing.Package + BriefingPath string + DataPackage promptinput.Package + DataPackagePath string + PreflightPath string + ReportPath string + OutputPath string + NotificationPath string + Metadata state.Metadata + MetadataPath string + PriorSnapshot *state.PriorSnapshot + RecentChanges []changes.Change + RenderResult *scriptorium.RenderResult + RunResult *scriptorium.RunResult + Notification *NotificationResult } type BatchResult struct { @@ -122,6 +123,7 @@ type BatchReportResult struct { NotificationStatus string `json:"notificationStatus,omitempty"` NotificationRunID string `json:"notificationRunId,omitempty"` NotificationError string `json:"notificationError,omitempty"` + NotificationPath string `json:"notificationPath,omitempty"` GeneratedAt time.Time `json:"generatedAt"` ValidPeriod timeutil.Period `json:"validPeriod"` BriefingPath string `json:"briefingPath,omitempty"` @@ -159,6 +161,7 @@ type NotificationRequest struct { IdempotencyKey string ReportPath string BundlePath string + CreatedAt time.Time } type NotificationResult struct { @@ -166,6 +169,14 @@ type NotificationResult struct { IdempotencyKey string RunID string Status string + UploadStatus string + StatusError string + PipelineID string + AcceptedAt time.Time + StartedAt *time.Time + FinishedAt *time.Time + Report []byte + Error string } type NotificationError struct { @@ -269,6 +280,9 @@ func RunBatchDetailed(ctx context.Context, req BatchRequest) (*BatchResult, erro if errors.As(err, ¬ificationErr) { item.NotificationStatus = "failed" item.NotificationError = notificationErr.Error() + if paths, pathErr := store.Paths(resolved); pathErr == nil { + item.NotificationPath = paths.Notification + } } result.Failed++ } else { @@ -279,6 +293,7 @@ func RunBatchDetailed(ctx context.Context, req BatchRequest) (*BatchResult, erro item.ReportPath = reportResult.ReportPath item.OutputPath = reportResult.OutputPath item.MetadataPath = reportResult.MetadataPath + item.NotificationPath = reportResult.NotificationPath if reportResult.Notification != nil { item.NotificationStatus = reportResult.Notification.Status item.NotificationRunID = reportResult.Notification.RunID @@ -540,46 +555,62 @@ func GenerateReport(ctx context.Context, req ReportRequest) (*ReportResult, erro return nil, runErr } - notification, err := notifyReport(ctx, req.Config, req.Resolved, reportPath, metadata, req.Notifier) + notification, notificationPath, err := notifyReport(ctx, req.Config, req.Resolved, reportPath, metadata, req.Notifier, store) + if notificationPath != "" { + metadata.NotificationPath = notificationPath + metadataPath, metadataErr = store.SaveMetadata(ctx, metadata) + if metadataErr != nil { + return nil, metadataErr + } + } if err != nil { return nil, err } return &ReportResult{ - Briefing: briefingPackage, - BriefingPath: briefingPath, - DataPackage: dataPackage, - DataPackagePath: dataPackagePath, - PreflightPath: preflightPath, - ReportPath: reportPath, - OutputPath: outputPath, - Metadata: metadata, - MetadataPath: metadataPath, - PriorSnapshot: priorSnapshot, - RecentChanges: recentChanges, - RenderResult: renderResult, - RunResult: runResult, - Notification: notification, + Briefing: briefingPackage, + BriefingPath: briefingPath, + DataPackage: dataPackage, + DataPackagePath: dataPackagePath, + PreflightPath: preflightPath, + ReportPath: reportPath, + OutputPath: outputPath, + NotificationPath: notificationPath, + Metadata: metadata, + MetadataPath: metadataPath, + PriorSnapshot: priorSnapshot, + RecentChanges: recentChanges, + RenderResult: renderResult, + RunResult: runResult, + Notification: notification, }, nil } -func notifyReport(ctx context.Context, cfg config.Config, resolved report.Resolved, reportPath string, metadata state.Metadata, notifier Notifier) (*NotificationResult, error) { +func notifyReport(ctx context.Context, cfg config.Config, resolved report.Resolved, reportPath string, metadata state.Metadata, notifier Notifier, store state.Store) (*NotificationResult, string, error) { notifier, enabled := reportNotifier(cfg, notifier) if !enabled { - return nil, nil + return nil, "", nil } notificationRequest, err := buildNotificationRequest(cfg, resolved, reportPath, metadata) if err != nil { - return nil, err + notificationPath, saveErr := saveNotificationArtifact(ctx, store, resolved, cfg, metadata, NotificationRequest{}, nil, err) + if saveErr != nil { + return nil, "", saveErr + } + return nil, notificationPath, err } result, err := notifier.Notify(ctx, notificationRequest) + notificationPath, saveErr := saveNotificationArtifact(ctx, store, resolved, cfg, metadata, notificationRequest, result, err) + if saveErr != nil { + return nil, "", saveErr + } if err != nil { - return nil, &NotificationError{ + return result, notificationPath, &NotificationError{ Request: notificationRequest, Err: fmt.Errorf("notify report %q run %q from managed report %q: %w", resolved.Definition.ID, metadata.RunID, reportPath, err), } } - return result, nil + return result, notificationPath, nil } func reportNotifier(cfg config.Config, notifier Notifier) (Notifier, bool) { @@ -622,9 +653,57 @@ func buildNotificationRequest(cfg config.Config, resolved report.Resolved, repor IdempotencyKey: idempotencyKey, ReportPath: reportPath, BundlePath: bundlePath, + CreatedAt: metadata.GeneratedAt, }, nil } +func saveNotificationArtifact(ctx context.Context, store state.Store, resolved report.Resolved, cfg config.Config, metadata state.Metadata, req NotificationRequest, result *NotificationResult, notifyErr error) (string, error) { + if store == nil { + return "", fmt.Errorf("state store is required") + } + artifact := state.DistributorNotificationArtifact{ + SchemaVersion: state.DistributorNotificationSchemaVersion, + RunID: metadata.RunID, + ReportID: resolved.Definition.ID, + AttemptedAt: time.Now(), + Endpoint: cfg.Notify.Distributor.Endpoint, + BundleID: req.BundleID, + IdempotencyKey: req.IdempotencyKey, + SourcePath: req.ReportPath, + BundlePath: req.BundlePath, + BundleCreated: req.CreatedAt, + Status: "attempted", + } + if result != nil { + artifact.Status = result.Status + artifact.Upload = &state.DistributorUploadResult{ + RunID: result.RunID, + Status: result.UploadStatus, + } + if result.PipelineID != "" || !result.AcceptedAt.IsZero() || result.StartedAt != nil || result.FinishedAt != nil || len(result.Report) > 0 || result.Error != "" { + artifact.RunStatus = &state.DistributorRunStatus{ + RunID: result.RunID, + PipelineID: result.PipelineID, + Status: result.Status, + AcceptedAt: result.AcceptedAt, + StartedAt: result.StartedAt, + FinishedAt: result.FinishedAt, + Report: append([]byte(nil), result.Report...), + Error: result.Error, + } + } + artifact.StatusError = result.StatusError + } + if notifyErr != nil { + artifact.Status = "failed" + artifact.Error = notifyErr.Error() + } + if artifact.Status == "" { + artifact.Status = "unknown" + } + return store.SaveDistributorNotification(ctx, resolved, artifact) +} + type noopNotifier struct{} func (noopNotifier) Notify(context.Context, NotificationRequest) (*NotificationResult, error) { @@ -641,16 +720,28 @@ func (n distributorNotifier) Notify(ctx context.Context, req NotificationRequest IdempotencyKey: req.IdempotencyKey, SourcePath: req.ReportPath, BundlePath: req.BundlePath, + CreatedAt: req.CreatedAt, }) - if err != nil { - return nil, err - } - return &NotificationResult{ + notification := &NotificationResult{ BundleID: req.BundleID, IdempotencyKey: req.IdempotencyKey, RunID: result.RunID, Status: result.Status, - }, nil + UploadStatus: result.UploadStatus, + StatusError: result.StatusError, + } + if result.RunStatus != nil { + notification.PipelineID = result.RunStatus.PipelineID + notification.AcceptedAt = result.RunStatus.AcceptedAt + notification.StartedAt = result.RunStatus.StartedAt + notification.FinishedAt = result.RunStatus.FinishedAt + notification.Report = append([]byte(nil), result.RunStatus.Report...) + notification.Error = result.RunStatus.Error + } + if err != nil { + return notification, err + } + return notification, nil } func BuildBriefing(req BriefingRequest, bundle *forecast.Bundle) (briefing.Package, error) { diff --git a/internal/app/app_test.go b/internal/app/app_test.go index d5f9d4f..ed0ec3c 100644 --- a/internal/app/app_test.go +++ b/internal/app/app_test.go @@ -297,7 +297,13 @@ func TestGenerateReportNotifiesManagedReportPath(t *testing.T) { t.Fatalf("ResolveGenerate() error = %v", err) } notifier := &recordingNotifier{ - result: &NotificationResult{RunID: "distributor-run", Status: "accepted"}, + result: &NotificationResult{ + RunID: "distributor-run", + Status: "succeeded", + UploadStatus: "accepted", + PipelineID: "reports", + Report: []byte(`{"actions":[{"action":"replace_older"}]}`), + }, } outputPath := filepath.Join(t.TempDir(), "daily-copy.md") @@ -314,8 +320,18 @@ func TestGenerateReportNotifiesManagedReportPath(t *testing.T) { if result.Notification == nil { t.Fatal("Notification = nil, want notification result") } - if result.Notification.RunID != "distributor-run" || result.Notification.Status != "accepted" { - t.Fatalf("Notification = %#v, want accepted distributor run", result.Notification) + if result.Notification.RunID != "distributor-run" || result.Notification.Status != "succeeded" { + t.Fatalf("Notification = %#v, want succeeded distributor run", result.Notification) + } + if result.NotificationPath == "" || result.Metadata.NotificationPath != result.NotificationPath { + t.Fatalf("NotificationPath result=%q metadata=%q, want linked artifact", result.NotificationPath, result.Metadata.NotificationPath) + } + notificationData, err := os.ReadFile(result.NotificationPath) + if err != nil { + t.Fatalf("read notification artifact: %v", err) + } + if !strings.Contains(string(notificationData), `"replace_older"`) || !strings.Contains(string(notificationData), `"bundleCreated"`) { + t.Fatalf("notification artifact missing status report or created timestamp:\n%s", string(notificationData)) } if len(notifier.requests) != 1 { t.Fatalf("notification requests = %d, want 1", len(notifier.requests)) @@ -339,6 +355,9 @@ func TestGenerateReportNotifiesManagedReportPath(t *testing.T) { if req.RunID != result.Metadata.RunID { t.Fatalf("notification RunID = %q, want report run id %q", req.RunID, result.Metadata.RunID) } + if !req.CreatedAt.Equal(result.Metadata.GeneratedAt) { + t.Fatalf("notification CreatedAt = %s, want generated at %s", req.CreatedAt, result.Metadata.GeneratedAt) + } } func TestGenerateReportNotificationFailureFailsReport(t *testing.T) { @@ -356,10 +375,15 @@ func TestGenerateReportNotificationFailureFailsReport(t *testing.T) { } notifier := &recordingNotifier{err: errors.New("upload rejected")} + store, err := state.NewFilesystemStore(cfg.Workspace) + if err != nil { + t.Fatalf("NewFilesystemStore() error = %v", err) + } _, err = GenerateReport(context.Background(), ReportRequest{ Config: cfg, Resolved: resolved, Renderer: successfulRenderer("# Daily Report\n"), + Store: store, Notifier: notifier, }) if err == nil { @@ -371,6 +395,21 @@ func TestGenerateReportNotificationFailureFailsReport(t *testing.T) { if len(notifier.requests) != 1 { t.Fatalf("notification requests = %d, want one attempted notification", len(notifier.requests)) } + paths, pathErr := store.Paths(resolved) + if pathErr != nil { + t.Fatalf("Paths() error = %v", pathErr) + } + notificationData, readErr := os.ReadFile(paths.Notification) + if readErr != nil { + t.Fatalf("read notification artifact after failure: %v", readErr) + } + var notification state.DistributorNotificationArtifact + if err := json.Unmarshal(notificationData, ¬ification); err != nil { + t.Fatalf("decode notification artifact: %v", err) + } + if notification.Status != "failed" || !strings.Contains(notification.Error, "upload rejected") { + t.Fatalf("notification failure artifact = %+v, want failed upload context", notification) + } } func TestGenerateReportDoesNotNotifyAfterRenderOrRunFailure(t *testing.T) { @@ -1467,6 +1506,7 @@ func (n *recordingNotifier) Notify(_ context.Context, req NotificationRequest) ( BundleID: req.BundleID, IdempotencyKey: req.IdempotencyKey, Status: "accepted", + UploadStatus: "accepted", }, nil } diff --git a/internal/config/config.go b/internal/config/config.go index 87a673f..72203c7 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -75,11 +75,12 @@ type ScriptoriumConfig struct { } type WorkspaceConfig struct { - Root string `yaml:"root"` - SnapshotsDir string `yaml:"snapshots_dir"` - ReportsDir string `yaml:"reports_dir"` - DataPackagesDir string `yaml:"data_packages_dir"` - PreflightDir string `yaml:"preflight_dir"` + Root string `yaml:"root"` + SnapshotsDir string `yaml:"snapshots_dir"` + ReportsDir string `yaml:"reports_dir"` + DataPackagesDir string `yaml:"data_packages_dir"` + PreflightDir string `yaml:"preflight_dir"` + NotificationsDir string `yaml:"notifications_dir"` } type DaypartConfig struct { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index d59a13f..9c345fe 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -96,6 +96,9 @@ func TestLoadMinimalExampleConfig(t *testing.T) { if cfg.Workspace.Root != "workspace" { t.Fatalf("Workspace.Root = %q, want default workspace", cfg.Workspace.Root) } + if cfg.Workspace.NotificationsDir != "notifications" { + t.Fatalf("Workspace.NotificationsDir = %q, want notifications", cfg.Workspace.NotificationsDir) + } if cfg.Location.Name != "Brentwood" { t.Fatalf("Location.Name = %q, want default Brentwood", cfg.Location.Name) } diff --git a/internal/config/defaults.go b/internal/config/defaults.go index 36d2361..c4f1c8f 100644 --- a/internal/config/defaults.go +++ b/internal/config/defaults.go @@ -42,11 +42,12 @@ func Defaults() Config { Timeout: 2 * time.Minute, }, Workspace: WorkspaceConfig{ - Root: "workspace", - SnapshotsDir: "snapshots", - ReportsDir: "reports", - DataPackagesDir: "data-packages", - PreflightDir: "preflight", + Root: "workspace", + SnapshotsDir: "snapshots", + ReportsDir: "reports", + DataPackagesDir: "data-packages", + PreflightDir: "preflight", + NotificationsDir: "notifications", }, Dayparts: []DaypartConfig{ {Name: "overnight", Start: "00:00", End: "06:00"}, diff --git a/internal/state/filesystem.go b/internal/state/filesystem.go index 9fa909a..20248a5 100644 --- a/internal/state/filesystem.go +++ b/internal/state/filesystem.go @@ -18,11 +18,12 @@ import ( ) type FilesystemStore struct { - root string - snapshotsDir string - reportsDir string - dataPackagesDir string - preflightDir string + root string + snapshotsDir string + reportsDir string + dataPackagesDir string + preflightDir string + notificationsDir string } type ArtifactPaths struct { @@ -30,6 +31,7 @@ type ArtifactPaths struct { Metadata string `json:"metadata"` DataPackage string `json:"dataPackage"` Preflight string `json:"preflight"` + Notification string `json:"notification,omitempty"` RenderedReport string `json:"renderedReport,omitempty"` } @@ -57,17 +59,19 @@ func NewFilesystemStore(cfg config.WorkspaceConfig) (*FilesystemStore, error) { "reports_dir": cfg.ReportsDir, "data_packages_dir": cfg.DataPackagesDir, "preflight_dir": cfg.PreflightDir, + "notifications_dir": cfg.NotificationsDir, } { if err := validateRelativeDir(name, value); err != nil { return nil, err } } return &FilesystemStore{ - root: filepath.Clean(cfg.Root), - snapshotsDir: filepath.Clean(cfg.SnapshotsDir), - reportsDir: filepath.Clean(cfg.ReportsDir), - dataPackagesDir: filepath.Clean(cfg.DataPackagesDir), - preflightDir: filepath.Clean(cfg.PreflightDir), + root: filepath.Clean(cfg.Root), + snapshotsDir: filepath.Clean(cfg.SnapshotsDir), + reportsDir: filepath.Clean(cfg.ReportsDir), + dataPackagesDir: filepath.Clean(cfg.DataPackagesDir), + preflightDir: filepath.Clean(cfg.PreflightDir), + notificationsDir: filepath.Clean(cfg.NotificationsDir), }, nil } @@ -90,6 +94,7 @@ func (s *FilesystemStore) Paths(resolved report.Resolved) (ArtifactPaths, error) Metadata: s.join(s.snapshotsDir, group, validDate, filenameBase+".metadata.json"), DataPackage: s.join(s.dataPackagesDir, group, validDate, filenameBase+".data_package.json"), Preflight: s.join(s.preflightDir, group, validDate, filenameBase+".render.json"), + Notification: s.join(s.notificationsDir, group, validDate, filenameBase+".distributor.json"), RenderedReport: s.join(s.reportsDir, group, filenameBase+".md"), }, nil } @@ -130,6 +135,20 @@ func (s *FilesystemStore) SavePreflight(_ context.Context, resolved report.Resol return paths.Preflight, nil } +func (s *FilesystemStore) SaveDistributorNotification(_ context.Context, resolved report.Resolved, artifact DistributorNotificationArtifact) (string, error) { + paths, err := s.Paths(resolved) + if err != nil { + return "", err + } + if artifact.SchemaVersion == "" { + artifact.SchemaVersion = DistributorNotificationSchemaVersion + } + if err := fileutil.WriteJSONAtomic(paths.Notification, artifact); err != nil { + return "", err + } + return paths.Notification, nil +} + func (s *FilesystemStore) PrepareRenderedReport(_ context.Context, resolved report.Resolved) (string, error) { paths, err := s.Paths(resolved) if err != nil { diff --git a/internal/state/filesystem_test.go b/internal/state/filesystem_test.go index 97192f6..4380a8e 100644 --- a/internal/state/filesystem_test.go +++ b/internal/state/filesystem_test.go @@ -30,6 +30,7 @@ func TestPathsUseRunIDAndWorkspace(t *testing.T) { filepath.Join("snapshots", "daily", "2026-05-29", "20260529T100000.000000000Z_daily_today.metadata.json"), filepath.Join("data-packages", "daily", "2026-05-29", "20260529T100000.000000000Z_daily_today.data_package.json"), filepath.Join("preflight", "daily", "2026-05-29", "20260529T100000.000000000Z_daily_today.render.json"), + filepath.Join("notifications", "daily", "2026-05-29", "20260529T100000.000000000Z_daily_today.distributor.json"), filepath.Join("reports", "daily", "20260529T100000.000000000Z_daily_today.md"), } { if !strings.Contains(pathsString(paths), want) { @@ -59,6 +60,22 @@ func TestSaveArtifactsAndMetadataRoundTrip(t *testing.T) { if err != nil { t.Fatalf("SavePreflight() error = %v", err) } + notificationPath, err := store.SaveDistributorNotification(context.Background(), resolved, DistributorNotificationArtifact{ + RunID: resolved.Metadata().RunID, + ReportID: resolved.Definition.ID, + AttemptedAt: resolved.GeneratedAt, + Endpoint: "https://distributor.example.test", + BundleID: "weatherreporter.home.daily.run", + IdempotencyKey: "weatherreporter.home.daily.run", + SourcePath: "/tmp/report.md", + BundlePath: "daily.md", + BundleCreated: resolved.GeneratedAt, + Status: "succeeded", + RunStatus: &DistributorRunStatus{RunID: "distributor-run", Status: "succeeded"}, + }) + if err != nil { + t.Fatalf("SaveDistributorNotification() error = %v", err) + } renderedReportPath, err := store.PrepareRenderedReport(context.Background(), resolved) if err != nil { t.Fatalf("PrepareRenderedReport() error = %v", err) @@ -77,6 +94,17 @@ func TestSaveArtifactsAndMetadataRoundTrip(t *testing.T) { if preflight.Stdout != `{"ok":true}` { t.Fatalf("preflight stdout = %q, want render stdout", preflight.Stdout) } + var notification DistributorNotificationArtifact + notificationData, err := os.ReadFile(notificationPath) + if err != nil { + t.Fatalf("read notification: %v", err) + } + if err := json.Unmarshal(notificationData, ¬ification); err != nil { + t.Fatalf("decode notification: %v", err) + } + if notification.SchemaVersion != DistributorNotificationSchemaVersion || notification.RunStatus == nil || notification.RunStatus.Status != "succeeded" { + t.Fatalf("notification = %#v, want persisted distributor status", notification) + } paths, err := store.Paths(resolved) if err != nil { t.Fatalf("Paths() error = %v", err) @@ -93,7 +121,7 @@ func TestSaveArtifactsAndMetadataRoundTrip(t *testing.T) { t.Fatalf("SaveMetadata() error = %v", err) } - for _, path := range []string{briefingPath, dataPackagePath, preflightPath, renderedReportPath, metadataPath} { + for _, path := range []string{briefingPath, dataPackagePath, preflightPath, notificationPath, renderedReportPath, metadataPath} { if _, err := os.Stat(path); err != nil { t.Fatalf("expected artifact %q: %v", path, err) } @@ -494,6 +522,7 @@ func pathsString(paths ArtifactPaths) string { paths.Metadata, paths.DataPackage, paths.Preflight, + paths.Notification, paths.RenderedReport, }, "\n") } diff --git a/internal/state/metadata.go b/internal/state/metadata.go index fc55d2f..01eedf9 100644 --- a/internal/state/metadata.go +++ b/internal/state/metadata.go @@ -29,6 +29,7 @@ type Metadata struct { BriefingPath string `json:"briefingPath"` DataPackagePath string `json:"dataPackagePath"` PreflightPath string `json:"preflightPath"` + NotificationPath string `json:"notificationPath,omitempty"` RenderedReportPath string `json:"renderedReportPath,omitempty"` } diff --git a/internal/state/store.go b/internal/state/store.go index 766a265..d1771c0 100644 --- a/internal/state/store.go +++ b/internal/state/store.go @@ -3,6 +3,8 @@ package state import ( "context" + "encoding/json" + "time" "gitea.maximumdirect.net/eric/weatherreporter/internal/briefing" "gitea.maximumdirect.net/eric/weatherreporter/internal/promptinput" @@ -14,6 +16,7 @@ type Store interface { SaveBriefing(context.Context, report.Resolved, briefing.Package) (string, error) 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) PrepareRenderedReport(context.Context, report.Resolved) (string, error) SaveMetadata(context.Context, Metadata) (string, error) FindPriorSnapshot(context.Context, report.Resolved) (*PriorSnapshot, error) @@ -33,3 +36,39 @@ type PreflightArtifact struct { StderrTruncated bool `json:"stderrTruncated,omitempty"` ExitCode int `json:"exitCode"` } + +const DistributorNotificationSchemaVersion = "weatherreporter.distributor_notification.v1" + +type DistributorNotificationArtifact struct { + SchemaVersion string `json:"schemaVersion"` + RunID string `json:"runId"` + ReportID report.ID `json:"reportId"` + AttemptedAt time.Time `json:"attemptedAt"` + Endpoint string `json:"endpoint"` + BundleID string `json:"bundleId,omitempty"` + IdempotencyKey string `json:"idempotencyKey,omitempty"` + SourcePath string `json:"sourcePath,omitempty"` + BundlePath string `json:"bundlePath,omitempty"` + BundleCreated time.Time `json:"bundleCreated,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 DistributorUploadResult struct { + RunID string `json:"runId,omitempty"` + Status string `json:"status,omitempty"` +} + +type DistributorRunStatus struct { + RunID string `json:"runId,omitempty"` + PipelineID string `json:"pipelineId,omitempty"` + Status string `json:"status,omitempty"` + AcceptedAt time.Time `json:"acceptedAt,omitempty"` + StartedAt *time.Time `json:"startedAt,omitempty"` + FinishedAt *time.Time `json:"finishedAt,omitempty"` + Report json.RawMessage `json:"report,omitempty"` + Error string `json:"error,omitempty"` +}