diff --git a/docs/roadmaps/2026-09-12-095059-stable-policy-orchestration-and-statistical-learning.md b/docs/roadmaps/2026-09-12-095059-stable-policy-orchestration-and-statistical-learning.md index aa46b2299..2c7bf2f38 100644 --- a/docs/roadmaps/2026-09-12-095059-stable-policy-orchestration-and-statistical-learning.md +++ b/docs/roadmaps/2026-09-12-095059-stable-policy-orchestration-and-statistical-learning.md @@ -292,6 +292,7 @@ All milestones use the routine route and one accountable owner. Technical/model/ Scope: minimal additive typed facts/references for pre-dispatch task snapshots, manifest/recipe identity, policy/window assignment and deviations; derive whole-issue analysis through the existing bounded private read path. Retain current receipt limits, SQLite writer/replay semantics, privacy and public allowlist. Do not copy execution intent into a new journal. Acceptance: repeatable input snapshots reproduce totals, coverage and assigned-treatment reports; unique costs reconcile across root/children/retry/review/repair; source fixtures cover reordering, duplicates, correction, reopened roots, incomplete CI, provider mismatch, unsupported usage and late arrival. Negative controls for double-counted cache/reasoning, missing child, redraw on retry and cross-workspace join stay visibly invalid/incomplete. Demonstrate bounded output and old-store/schema compatibility, plus source-owned checks appropriate to the change. Closure: source delivered and compatibility evidence retained. Publication, installed adoption and actual operating collection are separately reported pending; a green synthetic report is not a measured benefit. + C1 source window: [causal declaration receiver](learn-01-causal-receiver.md) records the immutable admission/1 byte contract, bounded snapshot/analyzer behavior and synthetic fixture controls. Local focused source qualification passed; source delivery, C3 exact composition, publication and installed receiver-first adoption remain pending. It does not close .2 or imply installed emission. Evidence: #3884 delivered the typed observations, bounded private snapshot analysis and retained native inventory source binding. Follow-on source adds schema 11 durable learning-fact order plus immutable `learn-shared-cost-allocation/1` and `learn-shared-cost-authority/1`: a sorted roster and fixed equal-largest-remainder rule must be persisted before every assignment, and later evidence must join the exact retained native source and observed provider total. Historical schema-10 facts retain unknown order. #3916 added protected host-config/2 credential roles and grants, receiver-owned receipt admissions, schema-12 first-fact provenance, a restricted native-collector fact scope, and private snapshot v4. Recovery retains the original authenticated role, replay cannot upgrade provenance, schema-11 history stays unknown, and generic full-corpus submissions remain unqualified. Follow-on hardening binds exact receipt retries to the original enrolled authority, rejects incompatible credentials for one scope, and refuses contradictory protected collector grants in analysis. A protected host command now accepts only dispatch/native selectors, resolves lineage and profile from durable admissions, loads executable/source/principal from private installation configuration, clears inherited environment, recomputes the roster/counters/binding, atomically retains the canonical envelope, and submits it through the restricted collector principal. The roadmap adapter selects an installed private Host configuration separately, sends only its durable dispatch/parent/native selectors, and preserves generic/local fallback as UNKNOWN for missing configuration, old Host and refusal; a protected success cannot copy authority into the generic producer. The resolver admits the adapter's exact `collaboration-spawn-agent` lineage and refuses other durable runtimes. Profile mismatch refuses before invocation and exact retry reuses retained bytes. A protected collector corpus passes the credential-provenance check but remains explicitly `native-source-verification-unavailable` and `snapshot-origin-unverified`; therefore every shared-cost comparison also remains `independent-shared-cost-authority-unavailable`. .2 stays open pending published installed configuration/credential, qualified capture custody, and trusted snapshot acquisition. Enrollment, live collection and any efficiency result remain pending. - [x] **LEARN-01.3 — Integrate fixed profiles and context assembly into the existing orchestrator — route: routine.** diff --git a/docs/roadmaps/learn-01-causal-receiver.md b/docs/roadmaps/learn-01-causal-receiver.md new file mode 100644 index 000000000..02dc20e06 --- /dev/null +++ b/docs/roadmaps/learn-01-causal-receiver.md @@ -0,0 +1,15 @@ +# LEARN-01.2 causal declaration receiver + +This is the receiver-owned C1 source window of LEARN-01.2. The [LEARN owning plan](2026-09-12-095059-stable-policy-orchestration-and-statistical-learning.md) remains the completion ledger. Coordination owns committed admission/1 and producer publication; this repository owns the versioned `execution-causal-admission/1` telemetry event, immutable retention and bounded analysis. + +The event binds its canonical identity, item and dispatch to the embedded admission/1 DTO. It carries canonical base64 of the exact decoded committed admission bytes, their SHA-256, and the verified route-binding and launch-intent SHA-256 references. The receiver parses the closed admission shape, bounded purpose/dependency/retry declaration and canonical invocation lineage; reserializing the DTO cannot replace the retained bytes. Declaration edges preserve their recorded order and source references. A source digest is evidence about committed bytes, not execution authority or access to another workspace. + +Schema-14 `ingest_facts` retains the event with first-receipt provenance. Exact replay is idempotent; changed content, revision or kind substitution involving a causal declaration refuses atomically. The existing `learningObservations` relation exposes the opaque canonical fact under its existing row, byte and selected-item bounds. Older batches and snapshots remain decodable; an older receiver or analyzer may refuse the new kind. No database migration, backfill or existing fact rewrite occurs. + +The private analyzer reports declarations returned, selected admitted invocations without declarations, missing or contradictory own joins, declared dependency coverage counts and unresolved references, with bounded diagnostic samples. `dependencyCoverage=complete` describes only a source declaration for that admission. Missing historical facts, parent/root/dependency targets and external-original references remain unresolved. The analyzer keeps `waitCoverage=unknown` and `criticalPath=not-evaluated`; declarations cannot complete the invocation census, usage, capture, allocation, enrollment or token comparison. + +The fixed fixtures in `tests/learn-01-analysis/fixtures/causal-admission/` are synthetic. Their manifest pins case count and each admission/event byte length and digest. The root admission intentionally reorders JSON properties and includes whitespace to catch raw-byte reserialization. Manifest route/intent digests are inert contract references, not hashes of supplied committed route/intent files. Coordination's real PostgreSQL controls must derive those digests from genuine retained route/intent bytes and pass emitted batches through an exact qualified receiver artifact. + +Local source qualification on the isolated C1 candidate passed 46 Python analyzer tests, three focused Core decoder tests, one CLI schema-14 temporary-store ingress/reopen/replay/refusal test and one Dashboard private-row projection test. The synthetic fixture manifest SHA-256 is `58de080758af83f2e243faf6995a44a529d00f810d43b2ac64753d14f59ce880`. The source-built CLI DLL SHA-256 is `e3a0f5949cb54c7a00d2883083fd05d0efcd239fc18eb07a4e4c850a1492ef87`; its runtime closure and C3 producer/receiver composition still require separate exact-artifact qualification. Earlier local restore/compiler/test-helper failures remain retained in the private qualification record. These checks used disposable source fixtures and temporary stores; they did not ingest into an installed observer or establish native provenance. + +Source order is merged A1–A3, C1 receiver, C2 producer, then C3 exact composition. Publication and installed adoption remain separate: adopt a compatible receiver before enabling new producer emission. This source window does not select a dependency wait or change fresh-workspace defaults. LEARN-01.2 and LEARN-01.4 remain open pending installed custody, independent population and shared-cost evidence, and the later qualified wait/lifecycle reduction. diff --git a/src/FS.GG.Coord.Cli/TelemetryStoreApplication.fs b/src/FS.GG.Coord.Cli/TelemetryStoreApplication.fs index ff547682f..153ebefd2 100644 --- a/src/FS.GG.Coord.Cli/TelemetryStoreApplication.fs +++ b/src/FS.GG.Coord.Cli/TelemetryStoreApplication.fs @@ -1630,6 +1630,7 @@ WHERE f.identity=$resource AND f.revision=$sourceRevision AND f.content_digest=$ | TelemetryStore.RuntimeResponseNativeInventory _ | TelemetryStore.RuntimeProviderObservation _ | TelemetryStore.RuntimeResponseUsage _ + | TelemetryStore.ExecutionCausalAdmission _ | TelemetryStore.RuntimeNativeInventorySource _ | TelemetryStore.LearnSharedCost _ | TelemetryStore.LearnSharedCostAuthority _ @@ -3456,6 +3457,13 @@ UNION ALL SELECT count(*) FROM runtime_turn_usage WHERE invocation_id=$invocatio | _ -> () match state with + | Some(oldKind, digest, revision) + when oldKind = "execution-causal-admission/1" + || fact.Kind = "execution-causal-admission/1" -> + if oldKind = fact.Kind && digest = fact.ContentDigest && revision = fact.Revision then + replayed <- replayed + 1L + else + invalidOp "execution-causal-admission/1 is immutable after persistence" | Some(_, digest, _) when digest = fact.ContentDigest -> replayed <- replayed + 1L | Some(_, _, revision) when fact.Revision <= revision -> invalidOp $"native fact identity conflict: %s{fact.Kind}/%s{fact.Identity}" @@ -3517,6 +3525,7 @@ UNION ALL SELECT count(*) FROM runtime_turn_usage WHERE invocation_id=$invocatio insert.ExecuteNonQuery() |> ignore if fact.Kind.StartsWith("learn-", StringComparison.Ordinal) + || fact.Kind = "execution-causal-admission/1" || fact.Kind = "runtime-native-inventory/1" || fact.Kind = "runtime-native-inventory-source/1" then use order = connection.CreateCommand() @@ -6360,7 +6369,7 @@ WHERE n.source_ref=$source; ItemId = itemId FactCount = runtimeScalar - "SELECT count(*) FROM current_ingest_facts WHERE item_id=$item AND kind NOT IN ('learn-task-snapshot','learn-context-manifest','learn-experiment-assignment','learn-accounting-inventory/1','runtime-native-inventory/1','runtime-native-inventory-source/1','learn-shared-cost/1','learn-shared-cost-allocation/1','learn-shared-cost-authority/1','learn-native-delivery-source/1','learn-installed-origin/1');" + "SELECT count(*) FROM current_ingest_facts WHERE item_id=$item AND kind NOT IN ('learn-task-snapshot','learn-context-manifest','learn-experiment-assignment','learn-accounting-inventory/1','runtime-native-inventory/1','runtime-native-inventory-source/1','learn-shared-cost/1','learn-shared-cost-allocation/1','learn-shared-cost-authority/1','learn-native-delivery-source/1','learn-installed-origin/1','execution-causal-admission/1');" UnknownUsageCounters = unknownUsageCounters UsageObservations = usageCount DeliveryObservations = count "delivery_observations" @@ -7375,7 +7384,7 @@ WHERE n.source_ref=$source; use learningCount = connection.CreateCommand() learningCount.Transaction <- transaction learningCount.CommandText <- - $"SELECT count(*) FROM ingest_facts WHERE kind IN ('learn-task-snapshot','learn-context-manifest','learn-experiment-assignment','learn-accounting-inventory/1','runtime-native-inventory/1','runtime-native-inventory-source/1','learn-shared-cost/1','learn-shared-cost-allocation/1','learn-shared-cost-authority/1','learn-native-delivery-source/1','learn-installed-origin/1')%s{learningItemFilter};" + $"SELECT count(*) FROM ingest_facts WHERE kind IN ('learn-task-snapshot','learn-context-manifest','learn-experiment-assignment','learn-accounting-inventory/1','runtime-native-inventory/1','runtime-native-inventory-source/1','learn-shared-cost/1','learn-shared-cost-allocation/1','learn-shared-cost-authority/1','learn-native-delivery-source/1','learn-installed-origin/1','execution-causal-admission/1')%s{learningItemFilter};" itemId |> Option.iter (parameter learningCount "$selected") if not compactCi && Convert.ToInt64(learningCount.ExecuteScalar()) > 10000L then @@ -7436,7 +7445,7 @@ WHERE n.source_ref=$source; JsonArray() else rows - ($"SELECT o.sequence AS ingest_order,a.producer AS receipt_producer,a.stream AS receipt_stream,a.authority_role AS receipt_role,a.grant_id AS receipt_grant_id,a.grant_generation AS receipt_grant_generation,a.receipt_key,a.envelope_digest AS receipt_envelope_digest,f.identity,f.kind,f.item_id,f.revision,f.content_digest,f.canonical FROM ingest_facts f LEFT JOIN learning_fact_order o ON o.identity=f.identity LEFT JOIN fact_admissions a ON a.identity=f.identity WHERE f.kind IN ('learn-task-snapshot','learn-context-manifest','learn-experiment-assignment','learn-accounting-inventory/1','runtime-native-inventory/1','runtime-native-inventory-source/1','learn-shared-cost/1','learn-shared-cost-allocation/1','learn-shared-cost-authority/1','learn-native-delivery-source/1','learn-installed-origin/1')%s{learningFactItemFilter} ORDER BY f.item_id,f.kind,f.identity LIMIT 10001;") + ($"SELECT o.sequence AS ingest_order,a.producer AS receipt_producer,a.stream AS receipt_stream,a.authority_role AS receipt_role,a.grant_id AS receipt_grant_id,a.grant_generation AS receipt_grant_generation,a.receipt_key,a.envelope_digest AS receipt_envelope_digest,f.identity,f.kind,f.item_id,f.revision,f.content_digest,f.canonical FROM ingest_facts f LEFT JOIN learning_fact_order o ON o.identity=f.identity LEFT JOIN fact_admissions a ON a.identity=f.identity WHERE f.kind IN ('learn-task-snapshot','learn-context-manifest','learn-experiment-assignment','learn-accounting-inventory/1','runtime-native-inventory/1','runtime-native-inventory-source/1','learn-shared-cost/1','learn-shared-cost-allocation/1','learn-shared-cost-authority/1','learn-native-delivery-source/1','learn-installed-origin/1','execution-causal-admission/1')%s{learningFactItemFilter} ORDER BY f.item_id,f.kind,f.identity LIMIT 10001;") [ "populations", @@ -8492,7 +8501,7 @@ WHERE n.source_ref=$source; use command = connection.CreateCommand() command.CommandText <- - "SELECT DISTINCT item_id FROM current_ingest_facts WHERE item_id IS NOT NULL AND kind NOT IN ('learn-task-snapshot','learn-context-manifest','learn-experiment-assignment','learn-accounting-inventory/1','runtime-native-inventory/1','runtime-native-inventory-source/1','learn-shared-cost/1','learn-shared-cost-allocation/1','learn-shared-cost-authority/1','learn-native-delivery-source/1','learn-installed-origin/1') ORDER BY item_id;" + "SELECT DISTINCT item_id FROM current_ingest_facts WHERE item_id IS NOT NULL AND kind NOT IN ('learn-task-snapshot','learn-context-manifest','learn-experiment-assignment','learn-accounting-inventory/1','runtime-native-inventory/1','runtime-native-inventory-source/1','learn-shared-cost/1','learn-shared-cost-allocation/1','learn-shared-cost-authority/1','learn-native-delivery-source/1','learn-installed-origin/1','execution-causal-admission/1') ORDER BY item_id;" use reader = command.ExecuteReader() let items = ResizeArray() diff --git a/src/FS.GG.Coord.Core/TelemetryStore.fs b/src/FS.GG.Coord.Core/TelemetryStore.fs index 59db0c62b..a39d32e80 100644 --- a/src/FS.GG.Coord.Core/TelemetryStore.fs +++ b/src/FS.GG.Coord.Core/TelemetryStore.fs @@ -122,6 +122,44 @@ module TelemetryStore = ProviderCreatedAt: string option Status: string } + type CausalDependency = + { OriginalItemId: string + InvocationId: string + SourceReference: string } + + type CausalDeclaration = + { Purpose: string + DependencyCoverage: string + Dependencies: CausalDependency array + RetryOfInvocationId: string option } + + type CausalAdmission = + { OriginalItemId: string + MemberItemId: string + AssignmentId: Guid + AttemptId: Guid + Generation: int64 + InvocationId: string + RootInvocationId: string + ParentInvocationId: string option + ParentAttemptId: Guid option + ParentGeneration: int64 option + RootAttemptId: Guid + RootGeneration: int64 + Relation: string + AdmittedAt: DateTimeOffset + RootAdmittedAt: DateTimeOffset + ClockProvenance: string + Declaration: CausalDeclaration } + + type CausalAdmissionEvidence = + { DispatchId: string + AdmissionBase64: string + AdmissionSha256: string + RouteBindingSha256: string + LaunchIntentSha256: string + Admission: CausalAdmission } + type Payload = | Item of featureId: string option | Feature of name: string @@ -160,6 +198,7 @@ module TelemetryStore = requestedModel: string option * requestedEffort: string option * backend: string option + | ExecutionCausalAdmission of CausalAdmissionEvidence | RuntimeStart of invocationId: string * threadId: string option * @@ -574,6 +613,144 @@ module TelemetryStore = requiredText label node name |> Result.bind (boundedText $"%s{label}.%s{name}" maximum) + // This decoder is an independent, closed mirror of the committed admission/1 + // contract. In particular, it validates the retained bytes rather than a + // serialization of the parsed DTO. + let private parseCausalAdmission (bytes: byte array) = + let reject reason = invalidOp ("execution-causal-admission: " + reason) + let field (node: JsonElement) (name: string) : JsonElement = + match node.TryGetProperty name with + | true, value -> value + | _ -> reject ("missing " + name) + let shape (node: JsonElement) (names: string list) = + if node.ValueKind <> JsonValueKind.Object then reject "object required" + let actual = node.EnumerateObject() |> Seq.map _.Name |> Seq.toArray + if actual.Length <> names.Length || Set.ofArray actual <> Set.ofList names then + reject "closed field set required" + let rec noDuplicates (node: JsonElement) = + match node.ValueKind with + | JsonValueKind.Object -> + let values = node.EnumerateObject() |> Seq.toArray + if values.Length <> (values |> Array.map _.Name |> Array.distinct |> Array.length) then + reject "duplicate property" + values |> Array.iter (fun value -> noDuplicates value.Value) + | JsonValueKind.Array -> node.EnumerateArray() |> Seq.iter noDuplicates + | _ -> () + let boundedString (maximum: int) (node: JsonElement) : string = + if node.ValueKind <> JsonValueKind.String then reject "string required" + let value = node.GetString() + if String.IsNullOrWhiteSpace value || value.Length > maximum || (value |> Seq.exists Char.IsControl) then + reject "bounded string required" + value + let text (node: JsonElement) (name: string) (maximum: int) = field node name |> boundedString maximum + let optionalText (node: JsonElement) (name: string) (maximum: int) = + let value = field node name + if value.ValueKind = JsonValueKind.Null then None else Some(boundedString maximum value) + let integer (node: JsonElement) = + if node.ValueKind <> JsonValueKind.Number then reject "integer required" + match node.TryGetInt64() with + | true, value when value >= 0L -> value + | _ -> reject "nonnegative integer required" + let number (node: JsonElement) (name: string) = field node name |> integer + let optionalNumber (node: JsonElement) (name: string) = + let value = field node name + if value.ValueKind = JsonValueKind.Null then None else Some(integer value) + let guid (node: JsonElement) = + let value = boundedString 64 node + match Guid.TryParse value with + | true, parsed when parsed <> Guid.Empty -> parsed + | _ -> reject "nonempty GUID required" + let requiredGuid (node: JsonElement) (name: string) = field node name |> guid + let optionalGuid (node: JsonElement) (name: string) = + let value = field node name + if value.ValueKind = JsonValueKind.Null then None else Some(guid value) + let timestamp (node: JsonElement) (name: string) = + let value = text node name 64 + if not (validTimestamp value) then reject "RFC3339 timestamp required" + DateTimeOffset.Parse(value, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind) + let identity (original: string) (attempt: Guid) (generation: int64) (suffix: string) (prefix: string) = + prefix + + CanonicalJson.sha256 + (Encoding.UTF8.GetBytes(original + "\u001f" + attempt.ToString("N") + "\u001f" + string generation + "\u001f" + suffix)) + try + if bytes.Length < 1 || bytes.Length > 4096 then reject "size outside 1..4096" + use document = JsonDocument.Parse bytes + let root = document.RootElement + noDuplicates root + shape root + [ "schema"; "originalItemId"; "memberItemId"; "assignmentId"; "attemptId"; "generation" + "invocationId"; "rootInvocationId"; "parentInvocationId"; "parentAttemptId" + "parentGeneration"; "rootAttemptId"; "rootGeneration"; "relation"; "admittedAt" + "rootAdmittedAt"; "clockProvenance"; "declaration" ] + if text root "schema" 128 <> "fsgg.orchestration.execution-causal-admission/1" then reject "schema" + let original = text root "originalItemId" 512 + let memberId = text root "memberItemId" 512 + let assignment = requiredGuid root "assignmentId" + let attempt = requiredGuid root "attemptId" + let generation = number root "generation" + let invocation = text root "invocationId" 128 + let rootInvocation = text root "rootInvocationId" 128 + let parentInvocation = optionalText root "parentInvocationId" 128 + let parentAttempt = optionalGuid root "parentAttemptId" + let parentGeneration = optionalNumber root "parentGeneration" + let rootAttempt = requiredGuid root "rootAttemptId" + let rootGeneration = number root "rootGeneration" + let relation = text root "relation" 16 + let admitted = timestamp root "admittedAt" + let rootAdmitted = timestamp root "rootAdmittedAt" + let clock = text root "clockProvenance" 32 + let declaration = field root "declaration" + shape declaration [ "purpose"; "dependencyCoverage"; "dependencies"; "retryOfInvocationId" ] + let purpose = text declaration "purpose" 32 + let coverage = text declaration "dependencyCoverage" 16 + let retry = optionalText declaration "retryOfInvocationId" 128 + let edges = field declaration "dependencies" + if edges.ValueKind <> JsonValueKind.Array || edges.GetArrayLength() > 16 then reject "dependencies bound" + let dependencies = + edges.EnumerateArray() + |> Seq.map (fun edge -> + shape edge [ "originalItemId"; "invocationId"; "sourceReference" ] + { OriginalItemId = text edge "originalItemId" 512 + InvocationId = text edge "invocationId" 128 + SourceReference = text edge "sourceReference" 512 }) + |> Seq.toArray + let pairs = dependencies |> Array.map (fun edge -> edge.OriginalItemId, edge.InvocationId) + if pairs.Length <> (pairs |> Array.distinct |> Array.length) + || (dependencies |> Array.exists (fun edge -> edge.InvocationId = invocation)) then + reject "duplicate or self dependency" + if not (Set.contains purpose (Set.ofList [ "planning"; "implementation"; "review"; "validation"; + "delivery"; "repair"; "operations"; "other"; "unclassified" ])) + || not (Set.contains coverage (Set.ofList [ "complete"; "partial"; "unknown" ])) + || clock <> "host-wall" || admitted = DateTimeOffset.MinValue + || rootAdmitted = DateTimeOffset.MinValue || rootGeneration > generation || rootAdmitted > admitted + || invocation <> identity original attempt generation "invocation" "invocation-" + || rootInvocation <> identity original rootAttempt rootGeneration "invocation" "invocation-" + || (relation = "root" && + (parentInvocation.IsSome || parentAttempt.IsSome || parentGeneration.IsSome + || invocation <> rootInvocation)) + || (relation <> "root" && + (not (Set.contains relation (Set.ofList [ "child"; "follow-up" ])) + || parentInvocation.IsNone || parentAttempt.IsNone || parentGeneration.IsNone + || generation <= rootGeneration + || (parentGeneration |> Option.defaultValue -1L) < rootGeneration + || (parentGeneration |> Option.defaultValue -1L) >= generation + || parentInvocation.Value <> identity original parentAttempt.Value parentGeneration.Value "invocation" "invocation-")) + || (retry.IsSome && retry <> parentInvocation) then + reject "lineage or declaration" + Ok + { OriginalItemId = original; MemberItemId = memberId; AssignmentId = assignment + AttemptId = attempt; Generation = generation; InvocationId = invocation + RootInvocationId = rootInvocation; ParentInvocationId = parentInvocation + ParentAttemptId = parentAttempt; ParentGeneration = parentGeneration + RootAttemptId = rootAttempt; RootGeneration = rootGeneration + Relation = relation; AdmittedAt = admitted; RootAdmittedAt = rootAdmitted + ClockProvenance = clock + Declaration = { Purpose = purpose; DependencyCoverage = coverage + Dependencies = dependencies; RetryOfInvocationId = retry } } + with + | :? JsonException as error -> Error("execution-causal-admission: " + error.Message) + | :? InvalidOperationException as error -> Error error.Message + let private requiredTextArray (label: string) (node: JsonElement) (name: string) = match node.TryGetProperty name with | true, value when value.ValueKind = JsonValueKind.Array && value.GetArrayLength() <= 8 -> @@ -1099,6 +1276,57 @@ module TelemetryStore = match requiredText label node "targetIdentity", requiredText label node "reason" with | Ok target, Ok reason -> make [ "targetIdentity"; "reason" ] (Correction(target, reason)) | values -> Error(sprintf "%A" values) + | "execution-causal-admission/1" -> + match + requiredText label node "dispatchId", + requiredText label node "admissionBase64", + requiredText label node "admissionSha256", + requiredText label node "routeBindingSha256", + requiredText label node "launchIntentSha256" + with + | Ok dispatch, Ok encoded, Ok admissionDigest, Ok routeDigest, Ok intentDigest -> + let isDigest (value: string) = Regex.IsMatch(value, "^[0-9a-f]{64}$") + let names = node.EnumerateObject() |> Seq.map _.Name |> Seq.toArray + if names.Length <> (names |> Array.distinct |> Array.length) + || not (names |> Array.contains "revision") + || revision <> 0L || not (List.forall isDigest [ admissionDigest; routeDigest; intentDigest ]) + || encoded.Length > 5464 then + Error $"%s{label} has invalid immutable admission metadata" + else + try + let raw = Convert.FromBase64String encoded + if raw.Length < 1 || raw.Length > 4096 + || Convert.ToBase64String raw <> encoded + || CanonicalJson.sha256 raw <> admissionDigest then + Error $"%s{label} has noncanonical or unbound admission bytes" + else + match parseCausalAdmission raw with + | Error reason -> Error reason + | Ok admission -> + let expectedDispatch = + "dispatch-" + + CanonicalJson.sha256 + (Encoding.UTF8.GetBytes( + admission.OriginalItemId + "\u001f" + + admission.AttemptId.ToString("N") + "\u001f" + + string admission.Generation + "\u001fdispatch")) + if itemId <> Some admission.OriginalItemId + || identity <> "execution-causal-admission-" + admission.InvocationId + || dispatch <> expectedDispatch then + Error $"%s{label} does not bind the canonical invocation and dispatch" + else + make + [ "dispatchId"; "admissionBase64"; "admissionSha256" + "routeBindingSha256"; "launchIntentSha256" ] + (ExecutionCausalAdmission + { DispatchId = dispatch; AdmissionBase64 = encoded + AdmissionSha256 = admissionDigest + RouteBindingSha256 = routeDigest + LaunchIntentSha256 = intentDigest + Admission = admission }) + with :? FormatException -> + Error $"%s{label}.admissionBase64 is invalid" + | values -> Error(sprintf "%A" values) | "runtime-admission" -> match requiredText label node "invocationId", diff --git a/src/FS.GG.Coord.Core/TelemetryStore.fsi b/src/FS.GG.Coord.Core/TelemetryStore.fsi index 778c70732..7201766fa 100644 --- a/src/FS.GG.Coord.Core/TelemetryStore.fsi +++ b/src/FS.GG.Coord.Core/TelemetryStore.fsi @@ -1,5 +1,7 @@ namespace FS.GG.Coord +open System + module TelemetryStore = [] val BatchSchema: string = "fsgg.telemetry.ingest/1" @@ -118,6 +120,44 @@ module TelemetryStore = ProviderCreatedAt: string option Status: string } + type CausalDependency = + { OriginalItemId: string + InvocationId: string + SourceReference: string } + + type CausalDeclaration = + { Purpose: string + DependencyCoverage: string + Dependencies: CausalDependency array + RetryOfInvocationId: string option } + + type CausalAdmission = + { OriginalItemId: string + MemberItemId: string + AssignmentId: Guid + AttemptId: Guid + Generation: int64 + InvocationId: string + RootInvocationId: string + ParentInvocationId: string option + ParentAttemptId: Guid option + ParentGeneration: int64 option + RootAttemptId: Guid + RootGeneration: int64 + Relation: string + AdmittedAt: DateTimeOffset + RootAdmittedAt: DateTimeOffset + ClockProvenance: string + Declaration: CausalDeclaration } + + type CausalAdmissionEvidence = + { DispatchId: string + AdmissionBase64: string + AdmissionSha256: string + RouteBindingSha256: string + LaunchIntentSha256: string + Admission: CausalAdmission } + type Payload = | Item of featureId: string option | Feature of name: string @@ -156,6 +196,7 @@ module TelemetryStore = requestedModel: string option * requestedEffort: string option * backend: string option + | ExecutionCausalAdmission of CausalAdmissionEvidence | RuntimeStart of invocationId: string * threadId: string option * diff --git a/tests/FS.GG.Coord.Cli.Tests/TelemetryStoreApplicationTests.fs b/tests/FS.GG.Coord.Cli.Tests/TelemetryStoreApplicationTests.fs index 23315006e..32f685884 100644 --- a/tests/FS.GG.Coord.Cli.Tests/TelemetryStoreApplicationTests.fs +++ b/tests/FS.GG.Coord.Cli.Tests/TelemetryStoreApplicationTests.fs @@ -15,6 +15,15 @@ open FS.GG.Coord.Cli module TelemetryStoreApplicationTests = let private approved = TelemetryStore.ApprovedLocalDurable + let private causalFixture name = + Path.Combine(__SOURCE_DIRECTORY__, "..", "learn-01-analysis", "fixtures", "causal-admission", name) + + let private causalBatch name event = + Encoding.UTF8.GetBytes + $"""{{"schema":"{TelemetryStore.BatchSchema}","ingestId":"causal-{name}","sourceIdentity":"coordination","generation":"synthetic-{name}","cursor":"causal-{name}","eventCount":1,"events":[{event}]}}""" + + let private causalEvent name = File.ReadAllText(causalFixture(name + ".event.json")).Trim() + let private unwrap = function | Ok value -> value @@ -35,6 +44,42 @@ module TelemetryStoreApplicationTests = }, path + [] + let ``causal declaration survives schema14 reopen and refuses immutable correction atomically`` () = + let cleanup, path = root () + use cleanup = cleanup + TelemetryStoreApplication.initialize path approved |> unwrap |> ignore + let rootEvent = causalEvent "root" + let batch = causalBatch "root" rootEvent + Assert.Contains("\"accepted\":1", TelemetryStoreApplication.ingest path approved batch |> unwrap) + Assert.Contains("\"replayed\":1", TelemetryStoreApplication.ingest path approved batch |> unwrap) + let snapshot () = + use envelope = JsonDocument.Parse(TelemetryStoreApplication.dashboardSnapshot path approved None |> unwrap) + let compressed = Convert.FromBase64String(envelope.RootElement.GetProperty("canonicalSnapshotGzip").GetString()) + use input = new MemoryStream(compressed) + use unzip = new GZipStream(input, CompressionMode.Decompress) + use plain = new MemoryStream() + unzip.CopyTo plain + JsonDocument.Parse(plain.ToArray()) + use first = snapshot () + let rows = first.RootElement.GetProperty("learningObservations") + Assert.Equal(1, rows.GetArrayLength()) + Assert.Equal("execution-causal-admission/1", rows.[0].GetProperty("kind").GetString()) + use originalEvent = JsonDocument.Parse rootEvent + use retainedEvent = JsonDocument.Parse(rows.[0].GetProperty("canonical").GetString()) + Assert.Equal(originalEvent.RootElement.GetProperty("admissionBase64").GetString(), + retainedEvent.RootElement.GetProperty("admissionBase64").GetString()) + let event = System.Text.Json.Nodes.JsonNode.Parse rootEvent + let admission = Convert.FromBase64String(event["admissionBase64"].GetValue()) + let changed = Encoding.UTF8.GetString(admission).Replace("\"implementation\"", "\"review\"") |> Encoding.UTF8.GetBytes + event["admissionBase64"] <- Convert.ToBase64String changed + event["admissionSha256"] <- CanonicalJson.sha256 changed + Assert.True(TelemetryStoreApplication.ingest path approved (causalBatch "changed" (event.ToJsonString())) |> Result.isError) + use after = snapshot () + Assert.Equal(first.RootElement.GetProperty("learningObservations").[0].GetProperty("content_digest").GetString(), + after.RootElement.GetProperty("learningObservations").[0].GetProperty("content_digest").GetString()) + Assert.Equal(14, after.RootElement.GetProperty("store").GetProperty("schemaVersion").GetInt32()) + let private pragma path value = use connection = new SqliteConnection( diff --git a/tests/FS.GG.Coord.Core.Tests/TelemetryStoreTests.fs b/tests/FS.GG.Coord.Core.Tests/TelemetryStoreTests.fs index dee309c13..2ed505db4 100644 --- a/tests/FS.GG.Coord.Core.Tests/TelemetryStoreTests.fs +++ b/tests/FS.GG.Coord.Core.Tests/TelemetryStoreTests.fs @@ -2,12 +2,94 @@ namespace FS.GG.Coord.Tests open System open System.Text +open System.IO +open System.Text.Json open Xunit open FS.GG.Coord module TelemetryStoreTests = let private bytes (value: string) = Encoding.UTF8.GetBytes value + let private causalFixture name = + Path.Combine(__SOURCE_DIRECTORY__, "..", "learn-01-analysis", "fixtures", "causal-admission", name) + + let private causalBatch eventText = + bytes $"""{{"schema":"{TelemetryStore.BatchSchema}","ingestId":"causal-fixture","sourceIdentity":"coordination","generation":"synthetic","cursor":"causal-fixture","eventCount":1,"events":[{eventText}]}}""" + + [] + let ``causal admission fixtures retain exact decoded bytes and distinct retry annotation`` () = + use manifest = JsonDocument.Parse(File.ReadAllBytes(causalFixture "manifest.json")) + let cases = manifest.RootElement.GetProperty("cases").EnumerateArray() |> Seq.toArray + Assert.Equal(4, manifest.RootElement.GetProperty("caseCount").GetInt32()) + Assert.Equal(4, cases.Length) + for item in cases do + let raw = File.ReadAllBytes(causalFixture(item.GetProperty("admissionFile").GetString())) + let eventBytes = File.ReadAllBytes(causalFixture(item.GetProperty("eventFile").GetString())) + Assert.Equal(item.GetProperty("admissionBytes").GetInt32(), raw.Length) + Assert.Equal(item.GetProperty("admissionSha256").GetString(), CanonicalJson.sha256 raw) + Assert.Equal(item.GetProperty("eventBytes").GetInt32(), eventBytes.Length) + Assert.Equal(item.GetProperty("eventSha256").GetString(), CanonicalJson.sha256 eventBytes) + let parsed = + TelemetryStore.parseBatch (causalBatch (Encoding.UTF8.GetString(eventBytes).Trim())) + |> Result.defaultWith (sprintf "%A" >> failwith) + match parsed.Facts with + | [ { Payload = TelemetryStore.ExecutionCausalAdmission evidence } ] -> + Assert.Equal(Convert.ToBase64String raw, evidence.AdmissionBase64) + Assert.Equal(item.GetProperty("invocationId").GetString(), evidence.Admission.InvocationId) + if item.GetProperty("name").GetString() = "explicit-retry" then + Assert.Equal(evidence.Admission.ParentInvocationId, evidence.Admission.Declaration.RetryOfInvocationId) + | _ -> Assert.Fail("causal fixture did not decode to one declaration") + + [] + let ``causal admission refuses changed bytes identity and shape before ingestion`` () = + let root = File.ReadAllText(causalFixture "root.event.json").Trim() + let node = System.Text.Json.Nodes.JsonNode.Parse root + let refuse (value: string) = Assert.True(TelemetryStore.parseBatch (causalBatch value) |> Result.isError) + refuse (root.Replace("\"revision\":0", "\"revision\":1")) + refuse (root.Replace("\"kind\":\"execution-causal-admission/1\"", "\"kind\":\"unknown/1\"")) + refuse (root.Replace("\"admissionSha256\":\"", "\"admissionSha256\":\"0")) + refuse (root.Replace("\"dispatchId\":\"", "\"dispatchId\":\"other-")) + refuse (root.Replace("\"revision\":0", "\"revision\":0,\"revision\":0")) + node["admissionBase64"] <- node["admissionBase64"].GetValue() + " " + refuse (node.ToJsonString()) + + [] + let ``causal admission enforces exact byte and dependency edge boundaries`` () = + let original = File.ReadAllBytes(causalFixture "root.admission.json") + let eventText = File.ReadAllText(causalFixture "root.event.json") + let eventFor (raw: byte array) = + let event = System.Text.Json.Nodes.JsonNode.Parse eventText + event["admissionBase64"] <- Convert.ToBase64String raw + event["admissionSha256"] <- CanonicalJson.sha256 raw + event.ToJsonString() + let accepts raw = TelemetryStore.parseBatch (causalBatch (eventFor raw)) |> Result.isOk + let rejects raw = TelemetryStore.parseBatch (causalBatch (eventFor raw)) |> Result.isError + let padded size = Array.append original (Array.create (size - original.Length) 0x20uy) + Assert.True(accepts (padded 4096)) + Assert.True(rejects (padded 4097)) + let edges count duplicate self = + let dto = System.Text.Json.Nodes.JsonNode.Parse(Encoding.UTF8.GetString original) + let declaration = dto["declaration"].AsObject() + let values = declaration["dependencies"].AsArray() + for index in 0 .. count - 1 do + let endpoint = + if self then dto["invocationId"].GetValue() + elif duplicate then "invocation-external-0" + else $"invocation-external-{index}" + values.Add(System.Text.Json.Nodes.JsonNode.Parse( + $"""{{"originalItemId":"LEARN-01.2-SYNTHETIC","invocationId":"{endpoint}","sourceReference":"synthetic"}}""")) + Encoding.UTF8.GetBytes(dto.ToJsonString()) + let sixteen = edges 16 false false + let seventeen = edges 17 false false + Assert.True(sixteen.Length <= 4096 && seventeen.Length <= 4096) + Assert.True(accepts sixteen) + Assert.True(rejects seventeen) + Assert.True(rejects (edges 2 true false)) + Assert.True(rejects (edges 1 false true)) + let unknown = System.Text.Json.Nodes.JsonNode.Parse(Encoding.UTF8.GetString original) + unknown["unrecognized"] <- true + Assert.True(rejects (Encoding.UTF8.GetBytes(unknown.ToJsonString()))) + let private validBatch ingest identity revision : byte array = bytes $"""{{"schema":"%s{TelemetryStore.BatchSchema}","ingestId":"%s{ingest}","sourceIdentity":"worker-a","generation":"g1","cursor":"c1","eventCount":1,"events":[{{"kind":"usage","identity":"%s{identity}","itemId":"UTEL-02","revision":%d{revision},"provider":"OpenAI","model":"sol","effort":"medium","input":10,"cachedInput":2,"cacheWriteInput":1,"output":5,"reasoning":2,"total":15,"responses":1,"sessions":1,"turns":1}}]}}""" diff --git a/tests/FS.GG.Telemetry.Dashboard.Tests/DashboardProjectionTests.fs b/tests/FS.GG.Telemetry.Dashboard.Tests/DashboardProjectionTests.fs index f1fa24672..1d753633b 100644 --- a/tests/FS.GG.Telemetry.Dashboard.Tests/DashboardProjectionTests.fs +++ b/tests/FS.GG.Telemetry.Dashboard.Tests/DashboardProjectionTests.fs @@ -158,6 +158,17 @@ module DashboardProjectionTests = JsonArray([| for index in 0..10000 -> JsonValue.Create(index) :> JsonNode |]) Assert.Equal(Error InvalidSnapshot, DashboardProjection.project "workspace-a" (scopedEnvelope "workspace-a" value)) + [] + let ``causal observation remains an opaque bounded private row in the existing dashboard projection`` () = + let value = responseSnapshot () + value["learningObservations"] <- + nodes + [| row """{"kind":"execution-causal-admission/1","identity":"execution-causal-admission-invocation-example","canonical":"private-admission-bytes","receipt_producer":"private-producer"}""" |] + let projected = DashboardProjection.project "workspace-a" (scopedEnvelope "workspace-a" value) |> unwrap + let text = Encoding.UTF8.GetString projected + Assert.DoesNotContain("private-admission-bytes", text) + Assert.DoesNotContain("private-producer", text) + [] let ``private item steps expose bounded timing and direct tokens without evidence text`` () = let value = snapshot "item-a" diff --git a/tests/learn-01-analysis/fixtures/causal-admission/child.admission.json b/tests/learn-01-analysis/fixtures/causal-admission/child.admission.json new file mode 100644 index 000000000..e884b5bfa --- /dev/null +++ b/tests/learn-01-analysis/fixtures/causal-admission/child.admission.json @@ -0,0 +1 @@ +{"schema":"fsgg.orchestration.execution-causal-admission/1","originalItemId":"LEARN-01.2-SYNTHETIC","memberItemId":"LEARN-01.2-SYNTHETIC","assignmentId":"11111111-1111-4111-8111-111111111111","attemptId":"33333333-3333-4333-8333-333333333333","generation":1,"invocationId":"invocation-435d85bb120f1036fc8abb6c96b1b713c5c0bd4d5226e99daadb7d902e6a8199","rootInvocationId":"invocation-e32a815d66260d1d7b8011c37130ae4bc6145252f1cd1fab33cbc6dee7029870","parentInvocationId":"invocation-e32a815d66260d1d7b8011c37130ae4bc6145252f1cd1fab33cbc6dee7029870","parentAttemptId":"22222222-2222-4222-8222-222222222222","parentGeneration":0,"rootAttemptId":"22222222-2222-4222-8222-222222222222","rootGeneration":0,"relation":"child","admittedAt":"2026-10-10T00:00:01Z","rootAdmittedAt":"2026-10-10T00:00:00Z","clockProvenance":"host-wall","declaration":{"purpose":"validation","dependencyCoverage":"partial","dependencies":[{"originalItemId":"LEARN-01.2-SYNTHETIC","invocationId":"invocation-e32a815d66260d1d7b8011c37130ae4bc6145252f1cd1fab33cbc6dee7029870","sourceReference":"synthetic:parent:child"}],"retryOfInvocationId":null}} \ No newline at end of file diff --git a/tests/learn-01-analysis/fixtures/causal-admission/child.event.json b/tests/learn-01-analysis/fixtures/causal-admission/child.event.json new file mode 100644 index 000000000..4a46f9878 --- /dev/null +++ b/tests/learn-01-analysis/fixtures/causal-admission/child.event.json @@ -0,0 +1 @@ +{"kind":"execution-causal-admission/1","identity":"execution-causal-admission-invocation-435d85bb120f1036fc8abb6c96b1b713c5c0bd4d5226e99daadb7d902e6a8199","itemId":"LEARN-01.2-SYNTHETIC","revision":0,"dispatchId":"dispatch-f7f020c548d3d6116834aac2ad92a4a91b2616f35c60fcc9652238170d09283a","admissionBase64":"eyJzY2hlbWEiOiJmc2dnLm9yY2hlc3RyYXRpb24uZXhlY3V0aW9uLWNhdXNhbC1hZG1pc3Npb24vMSIsIm9yaWdpbmFsSXRlbUlkIjoiTEVBUk4tMDEuMi1TWU5USEVUSUMiLCJtZW1iZXJJdGVtSWQiOiJMRUFSTi0wMS4yLVNZTlRIRVRJQyIsImFzc2lnbm1lbnRJZCI6IjExMTExMTExLTExMTEtNDExMS04MTExLTExMTExMTExMTExMSIsImF0dGVtcHRJZCI6IjMzMzMzMzMzLTMzMzMtNDMzMy04MzMzLTMzMzMzMzMzMzMzMyIsImdlbmVyYXRpb24iOjEsImludm9jYXRpb25JZCI6Imludm9jYXRpb24tNDM1ZDg1YmIxMjBmMTAzNmZjOGFiYjZjOTZiMWI3MTNjNWMwYmQ0ZDUyMjZlOTlkYWFkYjdkOTAyZTZhODE5OSIsInJvb3RJbnZvY2F0aW9uSWQiOiJpbnZvY2F0aW9uLWUzMmE4MTVkNjYyNjBkMWQ3YjgwMTFjMzcxMzBhZTRiYzYxNDUyNTJmMWNkMWZhYjMzY2JjNmRlZTcwMjk4NzAiLCJwYXJlbnRJbnZvY2F0aW9uSWQiOiJpbnZvY2F0aW9uLWUzMmE4MTVkNjYyNjBkMWQ3YjgwMTFjMzcxMzBhZTRiYzYxNDUyNTJmMWNkMWZhYjMzY2JjNmRlZTcwMjk4NzAiLCJwYXJlbnRBdHRlbXB0SWQiOiIyMjIyMjIyMi0yMjIyLTQyMjItODIyMi0yMjIyMjIyMjIyMjIiLCJwYXJlbnRHZW5lcmF0aW9uIjowLCJyb290QXR0ZW1wdElkIjoiMjIyMjIyMjItMjIyMi00MjIyLTgyMjItMjIyMjIyMjIyMjIyIiwicm9vdEdlbmVyYXRpb24iOjAsInJlbGF0aW9uIjoiY2hpbGQiLCJhZG1pdHRlZEF0IjoiMjAyNi0xMC0xMFQwMDowMDowMVoiLCJyb290QWRtaXR0ZWRBdCI6IjIwMjYtMTAtMTBUMDA6MDA6MDBaIiwiY2xvY2tQcm92ZW5hbmNlIjoiaG9zdC13YWxsIiwiZGVjbGFyYXRpb24iOnsicHVycG9zZSI6InZhbGlkYXRpb24iLCJkZXBlbmRlbmN5Q292ZXJhZ2UiOiJwYXJ0aWFsIiwiZGVwZW5kZW5jaWVzIjpbeyJvcmlnaW5hbEl0ZW1JZCI6IkxFQVJOLTAxLjItU1lOVEhFVElDIiwiaW52b2NhdGlvbklkIjoiaW52b2NhdGlvbi1lMzJhODE1ZDY2MjYwZDFkN2I4MDExYzM3MTMwYWU0YmM2MTQ1MjUyZjFjZDFmYWIzM2NiYzZkZWU3MDI5ODcwIiwic291cmNlUmVmZXJlbmNlIjoic3ludGhldGljOnBhcmVudDpjaGlsZCJ9XSwicmV0cnlPZkludm9jYXRpb25JZCI6bnVsbH19","admissionSha256":"28623513c4cdcd7a29b0327bf7ac2acef46a808fbdd6a1d82b8c9ae188088c7d","routeBindingSha256":"f6f5f48fe090419ba218caac8cce95b5974da7cc343c3aa3151c871eb7aba0c6","launchIntentSha256":"0c63638fa6d9b5d770ba179f894972ae69bed46a150536a8dc0d778990db9ee3"} diff --git a/tests/learn-01-analysis/fixtures/causal-admission/explicit-retry.admission.json b/tests/learn-01-analysis/fixtures/causal-admission/explicit-retry.admission.json new file mode 100644 index 000000000..0cd6f0b1f --- /dev/null +++ b/tests/learn-01-analysis/fixtures/causal-admission/explicit-retry.admission.json @@ -0,0 +1 @@ +{"schema":"fsgg.orchestration.execution-causal-admission/1","originalItemId":"LEARN-01.2-SYNTHETIC","memberItemId":"LEARN-01.2-SYNTHETIC","assignmentId":"11111111-1111-4111-8111-111111111111","attemptId":"55555555-5555-4555-8555-555555555555","generation":3,"invocationId":"invocation-a9e6d313a419cd36ec34881b3dc4b8adaccfdd3bb6ec6ac2fc6965198d776970","rootInvocationId":"invocation-e32a815d66260d1d7b8011c37130ae4bc6145252f1cd1fab33cbc6dee7029870","parentInvocationId":"invocation-a98f673543f854d4e92f4912ef831c874befb284e2312a27a6615e86c0f2d9a7","parentAttemptId":"44444444-4444-4444-8444-444444444444","parentGeneration":2,"rootAttemptId":"22222222-2222-4222-8222-222222222222","rootGeneration":0,"relation":"follow-up","admittedAt":"2026-10-10T00:00:03Z","rootAdmittedAt":"2026-10-10T00:00:00Z","clockProvenance":"host-wall","declaration":{"purpose":"repair","dependencyCoverage":"complete","dependencies":[{"originalItemId":"LEARN-01.2-SYNTHETIC","invocationId":"invocation-a98f673543f854d4e92f4912ef831c874befb284e2312a27a6615e86c0f2d9a7","sourceReference":"synthetic:parent:explicit-retry"}],"retryOfInvocationId":"invocation-a98f673543f854d4e92f4912ef831c874befb284e2312a27a6615e86c0f2d9a7"}} \ No newline at end of file diff --git a/tests/learn-01-analysis/fixtures/causal-admission/explicit-retry.event.json b/tests/learn-01-analysis/fixtures/causal-admission/explicit-retry.event.json new file mode 100644 index 000000000..11c004a28 --- /dev/null +++ b/tests/learn-01-analysis/fixtures/causal-admission/explicit-retry.event.json @@ -0,0 +1 @@ +{"kind":"execution-causal-admission/1","identity":"execution-causal-admission-invocation-a9e6d313a419cd36ec34881b3dc4b8adaccfdd3bb6ec6ac2fc6965198d776970","itemId":"LEARN-01.2-SYNTHETIC","revision":0,"dispatchId":"dispatch-b7564d171bbb56c7fa2a01e0cb70ab98bdebc36bd2a5fec66903a3921fdac09e","admissionBase64":"eyJzY2hlbWEiOiJmc2dnLm9yY2hlc3RyYXRpb24uZXhlY3V0aW9uLWNhdXNhbC1hZG1pc3Npb24vMSIsIm9yaWdpbmFsSXRlbUlkIjoiTEVBUk4tMDEuMi1TWU5USEVUSUMiLCJtZW1iZXJJdGVtSWQiOiJMRUFSTi0wMS4yLVNZTlRIRVRJQyIsImFzc2lnbm1lbnRJZCI6IjExMTExMTExLTExMTEtNDExMS04MTExLTExMTExMTExMTExMSIsImF0dGVtcHRJZCI6IjU1NTU1NTU1LTU1NTUtNDU1NS04NTU1LTU1NTU1NTU1NTU1NSIsImdlbmVyYXRpb24iOjMsImludm9jYXRpb25JZCI6Imludm9jYXRpb24tYTllNmQzMTNhNDE5Y2QzNmVjMzQ4ODFiM2RjNGI4YWRhY2NmZGQzYmI2ZWM2YWMyZmM2OTY1MTk4ZDc3Njk3MCIsInJvb3RJbnZvY2F0aW9uSWQiOiJpbnZvY2F0aW9uLWUzMmE4MTVkNjYyNjBkMWQ3YjgwMTFjMzcxMzBhZTRiYzYxNDUyNTJmMWNkMWZhYjMzY2JjNmRlZTcwMjk4NzAiLCJwYXJlbnRJbnZvY2F0aW9uSWQiOiJpbnZvY2F0aW9uLWE5OGY2NzM1NDNmODU0ZDRlOTJmNDkxMmVmODMxYzg3NGJlZmIyODRlMjMxMmEyN2E2NjE1ZTg2YzBmMmQ5YTciLCJwYXJlbnRBdHRlbXB0SWQiOiI0NDQ0NDQ0NC00NDQ0LTQ0NDQtODQ0NC00NDQ0NDQ0NDQ0NDQiLCJwYXJlbnRHZW5lcmF0aW9uIjoyLCJyb290QXR0ZW1wdElkIjoiMjIyMjIyMjItMjIyMi00MjIyLTgyMjItMjIyMjIyMjIyMjIyIiwicm9vdEdlbmVyYXRpb24iOjAsInJlbGF0aW9uIjoiZm9sbG93LXVwIiwiYWRtaXR0ZWRBdCI6IjIwMjYtMTAtMTBUMDA6MDA6MDNaIiwicm9vdEFkbWl0dGVkQXQiOiIyMDI2LTEwLTEwVDAwOjAwOjAwWiIsImNsb2NrUHJvdmVuYW5jZSI6Imhvc3Qtd2FsbCIsImRlY2xhcmF0aW9uIjp7InB1cnBvc2UiOiJyZXBhaXIiLCJkZXBlbmRlbmN5Q292ZXJhZ2UiOiJjb21wbGV0ZSIsImRlcGVuZGVuY2llcyI6W3sib3JpZ2luYWxJdGVtSWQiOiJMRUFSTi0wMS4yLVNZTlRIRVRJQyIsImludm9jYXRpb25JZCI6Imludm9jYXRpb24tYTk4ZjY3MzU0M2Y4NTRkNGU5MmY0OTEyZWY4MzFjODc0YmVmYjI4NGUyMzEyYTI3YTY2MTVlODZjMGYyZDlhNyIsInNvdXJjZVJlZmVyZW5jZSI6InN5bnRoZXRpYzpwYXJlbnQ6ZXhwbGljaXQtcmV0cnkifV0sInJldHJ5T2ZJbnZvY2F0aW9uSWQiOiJpbnZvY2F0aW9uLWE5OGY2NzM1NDNmODU0ZDRlOTJmNDkxMmVmODMxYzg3NGJlZmIyODRlMjMxMmEyN2E2NjE1ZTg2YzBmMmQ5YTcifX0=","admissionSha256":"d07fd175b1dd7729ecefcb4adba0249fa2645ebfaa0adbee8e296eefd3a07ec9","routeBindingSha256":"a800dd4e428f7d80fa932fc0cc44b5164b18f2f01246d8e910f95ca46bee2dcf","launchIntentSha256":"c2d679c34b84c28026ef8a1560cbf4670f8f34085b7e282b902fb69f2df1a10d"} diff --git a/tests/learn-01-analysis/fixtures/causal-admission/follow-up.admission.json b/tests/learn-01-analysis/fixtures/causal-admission/follow-up.admission.json new file mode 100644 index 000000000..edd340ea9 --- /dev/null +++ b/tests/learn-01-analysis/fixtures/causal-admission/follow-up.admission.json @@ -0,0 +1 @@ +{"schema":"fsgg.orchestration.execution-causal-admission/1","originalItemId":"LEARN-01.2-SYNTHETIC","memberItemId":"LEARN-01.2-SYNTHETIC","assignmentId":"11111111-1111-4111-8111-111111111111","attemptId":"44444444-4444-4444-8444-444444444444","generation":2,"invocationId":"invocation-a98f673543f854d4e92f4912ef831c874befb284e2312a27a6615e86c0f2d9a7","rootInvocationId":"invocation-e32a815d66260d1d7b8011c37130ae4bc6145252f1cd1fab33cbc6dee7029870","parentInvocationId":"invocation-435d85bb120f1036fc8abb6c96b1b713c5c0bd4d5226e99daadb7d902e6a8199","parentAttemptId":"33333333-3333-4333-8333-333333333333","parentGeneration":1,"rootAttemptId":"22222222-2222-4222-8222-222222222222","rootGeneration":0,"relation":"follow-up","admittedAt":"2026-10-10T00:00:02Z","rootAdmittedAt":"2026-10-10T00:00:00Z","clockProvenance":"host-wall","declaration":{"purpose":"review","dependencyCoverage":"unknown","dependencies":[],"retryOfInvocationId":null}} \ No newline at end of file diff --git a/tests/learn-01-analysis/fixtures/causal-admission/follow-up.event.json b/tests/learn-01-analysis/fixtures/causal-admission/follow-up.event.json new file mode 100644 index 000000000..e884e7f4c --- /dev/null +++ b/tests/learn-01-analysis/fixtures/causal-admission/follow-up.event.json @@ -0,0 +1 @@ +{"kind":"execution-causal-admission/1","identity":"execution-causal-admission-invocation-a98f673543f854d4e92f4912ef831c874befb284e2312a27a6615e86c0f2d9a7","itemId":"LEARN-01.2-SYNTHETIC","revision":0,"dispatchId":"dispatch-a108d56863a12ee5cad95f74a47f5affdd8e89dc1013a383d8e4cac7b551c6fa","admissionBase64":"eyJzY2hlbWEiOiJmc2dnLm9yY2hlc3RyYXRpb24uZXhlY3V0aW9uLWNhdXNhbC1hZG1pc3Npb24vMSIsIm9yaWdpbmFsSXRlbUlkIjoiTEVBUk4tMDEuMi1TWU5USEVUSUMiLCJtZW1iZXJJdGVtSWQiOiJMRUFSTi0wMS4yLVNZTlRIRVRJQyIsImFzc2lnbm1lbnRJZCI6IjExMTExMTExLTExMTEtNDExMS04MTExLTExMTExMTExMTExMSIsImF0dGVtcHRJZCI6IjQ0NDQ0NDQ0LTQ0NDQtNDQ0NC04NDQ0LTQ0NDQ0NDQ0NDQ0NCIsImdlbmVyYXRpb24iOjIsImludm9jYXRpb25JZCI6Imludm9jYXRpb24tYTk4ZjY3MzU0M2Y4NTRkNGU5MmY0OTEyZWY4MzFjODc0YmVmYjI4NGUyMzEyYTI3YTY2MTVlODZjMGYyZDlhNyIsInJvb3RJbnZvY2F0aW9uSWQiOiJpbnZvY2F0aW9uLWUzMmE4MTVkNjYyNjBkMWQ3YjgwMTFjMzcxMzBhZTRiYzYxNDUyNTJmMWNkMWZhYjMzY2JjNmRlZTcwMjk4NzAiLCJwYXJlbnRJbnZvY2F0aW9uSWQiOiJpbnZvY2F0aW9uLTQzNWQ4NWJiMTIwZjEwMzZmYzhhYmI2Yzk2YjFiNzEzYzVjMGJkNGQ1MjI2ZTk5ZGFhZGI3ZDkwMmU2YTgxOTkiLCJwYXJlbnRBdHRlbXB0SWQiOiIzMzMzMzMzMy0zMzMzLTQzMzMtODMzMy0zMzMzMzMzMzMzMzMiLCJwYXJlbnRHZW5lcmF0aW9uIjoxLCJyb290QXR0ZW1wdElkIjoiMjIyMjIyMjItMjIyMi00MjIyLTgyMjItMjIyMjIyMjIyMjIyIiwicm9vdEdlbmVyYXRpb24iOjAsInJlbGF0aW9uIjoiZm9sbG93LXVwIiwiYWRtaXR0ZWRBdCI6IjIwMjYtMTAtMTBUMDA6MDA6MDJaIiwicm9vdEFkbWl0dGVkQXQiOiIyMDI2LTEwLTEwVDAwOjAwOjAwWiIsImNsb2NrUHJvdmVuYW5jZSI6Imhvc3Qtd2FsbCIsImRlY2xhcmF0aW9uIjp7InB1cnBvc2UiOiJyZXZpZXciLCJkZXBlbmRlbmN5Q292ZXJhZ2UiOiJ1bmtub3duIiwiZGVwZW5kZW5jaWVzIjpbXSwicmV0cnlPZkludm9jYXRpb25JZCI6bnVsbH19","admissionSha256":"96c87ab913533a7380d34934f21b7e4dd50700f65be83b36333a8549c2fde590","routeBindingSha256":"d1545d7f6c9c10800532a6e368cb7f1e432af2ad719608cd6c9525655befaf88","launchIntentSha256":"060f5519be38f2eaf7f7f365b11c3252444104d8e49ca29fab5868d2c6952b81"} diff --git a/tests/learn-01-analysis/fixtures/causal-admission/manifest.json b/tests/learn-01-analysis/fixtures/causal-admission/manifest.json new file mode 100644 index 000000000..ee7d276a4 --- /dev/null +++ b/tests/learn-01-analysis/fixtures/causal-admission/manifest.json @@ -0,0 +1,47 @@ +{ + "schema": "fsgg.learn.causal-admission-fixtures/1", + "provenance": "synthetic-source-contract-not-native-installation", + "caseCount": 4, + "cases": [ + { + "name": "root", + "admissionFile": "root.admission.json", + "admissionBytes": 940, + "admissionSha256": "e7f021da96c018280575c6b4e56906259811d0a9d16cfcca96226aa144e33bf3", + "eventFile": "root.event.json", + "eventBytes": 1828, + "eventSha256": "309bbf168e9bfc09f95a262d555d88735a556738a32c08836d1e6af3920f74eb", + "invocationId": "invocation-e32a815d66260d1d7b8011c37130ae4bc6145252f1cd1fab33cbc6dee7029870" + }, + { + "name": "child", + "admissionFile": "child.admission.json", + "admissionBytes": 1116, + "admissionSha256": "28623513c4cdcd7a29b0327bf7ac2acef46a808fbdd6a1d82b8c9ae188088c7d", + "eventFile": "child.event.json", + "eventBytes": 2060, + "eventSha256": "b69488a5817921d4eddda3b095291e706ac136ad948dda6121802bfd4fb0846a", + "invocationId": "invocation-435d85bb120f1036fc8abb6c96b1b713c5c0bd4d5226e99daadb7d902e6a8199" + }, + { + "name": "follow-up", + "admissionFile": "follow-up.admission.json", + "admissionBytes": 939, + "admissionSha256": "96c87ab913533a7380d34934f21b7e4dd50700f65be83b36333a8549c2fde590", + "eventFile": "follow-up.event.json", + "eventBytes": 1824, + "eventSha256": "d34b4fe315cfaa87e65a42998ced6bae7ff19b96908edb840c571ceab5a896a5", + "invocationId": "invocation-a98f673543f854d4e92f4912ef831c874befb284e2312a27a6615e86c0f2d9a7" + }, + { + "name": "explicit-retry", + "admissionFile": "explicit-retry.admission.json", + "admissionBytes": 1199, + "admissionSha256": "d07fd175b1dd7729ecefcb4adba0249fa2645ebfaa0adbee8e296eefd3a07ec9", + "eventFile": "explicit-retry.event.json", + "eventBytes": 2172, + "eventSha256": "2168aae3c084af2148a36e2ff2bc925c70750838cc864908ca7f78daf4926762", + "invocationId": "invocation-a9e6d313a419cd36ec34881b3dc4b8adaccfdd3bb6ec6ac2fc6965198d776970" + } + ] +} diff --git a/tests/learn-01-analysis/fixtures/causal-admission/root.admission.json b/tests/learn-01-analysis/fixtures/causal-admission/root.admission.json new file mode 100644 index 000000000..b8128bb40 --- /dev/null +++ b/tests/learn-01-analysis/fixtures/causal-admission/root.admission.json @@ -0,0 +1,25 @@ +{ + "declaration": { + "purpose": "implementation", + "dependencyCoverage": "complete", + "dependencies": [], + "retryOfInvocationId": null + }, + "clockProvenance": "host-wall", + "rootAdmittedAt": "2026-10-10T00:00:00Z", + "admittedAt": "2026-10-10T00:00:00Z", + "relation": "root", + "rootGeneration": 0, + "rootAttemptId": "22222222-2222-4222-8222-222222222222", + "parentGeneration": null, + "parentAttemptId": null, + "parentInvocationId": null, + "rootInvocationId": "invocation-e32a815d66260d1d7b8011c37130ae4bc6145252f1cd1fab33cbc6dee7029870", + "invocationId": "invocation-e32a815d66260d1d7b8011c37130ae4bc6145252f1cd1fab33cbc6dee7029870", + "generation": 0, + "attemptId": "22222222-2222-4222-8222-222222222222", + "assignmentId": "11111111-1111-4111-8111-111111111111", + "memberItemId": "LEARN-01.2-SYNTHETIC", + "originalItemId": "LEARN-01.2-SYNTHETIC", + "schema": "fsgg.orchestration.execution-causal-admission/1" +} diff --git a/tests/learn-01-analysis/fixtures/causal-admission/root.event.json b/tests/learn-01-analysis/fixtures/causal-admission/root.event.json new file mode 100644 index 000000000..a05ea795f --- /dev/null +++ b/tests/learn-01-analysis/fixtures/causal-admission/root.event.json @@ -0,0 +1 @@ +{"kind":"execution-causal-admission/1","identity":"execution-causal-admission-invocation-e32a815d66260d1d7b8011c37130ae4bc6145252f1cd1fab33cbc6dee7029870","itemId":"LEARN-01.2-SYNTHETIC","revision":0,"dispatchId":"dispatch-8553d997674ee3b174abe2d9fdda32c45cb78bd668a1a2976dfa9251d74851bf","admissionBase64":"ewogICJkZWNsYXJhdGlvbiI6IHsKICAgICJwdXJwb3NlIjogImltcGxlbWVudGF0aW9uIiwKICAgICJkZXBlbmRlbmN5Q292ZXJhZ2UiOiAiY29tcGxldGUiLAogICAgImRlcGVuZGVuY2llcyI6IFtdLAogICAgInJldHJ5T2ZJbnZvY2F0aW9uSWQiOiBudWxsCiAgfSwKICAiY2xvY2tQcm92ZW5hbmNlIjogImhvc3Qtd2FsbCIsCiAgInJvb3RBZG1pdHRlZEF0IjogIjIwMjYtMTAtMTBUMDA6MDA6MDBaIiwKICAiYWRtaXR0ZWRBdCI6ICIyMDI2LTEwLTEwVDAwOjAwOjAwWiIsCiAgInJlbGF0aW9uIjogInJvb3QiLAogICJyb290R2VuZXJhdGlvbiI6IDAsCiAgInJvb3RBdHRlbXB0SWQiOiAiMjIyMjIyMjItMjIyMi00MjIyLTgyMjItMjIyMjIyMjIyMjIyIiwKICAicGFyZW50R2VuZXJhdGlvbiI6IG51bGwsCiAgInBhcmVudEF0dGVtcHRJZCI6IG51bGwsCiAgInBhcmVudEludm9jYXRpb25JZCI6IG51bGwsCiAgInJvb3RJbnZvY2F0aW9uSWQiOiAiaW52b2NhdGlvbi1lMzJhODE1ZDY2MjYwZDFkN2I4MDExYzM3MTMwYWU0YmM2MTQ1MjUyZjFjZDFmYWIzM2NiYzZkZWU3MDI5ODcwIiwKICAiaW52b2NhdGlvbklkIjogImludm9jYXRpb24tZTMyYTgxNWQ2NjI2MGQxZDdiODAxMWMzNzEzMGFlNGJjNjE0NTI1MmYxY2QxZmFiMzNjYmM2ZGVlNzAyOTg3MCIsCiAgImdlbmVyYXRpb24iOiAwLAogICJhdHRlbXB0SWQiOiAiMjIyMjIyMjItMjIyMi00MjIyLTgyMjItMjIyMjIyMjIyMjIyIiwKICAiYXNzaWdubWVudElkIjogIjExMTExMTExLTExMTEtNDExMS04MTExLTExMTExMTExMTExMSIsCiAgIm1lbWJlckl0ZW1JZCI6ICJMRUFSTi0wMS4yLVNZTlRIRVRJQyIsCiAgIm9yaWdpbmFsSXRlbUlkIjogIkxFQVJOLTAxLjItU1lOVEhFVElDIiwKICAic2NoZW1hIjogImZzZ2cub3JjaGVzdHJhdGlvbi5leGVjdXRpb24tY2F1c2FsLWFkbWlzc2lvbi8xIgp9Cg==","admissionSha256":"e7f021da96c018280575c6b4e56906259811d0a9d16cfcca96226aa144e33bf3","routeBindingSha256":"50ad44bebd0cbb65036a2251757dcbc578bfacf277599723c33f354c3787ef90","launchIntentSha256":"a45ceaed7093431c1efc167281e7761f5257650206a9047165c7e94665a63485"} diff --git a/tests/learn-01-analysis/test_analysis.py b/tests/learn-01-analysis/test_analysis.py index e6456c4e7..a84ae6454 100644 --- a/tests/learn-01-analysis/test_analysis.py +++ b/tests/learn-01-analysis/test_analysis.py @@ -1330,6 +1330,101 @@ def test_delayed_repair_must_follow_acceptance(self): with self.assertRaisesRegex(MODULE.Refusal, "repair does not follow acceptance"): MODULE.validate_corpus(CONTRACT, corpus) + def test_causal_fixture_manifest_binds_exact_raw_and_event_bytes(self): + fixture = ROOT / "tests/learn-01-analysis/fixtures/causal-admission" + manifest = json.loads((fixture / "manifest.json").read_text()) + self.assertEqual(4, manifest["caseCount"]) + self.assertEqual(manifest["caseCount"], len(manifest["cases"])) + for case in manifest["cases"]: + raw = (fixture / case["admissionFile"]).read_bytes() + event_bytes = (fixture / case["eventFile"]).read_bytes() + self.assertEqual((case["admissionBytes"], case["admissionSha256"]), + (len(raw), hashlib.sha256(raw).hexdigest())) + self.assertEqual((case["eventBytes"], case["eventSha256"]), + (len(event_bytes), hashlib.sha256(event_bytes).hexdigest())) + event = json.loads(event_bytes) + self.assertEqual(raw, base64.b64decode(event["admissionBase64"], validate=True)) + self.assertEqual(case["invocationId"], MODULE.decode_causal_declaration(event)["invocationId"]) + root = (fixture / "root.admission.json").read_bytes() + self.assertNotEqual(root, json.dumps(json.loads(root), separators=(",", ":")).encode()) + + def test_causal_declaration_refuses_mutation_and_keeps_wait_unknown(self): + fixture = ROOT / "tests/learn-01-analysis/fixtures/causal-admission" + events = [json.loads((fixture / (name + ".event.json")).read_text()) + for name in ("root", "child", "follow-up", "explicit-retry")] + original = events[0] + for mutation in ( + {"revision": 1}, + {"identity": "execution-causal-admission-other"}, + {"admissionBase64": original["admissionBase64"] + " "}, + {"admissionSha256": "0" * 64}, + {"unknown": "field"}): + changed = {**original, **mutation} + with self.assertRaises(MODULE.Refusal): + MODULE.decode_causal_declaration(changed) + raw = bytearray(base64.b64decode(original["admissionBase64"])) + raw.extend(b" ") + changed = {**original, "admissionBase64": base64.b64encode(raw).decode(), + "admissionSha256": hashlib.sha256(raw).hexdigest()} + self.assertEqual(events[0]["identity"], original["identity"]) + self.assertEqual(original["itemId"], MODULE.decode_causal_declaration(changed)["originalItemId"]) + summary = MODULE.causal_observation_summary( + {"admissions": [{"invocation_id": event["identity"].removeprefix("execution-causal-admission-"), + "item_id": event["itemId"]} for event in events]}, + events) + self.assertEqual(4, summary["declarationsReturned"]) + self.assertEqual({"complete": 2, "partial": 1, "unknown": 1}, + summary["declaredDependencyCoverage"]) + self.assertEqual("unknown", summary["waitCoverage"]) + self.assertEqual("not-evaluated", summary["criticalPath"]) + self.assertEqual(4, summary["missingOwnJoins"]["count"]) + + def test_causal_own_lineage_conflict_is_visible_without_cost_authority(self): + fixture = ROOT / "tests/learn-01-analysis/fixtures/causal-admission" + event = json.loads((fixture / "root.event.json").read_text()) + admission = MODULE.decode_causal_declaration(event) + summary = MODULE.causal_observation_summary({ + "admissions": [{"invocation_id": admission["invocationId"], "item_id": admission["originalItemId"]}], + "lineage": [{"invocation_id": admission["invocationId"], "item_id": admission["originalItemId"], + "dispatch_id": event["dispatchId"], "root_invocation_id": "wrong", + "parent_invocation_id": None, "relation": "root"}], + "expectedDispatches": [{"dispatch_id": event["dispatchId"], "item_id": admission["originalItemId"], + "relation": "root"}], + }, [event]) + self.assertEqual(1, summary["contradictoryOwnJoins"]["count"]) + self.assertEqual(0, summary["missingOwnJoins"]["count"]) + self.assertNotIn("tokenTotal", summary) + duplicated = copy.deepcopy({ + "admissions": [{"invocation_id": admission["invocationId"], "item_id": admission["originalItemId"]}], + "lineage": [{"invocation_id": admission["invocationId"], "item_id": admission["originalItemId"], + "dispatch_id": event["dispatchId"], "root_invocation_id": admission["rootInvocationId"], + "parent_invocation_id": None, "relation": "root"}] * 2, + "expectedDispatches": [{"dispatch_id": event["dispatchId"], "item_id": admission["originalItemId"], + "relation": "root"}], + }) + self.assertEqual(1, MODULE.causal_observation_summary(duplicated, [event])["contradictoryOwnJoins"]["count"]) + + def test_causal_rows_only_add_inert_snapshot_observation(self): + fixture = ROOT / "tests/learn-01-analysis/fixtures/causal-admission" + event = json.loads((fixture / "root.event.json").read_text()) + baseline = { + "learningObservations": [ + {"canonical": json.dumps(row, separators=(",", ":"), sort_keys=True)} + for row in OBSERVATIONS["events"] + ], + "populations": [], "admissions": [], "expectedDispatches": [], "lineage": [], + "usage": [], "terminals": [], "runtimeGaps": [], + } + before = MODULE.analyze_private_snapshot(CONTRACT, private_envelope(baseline, version=4)) + with_causal = copy.deepcopy(baseline) + with_causal["learningObservations"].append({"canonical": json.dumps(event, separators=(",", ":"), sort_keys=True)}) + after = MODULE.analyze_private_snapshot(CONTRACT, private_envelope(with_causal, version=4)) + self.assertEqual(1, after["causalObservation"]["declarationsReturned"]) + self.assertEqual("unknown", after["causalObservation"]["waitCoverage"]) + for key in ("tokenComparisonQualified", "providerTotalTokensByArm", "incompleteTokenReasons", + "assignedOriginalItemsByArm"): + self.assertEqual(before[key], after[key]) + if __name__ == "__main__": unittest.main() diff --git a/tools/learn-01-analysis.py b/tools/learn-01-analysis.py index e62bcc2e5..be062588c 100755 --- a/tools/learn-01-analysis.py +++ b/tools/learn-01-analysis.py @@ -17,6 +17,8 @@ import subprocess import sys import time +import unicodedata +import uuid from collections import Counter, defaultdict from datetime import datetime, timedelta @@ -725,6 +727,205 @@ def scope_disposition(value: dict) -> str: } +CAUSAL_KIND = "execution-causal-admission/1" +CAUSAL_SCHEMA = "fsgg.orchestration.execution-causal-admission/1" +CAUSAL_PURPOSES = {"planning", "implementation", "review", "validation", "delivery", + "repair", "operations", "other", "unclassified"} + + +def decode_causal_declaration(event: dict) -> dict: + """Validate the immutable source bytes without normalizing their JSON layout.""" + required = {"kind", "identity", "itemId", "revision", "dispatchId", "admissionBase64", + "admissionSha256", "routeBindingSha256", "launchIntentSha256"} + if set(event) != required or event["kind"] != CAUSAL_KIND or type(event["revision"]) is not int or event["revision"] != 0: + raise Refusal("causal event has an unsupported shape or revision") + for name in ("admissionSha256", "routeBindingSha256", "launchIntentSha256"): + if not isinstance(event[name], str) or not re.fullmatch(r"[0-9a-f]{64}", event[name]): + raise Refusal("causal source digest is malformed") + encoded = event["admissionBase64"] + if not isinstance(encoded, str) or len(encoded) > 5464: + raise Refusal("causal admission base64 bound") + try: + raw = base64.b64decode(encoded, validate=True) + except (ValueError, base64.binascii.Error) as error: + raise Refusal("causal admission base64 is invalid") from error + if not 1 <= len(raw) <= 4096 or base64.b64encode(raw).decode() != encoded: + raise Refusal("causal admission base64 is noncanonical or oversized") + if hashlib.sha256(raw).hexdigest() != event["admissionSha256"]: + raise Refusal("causal admission digest does not bind retained bytes") + def unique_pairs(pairs): + value = {} + for key, item in pairs: + if key in value: + raise Refusal("causal admission contains duplicate property") + value[key] = item + return value + try: + admission = json.loads(raw, object_pairs_hook=unique_pairs) + except (UnicodeDecodeError, json.JSONDecodeError) as error: + raise Refusal("causal admission JSON is invalid") from error + fields = {"schema", "originalItemId", "memberItemId", "assignmentId", "attemptId", + "generation", "invocationId", "rootInvocationId", "parentInvocationId", + "parentAttemptId", "parentGeneration", "rootAttemptId", "rootGeneration", + "relation", "admittedAt", "rootAdmittedAt", "clockProvenance", "declaration"} + if not isinstance(admission, dict) or set(admission) != fields or admission["schema"] != CAUSAL_SCHEMA: + raise Refusal("causal admission requires the closed admission/1 shape") + def text(value, limit): + if (not isinstance(value, str) or not value.strip() or len(value) > limit or + any(unicodedata.category(char) == "Cc" for char in value)): + raise Refusal("causal admission contains invalid bounded text") + return value + def number(value): + if type(value) is not int or value < 0 or value > 2**63 - 1: + raise Refusal("causal admission generation is invalid") + return value + def guid(value): + text(value, 64) + try: + parsed = uuid.UUID(value) + except ValueError as error: + raise Refusal("causal admission GUID is invalid") from error + if parsed.int == 0: + raise Refusal("causal admission GUID is empty") + return parsed + def clock(value): + text(value, 64) + if not re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(?:\.[0-9]+)?(?:Z|[+-][0-9]{2}:[0-9]{2})", value): + raise Refusal("causal admission timestamp is malformed") + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as error: + raise Refusal("causal admission timestamp is invalid") from error + original = text(admission["originalItemId"], 512) + text(admission["memberItemId"], 512) + guid(admission["assignmentId"]) + attempt = guid(admission["attemptId"]) + generation = number(admission["generation"]) + invocation = text(admission["invocationId"], 128) + root_invocation = text(admission["rootInvocationId"], 128) + root_attempt = guid(admission["rootAttemptId"]) + root_generation = number(admission["rootGeneration"]) + parent = admission["parentInvocationId"] + parent_attempt = admission["parentAttemptId"] + parent_generation = admission["parentGeneration"] + if parent is not None: + text(parent, 128) + if parent_attempt is not None: + parent_attempt = guid(parent_attempt) + if parent_generation is not None: + parent_generation = number(parent_generation) + admitted = clock(admission["admittedAt"]) + root_admitted = clock(admission["rootAdmittedAt"]) + if (admitted == datetime.min.replace(tzinfo=admitted.tzinfo) or + root_admitted > admitted or + admission["clockProvenance"] != "host-wall" or + root_generation > generation): + raise Refusal("causal admission clock or generation is inconsistent") + derive = lambda current_attempt, current_generation: ( + "invocation-" + hashlib.sha256( + (original + "\x1f" + current_attempt.hex + "\x1f" + str(current_generation) + "\x1finvocation").encode() + ).hexdigest()) + if invocation != derive(attempt, generation) or root_invocation != derive(root_attempt, root_generation): + raise Refusal("causal admission invocation identity is not canonical") + relation = admission["relation"] + if relation == "root": + if parent is not None or parent_attempt is not None or parent_generation is not None or invocation != root_invocation: + raise Refusal("causal root lineage is inconsistent") + elif relation in {"child", "follow-up"}: + if (parent is None or parent_attempt is None or parent_generation is None or + generation <= root_generation or not root_generation <= parent_generation < generation or + parent != derive(parent_attempt, parent_generation)): + raise Refusal("causal descendant lineage is inconsistent") + else: + raise Refusal("causal relation is unsupported") + declaration = admission["declaration"] + if not isinstance(declaration, dict) or set(declaration) != {"purpose", "dependencyCoverage", "dependencies", "retryOfInvocationId"}: + raise Refusal("causal declaration shape is invalid") + if (not isinstance(declaration["purpose"], str) or declaration["purpose"] not in CAUSAL_PURPOSES or + not isinstance(declaration["dependencyCoverage"], str) or + declaration["dependencyCoverage"] not in {"complete", "partial", "unknown"}): + raise Refusal("causal declaration vocabulary is invalid") + retry = declaration["retryOfInvocationId"] + if retry is not None and (not isinstance(retry, str) or retry != parent): + raise Refusal("causal retry annotation must name the parent") + dependencies = declaration["dependencies"] + if not isinstance(dependencies, list) or len(dependencies) > 16: + raise Refusal("causal dependencies exceed 16") + endpoints = set() + for edge in dependencies: + if not isinstance(edge, dict) or set(edge) != {"originalItemId", "invocationId", "sourceReference"}: + raise Refusal("causal dependency shape is invalid") + endpoint = (text(edge["originalItemId"], 512), text(edge["invocationId"], 128)) + text(edge["sourceReference"], 512) + if endpoint in endpoints or endpoint[1] == invocation: + raise Refusal("causal dependency repeats or names self") + endpoints.add(endpoint) + dispatch = "dispatch-" + hashlib.sha256( + (original + "\x1f" + attempt.hex + "\x1f" + str(generation) + "\x1fdispatch").encode() + ).hexdigest() + if (event["identity"] != "execution-causal-admission-" + invocation or + event["itemId"] != original or event["dispatchId"] != dispatch): + raise Refusal("causal event identity does not bind admission") + return admission + + +def causal_observation_summary(content: dict, events: list[dict]) -> dict: + declarations = [decode_causal_declaration(event) for event in events if event.get("kind") == CAUSAL_KIND] + by_invocation = {} + for admission in declarations: + invocation = admission["invocationId"] + if invocation in by_invocation: + raise Refusal("duplicate causal declaration for invocation") + by_invocation[invocation] = admission + admitted = {row.get("invocation_id"): row for row in content.get("admissions", [])} + lineage = defaultdict(list) + for row in content.get("lineage", []): + lineage[row.get("invocation_id")].append(row) + expected = {row.get("dispatch_id"): row for row in content.get("expectedDispatches", [])} + missing = sorted(invocation for invocation in admitted if invocation not in by_invocation and isinstance(invocation, str)) + missing_joins, conflicts, unresolved = set(), set(), set() + coverage = Counter(admission["declaration"]["dependencyCoverage"] for admission in declarations) + for invocation, declaration in by_invocation.items(): + original = declaration["originalItemId"] + row = admitted.get(invocation) + own_lineages = lineage.get(invocation, []) + dispatch = "dispatch-" + hashlib.sha256( + (original + "\x1f" + uuid.UUID(declaration["attemptId"]).hex + "\x1f" + + str(declaration["generation"]) + "\x1fdispatch").encode()).hexdigest() + expected_row = expected.get(dispatch) + if row is None or not own_lineages or expected_row is None: + missing_joins.add(invocation) + if row is not None and row.get("item_id") != original: + conflicts.add(invocation) + if len(own_lineages) > 1: + conflicts.add(invocation) + for own_lineage in own_lineages: + if (own_lineage.get("item_id") != original or own_lineage.get("dispatch_id") != dispatch or + own_lineage.get("root_invocation_id") != declaration["rootInvocationId"] or + own_lineage.get("parent_invocation_id") != declaration["parentInvocationId"] or + own_lineage.get("relation") != declaration["relation"]): + conflicts.add(invocation) + if expected_row is not None and (expected_row.get("item_id") != original or + expected_row.get("relation") != declaration["relation"]): + conflicts.add(invocation) + for target in [declaration["rootInvocationId"], declaration["parentInvocationId"]]: + if target and target not in by_invocation: + unresolved.add((original, target)) + for edge in declaration["declaration"]["dependencies"]: + endpoint = (edge["originalItemId"], edge["invocationId"]) + if endpoint[0] != original or endpoint[1] not in by_invocation: + unresolved.add(endpoint) + def bounded(values): + ordered = sorted(values) + return {"sample": ordered[:20], "omitted": max(0, len(ordered) - 20), "count": len(ordered)} + return {"declarationsReturned": len(declarations), "admittedInvocationsMissingDeclarations": bounded(missing), + "missingOwnJoins": bounded(missing_joins), "contradictoryOwnJoins": bounded(conflicts), + "declaredDependencyCoverage": {key: coverage[key] for key in ("complete", "partial", "unknown")}, + "unresolvedReferences": bounded([f"{original}:{invocation}" for original, invocation in unresolved]), + "waitCoverage": "unknown", "criticalPath": "not-evaluated", + "populationScope": "bounded-selected-snapshot-only"} + + def analyze_private_snapshot(contract: dict, envelope: dict, *, _protected_captures: list | None = None) -> dict: """Derive assigned-treatment token coverage only from one retained private snapshot.""" content, workspace = decode_private_snapshot(envelope) @@ -740,11 +941,12 @@ def analyze_private_snapshot(contract: dict, envelope: dict, *, _protected_captu "learn-accounting-inventory/1", "runtime-native-inventory/1", "runtime-native-inventory-source/1", "learn-shared-cost/1", "learn-shared-cost-allocation/1", "learn-shared-cost-authority/1", - "learn-native-delivery-source/1" + "learn-native-delivery-source/1", CAUSAL_KIND } unknown_kinds = {event.get("kind") for event in events} - supported_kinds if unknown_kinds: raise Refusal("private snapshot contains unsupported learning fact kinds: " + ", ".join(sorted(unknown_kinds))) + causal_summary = causal_observation_summary(content, events) observation_events = [event for event in events if event.get("kind") in observation_kinds] assignments = { event.get("itemId"): event @@ -1374,6 +1576,7 @@ def analyze_private_snapshot(contract: dict, envelope: dict, *, _protected_captu "contractId": contract["contractId"], "workspaceId": workspace, "snapshotRevision": envelope["revision"], + "causalObservation": causal_summary, **observation_result, "assignedOriginalItemsByArm": { arm: sum(1 for value in arms.values() if value == arm) for arm in ("current", "focused")