From 748e02db80dda0d901659c825f60334be9c857b7 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Wed, 22 Jul 2026 14:24:03 +0000 Subject: [PATCH] Harden checkpoint reuse and combat validation --- docs/internal/pipeline.md | 28 +- docs/internal/state.md | 16 +- docs/operations.md | 7 +- internal/cli/dnd_combat_contract_test.go | 4 +- internal/cli/production_contract_test.go | 4 +- internal/framework/checkpoint/loader.go | 70 ++-- internal/framework/pipeline/checkpoint.go | 100 ++++-- .../runner_accepted_checkpoint_test.go | 118 ++++++- .../framework/pipeline/runner_concurrent.go | 315 +++++++++++------- internal/framework/pipeline/runner_typed.go | 20 +- .../pipeline/runner_typed_checkpoint_test.go | 27 +- .../assets/prompts/instructions.md | 6 +- internal/modules/dnd/register/chains.go | 4 +- .../modules/dnd/register/register_test.go | 4 +- .../integration/dnd_combat_runner_test.go | 49 ++- 15 files changed, 520 insertions(+), 252 deletions(-) diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 495db01..74b0892 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -193,10 +193,12 @@ the runner returns. The pipeline-wide coordinator owns the ordered step loop, generated-reference sets at each barrier, and deterministic merging of step outcomes. For one step, -the lane engine initializes checkpoint state in lane order, dispatches bounded -extract work, advances terminal lanes through serial merge and normalize work, -selects failures by stable pipeline scope, and merges lane-local outcomes back -in resolved order. Completion timing never becomes public ordering. +one run-local lane engine owns worker lifecycle, cancellation, dispatch, +continuation queues, and result collection. It initializes checkpoint state in +lane order, dispatches bounded extract work, advances terminal lanes through +serial merge and normalize work, selects failures by stable pipeline scope, and +merges lane-local outcomes back in resolved order. Completion timing never +becomes public ordering. The runner: @@ -329,14 +331,16 @@ current non-empty checkpoint identity to match, so the invocation identity still binds the input, resolved topology and configuration, references, runtime overrides, profiles, and component fingerprints. -The runner decodes that accepted normalized artifact with the prepared codec, -re-encodes it, and requires exact kind, schema identity and digest, media type, -canonical bytes, content digest, and producer provenance. A valid result becomes -a runner-owned cloned normalized output, restores only normalize-checkpoint -warnings, and records one `accepted_artifact_reused` normalize decision. It does -not invoke or record extract, merge, normalize, or their validators. Invalid or -unavailable accepted state records its decision and fails the producer step; -the dependent step never starts and the producer is not implicitly rerun. +The runner decodes and canonically re-encodes each reusable artifact once with +the prepared codec, requiring exact kind, schema identity and digest, media +type, canonical bytes, content digest, and producer provenance. A valid accepted +producer becomes a runner-owned cloned normalized output, restores only +normalize-checkpoint warnings, and records one `accepted_artifact_reused` +normalize decision. It does not invoke or record extract, merge, normalize, or +their validators. Invalid or unavailable accepted state records its decision +and fails the producer step; the dependent step never starts and the producer +is not implicitly rerun. If a later required lane fails during initialization, +already hydrated terminal lanes remain in the failed output in resolved order. Generated references add downstream dependencies containing the producer's artifact kind, complete schema identity, media type, canonical content digest, diff --git a/docs/internal/state.md b/docs/internal/state.md index 9faac72..55c0c03 100644 --- a/docs/internal/state.md +++ b/docs/internal/state.md @@ -57,14 +57,14 @@ records the decision, and stops without executing the producer or consumer. The loader assigns a typed category and reason code at each validation site; diagnostic prose is not classified after the fact. The runner then applies forced-execution policy, validates reusable artifact bytes through the prepared -codec, and records the final decision before enforcing a required-predecessor -failure. That failure names only the step, lane, and stable reason code. Decision -detail passes through one UTF-8-safe bounded sanitizer and contains only -allowlisted diagnostic context, never payloads, references, credentials, -environment values, or physical paths. Typed categories and codes remain intact -through pipeline events and become strings only in manifest and debug-summary -JSON. [Operations](../operations.md#resume-and-selective-recompute) is the -canonical operator-facing reason-code reference. +codec once, returns the canonical hydrated value to the stage, and records the +final decision before enforcing a required-predecessor failure. That failure +names only the step, lane, and stable reason code. Decision detail is selected +from code-owned descriptions by reason code and then UTF-8 normalized and +bounded; callers cannot supply arbitrary diagnostic prose. Typed categories and +codes remain intact through pipeline events and become strings only in manifest +and debug-summary JSON. [Operations](../operations.md#resume-and-selective-recompute) +is the canonical operator-facing reason-code reference. `internal/core/fileio` provides confined atomic file writes used by state collaborators. The chunk-plan store retains its stronger entry validation. diff --git a/docs/operations.md b/docs/operations.md index 0225c6d..1588ab3 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -209,9 +209,10 @@ Checkpoint reason codes are stable diagnostic identifiers: | `accepted_artifact_reused` | A required producer's accepted normalized artifact was canonically validated and hydrated. | | `recompute_step` | Selective recomputation forced execution of this lane. | -Decision detail is bounded explanatory text, not a data-recovery channel. It -never contains checkpoint paths, artifact or reference content, source content, -credentials, or environment values. +Decision detail is bounded explanatory text derived from the stable reason code, +not caller-supplied prose or a data-recovery channel. It never contains +checkpoint paths, artifact or reference content, source content, credentials, +or environment values. ## Debug Bundles diff --git a/internal/cli/dnd_combat_contract_test.go b/internal/cli/dnd_combat_contract_test.go index f8f0480..4d6b1f6 100644 --- a/internal/cli/dnd_combat_contract_test.go +++ b/internal/cli/dnd_combat_contract_test.go @@ -55,17 +55,17 @@ func TestProductionCombatConfigurationResolvesTypedLane(t *testing.T) { wantExtractChain := []pipeline.ModuleBinding{ pipeline.Binding("generic/valid_json"), - pipeline.Binding("generic/valid_json_schema"), pipeline.Binding("extract/dnd/combat-turns/shape"), pipeline.Binding("extract/dnd/combat-turns/source_refs"), + pipeline.Binding("generic/valid_json_schema"), pipeline.Binding("extract/dnd/combat-turns/source_relatedness"), } wantNormalizeChain := []pipeline.ModuleBinding{ pipeline.Binding("generic/valid_json"), - pipeline.Binding("generic/valid_json_schema"), pipeline.Binding("extract/dnd/combat-turns/shape"), pipeline.Binding("normalize/dnd/combat-turns/invariants"), pipeline.Binding("extract/dnd/combat-turns/source_refs"), + pipeline.Binding("generic/valid_json_schema"), pipeline.Binding("extract/dnd/combat-turns/source_relatedness"), } if got := validatorChain(effective.ResolvedPipeline, pipeline.StageExtract, combatextract.Key); !reflect.DeepEqual(got, wantExtractChain) { diff --git a/internal/cli/production_contract_test.go b/internal/cli/production_contract_test.go index cd0c16f..300fa96 100644 --- a/internal/cli/production_contract_test.go +++ b/internal/cli/production_contract_test.go @@ -77,17 +77,17 @@ func TestProductionCatalogCoversMaintainedConfigurations(t *testing.T) { } combatExtractChain := []pipeline.ModuleBinding{ pipeline.Binding("generic/valid_json"), - pipeline.Binding("generic/valid_json_schema"), pipeline.Binding("extract/dnd/combat-turns/shape"), pipeline.Binding("extract/dnd/combat-turns/source_refs"), + pipeline.Binding("generic/valid_json_schema"), pipeline.Binding("extract/dnd/combat-turns/source_relatedness"), } combatNormalizeChain := []pipeline.ModuleBinding{ pipeline.Binding("generic/valid_json"), - pipeline.Binding("generic/valid_json_schema"), pipeline.Binding("extract/dnd/combat-turns/shape"), pipeline.Binding("normalize/dnd/combat-turns/invariants"), pipeline.Binding("extract/dnd/combat-turns/source_refs"), + pipeline.Binding("generic/valid_json_schema"), pipeline.Binding("extract/dnd/combat-turns/source_relatedness"), } if got := registries.ValidatorChains.Validators(pipeline.StageExtract, combatextract.Key); !reflect.DeepEqual(got, combatExtractChain) { diff --git a/internal/framework/checkpoint/loader.go b/internal/framework/checkpoint/loader.go index 5c9670b..accfacb 100644 --- a/internal/framework/checkpoint/loader.go +++ b/internal/framework/checkpoint/loader.go @@ -52,13 +52,13 @@ func (l *FilesystemLoader) Source(moduleKey string) (pipeline.SourceCheckpoint, } doc := cloneSourceDocument(payload.Document) if err := source.ValidateDocument(&doc); err != nil { - return pipeline.SourceCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactPayloadInvalid, "source checkpoint document is invalid") + return pipeline.SourceCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactPayloadInvalid) } if strings.TrimSpace(manifest.SourceID) != "" && manifest.SourceID != doc.ID { - return pipeline.SourceCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactPayloadInvalid, "source checkpoint identity does not match its payload") + return pipeline.SourceCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactPayloadInvalid) } if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), digestFingerprints("source_document", doc.Digest)) { - return pipeline.SourceCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch, "source checkpoint output digest does not match its payload") + return pipeline.SourceCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch) } return pipeline.SourceCheckpoint{Document: &doc}, reusedDecision() } @@ -84,7 +84,7 @@ func (l *FilesystemLoader) ExtractForStep(stepID, laneID, moduleKey string, depe return pipeline.ExtractCheckpoint{}, artifactDecision(err, "extract checkpoint artifact is invalid") } if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(outputs)) { - return pipeline.ExtractCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch, "extract checkpoint output digest does not match its payload") + return pipeline.ExtractCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch) } return pipeline.ExtractCheckpoint{Outputs: outputs, Rejected: cloneRejectedOutputs(payload.Rejected), Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() } @@ -110,7 +110,7 @@ func (l *FilesystemLoader) MergeForStep(stepID, laneID, moduleKey string, depend return pipeline.MergeCheckpoint{}, artifactDecision(err, "merge checkpoint artifact is invalid") } if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) { - return pipeline.MergeCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch, "merge checkpoint output digest does not match its payload") + return pipeline.MergeCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch) } return pipeline.MergeCheckpoint{Output: values[0], Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() } @@ -136,7 +136,7 @@ func (l *FilesystemLoader) NormalizeForStep(stepID, laneID, moduleKey string, de return pipeline.NormalizeCheckpoint{}, artifactDecision(err, "normalize checkpoint artifact is invalid") } if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) { - return pipeline.NormalizeCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch, "normalize checkpoint output digest does not match its payload") + return pipeline.NormalizeCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch) } return pipeline.NormalizeCheckpoint{Output: values[0], Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() } @@ -158,36 +158,36 @@ func (l *FilesystemLoader) AcceptedNormalize(stepID, laneID, moduleKey string) ( return pipeline.NormalizeCheckpoint{}, artifactDecision(err, "accepted normalize checkpoint artifact is invalid") } if len(values) != 1 { - return pipeline.NormalizeCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactPayloadInvalid, "accepted normalize checkpoint payload is invalid") + return pipeline.NormalizeCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactPayloadInvalid) } if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) { - return pipeline.NormalizeCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch, "accepted normalize checkpoint digest does not match its payload") + return pipeline.NormalizeCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch) } - return pipeline.NormalizeCheckpoint{Output: values[0], Warnings: cloneWarnings(payload.Warnings)}, decision(pipeline.CheckpointDecisionReused, pipeline.CheckpointReasonAcceptedArtifactReused, "accepted normalized artifact is reusable") + return pipeline.NormalizeCheckpoint{Output: values[0], Warnings: cloneWarnings(payload.Warnings)}, decision(pipeline.CheckpointDecisionReused, pipeline.CheckpointReasonAcceptedArtifactReused) } func (l *FilesystemLoader) validateAcceptedNormalizeManifest(manifest StageManifest, stepID, laneID, moduleKey string) pipeline.CheckpointDecision { if manifest.WorkspaceSchemaVersion != WorkspaceSchemaVersion { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonWorkspaceSchemaIncompatible, "checkpoint workspace schema is incompatible") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonWorkspaceSchemaIncompatible) } identity := strings.TrimSpace(l.identityDigest) if identity == "" || strings.TrimSpace(manifest.Metadata["checkpoint_identity_digest"]) == "" || manifest.Metadata["checkpoint_identity_digest"] != identity { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonIdentityMismatch, "checkpoint identity is unavailable or does not match the current invocation") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonIdentityMismatch) } if manifest.Stage != StageNormalize { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStageMismatch, "checkpoint stage does not match normalize") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStageMismatch) } if strings.TrimSpace(stepID) == "" || manifest.StepID != stepID { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStepMismatch, "checkpoint step does not match the requested step") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStepMismatch) } if strings.TrimSpace(laneID) == "" || manifest.LaneID != laneID { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonLaneMismatch, "checkpoint lane does not match the requested lane") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonLaneMismatch) } if strings.TrimSpace(moduleKey) == "" || manifest.ModuleKey != moduleKey { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonModuleMismatch, "checkpoint module does not match the requested normalizer") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonModuleMismatch) } if manifest.Status != StatusSucceeded { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStatusNotReusable, "checkpoint status cannot provide an accepted normalized artifact") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStatusNotReusable) } return reusedDecision() } @@ -212,21 +212,21 @@ func artifactCheckpointOutputs(values []artifactCheckpointEnvelope) ([]pipeline. func (l *FilesystemLoader) readJSON(name string, out any) pipeline.CheckpointDecision { if !l.Enabled() { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonLoadingDisabled, "checkpoint loading disabled") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonLoadingDisabled) } target, err := fileio.SafePath(l.root, name) if err != nil { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonPathInvalid, "checkpoint path is invalid") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonPathInvalid) } data, err := os.ReadFile(target) if err != nil { if os.IsNotExist(err) { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonMissing, "checkpoint artifact is missing") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonMissing) } - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonReadFailed, "checkpoint artifact could not be read") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonReadFailed) } if err := json.Unmarshal(data, out); err != nil { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonDecodeFailed, "checkpoint artifact could not be decoded") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonDecodeFailed) } return reusedDecision() } @@ -237,28 +237,28 @@ func (l *FilesystemLoader) validateManifest(manifest StageManifest, stage StageN func (l *FilesystemLoader) validateLaneManifest(manifest StageManifest, stage StageName, stepID string, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, statuses ...StageStatus) pipeline.CheckpointDecision { if manifest.WorkspaceSchemaVersion == WorkspaceSchemaVersionV1 { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonWorkspaceSchemaIncompatible, "checkpoint workspace schema is incompatible") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonWorkspaceSchemaIncompatible) } if manifest.WorkspaceSchemaVersion == WorkspaceSchemaVersionV2 { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonWorkspaceSchemaIncompatible, "checkpoint workspace schema is incompatible") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonWorkspaceSchemaIncompatible) } if manifest.WorkspaceSchemaVersion != WorkspaceSchemaVersion { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonWorkspaceSchemaIncompatible, "checkpoint workspace schema is incompatible") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonWorkspaceSchemaIncompatible) } if strings.TrimSpace(l.identityDigest) != "" && manifest.Metadata["checkpoint_identity_digest"] != l.identityDigest { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonIdentityMismatch, "checkpoint identity does not match the current invocation") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonIdentityMismatch) } if manifest.Stage != stage { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStageMismatch, "checkpoint stage does not match the requested stage") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStageMismatch) } if strings.TrimSpace(stepID) != "" && manifest.StepID != stepID { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStepMismatch, "checkpoint step does not match the requested step") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStepMismatch) } if strings.TrimSpace(laneID) != "" && manifest.LaneID != laneID { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonLaneMismatch, "checkpoint lane does not match the requested lane") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonLaneMismatch) } if strings.TrimSpace(moduleKey) != "" && manifest.ModuleKey != moduleKey { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonModuleMismatch, "checkpoint module does not match the requested module") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonModuleMismatch) } statusOK := false for _, status := range statuses { @@ -268,10 +268,10 @@ func (l *FilesystemLoader) validateLaneManifest(manifest StageManifest, stage St } } if !statusOK { - return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStatusNotReusable, "checkpoint status cannot be reused") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStatusNotReusable) } if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.DependencyFingerprints), dependencies) { - return decision(pipeline.CheckpointDecisionDependencyInvalidated, pipeline.CheckpointReasonDependencyMismatch, "checkpoint dependencies do not match") + return decision(pipeline.CheckpointDecisionDependencyInvalidated, pipeline.CheckpointReasonDependencyMismatch) } return reusedDecision() } @@ -313,11 +313,11 @@ func fingerprintsEqual(a []pipeline.CheckpointFingerprint, b []pipeline.Checkpoi } func reusedDecision() pipeline.CheckpointDecision { - return decision(pipeline.CheckpointDecisionReused, pipeline.CheckpointReasonReused, "checkpoint is reusable") + return decision(pipeline.CheckpointDecisionReused, pipeline.CheckpointReasonReused) } -func decision(category pipeline.CheckpointDecisionCategory, code pipeline.CheckpointReasonCode, detail string) pipeline.CheckpointDecision { - return pipeline.NewCheckpointDecision(category, code, detail) +func decision(category pipeline.CheckpointDecisionCategory, code pipeline.CheckpointReasonCode) pipeline.CheckpointDecision { + return pipeline.NewCheckpointDecision(category, code) } type artifactPayloadError struct { @@ -332,5 +332,5 @@ func artifactDecision(err error, detail string) pipeline.CheckpointDecision { if payloadErr, ok := err.(*artifactPayloadError); ok { code = payloadErr.code } - return decision(pipeline.CheckpointDecisionExecuted, code, detail) + return decision(pipeline.CheckpointDecisionExecuted, code) } diff --git a/internal/framework/pipeline/checkpoint.go b/internal/framework/pipeline/checkpoint.go index 6d0fe82..0aa4e41 100644 --- a/internal/framework/pipeline/checkpoint.go +++ b/internal/framework/pipeline/checkpoint.go @@ -97,18 +97,18 @@ type CheckpointDecision struct { ReasonCode CheckpointReasonCode `json:"reason_code,omitempty"` Detail string `json:"detail,omitempty"` // Reason is retained as a compatibility/debug field for existing callers. - // New checkpoint stores should put bounded, non-sensitive text in Detail. + // Detail and Reason are derived from the stable reason code. Reason string `json:"reason,omitempty"` } const checkpointDecisionDetailLimit = 512 -func NewCheckpointDecision(category CheckpointDecisionCategory, reasonCode CheckpointReasonCode, detail string) CheckpointDecision { - return checkpointDecision(category, reasonCode, detail) +func NewCheckpointDecision(category CheckpointDecisionCategory, reasonCode CheckpointReasonCode) CheckpointDecision { + return checkpointDecision(category, reasonCode) } -func checkpointDecision(category CheckpointDecisionCategory, reasonCode CheckpointReasonCode, detail string) CheckpointDecision { - detail = sanitizeCheckpointDecisionDetail(detail) +func checkpointDecision(category CheckpointDecisionCategory, reasonCode CheckpointReasonCode) CheckpointDecision { + detail := normalizeCheckpointDecisionDetail(checkpointDecisionDetail(reasonCode)) return CheckpointDecision{ Reused: category == CheckpointDecisionReused, Category: category, @@ -118,12 +118,55 @@ func checkpointDecision(category CheckpointDecisionCategory, reasonCode Checkpoi } } -func sanitizeCheckpointDecisionDetail(detail string) string { - detail = strings.TrimSpace(strings.ToValidUTF8(detail, "?")) - lower := strings.ToLower(detail) - if strings.ContainsAny(detail, `/\\`) || strings.Contains(lower, "secret") || strings.Contains(lower, "token") || strings.Contains(lower, "password") || strings.Contains(lower, "credential") || strings.Contains(lower, "environment") { - return "checkpoint decision detail redacted" +func checkpointDecisionDetail(reasonCode CheckpointReasonCode) string { + switch reasonCode { + case CheckpointReasonLoadingDisabled: + return "checkpoint loading is disabled" + case CheckpointReasonMissing: + return "checkpoint artifact is missing" + case CheckpointReasonPathInvalid: + return "checkpoint location is invalid" + case CheckpointReasonReadFailed: + return "checkpoint artifact could not be read" + case CheckpointReasonDecodeFailed: + return "checkpoint artifact could not be decoded" + case CheckpointReasonWorkspaceSchemaIncompatible: + return "checkpoint workspace schema is incompatible" + case CheckpointReasonIdentityMismatch: + return "checkpoint identity does not match the current invocation" + case CheckpointReasonStageMismatch: + return "checkpoint stage does not match" + case CheckpointReasonStepMismatch: + return "checkpoint step does not match" + case CheckpointReasonLaneMismatch: + return "checkpoint lane does not match" + case CheckpointReasonModuleMismatch: + return "checkpoint module does not match" + case CheckpointReasonStatusNotReusable: + return "checkpoint status is not reusable" + case CheckpointReasonDependencyMismatch: + return "checkpoint dependencies do not match" + case CheckpointReasonArtifactPayloadInvalid: + return "checkpoint artifact payload is invalid" + case CheckpointReasonArtifactDigestMismatch: + return "checkpoint artifact digest does not match" + case CheckpointReasonArtifactCodecIncompatible: + return "checkpoint artifact is incompatible with the registered codec" + case CheckpointReasonArtifactNotCanonical: + return "checkpoint artifact is not canonical" + case CheckpointReasonReused: + return "checkpoint is reusable" + case CheckpointReasonAcceptedArtifactReused: + return "accepted normalized artifact is reusable" + case CheckpointReasonRecomputeStep: + return "selected step requires execution" + default: + return "checkpoint decision" } +} + +func normalizeCheckpointDecisionDetail(detail string) string { + detail = strings.TrimSpace(strings.ToValidUTF8(detail, "?")) var b strings.Builder for _, r := range detail { if r < 0x20 || r == 0x7f { @@ -170,7 +213,7 @@ func (policy CheckpointExecutionPolicy) requiresReusable(stepID, laneID string) func forceCheckpointDecision(policy CheckpointExecutionPolicy, stepID, laneID string, decision CheckpointDecision) CheckpointDecision { if policy.forced(stepID, laneID) { - return checkpointDecision(CheckpointDecisionForcedRecompute, CheckpointReasonRecomputeStep, "selected step requires execution") + return checkpointDecision(CheckpointDecisionForcedRecompute, CheckpointReasonRecomputeStep) } return decision } @@ -184,23 +227,38 @@ func requireReusableCheckpoint(policy CheckpointExecutionPolicy, stepID, laneID // resolveCheckpointDecision applies runner policy and canonical payload // validation at the single point where a stage's observable decision is made. -func resolveCheckpointDecision(output *RunOutput, loader CheckpointLoader, policy CheckpointExecutionPolicy, stage ModuleStage, stepID, laneID, moduleKey string, decision CheckpointDecision, codec artifactCodecEntry, artifacts []CheckpointArtifact) (CheckpointDecision, error) { +type checkpointResolution struct { + decision CheckpointDecision + values []any + artifacts []CheckpointArtifact +} + +func resolveCheckpointDecision(output *RunOutput, loader CheckpointLoader, policy CheckpointExecutionPolicy, stage ModuleStage, stepID, laneID, moduleKey string, decision CheckpointDecision, codec artifactCodecEntry, artifacts []CheckpointArtifact) (checkpointResolution, error) { decision = forceCheckpointDecision(policy, stepID, laneID, decision) + resolution := checkpointResolution{decision: decision} if decision.Reused { + resolution.values = make([]any, 0, len(artifacts)) + resolution.artifacts = make([]CheckpointArtifact, 0, len(artifacts)) for _, artifact := range artifacts { - if _, _, err := decodeCanonicalCheckpointArtifact(codec, artifact); err != nil { - decision = checkpointDecision(CheckpointDecisionExecuted, checkpointArtifactReasonCode(err), "stored "+string(stage)+" artifact failed canonical codec validation") + value, hydrated, err := decodeCanonicalCheckpointArtifact(codec, artifact) + if err != nil { + decision = checkpointDecision(CheckpointDecisionExecuted, checkpointArtifactReasonCode(err)) + resolution.values = nil + resolution.artifacts = nil break } + resolution.values = append(resolution.values, value) + resolution.artifacts = append(resolution.artifacts, hydrated) } } + resolution.decision = decision if output != nil { recordCheckpointEvent(output, loader, string(stage), stepID, laneID, moduleKey, decision) } if err := requireReusableCheckpoint(policy, stepID, laneID, decision); err != nil { - return decision, err + return resolution, err } - return decision, nil + return resolution, nil } type SourceCheckpoint struct { @@ -296,19 +354,19 @@ func (noopCheckpointRecorder) NormalizeFailed(string, string, []CheckpointFinger func (noopCheckpointLoader) Enabled() bool { return false } func (noopCheckpointLoader) Source(string) (SourceCheckpoint, CheckpointDecision) { - return SourceCheckpoint{}, checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonLoadingDisabled, "checkpoint loading disabled") + return SourceCheckpoint{}, checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonLoadingDisabled) } func (noopCheckpointLoader) Extract(string, string, []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { - return ExtractCheckpoint{}, checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonLoadingDisabled, "checkpoint loading disabled") + return ExtractCheckpoint{}, checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonLoadingDisabled) } func (noopCheckpointLoader) Merge(string, string, []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) { - return MergeCheckpoint{}, checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonLoadingDisabled, "checkpoint loading disabled") + return MergeCheckpoint{}, checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonLoadingDisabled) } func (noopCheckpointLoader) Normalize(string, string, []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) { - return NormalizeCheckpoint{}, checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonLoadingDisabled, "checkpoint loading disabled") + return NormalizeCheckpoint{}, checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonLoadingDisabled) } func (noopCheckpointLoader) AcceptedNormalize(string, string, string) (NormalizeCheckpoint, CheckpointDecision) { - return NormalizeCheckpoint{}, checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonLoadingDisabled, "checkpoint loading disabled") + return NormalizeCheckpoint{}, checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonLoadingDisabled) } func checkpointExtractRunning(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint) error { diff --git a/internal/framework/pipeline/runner_accepted_checkpoint_test.go b/internal/framework/pipeline/runner_accepted_checkpoint_test.go index 0613730..4fbe74c 100644 --- a/internal/framework/pipeline/runner_accepted_checkpoint_test.go +++ b/internal/framework/pipeline/runner_accepted_checkpoint_test.go @@ -33,13 +33,13 @@ func (l *acceptedCheckpointLoader) AcceptedNormalize(stepID, laneID, _ string) ( } func (l *acceptedCheckpointLoader) Extract(laneID string, _ string, dependencies []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { l.extractDeps[laneID] = append([]CheckpointFingerprint(nil), dependencies...) - return ExtractCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing, "checkpoint missing") + return ExtractCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing) } func (l *acceptedCheckpointLoader) Merge(_ string, _ string, _ []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) { - return MergeCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing, "checkpoint missing") + return MergeCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing) } func (l *acceptedCheckpointLoader) Normalize(_ string, _ string, _ []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) { - return NormalizeCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing, "checkpoint missing") + return NormalizeCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing) } func TestRunnerHydratesRequiredNormalizedArtifact(t *testing.T) { @@ -85,7 +85,7 @@ func TestRunnerHydratesRequiredNormalizedArtifact(t *testing.T) { producerKey := CheckpointLaneKey(producer.resolved.StepID, producer.resolved.ID) consumerKey := CheckpointLaneKey(consumer.resolved.StepID, consumer.resolved.ID) loader.accepted[producerKey] = NormalizeCheckpoint{Output: stored, Warnings: []contracts.Warning{{Scope: "normalize", ReasonCode: "stored-warning", Message: "stored normalize warning"}}} - loader.acceptedDecision[producerKey] = NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused, "accepted artifact reusable") + loader.acceptedDecision[producerKey] = NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused) policy := CheckpointExecutionPolicy{ RequireReusableLanes: map[string]struct{}{producerKey: {}}, ForcedLanes: map[string]struct{}{consumerKey: {}}, @@ -154,12 +154,12 @@ func TestRunnerRejectsInvalidRequiredNormalizedArtifactBeforeConsumer(t *testing mutate func(*CheckpointArtifact) wantCode CheckpointReasonCode }{ - {"missing", NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing, "checkpoint missing"), nil, CheckpointReasonMissing}, - {"rejected status", NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonStatusNotReusable, "status rejected"), nil, CheckpointReasonStatusNotReusable}, - {"corrupt payload", NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused, "accepted artifact reusable"), func(v *CheckpointArtifact) { v.Artifact.Content = []byte(`{"items":[`) }, CheckpointReasonArtifactPayloadInvalid}, - {"non canonical", NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused, "accepted artifact reusable"), func(v *CheckpointArtifact) { v.Artifact.Content = []byte(`{"items": ["stored"]}`) }, CheckpointReasonArtifactNotCanonical}, - {"wrong codec identity", NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused, "accepted artifact reusable"), func(v *CheckpointArtifact) { v.Artifact.Kind = "test/score" }, CheckpointReasonArtifactCodecIncompatible}, - {"wrong content digest", NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonArtifactDigestMismatch, "content digest mismatch"), nil, CheckpointReasonArtifactDigestMismatch}, + {"missing", NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing), nil, CheckpointReasonMissing}, + {"rejected status", NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonStatusNotReusable), nil, CheckpointReasonStatusNotReusable}, + {"corrupt payload", NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused), func(v *CheckpointArtifact) { v.Artifact.Content = []byte(`{"items":[`) }, CheckpointReasonArtifactPayloadInvalid}, + {"non canonical", NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused), func(v *CheckpointArtifact) { v.Artifact.Content = []byte(`{"items": ["stored"]}`) }, CheckpointReasonArtifactNotCanonical}, + {"wrong codec identity", NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused), func(v *CheckpointArtifact) { v.Artifact.Kind = "test/score" }, CheckpointReasonArtifactCodecIncompatible}, + {"wrong content digest", NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonArtifactDigestMismatch), nil, CheckpointReasonArtifactDigestMismatch}, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { @@ -205,6 +205,100 @@ func TestRunnerRejectsInvalidRequiredNormalizedArtifactBeforeConsumer(t *testing } } +func TestRunnerRetainsEarlierHydratedProducerWhenLaterRequiredProducerFails(t *testing.T) { + prepared := preparedPipelineWithSharedProducerStep(t) + first := &prepared.Steps[0].lanes[0] + second := &prepared.Steps[0].lanes[1] + consumer := &prepared.Steps[1].lanes[0] + doc := prepared.input.(*typedTestInput).doc + + stored, err := checkpointArtifact(first.typed.codec, first.resolved.ID, first.resolved.Normalize.Module, doc.ID, codecNotes{Items: []string{"retained producer"}}) + if err != nil { + t.Fatal(err) + } + consumerCalls := 0 + consumer.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + consumerCalls++ + return erasedTypedResult{Value: codecScore{Value: 1}}, nil + } + + loader := newAcceptedCheckpointLoader() + firstKey := CheckpointLaneKey(first.resolved.StepID, first.resolved.ID) + secondKey := CheckpointLaneKey(second.resolved.StepID, second.resolved.ID) + loader.accepted[firstKey] = NormalizeCheckpoint{Output: stored, Warnings: []contracts.Warning{{Scope: "normalize", ReasonCode: "retained-warning", Message: "retained warning"}}} + loader.acceptedDecision[firstKey] = NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused) + loader.acceptedDecision[secondKey] = NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing) + policy := CheckpointExecutionPolicy{RequireReusableLanes: map[string]struct{}{firstKey: {}, secondKey: {}}} + + output, runErr := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoint: loader, CheckpointPolicy: policy}) + if runErr == nil || !strings.Contains(runErr.Error(), string(CheckpointReasonMissing)) { + t.Fatalf("Run() error = %v, want missing required producer", runErr) + } + if consumerCalls != 0 { + t.Fatalf("consumer calls = %d, want zero", consumerCalls) + } + if len(output.NormalizeOutputs) != 1 || output.NormalizeOutputs[0].LaneID != first.resolved.ID || string(output.NormalizeOutputs[0].Artifact.Content) != string(stored.Artifact.Content) { + t.Fatalf("retained normalize outputs = %#v, want first producer", output.NormalizeOutputs) + } + if len(output.Warnings) != 1 || output.Warnings[0].ReasonCode != "retained-warning" { + t.Fatalf("retained warnings = %#v", output.Warnings) + } + type decisionExpectation struct { + step, lane string + category CheckpointDecisionCategory + reason CheckpointReasonCode + } + want := []decisionExpectation{ + {first.resolved.StepID, first.resolved.ID, CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused}, + {second.resolved.StepID, second.resolved.ID, CheckpointDecisionExecuted, CheckpointReasonMissing}, + } + var got []decisionExpectation + for _, event := range output.CheckpointEvents { + if event.Stage == string(StageNormalize) { + got = append(got, decisionExpectation{event.StepID, event.LaneID, event.Category, event.ReasonCode}) + } + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("normalize decisions = %#v, want %#v", got, want) + } + var manifestGot []decisionExpectation + for _, decision := range output.Manifest.CheckpointDecisions { + if decision.Stage == string(StageNormalize) { + manifestGot = append(manifestGot, decisionExpectation{ + decision.StepID, decision.LaneID, + CheckpointDecisionCategory(decision.Category), CheckpointReasonCode(decision.ReasonCode), + }) + } + } + if !reflect.DeepEqual(manifestGot, want) { + t.Fatalf("manifest normalize decisions = %#v, want %#v", manifestGot, want) + } +} + +func preparedPipelineWithSharedProducerStep(t *testing.T) *PreparedPipeline { + t.Helper() + catalog := typedResolutionCatalog(t, completeTypedCatalogOptions()) + base := typedResolutionProfile() + profile := base + profile.Artifacts = nil + profile.Steps = []PipelineStepProfile{ + {ID: "producers", Artifacts: map[string]ArtifactLaneProfile{"first": base.Artifacts["notes"], "second": base.Artifacts["notes"]}}, + {ID: "consumer", Artifacts: map[string]ArtifactLaneProfile{"score": base.Artifacts["score"]}}, + } + resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v", err) + } + prepared, err := Prepare(resolved, registriesFromModuleCatalog(catalog), ModuleDependencies{}) + if err != nil { + t.Fatalf("Prepare() error = %v", err) + } + doc := typedTestDocumentWithUnits(1) + prepared.input.(*typedTestInput).doc = doc + prepared.chunker = &typedTestChunker{key: "typed/chunk", plan: typedTestPlan(doc)} + return prepared +} + func TestForcedRequiredLaneExecutesInsteadOfHydrating(t *testing.T) { prepared := preparedOrderedPipeline(t, 1, orderedLaneSpec{id: "unrelated", profile: "score"}, @@ -247,9 +341,9 @@ func TestForcedRequiredLaneExecutesInsteadOfHydrating(t *testing.T) { producerKey := CheckpointLaneKey(producer.resolved.StepID, producer.resolved.ID) consumerKey := CheckpointLaneKey(consumer.resolved.StepID, consumer.resolved.ID) loader.accepted[unrelatedKey] = NormalizeCheckpoint{Output: unrelatedArtifact} - loader.acceptedDecision[unrelatedKey] = NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused, "accepted artifact reusable") + loader.acceptedDecision[unrelatedKey] = NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused) loader.accepted[producerKey] = NormalizeCheckpoint{Output: producerArtifact} - loader.acceptedDecision[producerKey] = NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused, "accepted artifact reusable") + loader.acceptedDecision[producerKey] = NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused) policy := CheckpointExecutionPolicy{ ForcedLanes: map[string]struct{}{producerKey: {}, consumerKey: {}}, RequireReusableLanes: map[string]struct{}{unrelatedKey: {}, producerKey: {}}, diff --git a/internal/framework/pipeline/runner_concurrent.go b/internal/framework/pipeline/runner_concurrent.go index 30c6cc7..d7a8875 100644 --- a/internal/framework/pipeline/runner_concurrent.go +++ b/internal/framework/pipeline/runner_concurrent.go @@ -81,9 +81,15 @@ func (r *Runner) runLanes(parent context.Context, input RunInput, step PreparedP output := RunOutput{Manifest: manifestFromPipeline(input)} states, err := initializeLaneStates(input, step, checkpoints, loader, doc, chunks, &output) if err != nil { + if mergeErr := mergeTerminalLaneStates(&output, states); mergeErr != nil { + return output, errors.Join(err, mergeErr) + } return output, err } - completedOutputs, runErrors := r.runLaneEngine(parent, input, checkpoints, loader, doc, sourceInput, sessionID, chunks, states) + completedOutputs, runErrors := r.runLaneEngine(parent, laneEngineConfig{ + input: input, checkpoints: checkpoints, loader: loader, doc: doc, + sourceInput: sourceInput, sessionID: sessionID, chunks: chunks, states: states, + }) if err := mergeCompletedLanes(&output, completedOutputs); err != nil { return output, err } @@ -97,183 +103,232 @@ func initializeLaneStates(input RunInput, step PreparedPipelineStep, checkpoints states := make([]*laneExtractState, len(step.lanes)) for i, prepared := range step.lanes { if prepared.typed == nil { - return nil, fmt.Errorf("typed lane %q executor is not prepared", prepared.resolved.ID) + return states, fmt.Errorf("typed lane %q executor is not prepared", prepared.resolved.ID) } if err := setTypedLaneManifestMetadata(output, prepared.resolved.ID, prepared.typed.extractor, prepared.typed.merger, prepared.typed.normalizer); err != nil { - return nil, err + return states, err } if input.CheckpointPolicy.requiresReusable(input.stepID, prepared.resolved.ID) && !input.CheckpointPolicy.forced(input.stepID, prepared.resolved.ID) { - state, err := hydrateRequiredLane(input, loader, doc, i, prepared, output) - if err != nil { - return nil, err - } + state, err := hydrateRequiredLane(input, loader, doc, i, prepared) states[i] = state + if err != nil { + return states, err + } continue } state, err := prepareLaneExtract(input, loader, doc, chunks, i, prepared, output) if err != nil { - return nil, err + return states, err } if !state.decision.Reused { if err := checkpointExtractRunning(checkpoints, input.stepID, prepared.resolved.ID, prepared.resolved.Extract.Module, state.deps); err != nil { - return nil, fmt.Errorf("write extract checkpoint for lane %q: %w", prepared.resolved.ID, err) + return states, fmt.Errorf("write extract checkpoint for lane %q: %w", prepared.resolved.ID, err) } } else if err := finalizeLaneExtract(checkpoints, input.stepID, state); err != nil { - return nil, err + return states, err } states[i] = state } return states, nil } -func (r *Runner) runLaneEngine(parent context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, states []*laneExtractState) ([]RunOutput, []orderedRunError) { - workerCount := input.ExtractWorkers +type laneEngineConfig struct { + input RunInput + checkpoints CheckpointRecorder + loader CheckpointLoader + doc *source.SourceDocument + sourceInput contracts.LLMInputMaterial + sessionID string + chunks []source.Chunk + states []*laneExtractState +} + +type laneEngine struct { + runner *Runner + laneEngineConfig + ctx context.Context + cancel context.CancelFunc + workerCount int + jobs chan extractJob + results chan extractJobResult + completions chan laneCompletion + continuations chan *laneExtractState + continuationWorkers sync.WaitGroup + pending []*laneExtractState + completedOutputs []RunOutput + runErrors []orderedRunError + launched int + completed int +} + +func (r *Runner) runLaneEngine(parent context.Context, config laneEngineConfig) ([]RunOutput, []orderedRunError) { + engine := newLaneEngine(r, parent, config) + return engine.run() +} + +func newLaneEngine(r *Runner, parent context.Context, config laneEngineConfig) *laneEngine { + workerCount := config.input.ExtractWorkers if workerCount < 1 { workerCount = 1 } ctx, cancel := context.WithCancel(parent) - defer cancel() - jobs := make(chan extractJob, workerCount) - results := make(chan extractJobResult, workerCount) - completions := make(chan laneCompletion, len(states)) - continuations := make(chan *laneExtractState, workerCount) + return &laneEngine{ + runner: r, laneEngineConfig: config, ctx: ctx, cancel: cancel, workerCount: workerCount, + jobs: make(chan extractJob, workerCount), results: make(chan extractJobResult, workerCount), + completions: make(chan laneCompletion, len(config.states)), continuations: make(chan *laneExtractState, workerCount), + completedOutputs: make([]RunOutput, len(config.states)), + } +} +func (e *laneEngine) run() ([]RunOutput, []orderedRunError) { + defer e.cancel() + e.initializeCollection() + e.startExtractWorkers() + e.startContinuationWorkers() + e.collect() + close(e.continuations) + e.continuationWorkers.Wait() + return e.completedOutputs, e.runErrors +} + +func (e *laneEngine) initializeCollection() { + for _, state := range e.states { + if state.terminal { + e.completedOutputs[state.index] = state.output + } else if state.decision.Reused { + e.pending = append(e.pending, state) + } + } +} + +func (e *laneEngine) startExtractWorkers() { var workers sync.WaitGroup - for i := 0; i < workerCount; i++ { + for i := 0; i < e.workerCount; i++ { workers.Add(1) go func() { defer workers.Done() - for job := range jobs { - if ctx.Err() != nil { + for job := range e.jobs { + if e.ctx.Err() != nil { continue } - result := r.runExtractJob(ctx, input, doc, sourceInput, sessionID, job) - results <- result + e.results <- e.runner.runExtractJob(e.ctx, e.input, e.doc, e.sourceInput, e.sessionID, job) } }() } - go func() { - defer close(jobs) - for chunkIndex := range chunks { - for laneIndex := range states { - state := states[laneIndex] - if state.terminal || state.decision.Reused { - continue - } - select { - case jobs <- extractJob{lane: state, chunk: chunks[chunkIndex]}: - case <-ctx.Done(): - return - } - } - } - }() - go func() { workers.Wait(); close(results) }() - var continuationWorkers sync.WaitGroup - for i := 0; i < workerCount; i++ { - continuationWorkers.Add(1) - go func() { - defer continuationWorkers.Done() - for state := range continuations { - if err := ctx.Err(); err != nil { - completions <- laneCompletion{index: state.index, err: err} - continue - } - laneOutput, err := r.continueLane(ctx, input, checkpoints, loader, doc, sourceInput, sessionID, chunks, state) - completions <- laneCompletion{index: state.index, output: laneOutput, err: err} - } - }() - } - - completedOutputs, runErrors := collectLaneResults(ctx, cancel, input, checkpoints, chunks, states, results, completions, continuations) - close(continuations) - continuationWorkers.Wait() - return completedOutputs, runErrors + go e.dispatchExtractJobs() + go func() { workers.Wait(); close(e.results) }() } -func collectLaneResults(ctx context.Context, cancel context.CancelFunc, input RunInput, checkpoints CheckpointRecorder, chunks []source.Chunk, states []*laneExtractState, results <-chan extractJobResult, completions <-chan laneCompletion, continuations chan<- *laneExtractState) ([]RunOutput, []orderedRunError) { - completedOutputs := make([]RunOutput, len(states)) - var runErrors []orderedRunError - var pendingContinuations []*laneExtractState - launched, completed := 0, 0 - for _, state := range states { - if state.terminal { - completedOutputs[state.index] = state.output - } else if state.decision.Reused { - pendingContinuations = append(pendingContinuations, state) +func (e *laneEngine) dispatchExtractJobs() { + defer close(e.jobs) + for chunkIndex := range e.chunks { + for laneIndex := range e.states { + state := e.states[laneIndex] + if state.terminal || state.decision.Reused { + continue + } + select { + case e.jobs <- extractJob{lane: state, chunk: e.chunks[chunkIndex]}: + case <-e.ctx.Done(): + return + } } } - resultChannel := results - for resultChannel != nil || len(pendingContinuations) > 0 || completed < launched { +} + +func (e *laneEngine) startContinuationWorkers() { + for i := 0; i < e.workerCount; i++ { + e.continuationWorkers.Add(1) + go func() { + defer e.continuationWorkers.Done() + for state := range e.continuations { + if err := e.ctx.Err(); err != nil { + e.completions <- laneCompletion{index: state.index, err: err} + continue + } + laneOutput, err := e.runner.continueLane(e.ctx, e.input, e.checkpoints, e.loader, e.doc, e.sourceInput, e.sessionID, e.chunks, state) + e.completions <- laneCompletion{index: state.index, output: laneOutput, err: err} + } + }() + } +} + +func (e *laneEngine) collect() { + resultChannel := (<-chan extractJobResult)(e.results) + for resultChannel != nil || len(e.pending) > 0 || e.completed < e.launched { var continuationChannel chan<- *laneExtractState var nextContinuation *laneExtractState - if len(pendingContinuations) > 0 && ctx.Err() == nil { - continuationChannel = continuations - nextContinuation = pendingContinuations[0] - } else if ctx.Err() != nil { - pendingContinuations = nil + if len(e.pending) > 0 && e.ctx.Err() == nil { + continuationChannel = e.continuations + nextContinuation = e.pending[0] + } else if e.ctx.Err() != nil { + e.pending = nil } select { case continuationChannel <- nextContinuation: - pendingContinuations = pendingContinuations[1:] - launched++ + e.pending = e.pending[1:] + e.launched++ case result, ok := <-resultChannel: if !ok { resultChannel = nil continue } - state := states[result.laneIndex] - state.remaining-- - if result.err != nil { - state.failed = true - runErrors = append(runErrors, orderedRunError{stage: 0, lane: state.index, chunk: result.chunkIndex, err: result.err}) - _ = checkpointExtractFailed(checkpoints, input.stepID, state.prepared.resolved.ID, state.prepared.resolved.Extract.Module, state.deps, result.err) - cancel() - } else { - state.results[result.chunkIndex] = result - } - if state.remaining == 0 && !state.failed && ctx.Err() == nil { - if err := finalizeLaneExtract(checkpoints, input.stepID, state); err != nil { - state.failed = true - runErrors = append(runErrors, orderedRunError{stage: 0, lane: state.index, chunk: len(chunks), err: err}) - cancel() - } else { - pendingContinuations = append(pendingContinuations, state) - } - } - case completion := <-completions: - completed++ - completedOutputs[completion.index] = completion.output - if completion.err != nil { - runErrors = append(runErrors, classifyLaneError(completion.index, len(chunks), completion.err)) - cancel() - } + e.handleExtractResult(result) + case completion := <-e.completions: + e.handleCompletion(completion) } } - return completedOutputs, runErrors } -func hydrateRequiredLane(input RunInput, loader CheckpointLoader, doc *source.SourceDocument, index int, prepared preparedLaneExecutor, output *RunOutput) (*laneExtractState, error) { +func (e *laneEngine) handleExtractResult(result extractJobResult) { + state := e.states[result.laneIndex] + state.remaining-- + if result.err != nil { + state.failed = true + e.runErrors = append(e.runErrors, orderedRunError{stage: 0, lane: state.index, chunk: result.chunkIndex, err: result.err}) + _ = checkpointExtractFailed(e.checkpoints, e.input.stepID, state.prepared.resolved.ID, state.prepared.resolved.Extract.Module, state.deps, result.err) + e.cancel() + } else { + state.results[result.chunkIndex] = result + } + if state.remaining != 0 || state.failed || e.ctx.Err() != nil { + return + } + if err := finalizeLaneExtract(e.checkpoints, e.input.stepID, state); err != nil { + state.failed = true + e.runErrors = append(e.runErrors, orderedRunError{stage: 0, lane: state.index, chunk: len(e.chunks), err: err}) + e.cancel() + return + } + e.pending = append(e.pending, state) +} + +func (e *laneEngine) handleCompletion(completion laneCompletion) { + e.completed++ + e.completedOutputs[completion.index] = completion.output + if completion.err != nil { + e.runErrors = append(e.runErrors, classifyLaneError(completion.index, len(e.chunks), completion.err)) + e.cancel() + } +} + +func hydrateRequiredLane(input RunInput, loader CheckpointLoader, doc *source.SourceDocument, index int, prepared preparedLaneExecutor) (*laneExtractState, error) { lane, typed := prepared.resolved, prepared.typed local := RunOutput{Manifest: manifestFromPipeline(input)} checkpoint, decision := loader.AcceptedNormalize(input.stepID, lane.ID, lane.Normalize.Module) if decision.Reused { - decision = checkpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused, "accepted normalized artifact is reusable") + decision = checkpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused) if checkpoint.Output.LaneID != lane.ID || checkpoint.Output.ModuleKey != lane.Normalize.Module || checkpoint.Output.SourceID != doc.ID { - decision = checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonArtifactPayloadInvalid, "accepted normalized artifact provenance does not match the producer lane") + decision = checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonArtifactPayloadInvalid) } } - decision, err := resolveCheckpointDecision(&local, loader, input.CheckpointPolicy, StageNormalize, input.stepID, lane.ID, lane.Normalize.Module, decision, typed.codec, []CheckpointArtifact{checkpoint.Output}) + resolution, err := resolveCheckpointDecision(&local, loader, input.CheckpointPolicy, StageNormalize, input.stepID, lane.ID, lane.Normalize.Module, decision, typed.codec, []CheckpointArtifact{checkpoint.Output}) + decision = resolution.decision + state := &laneExtractState{index: index, prepared: prepared, decision: decision, terminal: true, output: local} if err != nil { - if mergeErr := mergeLaneOutput(output, local); mergeErr != nil { - return nil, mergeErr - } - return nil, err - } - _, hydrated, err := decodeCanonicalCheckpointArtifact(typed.codec, checkpoint.Output) - if err != nil { - return nil, fmt.Errorf("hydrate accepted normalized artifact for step %q lane %q: %w", input.stepID, lane.ID, err) + return state, err } + hydrated := resolution.artifacts[0] local.Warnings = append(local.Warnings, cloneWarnings(checkpoint.Warnings)...) local.NormalizeOutputs = append(local.NormalizeOutputs, contracts.SerializedOutput{ StepID: input.stepID, @@ -282,7 +337,20 @@ func hydrateRequiredLane(input RunInput, loader CheckpointLoader, doc *source.So SourceID: doc.ID, Artifact: contracts.CloneSerializedArtifact(hydrated.Artifact), }) - return &laneExtractState{index: index, prepared: prepared, decision: decision, terminal: true, output: local}, nil + state.output = local + return state, nil +} + +func mergeTerminalLaneStates(output *RunOutput, states []*laneExtractState) error { + for _, state := range states { + if state == nil || !state.terminal { + continue + } + if err := mergeLaneOutput(output, state.output); err != nil { + return err + } + } + return nil } func mergeCompletedLanes(output *RunOutput, completedOutputs []RunOutput) error { @@ -304,19 +372,16 @@ func prepareLaneExtract(input RunInput, loader CheckpointLoader, doc *source.Sou deps := append(digestFingerprints("chunks", digest), generatedReferenceDependencies(extractReferences)...) state := &laneExtractState{index: index, prepared: prepared, deps: normalizeCheckpointFingerprints(deps), remaining: len(chunks), results: make(map[int]extractJobResult, len(chunks))} cp, decision := loadExtract(loader, input.stepID, lane.ID, lane.Extract.Module, state.deps) - decision, err = resolveCheckpointDecision(output, loader, input.CheckpointPolicy, StageExtract, input.stepID, lane.ID, lane.Extract.Module, decision, typed.codec, cp.Outputs) + resolution, err := resolveCheckpointDecision(output, loader, input.CheckpointPolicy, StageExtract, input.stepID, lane.ID, lane.Extract.Module, decision, typed.codec, cp.Outputs) if err != nil { return nil, err } + decision = resolution.decision state.decision = decision if decision.Reused { state.remaining = 0 - for _, stored := range cp.Outputs { - value, hydrated, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, stored) - if decodeErr != nil { - return nil, fmt.Errorf("decode extract checkpoint for lane %q: %w", lane.ID, decodeErr) - } - stored = hydrated + for i, stored := range resolution.artifacts { + value := resolution.values[i] artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: stored.ChunkID, ChunkIndex: stored.ChunkIndex, ChunkRef: stored.ChunkRef, Value: value} if stored.ChunkIndex >= 0 && stored.ChunkIndex < len(chunks) && artifact.ChunkRef == (source.SourceRef{}) { artifact.ChunkRef = chunks[stored.ChunkIndex].Ref diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index ed41725..7187a33 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -227,10 +227,11 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints mergeDeps := append(artifactCheckpointDigests(extracts.serialized), generatedReferenceDependencies(mergeReferences)...) mergeDeps = normalizeCheckpointFingerprints(mergeDeps) mergeCP, mergeDecision := loadMerge(loader, input.stepID, lane.ID, lane.Merge.Module, mergeDeps) - mergeDecision, err := resolveCheckpointDecision(output, loader, input.CheckpointPolicy, StageMerge, input.stepID, lane.ID, lane.Merge.Module, mergeDecision, typed.codec, []CheckpointArtifact{mergeCP.Output}) + mergeResolution, err := resolveCheckpointDecision(output, loader, input.CheckpointPolicy, StageMerge, input.stepID, lane.ID, lane.Merge.Module, mergeDecision, typed.codec, []CheckpointArtifact{mergeCP.Output}) if err != nil { return stageResult, err } + mergeDecision = mergeResolution.decision if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "decision": mergeDecision, "source": debugSourceDocumentEnvelope(doc), "extract_outputs": debugCheckpointArtifacts(extracts.serialized), "options": redactSensitiveMap(lane.Merge.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil { return stageResult, err } @@ -238,12 +239,8 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints var serializedMerge CheckpointArtifact var mergeWarnings []contracts.Warning if mergeDecision.Reused { - value, hydrated, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, mergeCP.Output) - if decodeErr != nil { - return stageResult, fmt.Errorf("decode merge checkpoint for lane %q: %w", lane.ID, decodeErr) - } - merged = erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: value} - serializedMerge = hydrated + merged = erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: mergeResolution.values[0]} + serializedMerge = mergeResolution.artifacts[0] mergeWarnings = cloneWarnings(mergeCP.Warnings) output.Warnings = append(output.Warnings, mergeWarnings...) } else { @@ -327,21 +324,18 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi normalizeDeps := append(artifactCheckpointDigests([]CheckpointArtifact{serializedMerge}), generatedReferenceDependencies(normalizeReferences)...) normalizeDeps = normalizeCheckpointFingerprints(normalizeDeps) normalizeCP, normalizeDecision := loadNormalize(loader, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps) - normalizeDecision, err := resolveCheckpointDecision(output, loader, input.CheckpointPolicy, StageNormalize, input.stepID, lane.ID, lane.Normalize.Module, normalizeDecision, typed.codec, []CheckpointArtifact{normalizeCP.Output}) + normalizeResolution, err := resolveCheckpointDecision(output, loader, input.CheckpointPolicy, StageNormalize, input.stepID, lane.ID, lane.Normalize.Module, normalizeDecision, typed.codec, []CheckpointArtifact{normalizeCP.Output}) if err != nil { return stageResult, err } + normalizeDecision = normalizeResolution.decision if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "decision": normalizeDecision, "source": debugSourceDocumentEnvelope(doc), "merge_output": debugCheckpointArtifact(serializedMerge), "options": redactSensitiveMap(lane.Normalize.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil { return stageResult, err } var serializedNormalize CheckpointArtifact var normalizeWarnings []contracts.Warning if normalizeDecision.Reused { - _, hydrated, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, normalizeCP.Output) - if decodeErr != nil { - return stageResult, fmt.Errorf("decode normalize checkpoint for lane %q: %w", lane.ID, decodeErr) - } - serializedNormalize = hydrated + serializedNormalize = normalizeResolution.artifacts[0] normalizeWarnings = cloneWarnings(normalizeCP.Warnings) output.Warnings = append(output.Warnings, normalizeWarnings...) } else { diff --git a/internal/framework/pipeline/runner_typed_checkpoint_test.go b/internal/framework/pipeline/runner_typed_checkpoint_test.go index 5bd6cd4..23e0bd5 100644 --- a/internal/framework/pipeline/runner_typed_checkpoint_test.go +++ b/internal/framework/pipeline/runner_typed_checkpoint_test.go @@ -28,7 +28,6 @@ func (l requiredCheckpointLoader) AcceptedNormalize(string, string, string) (Nor } func TestRequiredCheckpointFailureRetainsDecision(t *testing.T) { - const unsafeDetail = "unsafe-loader-detail-/private/checkpoint/path" tests := []struct { name string decision CheckpointDecision @@ -36,10 +35,10 @@ func TestRequiredCheckpointFailureRetainsDecision(t *testing.T) { wantCode CheckpointReasonCode wantAction CheckpointDecisionCategory }{ - {"missing", NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing, unsafeDetail), false, CheckpointReasonMissing, CheckpointDecisionExecuted}, - {"corrupt", NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonReused, "checkpoint reusable"), true, CheckpointReasonArtifactNotCanonical, CheckpointDecisionExecuted}, - {"incompatible", NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonArtifactCodecIncompatible, unsafeDetail), false, CheckpointReasonArtifactCodecIncompatible, CheckpointDecisionExecuted}, - {"dependency invalidated", NewCheckpointDecision(CheckpointDecisionDependencyInvalidated, CheckpointReasonDependencyMismatch, unsafeDetail), false, CheckpointReasonDependencyMismatch, CheckpointDecisionDependencyInvalidated}, + {"missing", NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing), false, CheckpointReasonMissing, CheckpointDecisionExecuted}, + {"corrupt", NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonReused), true, CheckpointReasonArtifactNotCanonical, CheckpointDecisionExecuted}, + {"incompatible", NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonArtifactCodecIncompatible), false, CheckpointReasonArtifactCodecIncompatible, CheckpointDecisionExecuted}, + {"dependency invalidated", NewCheckpointDecision(CheckpointDecisionDependencyInvalidated, CheckpointReasonDependencyMismatch), false, CheckpointReasonDependencyMismatch, CheckpointDecisionDependencyInvalidated}, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { @@ -62,9 +61,6 @@ func TestRequiredCheckpointFailureRetainsDecision(t *testing.T) { if err == nil || !strings.Contains(err.Error(), step.ID) || !strings.Contains(err.Error(), lane.resolved.ID) || !strings.Contains(err.Error(), string(test.wantCode)) { t.Fatalf("Run() error = %v, want step, lane, and reason code %q", err, test.wantCode) } - if strings.Contains(err.Error(), unsafeDetail) { - t.Fatalf("Run() error leaked loader detail: %v", err) - } var found bool for _, event := range output.CheckpointEvents { if event.Stage == string(StageNormalize) && event.StepID == step.ID && event.LaneID == lane.resolved.ID { @@ -72,8 +68,8 @@ func TestRequiredCheckpointFailureRetainsDecision(t *testing.T) { if event.Action != test.wantAction || event.ReasonCode != test.wantCode { t.Fatalf("checkpoint event = %#v, want action %q and reason %q", event, test.wantAction, test.wantCode) } - if strings.Contains(event.Detail, unsafeDetail) { - t.Fatalf("checkpoint event leaked loader detail: %#v", event) + if event.Detail == "" || event.Detail != event.Reason { + t.Fatalf("checkpoint event detail is not code-owned compatibility text: %#v", event) } } } @@ -97,6 +93,17 @@ func TestRequiredCheckpointFailureRetainsDecision(t *testing.T) { } } +func TestCheckpointDecisionDetailDoesNotCopyUnknownReasonCode(t *testing.T) { + const sentinel = "opaque-sensitive-value-78421" + decision := NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonCode(sentinel)) + if decision.Detail == "" || decision.Detail != decision.Reason { + t.Fatalf("decision detail is not bounded compatibility text: %#v", decision) + } + if strings.Contains(decision.Detail, sentinel) || strings.Contains(decision.Reason, sentinel) { + t.Fatalf("decision detail copied caller-controlled reason code: %#v", decision) + } +} + func TestDecodeCheckpointArtifactRejectsIncompatibleCodecIdentityAndBytes(t *testing.T) { registry := NewArtifactCodecRegistry() if err := RegisterArtifactCodec(registry, notesCodec()); err != nil { diff --git a/internal/modules/dnd/extract/combatturns/assets/prompts/instructions.md b/internal/modules/dnd/extract/combatturns/assets/prompts/instructions.md index 358e46c..3ec6129 100644 --- a/internal/modules/dnd/extract/combatturns/assets/prompts/instructions.md +++ b/internal/modules/dnd/extract/combatturns/assets/prompts/instructions.md @@ -1,5 +1,7 @@ Return the combat_turns array even when no combat turn is established. Return -one or more actions for every turn. Use one of the supported turn_kind and -action category values. Set round to null when the transcript does not state an +one or more actions for every turn. For turn_kind, use exactly one of: turn, +reaction, legendary_action, lair_action, or other. For each action category, +use exactly one of: attack, spell, movement, item, ability_check, saving_throw, +condition, or other. Set round to null when the transcript does not state an explicit or unambiguous positive round number. Set resolution to null when the transcript establishes the declaration but not an immediate resolution. diff --git a/internal/modules/dnd/register/chains.go b/internal/modules/dnd/register/chains.go index e6c053e..dd54e89 100644 --- a/internal/modules/dnd/register/chains.go +++ b/internal/modules/dnd/register/chains.go @@ -87,9 +87,9 @@ func registerDefaultChains(registry *pipeline.ValidatorChainRegistry) error { Module: combatextract.Key, Validators: []pipeline.ModuleBinding{ pipeline.Binding(validjson.Key), - pipeline.Binding(validjsonschema.Key), pipeline.Binding(combatshape.Key), pipeline.Binding(combatsourcerefs.Key), + pipeline.Binding(validjsonschema.Key), pipeline.Binding(combatrelatedness.Key), }, }) @@ -100,10 +100,10 @@ func registerDefaultChains(registry *pipeline.ValidatorChainRegistry) error { Module: combatnormalize.Key, Validators: []pipeline.ModuleBinding{ pipeline.Binding(validjson.Key), - pipeline.Binding(validjsonschema.Key), pipeline.Binding(combatshape.Key), pipeline.Binding(combatinvariants.Key), pipeline.Binding(combatsourcerefs.Key), + pipeline.Binding(validjsonschema.Key), pipeline.Binding(combatrelatedness.Key), }, }) diff --git a/internal/modules/dnd/register/register_test.go b/internal/modules/dnd/register/register_test.go index cfa124e..f1b5670 100644 --- a/internal/modules/dnd/register/register_test.go +++ b/internal/modules/dnd/register/register_test.go @@ -87,9 +87,9 @@ func TestRegisterAddsDNDFamily(t *testing.T) { } combatExtractChain := []pipeline.ModuleBinding{ pipeline.Binding("generic/valid_json"), - pipeline.Binding("generic/valid_json_schema"), pipeline.Binding("extract/dnd/combat-turns/shape"), pipeline.Binding("extract/dnd/combat-turns/source_refs"), + pipeline.Binding("generic/valid_json_schema"), pipeline.Binding("extract/dnd/combat-turns/source_relatedness"), } if got := registries.ValidatorChains.Validators(pipeline.StageExtract, combatextract.Key); !reflect.DeepEqual(got, combatExtractChain) { @@ -97,10 +97,10 @@ func TestRegisterAddsDNDFamily(t *testing.T) { } combatNormalizeChain := []pipeline.ModuleBinding{ pipeline.Binding("generic/valid_json"), - pipeline.Binding("generic/valid_json_schema"), pipeline.Binding("extract/dnd/combat-turns/shape"), pipeline.Binding("normalize/dnd/combat-turns/invariants"), pipeline.Binding("extract/dnd/combat-turns/source_refs"), + pipeline.Binding("generic/valid_json_schema"), pipeline.Binding("extract/dnd/combat-turns/source_relatedness"), } if got := registries.ValidatorChains.Validators(pipeline.StageNormalize, combatnormalize.Key); !reflect.DeepEqual(got, combatNormalizeChain) { diff --git a/internal/modules/integration/dnd_combat_runner_test.go b/internal/modules/integration/dnd_combat_runner_test.go index 36361e1..c9081c6 100644 --- a/internal/modules/integration/dnd_combat_runner_test.go +++ b/internal/modules/integration/dnd_combat_runner_test.go @@ -22,6 +22,7 @@ import ( combatnormalize "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/normalize/combatturns" "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/npcs/identity" npcregistry "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/npcs/registry" + combatshape "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/validate/combatturns/shape" "gitea.maximumdirect.net/eric/notarius/internal/modules/seriatim/input/transcript" ) @@ -47,7 +48,7 @@ func TestProductionCombatPipelineRetriesMergesNormalizesAndWritesJSON(t *testing } client := &fakeCombatLLMClient{responses: []string{ - combatTestInvalidResponse(), + combatTestInvalidEnumResponse("unsupported", "attack"), combatTestTurnResponse("The Greencloak", "turn", "watches", "The Greencloak", 1, 1), combatTestTurnResponse("Mira Thorn", "reaction", "asks", "Hooded Guard", 2, 2), combatTestTurnResponse("Hooded Guard", "turn", "attacks", "The Greencloak", 3, 3), @@ -125,6 +126,48 @@ func TestProductionCombatPipelineRetriesMergesNormalizesAndWritesJSON(t *testing } } +func TestProductionCombatPipelineAttributesExhaustedInvalidEnumsToShapeValidation(t *testing.T) { + registries := productionNPCRegistries(t) + configValue := combatOnlyConfig() + profile := configValue.Pipelines["dnd-combat-fixture"] + profile.Chunk.Options["max_units"] = 100 + configValue.Pipelines["dnd-combat-fixture"] = profile + effective, err := configValue.Resolve(config.ResolveInput{ + PipelineID: "dnd-combat-fixture", + Catalog: moduleCatalog(registries), + }) + if err != nil { + t.Fatalf("Resolve() error = %v, want nil", err) + } + client := &fakeCombatLLMClient{responses: []string{ + combatTestInvalidEnumResponse("unsupported", "attack"), + combatTestInvalidEnumResponse("turn", "unsupported"), + combatTestInvalidEnumResponse("unsupported", "unsupported"), + }} + prepared, err := pipeline.Prepare(effective.ResolvedPipeline, registries, pipeline.ModuleDependencies{LLM: client}) + if err != nil { + t.Fatalf("Prepare() error = %v, want nil", err) + } + output, err := pipeline.New().Run(context.Background(), pipeline.RunInput{ + Prepared: prepared, + RawInput: readNPCFixture(t), + ExtractWorkers: 1, + }) + if err != nil { + t.Fatalf("Run() error = %v, want non-fatal rejected output", err) + } + if len(client.requests) != 3 || len(output.Rejected) != 1 { + t.Fatalf("LLM requests = %d rejected = %#v, want exhausted retry and one rejection", len(client.requests), output.Rejected) + } + rejected := output.Rejected[0] + if rejected.ReasonCode != combatshape.ReasonCode || rejected.ValidatorName != combatshape.Key || rejected.AttemptCount != 3 { + t.Fatalf("rejected output = %#v, want exhausted combat shape rejection", rejected) + } + if output.Manifest.ValidationStatus != "rejected" || len(output.NormalizeOutputs) != 0 { + t.Fatalf("output = %#v, want non-fatal rejected combat result without normalized artifacts", output) + } +} + func TestCombatNormalizerRejectsCampaignReferenceBinding(t *testing.T) { registries := productionNPCRegistries(t) catalog := moduleCatalog(registries) @@ -269,8 +312,8 @@ func (client *fakeCombatLLMClient) CompleteStructured(ctx context.Context, req c return contracts.StructuredCompletionResponse{Content: content, Provider: "test", Model: "combat-fake"}, nil } -func combatTestInvalidResponse() string { - return `{"combat_turns":[{"actor":"","turn_kind":"turn","round":1,"actions":[{"category":"attack","declaration":"watches","targets":[],"resolution":null}],"summary":"invalid candidate","source_refs":[{"start_unit_id":1,"end_unit_id":1}]}]}` +func combatTestInvalidEnumResponse(turnKind, category string) string { + return fmt.Sprintf(`{"combat_turns":[{"actor":"Aria","turn_kind":%q,"round":1,"actions":[{"category":%q,"declaration":"watches","targets":["Mira"],"resolution":null}],"summary":"invalid candidate","source_refs":[{"start_unit_id":1,"end_unit_id":1}]}]}`, turnKind, category) } func combatTestTurnResponse(actor, turnKind, declaration, target string, round, unit int) string {