diff --git a/.github/workflows/lotus-family-conformance.yml b/.github/workflows/lotus-family-conformance.yml index 69b28dd1..2d9464f6 100644 --- a/.github/workflows/lotus-family-conformance.yml +++ b/.github/workflows/lotus-family-conformance.yml @@ -14,6 +14,9 @@ on: permissions: contents: read +env: + LOTUS_SOURCE_SHA: ${{ github.event.pull_request.head.sha || github.sha }} + jobs: conformance: runs-on: ubuntu-latest @@ -21,16 +24,58 @@ jobs: - name: Check out repository uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 with: + ref: ${{ env.LOTUS_SOURCE_SHA }} persist-credentials: false + - name: Verify checked-out source identity + run: test "$(git rev-parse HEAD)" = "$LOTUS_SOURCE_SHA" + - name: Set up Python + id: python uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0 with: python-version: "3.12" - name: Run Lotus Family conformance suite + id: conformance run: | python -m unittest discover \ -s standards/lotus-family/conformance \ -p 'test_*.py' \ -v + + - name: Emit immutable CI causal observation + if: always() + env: + LOTUS_COMMIT_SHA: ${{ env.LOTUS_SOURCE_SHA }} + LOTUS_REF: ${{ github.head_ref || github.ref }} + LOTUS_CONCLUSION: ${{ steps.conformance.outcome }} + LOTUS_RUNNER_OS: ${{ runner.os }} + LOTUS_RUNNER_ARCH: ${{ runner.arch }} + LOTUS_PYTHON_VERSION: ${{ steps.python.outputs.python-version }} + run: | + python standards/lotus-family/conformance/lotus_family_ci_observation.py \ + --repository "$GITHUB_REPOSITORY" \ + --ref "$LOTUS_REF" \ + --commit-sha "$LOTUS_COMMIT_SHA" \ + --workflow "$GITHUB_WORKFLOW" \ + --workflow-run-id "$GITHUB_RUN_ID" \ + --workflow-run-attempt "$GITHUB_RUN_ATTEMPT" \ + --job "$GITHUB_JOB" \ + --step "Run Lotus Family conformance suite" \ + --command "python -m unittest discover -s standards/lotus-family/conformance -p 'test_*.py' -v" \ + --test-target "standards/lotus-family/conformance/test_*.py" \ + --conclusion "$LOTUS_CONCLUSION" \ + --runner-os "$LOTUS_RUNNER_OS" \ + --runner-arch "$LOTUS_RUNNER_ARCH" \ + --python-version "$LOTUS_PYTHON_VERSION" \ + --output artifacts/lotus-ci-memory/ci-causal-observation-v0.1.json + + - name: Upload immutable CI causal observation + if: always() + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2 + with: + name: lotus-ci-causal-observation-${{ github.run_id }}-${{ github.run_attempt }} + path: artifacts/lotus-ci-memory/ci-causal-observation-v0.1.json + if-no-files-found: error + retention-days: 30 diff --git a/standards/lotus-family/ci-memory/SCHEMA.md b/standards/lotus-family/ci-memory/SCHEMA.md new file mode 100644 index 00000000..f984e20d --- /dev/null +++ b/standards/lotus-family/ci-memory/SCHEMA.md @@ -0,0 +1,92 @@ +# CI Causal Memory v0.1 + +CI Causal Memory records one immutable observation for each Lotus Family conformance run. It connects execution space and run time without allowing learned state to modify protected CI. + +## Safety boundary + +The observation layer is advisory and append-only. + +It may: + +- record exact run identity and execution location; +- normalize a stable failure signature; +- link later observations to prior runs; +- support human-reviewed regression proposals. + +It may not: + +- rewrite workflows, tests, policies, or protected branches; +- approve, merge, deploy, deliver, or close findings; +- mark a causal hypothesis confirmed from correlation alone; +- store secrets, raw environment values, or unbounded logs. + +`PASS != APPROVED != MERGED`. + +## Spacetime coordinates + +Every observation has a stable spatial path: + +```text +repository + → ref + → exact commit + → workflow + → job + → step + → command/test target +``` + +Every observation also has a temporal identity: + +```text +workflow run ID + → run attempt + → observed timestamp + → optional predecessor observation +``` + +## Causal states + +The first slice supports these deterministic learning states: + +- `observed_once` +- `repeated` +- `reproduced` +- `fix_correlated` +- `fix_validated` +- `regression_protected` + +A single CI run always starts at `observed_once`. Promotion requires a separate trusted aggregator and explicit evidence. + +## Failure signature + +For non-success conclusions, the emitter computes SHA-256 over a canonical tuple: + +```text +schema version +repository +workflow +job +step +command +conclusion +reason code +``` + +The signature deliberately excludes timestamps, run IDs, secrets, and raw logs so equivalent failures can be linked across runs. + +## Artifact + +The workflow writes: + +```text +artifacts/lotus-ci-memory/ci-causal-observation-v0.1.json +``` + +The artifact is uploaded with `if: always()` after the conformance step. The conformance step remains failure-gating; artifact generation cannot turn a failed test into a successful job. + +## Trust model + +The PR workflow can emit only its own immutable observation artifact. Durable cross-run aggregation belongs in a later default-branch `workflow_run` workflow that does not execute PR code. + +The initial artifact may contain caller/platform claims for repository, ref, and commit. Trusted repository/tree binding remains the responsibility of Lotus Family v0.2 materialization. diff --git a/standards/lotus-family/ci-memory/ci-causal-memory-graph-v0.1.json b/standards/lotus-family/ci-memory/ci-causal-memory-graph-v0.1.json new file mode 100644 index 00000000..367a5ada --- /dev/null +++ b/standards/lotus-family/ci-memory/ci-causal-memory-graph-v0.1.json @@ -0,0 +1,138 @@ +{ + "schema_version": "lotus.ci_causal_memory_graph.v0.1", + "graph_id": "ci-causal-memory-v0.1", + "model_kind": "causal_spacetime_learning_advisory", + "dimensions": { + "spatial": "Where an observation sits in repository, commit, workflow, job, step, command, test, and artifact topology", + "temporal": "How immutable runs precede, supersede, repeat, validate, or stale prior evidence", + "causal": "How observations support hypotheses, fixes, validations, and regression guards without equating correlation with cause", + "hierarchy": "How evidence, learning state, proposals, and human authority remain separated" + }, + "node_columns": [ + "id", + "type", + "space", + "time", + "authority" + ], + "nodes": [ + ["space.repository", "location", "repository", "stable", "none"], + ["space.ref", "location", "ref", "run", "none"], + ["space.commit", "location", "commit", "run", "none"], + ["space.workflow", "location", "workflow", "run", "none"], + ["space.job", "location", "job", "run", "none"], + ["space.step", "location", "step", "run", "none"], + ["space.command", "location", "command", "run", "none"], + ["space.test_target", "location", "test", "run", "none"], + ["evidence.observation", "evidence", "artifact", "run", "advisory"], + ["evidence.failure_signature", "evidence", "memory", "cross_run", "advisory"], + ["time.prior_observation", "temporal", "memory", "prior", "none"], + ["time.current_observation", "temporal", "memory", "current", "none"], + ["time.evidence_stale", "temporal", "memory", "superseded", "none"], + ["memory.observed_once", "learning_state", "memory", "current", "advisory"], + ["memory.pattern_repeated", "learning_state", "memory", "cross_run", "advisory"], + ["cause.hypothesis", "causal_state", "memory", "cross_run", "advisory"], + ["cause.confirmed", "causal_state", "memory", "validated", "human_required"], + ["fix.applied", "change", "commit", "later", "human"], + ["fix.validated", "validation", "run", "later", "advisory"], + ["guard.regression_added", "guard", "test", "later", "human"], + ["memory.proposal_created", "proposal", "issue_or_draft_pr", "later", "human_required"], + ["authority.human_acceptance_required", "boundary", "governance", "always", "human"] + ], + "edge_columns": [ + "id", + "source", + "target", + "relation", + "dimension" + ], + "edges": [ + ["M01", "space.repository", "space.ref", "contains", "spatial"], + ["M02", "space.ref", "space.commit", "resolves_to_claim", "spatial"], + ["M03", "space.commit", "space.workflow", "evaluated_by", "spatial"], + ["M04", "space.workflow", "space.job", "contains", "spatial"], + ["M05", "space.job", "space.step", "contains", "spatial"], + ["M06", "space.step", "space.command", "executes", "spatial"], + ["M07", "space.command", "space.test_target", "tests", "spatial"], + ["M08", "space.test_target", "evidence.observation", "observed_at", "spatial"], + ["M09", "time.prior_observation", "time.current_observation", "preceded_by", "temporal"], + ["M10", "time.current_observation", "evidence.observation", "materializes", "temporal"], + ["M11", "evidence.observation", "evidence.failure_signature", "normalizes_to", "causal"], + ["M12", "evidence.failure_signature", "memory.observed_once", "first_seen_as", "temporal"], + ["M13", "evidence.failure_signature", "memory.pattern_repeated", "recurred_as", "temporal"], + ["M14", "memory.observed_once", "cause.hypothesis", "may_support", "causal"], + ["M15", "memory.pattern_repeated", "cause.hypothesis", "strengthens_without_confirming", "causal"], + ["M16", "cause.hypothesis", "cause.confirmed", "requires_reproduction_and_validation", "causal"], + ["M17", "cause.confirmed", "fix.applied", "mitigated_by", "causal"], + ["M18", "fix.applied", "fix.validated", "validated_by_later_run", "temporal"], + ["M19", "fix.validated", "guard.regression_added", "protected_by", "causal"], + ["M20", "memory.pattern_repeated", "memory.proposal_created", "proposes_regression", "causal"], + ["M21", "memory.proposal_created", "authority.human_acceptance_required", "requires", "hierarchy"], + ["M22", "cause.confirmed", "authority.human_acceptance_required", "requires", "hierarchy"], + ["M23", "guard.regression_added", "authority.human_acceptance_required", "requires", "hierarchy"], + ["M24", "time.evidence_stale", "cause.hypothesis", "invalidates_automatic_reuse", "temporal"], + ["M25", "cause.hypothesis", "authority.human_acceptance_required", "requires_before_action", "hierarchy"] + ], + "trajectories": [ + { + "id": "new-failure", + "path": [ + "space.repository", + "space.ref", + "space.commit", + "space.workflow", + "space.job", + "space.step", + "space.command", + "space.test_target", + "evidence.observation", + "evidence.failure_signature", + "memory.observed_once", + "cause.hypothesis", + "authority.human_acceptance_required" + ], + "outcome": "advisory_investigation" + }, + { + "id": "recurrence", + "path": [ + "time.prior_observation", + "time.current_observation", + "evidence.observation", + "evidence.failure_signature", + "memory.pattern_repeated", + "memory.proposal_created", + "authority.human_acceptance_required" + ], + "outcome": "advisory_regression_proposal" + }, + { + "id": "validated-fix", + "path": [ + "cause.hypothesis", + "cause.confirmed", + "fix.applied", + "fix.validated", + "guard.regression_added", + "authority.human_acceptance_required" + ], + "outcome": "human_reviewed_durable_guard" + }, + { + "id": "stale-learning", + "path": [ + "time.evidence_stale", + "cause.hypothesis", + "authority.human_acceptance_required" + ], + "outcome": "no_automatic_reuse" + } + ], + "invariants": { + "correlation_is_not_confirmation": true, + "single_run_cause_state": "unconfirmed", + "automatic_mutation_allowed": false, + "human_acceptance_required": true, + "pass_is_not_approval_or_merge": true + } +} diff --git a/standards/lotus-family/ci-memory/ci-causal-observation-v0.1.schema.json b/standards/lotus-family/ci-memory/ci-causal-observation-v0.1.schema.json new file mode 100644 index 00000000..bb9c5842 --- /dev/null +++ b/standards/lotus-family/ci-memory/ci-causal-observation-v0.1.schema.json @@ -0,0 +1,173 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/safal207/pythiaLabs/standards/lotus-family/ci-memory/ci-causal-observation-v0.1.schema.json", + "title": "Lotus CI causal observation v0.1", + "type": "object", + "additionalProperties": false, + "required": [ + "schema_version", + "observation_id", + "spatial", + "temporal", + "causal", + "evidence", + "learning", + "limitations", + "authority" + ], + "properties": { + "schema_version": { + "const": "lotus.ci_causal_observation.v0.1" + }, + "observation_id": { + "type": "string", + "pattern": "^obs-[0-9a-f]{64}$" + }, + "spatial": { + "type": "object", + "additionalProperties": false, + "required": [ + "repository", + "ref", + "commit_sha", + "workflow", + "job", + "step", + "command", + "test_target", + "changed_paths" + ], + "properties": { + "repository": {"type": "string", "minLength": 1}, + "ref": {"type": "string", "minLength": 1}, + "commit_sha": {"type": "string", "pattern": "^[0-9a-f]{40}$"}, + "workflow": {"type": "string", "minLength": 1}, + "job": {"type": "string", "minLength": 1}, + "step": {"type": "string", "minLength": 1}, + "command": {"type": "string", "minLength": 1}, + "test_target": {"type": ["string", "null"]}, + "changed_paths": { + "type": "array", + "items": {"type": "string", "minLength": 1}, + "uniqueItems": true + } + } + }, + "temporal": { + "type": "object", + "additionalProperties": false, + "required": [ + "workflow_run_id", + "workflow_run_attempt", + "observed_at", + "predecessor_observation_id" + ], + "properties": { + "workflow_run_id": {"type": "string", "pattern": "^[0-9]+$"}, + "workflow_run_attempt": {"type": "integer", "minimum": 1}, + "observed_at": {"type": "string", "format": "date-time"}, + "predecessor_observation_id": { + "type": ["string", "null"], + "pattern": "^obs-[0-9a-f]{64}$" + } + } + }, + "causal": { + "type": "object", + "additionalProperties": false, + "required": ["conclusion", "reason_code", "failure_signature", "cause_state"], + "properties": { + "conclusion": { + "enum": ["success", "failure", "cancelled", "skipped", "unknown"] + }, + "reason_code": {"type": ["string", "null"]}, + "failure_signature": { + "oneOf": [ + {"type": "null"}, + { + "type": "object", + "additionalProperties": false, + "required": ["algorithm", "digest", "basis"], + "properties": { + "algorithm": {"const": "sha256"}, + "digest": {"type": "string", "pattern": "^[0-9a-f]{64}$"}, + "basis": {"type": "string", "minLength": 1} + } + } + ] + }, + "cause_state": { + "enum": ["unconfirmed", "hypothesis", "confirmed"] + } + } + }, + "evidence": { + "type": "object", + "additionalProperties": false, + "required": ["command_sha256", "detector_versions", "environment_fingerprint"], + "properties": { + "command_sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"}, + "detector_versions": { + "type": "object", + "additionalProperties": {"type": "string"} + }, + "environment_fingerprint": { + "type": "object", + "additionalProperties": false, + "required": ["runner_os", "runner_arch", "python_version"], + "properties": { + "runner_os": {"type": ["string", "null"]}, + "runner_arch": {"type": ["string", "null"]}, + "python_version": {"type": ["string", "null"]} + } + } + } + }, + "learning": { + "type": "object", + "additionalProperties": false, + "required": ["confidence", "proposal_allowed", "automatic_mutation_allowed"], + "properties": { + "confidence": { + "enum": [ + "observed_once", + "repeated", + "reproduced", + "fix_correlated", + "fix_validated", + "regression_protected" + ] + }, + "proposal_allowed": {"const": true}, + "automatic_mutation_allowed": {"const": false} + } + }, + "limitations": { + "type": "array", + "items": {"type": "string", "minLength": 1}, + "uniqueItems": true + }, + "authority": { + "type": "object", + "additionalProperties": false, + "required": [ + "mode", + "ownership", + "approval", + "execution", + "delivery", + "deployment", + "merge" + ], + "properties": { + "mode": {"const": "advisory_only"}, + "ownership": {"const": false}, + "approval": {"const": false}, + "execution": {"const": false}, + "delivery": {"const": false}, + "deployment": {"const": false}, + "merge": {"const": false} + } + } + } +} diff --git a/standards/lotus-family/conformance/lotus_family_ci_memory.py b/standards/lotus-family/conformance/lotus_family_ci_memory.py new file mode 100644 index 00000000..3e9fa98f --- /dev/null +++ b/standards/lotus-family/conformance/lotus_family_ci_memory.py @@ -0,0 +1,301 @@ +"""Aggregate immutable Lotus CI observations into advisory causal memory.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import sys +from collections import defaultdict +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Iterable, Mapping + +from lotus_family_ci_observation import AUTHORITY_GRANTS, validate_observation + +MEMORY_SCHEMA_VERSION = "lotus.ci_causal_memory.v0.1" +GRAPH_ID = "ci-causal-memory-v0.1" + + +def _canonical(value: object) -> str: + """Return canonical JSON for deterministic equality and hashing.""" + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + + +def _digest(parts: Iterable[object]) -> str: + """Hash a canonical tuple for stable proposal identities.""" + return hashlib.sha256(_canonical(list(parts)).encode("utf-8")).hexdigest() + + +def _observation_sort_key(observation: Mapping[str, Any]) -> tuple[object, ...]: + """Order observations by timestamp, run identity, and immutable ID.""" + temporal = observation["temporal"] + observed_at = datetime.fromisoformat( + str(temporal["observed_at"]).replace("Z", "+00:00") + ).astimezone(timezone.utc) + return ( + observed_at, + int(temporal["workflow_run_id"]), + temporal["workflow_run_attempt"], + observation["observation_id"], + ) + + +def _context_key(observation: Mapping[str, Any]) -> tuple[str, ...]: + """Return the stable spatial context used for temporal linking.""" + spatial = observation["spatial"] + return ( + spatial["repository"], + spatial["workflow"], + spatial["job"], + spatial["step"], + spatial["command"], + ) + + +def _proposal(signature: Mapping[str, Any]) -> dict[str, Any]: + """Create a deterministic advisory regression proposal for a recurrence.""" + digest = str(signature["digest"]) + proposal_id = f"proposal-{_digest(('add_regression_fixture', digest))}" + count = int(signature["occurrence_count"]) + first_seen = str(signature["first_seen_in"]) + last_seen = str(signature["last_seen_in"]) + title = f"Add regression guard for repeated CI signature {digest[:12]}" + body = ( + "A normalized CI failure signature recurred across immutable Lotus " + f"observations.\n\n- Signature: `{digest}`\n" + f"- Occurrences: {count}\n" + f"- First seen: `{first_seen}`\n" + f"- Last seen: `{last_seen}`\n\n" + "Suggested human action: reproduce the failure, identify a confirmed " + "cause, and add or strengthen a regression fixture. Correlation alone " + "does not confirm causality. This proposal cannot modify CI, approve, " + "merge, deploy, or close findings." + ) + return { + "proposal_id": proposal_id, + "kind": "add_regression_fixture", + "signature_digest": digest, + "status": "advisory", + "title": title, + "body": body, + "human_acceptance_required": True, + "automatic_mutation_allowed": False, + } + + +def aggregate_observations( + observations: Iterable[Mapping[str, Any]], +) -> dict[str, Any]: + """Build deterministic cross-run memory without confirming causal claims.""" + unique: dict[str, dict[str, Any]] = {} + for raw in observations: + normalized = json.loads(_canonical(raw)) + validate_observation(normalized) + observation_id = normalized["observation_id"] + existing = unique.get(observation_id) + if existing is not None and _canonical(existing) != _canonical(normalized): + raise ValueError(f"conflicting observation payload: {observation_id}") + unique[observation_id] = normalized + + ordered = sorted(unique.values(), key=_observation_sort_key) + temporal_edges: list[dict[str, str]] = [] + validation_links: list[dict[str, str]] = [] + + by_context: dict[tuple[str, ...], list[dict[str, Any]]] = defaultdict(list) + for observation in ordered: + by_context[_context_key(observation)].append(observation) + + for context in sorted(by_context): + sequence = by_context[context] + for previous, current in zip(sequence, sequence[1:], strict=False): + temporal_edges.append( + { + "source": previous["observation_id"], + "target": current["observation_id"], + "relation": "preceded_by", + } + ) + previous_conclusion = previous["causal"]["conclusion"] + current_conclusion = current["causal"]["conclusion"] + if previous_conclusion != "success" and current_conclusion == "success": + validation_links.append( + { + "source": previous["observation_id"], + "target": current["observation_id"], + "relation": "success_after_failure", + "state": "fix_correlated", + "cause_state": "unconfirmed", + } + ) + + signature_groups: dict[str, list[dict[str, Any]]] = defaultdict(list) + for observation in ordered: + signature = observation["causal"]["failure_signature"] + if signature is None: + continue + signature_groups[signature["digest"]].append(observation) + + signatures: list[dict[str, Any]] = [] + for digest in sorted(signature_groups): + group = signature_groups[digest] + observation_ids = [item["observation_id"] for item in group] + signatures.append( + { + "digest": digest, + "algorithm": "sha256", + "basis": group[0]["causal"]["failure_signature"]["basis"], + "occurrence_count": len(group), + "state": "repeated" if len(group) > 1 else "observed_once", + "first_seen_in": observation_ids[0], + "last_seen_in": observation_ids[-1], + "observation_ids": observation_ids, + "reason_codes": sorted( + { + item["causal"]["reason_code"] + for item in group + if item["causal"]["reason_code"] is not None + } + ), + "cause_state": "unconfirmed", + } + ) + + proposals = [_proposal(row) for row in signatures if row["state"] == "repeated"] + memory = { + "schema_version": MEMORY_SCHEMA_VERSION, + "graph_id": GRAPH_ID, + "observation_count": len(ordered), + "observation_ids": [row["observation_id"] for row in ordered], + "temporal_edges": temporal_edges, + "validation_links": validation_links, + "signatures": signatures, + "proposals": proposals, + "learning": { + "highest_confidence": ( + "repeated" if any(row["state"] == "repeated" for row in signatures) + else "observed_once" + if signatures + else "no_failure_observed" + ), + "confirmed_cause_count": 0, + "automatic_mutation_allowed": False, + }, + "limitations": [ + "correlation_does_not_confirm_cause", + "trusted_materialization_not_established", + "human_acceptance_required_for_proposals", + ], + "authority": { + "mode": "advisory_only", + **{grant: False for grant in AUTHORITY_GRANTS}, + }, + } + validate_memory(memory) + return memory + + +def validate_memory(memory: Mapping[str, Any]) -> None: + """Validate cross-run learning and authority invariants.""" + if memory.get("schema_version") != MEMORY_SCHEMA_VERSION: + raise ValueError("memory schema_version mismatch") + if memory.get("graph_id") != GRAPH_ID: + raise ValueError("memory graph_id mismatch") + observation_ids = memory.get("observation_ids") + if not isinstance(observation_ids, list) or len(observation_ids) != len( + set(observation_ids) + ): + raise ValueError("memory observation_ids must be a unique list") + if memory.get("observation_count") != len(observation_ids): + raise ValueError("memory observation_count mismatch") + + signatures = memory.get("signatures") + proposals = memory.get("proposals") + if not isinstance(signatures, list) or not isinstance(proposals, list): + raise ValueError("memory signatures and proposals must be lists") + for signature in signatures: + if signature.get("cause_state") != "unconfirmed": + raise ValueError("aggregated correlation cannot confirm a cause") + count = signature.get("occurrence_count") + state = signature.get("state") + if not isinstance(count, int) or count < 1: + raise ValueError("signature occurrence_count is invalid") + expected = "repeated" if count > 1 else "observed_once" + if state != expected: + raise ValueError("signature state does not match occurrence_count") + + for proposal in proposals: + if proposal.get("status") != "advisory": + raise ValueError("proposal status must be advisory") + if proposal.get("human_acceptance_required") is not True: + raise ValueError("proposal requires human acceptance") + if proposal.get("automatic_mutation_allowed") is not False: + raise ValueError("proposal cannot mutate CI automatically") + + learning = memory.get("learning") + if not isinstance(learning, Mapping): + raise ValueError("memory learning must be a mapping") + if learning.get("confirmed_cause_count") != 0: + raise ValueError("aggregator cannot confirm causes") + if learning.get("automatic_mutation_allowed") is not False: + raise ValueError("memory cannot mutate CI automatically") + + authority = memory.get("authority") + if not isinstance(authority, Mapping) or authority.get("mode") != "advisory_only": + raise ValueError("memory authority mode must be advisory_only") + for grant in AUTHORITY_GRANTS: + if authority.get(grant) is not False: + raise ValueError(f"memory authority grant must remain false: {grant}") + + +def _load_observations(paths: Iterable[Path]) -> list[Mapping[str, Any]]: + """Load observation objects from JSON files.""" + observations: list[Mapping[str, Any]] = [] + for path in paths: + payload = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(payload, Mapping): + raise ValueError(f"observation file must contain an object: {path}") + observations.append(payload) + return observations + + +def main(argv: list[str] | None = None) -> int: + """Aggregate observation files and optionally render draft issue proposals.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--observation", type=Path, action="append", required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--proposal-dir", type=Path) + args = parser.parse_args(argv) + + try: + memory = aggregate_observations(_load_observations(args.observation)) + except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc: + print(f"cannot aggregate CI causal memory: {exc}", file=sys.stderr) + return 2 + + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text( + json.dumps(memory, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + + if args.proposal_dir is not None: + args.proposal_dir.mkdir(parents=True, exist_ok=True) + for proposal in memory["proposals"]: + path = args.proposal_dir / f"{proposal['proposal_id']}.md" + path.write_text( + f"# {proposal['title']}\n\n{proposal['body']}\n", + encoding="utf-8", + ) + + print(json.dumps(memory, ensure_ascii=False, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/standards/lotus-family/conformance/lotus_family_ci_observation.py b/standards/lotus-family/conformance/lotus_family_ci_observation.py new file mode 100644 index 00000000..b86a4ea8 --- /dev/null +++ b/standards/lotus-family/conformance/lotus_family_ci_observation.py @@ -0,0 +1,375 @@ +"""Emit deterministic, advisory CI causal observations for Lotus Family runs.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import sys +from datetime import datetime, timezone +from pathlib import Path, PurePosixPath +from typing import Any, Iterable, Mapping + +SCHEMA_VERSION = "lotus.ci_causal_observation.v0.1" +EMITTER_VERSION = "0.1" +ALLOWED_CONCLUSIONS = {"success", "failure", "cancelled", "skipped", "unknown"} +AUTHORITY_GRANTS = ( + "ownership", + "approval", + "execution", + "delivery", + "deployment", + "merge", +) +_HEX_40 = re.compile(r"^[0-9a-f]{40}$") +_HEX_64 = re.compile(r"^[0-9a-f]{64}$") +_OBSERVATION_ID = re.compile(r"^obs-[0-9a-f]{64}$") + + +def _required_text(value: object, field: str) -> str: + """Return a stripped non-empty string or raise a stable validation error.""" + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{field} must be a non-empty string") + return value.strip() + + +def _commit_sha(value: object) -> str: + """Normalize and validate an exact 40-character commit SHA claim.""" + normalized = _required_text(value, "commit_sha").lower() + if not _HEX_40.fullmatch(normalized): + raise ValueError("commit_sha must be exactly 40 hexadecimal characters") + return normalized + + +def _timestamp(value: object) -> str: + """Validate an RFC3339 timestamp with an explicit timezone.""" + text = _required_text(value, "observed_at") + candidate = text[:-1] + "+00:00" if text.endswith("Z") else text + try: + parsed = datetime.fromisoformat(candidate) + except ValueError as exc: + raise ValueError("observed_at must be an RFC3339 timestamp") from exc + if parsed.tzinfo is None: + raise ValueError("observed_at must include a timezone") + return text + + +def _safe_paths(values: Iterable[str]) -> list[str]: + """Normalize repository-relative changed paths without traversal.""" + normalized: set[str] = set() + for raw in values: + value = _required_text(raw, "changed_path").replace("\\", "/") + path = PurePosixPath(value) + if path.is_absolute() or ".." in path.parts or path == PurePosixPath("."): + raise ValueError(f"changed_path escapes repository: {value}") + normalized.add(path.as_posix()) + return sorted(normalized) + + +def _digest(parts: Iterable[object]) -> str: + """Hash a canonical JSON tuple without timestamps or secret-bearing logs.""" + payload = json.dumps( + list(parts), + ensure_ascii=False, + separators=(",", ":"), + ).encode("utf-8") + return hashlib.sha256(payload).hexdigest() + + +def _optional_text(value: object) -> str | None: + """Return a stripped optional string.""" + if value is None: + return None + if not isinstance(value, str): + raise ValueError("optional text fields must be strings or null") + stripped = value.strip() + return stripped or None + + +def build_observation( + *, + repository: str, + ref: str, + commit_sha: str, + workflow: str, + workflow_run_id: str, + workflow_run_attempt: int, + job: str, + step: str, + command: str, + test_target: str | None, + conclusion: str, + reason_code: str | None, + changed_paths: Iterable[str], + runner_os: str | None, + runner_arch: str | None, + python_version: str | None, + observed_at: str, + predecessor_observation_id: str | None = None, +) -> dict[str, Any]: + """Build one immutable observation with separate run and failure identities.""" + repository_value = _required_text(repository, "repository") + ref_value = _required_text(ref, "ref") + sha_value = _commit_sha(commit_sha) + workflow_value = _required_text(workflow, "workflow") + run_id_value = _required_text(workflow_run_id, "workflow_run_id") + if not run_id_value.isdigit(): + raise ValueError("workflow_run_id must contain only digits") + if not isinstance(workflow_run_attempt, int) or workflow_run_attempt < 1: + raise ValueError("workflow_run_attempt must be a positive integer") + job_value = _required_text(job, "job") + step_value = _required_text(step, "step") + command_value = _required_text(command, "command") + conclusion_value = _required_text(conclusion, "conclusion").lower() + if conclusion_value not in ALLOWED_CONCLUSIONS: + raise ValueError(f"unsupported conclusion: {conclusion_value}") + reason_value = _optional_text(reason_code) + target_value = _optional_text(test_target) + observed_value = _timestamp(observed_at) + changed_values = _safe_paths(changed_paths) + + predecessor_value = _optional_text(predecessor_observation_id) + if predecessor_value is not None and not _OBSERVATION_ID.fullmatch( + predecessor_value + ): + raise ValueError("predecessor_observation_id is invalid") + + observation_digest = _digest( + ( + SCHEMA_VERSION, + repository_value, + ref_value, + sha_value, + workflow_value, + run_id_value, + workflow_run_attempt, + job_value, + step_value, + ) + ) + + failure_signature: dict[str, str] | None = None + if conclusion_value != "success": + failure_signature = { + "algorithm": "sha256", + "digest": _digest( + ( + SCHEMA_VERSION, + repository_value, + workflow_value, + job_value, + step_value, + command_value, + conclusion_value, + reason_value or "", + ) + ), + "basis": ( + "schema_version|repository|workflow|job|step|command|" + "conclusion|reason_code" + ), + } + + limitations = [ + "repository_ref_and_commit_are_platform_or_caller_claims", + "raw_logs_are_not_persisted", + "cross_run_cause_is_not_confirmed", + ] + if not changed_values: + limitations.append("changed_paths_unavailable") + if predecessor_value is None: + limitations.append("predecessor_not_linked") + + observation = { + "schema_version": SCHEMA_VERSION, + "observation_id": f"obs-{observation_digest}", + "spatial": { + "repository": repository_value, + "ref": ref_value, + "commit_sha": sha_value, + "workflow": workflow_value, + "job": job_value, + "step": step_value, + "command": command_value, + "test_target": target_value, + "changed_paths": changed_values, + }, + "temporal": { + "workflow_run_id": run_id_value, + "workflow_run_attempt": workflow_run_attempt, + "observed_at": observed_value, + "predecessor_observation_id": predecessor_value, + }, + "causal": { + "conclusion": conclusion_value, + "reason_code": reason_value, + "failure_signature": failure_signature, + "cause_state": "unconfirmed", + }, + "evidence": { + "command_sha256": hashlib.sha256( + command_value.encode("utf-8") + ).hexdigest(), + "detector_versions": { + "lotus_ci_observation": EMITTER_VERSION, + "lotus_workflow_policy": "v3", + }, + "environment_fingerprint": { + "runner_os": _optional_text(runner_os), + "runner_arch": _optional_text(runner_arch), + "python_version": _optional_text(python_version), + }, + }, + "learning": { + "confidence": "observed_once", + "proposal_allowed": True, + "automatic_mutation_allowed": False, + }, + "limitations": sorted(limitations), + "authority": { + "mode": "advisory_only", + **{grant: False for grant in AUTHORITY_GRANTS}, + }, + } + validate_observation(observation) + return observation + + +def validate_observation(observation: Mapping[str, Any]) -> None: + """Validate the invariants required before an observation can be emitted.""" + if observation.get("schema_version") != SCHEMA_VERSION: + raise ValueError("schema_version mismatch") + observation_id = observation.get("observation_id") + if not isinstance(observation_id, str) or not _OBSERVATION_ID.fullmatch( + observation_id + ): + raise ValueError("observation_id is invalid") + + spatial = observation.get("spatial") + temporal = observation.get("temporal") + causal = observation.get("causal") + evidence = observation.get("evidence") + learning = observation.get("learning") + authority = observation.get("authority") + if not all( + isinstance(section, Mapping) + for section in (spatial, temporal, causal, evidence, learning, authority) + ): + raise ValueError("observation sections must be mappings") + + assert isinstance(spatial, Mapping) + if not _HEX_40.fullmatch(str(spatial.get("commit_sha", ""))): + raise ValueError("spatial.commit_sha is invalid") + _safe_paths(spatial.get("changed_paths", [])) + + assert isinstance(temporal, Mapping) + if not str(temporal.get("workflow_run_id", "")).isdigit(): + raise ValueError("temporal.workflow_run_id is invalid") + if not isinstance(temporal.get("workflow_run_attempt"), int): + raise ValueError("temporal.workflow_run_attempt is invalid") + _timestamp(temporal.get("observed_at")) + + assert isinstance(causal, Mapping) + conclusion = causal.get("conclusion") + if conclusion not in ALLOWED_CONCLUSIONS: + raise ValueError("causal.conclusion is invalid") + signature = causal.get("failure_signature") + if conclusion == "success" and signature is not None: + raise ValueError("successful observations cannot have a failure signature") + if conclusion != "success": + if not isinstance(signature, Mapping) or not _HEX_64.fullmatch( + str(signature.get("digest", "")) + ): + raise ValueError("non-success observations require a failure signature") + if causal.get("cause_state") != "unconfirmed": + raise ValueError("single-run observations cannot confirm a cause") + + assert isinstance(evidence, Mapping) + if not _HEX_64.fullmatch(str(evidence.get("command_sha256", ""))): + raise ValueError("evidence.command_sha256 is invalid") + + assert isinstance(learning, Mapping) + if learning.get("confidence") != "observed_once": + raise ValueError("single-run confidence must be observed_once") + if learning.get("proposal_allowed") is not True: + raise ValueError("advisory proposals must remain available") + if learning.get("automatic_mutation_allowed") is not False: + raise ValueError("automatic mutation is forbidden") + + assert isinstance(authority, Mapping) + if authority.get("mode") != "advisory_only": + raise ValueError("authority mode must be advisory_only") + for grant in AUTHORITY_GRANTS: + if authority.get(grant) is not False: + raise ValueError(f"authority grant must remain false: {grant}") + + +def _now() -> str: + """Return an RFC3339 UTC timestamp at second precision.""" + return datetime.now(timezone.utc).isoformat(timespec="seconds").replace( + "+00:00", "Z" + ) + + +def main(argv: list[str] | None = None) -> int: + """Write one schema-valid CI causal observation JSON artifact.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repository", required=True) + parser.add_argument("--ref", required=True) + parser.add_argument("--commit-sha", required=True) + parser.add_argument("--workflow", required=True) + parser.add_argument("--workflow-run-id", required=True) + parser.add_argument("--workflow-run-attempt", type=int, required=True) + parser.add_argument("--job", required=True) + parser.add_argument("--step", required=True) + parser.add_argument("--command", required=True) + parser.add_argument("--test-target") + parser.add_argument("--conclusion", required=True) + parser.add_argument("--reason-code") + parser.add_argument("--changed-path", action="append", default=[]) + parser.add_argument("--runner-os") + parser.add_argument("--runner-arch") + parser.add_argument("--python-version") + parser.add_argument("--observed-at", default=None) + parser.add_argument("--predecessor-observation-id") + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args(argv) + + try: + observation = build_observation( + repository=args.repository, + ref=args.ref, + commit_sha=args.commit_sha, + workflow=args.workflow, + workflow_run_id=args.workflow_run_id, + workflow_run_attempt=args.workflow_run_attempt, + job=args.job, + step=args.step, + command=args.command, + test_target=args.test_target, + conclusion=args.conclusion, + reason_code=args.reason_code, + changed_paths=args.changed_path, + runner_os=args.runner_os, + runner_arch=args.runner_arch, + python_version=args.python_version, + observed_at=args.observed_at or _now(), + predecessor_observation_id=args.predecessor_observation_id, + ) + except ValueError as exc: + print(f"invalid CI causal observation: {exc}", file=sys.stderr) + return 2 + + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text( + json.dumps(observation, ensure_ascii=False, indent=2, sort_keys=True) + + "\n", + encoding="utf-8", + ) + print(json.dumps(observation, ensure_ascii=False, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/standards/lotus-family/conformance/test_ci_memory.py b/standards/lotus-family/conformance/test_ci_memory.py new file mode 100644 index 00000000..87114375 --- /dev/null +++ b/standards/lotus-family/conformance/test_ci_memory.py @@ -0,0 +1,273 @@ +"""Regression coverage for cross-run CI causal memory learning.""" + +from __future__ import annotations + +import copy +import json +import tempfile +import unittest +from pathlib import Path + +from lotus_family_ci_memory import ( + GRAPH_ID, + MEMORY_SCHEMA_VERSION, + aggregate_observations, + main, + validate_memory, +) +from lotus_family_ci_observation import AUTHORITY_GRANTS, build_observation + +HERE = Path(__file__).resolve().parent +GRAPH_PATH = HERE.parent / "ci-memory" / "ci-causal-memory-graph-v0.1.json" +COMMAND = "python -m unittest discover -s standards/lotus-family/conformance -p 'test_*.py' -v" + + +def observation( + *, + run_id: str, + commit: str, + observed_at: str, + conclusion: str, + reason_code: str | None = None, + attempt: int = 1, +) -> dict: + return build_observation( + repository="safal207/pythiaLabs", + ref="agent/ci-causal-memory-v0-1", + commit_sha=commit, + workflow="Lotus Family conformance", + workflow_run_id=run_id, + workflow_run_attempt=attempt, + job="conformance", + step="Run Lotus Family conformance suite", + command=COMMAND, + test_target="standards/lotus-family/conformance/test_*.py", + conclusion=conclusion, + reason_code=reason_code, + changed_paths=[], + runner_os="Linux", + runner_arch="X64", + python_version="3.12.13", + observed_at=observed_at, + ) + + +class CiMemoryTest(unittest.TestCase): + """Keep recurrence learning deterministic, advisory, and graph-bound.""" + + def test_graph_routes_use_existing_nodes_and_edges(self) -> None: + graph = json.loads(GRAPH_PATH.read_text(encoding="utf-8")) + self.assertEqual(graph["graph_id"], GRAPH_ID) + node_ids = [row[0] for row in graph["nodes"]] + edge_ids = [row[0] for row in graph["edges"]] + self.assertEqual(len(node_ids), len(set(node_ids))) + self.assertEqual(len(edge_ids), len(set(edge_ids))) + + node_set = set(node_ids) + edge_pairs = {(row[1], row[2]) for row in graph["edges"]} + for edge in graph["edges"]: + self.assertIn(edge[1], node_set) + self.assertIn(edge[2], node_set) + for trajectory in graph["trajectories"]: + path = trajectory["path"] + self.assertGreaterEqual(len(path), 2) + for source, target in zip(path, path[1:], strict=False): + self.assertIn( + (source, target), + edge_pairs, + msg=f"route gap in {trajectory['id']}: {source} -> {target}", + ) + + invariants = graph["invariants"] + self.assertTrue(invariants["correlation_is_not_confirmation"]) + self.assertFalse(invariants["automatic_mutation_allowed"]) + self.assertTrue(invariants["human_acceptance_required"]) + + def test_repeated_signature_creates_one_advisory_proposal(self) -> None: + first = observation( + run_id="29660131336", + commit="a" * 40, + observed_at="2026-07-18T20:39:58Z", + conclusion="failure", + reason_code="UNITTEST_FAILURE", + ) + second = observation( + run_id="29660140000", + commit="b" * 40, + observed_at="2026-07-18T20:45:00Z", + conclusion="failure", + reason_code="UNITTEST_FAILURE", + ) + + memory = aggregate_observations([second, first]) + + self.assertEqual(memory["schema_version"], MEMORY_SCHEMA_VERSION) + self.assertEqual(memory["observation_count"], 2) + self.assertEqual(len(memory["signatures"]), 1) + signature = memory["signatures"][0] + self.assertEqual(signature["occurrence_count"], 2) + self.assertEqual(signature["state"], "repeated") + self.assertEqual(signature["cause_state"], "unconfirmed") + self.assertEqual(len(memory["proposals"]), 1) + proposal = memory["proposals"][0] + self.assertEqual(proposal["status"], "advisory") + self.assertTrue(proposal["human_acceptance_required"]) + self.assertFalse(proposal["automatic_mutation_allowed"]) + self.assertIn(signature["digest"], proposal["body"]) + self.assertEqual(memory["learning"]["confirmed_cause_count"], 0) + validate_memory(memory) + + def test_failure_then_success_is_only_fix_correlated(self) -> None: + failure = observation( + run_id="29660131336", + commit="a" * 40, + observed_at="2026-07-18T20:39:58Z", + conclusion="failure", + reason_code="IMPORT_ERROR", + ) + success = observation( + run_id="29660150000", + commit="c" * 40, + observed_at="2026-07-18T20:50:00Z", + conclusion="success", + ) + + memory = aggregate_observations([success, failure]) + + self.assertEqual(len(memory["temporal_edges"]), 1) + self.assertEqual(len(memory["validation_links"]), 1) + link = memory["validation_links"][0] + self.assertEqual(link["relation"], "success_after_failure") + self.assertEqual(link["state"], "fix_correlated") + self.assertEqual(link["cause_state"], "unconfirmed") + self.assertEqual(memory["learning"]["confirmed_cause_count"], 0) + self.assertEqual(len(memory["proposals"]), 0) + + def test_mixed_offsets_are_ordered_by_instant(self) -> None: + later = observation( + run_id="2", + commit="b" * 40, + observed_at="2026-07-18T10:00:00-05:00", + conclusion="success", + ) + earlier = observation( + run_id="1", + commit="a" * 40, + observed_at="2026-07-18T14:30:00Z", + conclusion="success", + ) + + memory = aggregate_observations([later, earlier]) + + self.assertEqual( + memory["observation_ids"], + [earlier["observation_id"], later["observation_id"]], + ) + self.assertEqual( + memory["temporal_edges"], + [ + { + "source": earlier["observation_id"], + "target": later["observation_id"], + "relation": "preceded_by", + } + ], + ) + + def test_distinct_signatures_do_not_create_recurrence(self) -> None: + first = observation( + run_id="1", + commit="a" * 40, + observed_at="2026-07-18T20:00:00Z", + conclusion="failure", + reason_code="IMPORT_ERROR", + ) + second = observation( + run_id="2", + commit="b" * 40, + observed_at="2026-07-18T20:01:00Z", + conclusion="failure", + reason_code="ASSERTION_FAILURE", + ) + + memory = aggregate_observations([first, second]) + + self.assertEqual(len(memory["signatures"]), 2) + self.assertTrue( + all(row["state"] == "observed_once" for row in memory["signatures"]) + ) + self.assertEqual(memory["proposals"], []) + + def test_duplicate_observation_is_deduplicated_and_conflict_rejected(self) -> None: + first = observation( + run_id="1", + commit="a" * 40, + observed_at="2026-07-18T20:00:00Z", + conclusion="failure", + reason_code="IMPORT_ERROR", + ) + memory = aggregate_observations([first, copy.deepcopy(first)]) + self.assertEqual(memory["observation_count"], 1) + + conflicting = copy.deepcopy(first) + conflicting["limitations"].append("conflicting_copy") + with self.assertRaisesRegex(ValueError, "conflicting observation payload"): + aggregate_observations([first, conflicting]) + + def test_memory_preserves_full_authority_boundary(self) -> None: + memory = aggregate_observations([]) + self.assertEqual(memory["authority"]["mode"], "advisory_only") + for grant in AUTHORITY_GRANTS: + self.assertFalse(memory["authority"][grant]) + self.assertFalse(memory["learning"]["automatic_mutation_allowed"]) + self.assertEqual(memory["learning"]["confirmed_cause_count"], 0) + + def test_cli_writes_memory_and_draft_proposal(self) -> None: + first = observation( + run_id="1", + commit="a" * 40, + observed_at="2026-07-18T20:00:00Z", + conclusion="failure", + reason_code="UNITTEST_FAILURE", + ) + second = observation( + run_id="2", + commit="b" * 40, + observed_at="2026-07-18T20:01:00Z", + conclusion="failure", + reason_code="UNITTEST_FAILURE", + ) + + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + first_path = root / "first.json" + second_path = root / "second.json" + output = root / "memory.json" + proposal_dir = root / "proposals" + first_path.write_text(json.dumps(first), encoding="utf-8") + second_path.write_text(json.dumps(second), encoding="utf-8") + + exit_code = main( + [ + "--observation", + str(second_path), + "--observation", + str(first_path), + "--output", + str(output), + "--proposal-dir", + str(proposal_dir), + ] + ) + memory = json.loads(output.read_text(encoding="utf-8")) + proposal_files = sorted(proposal_dir.glob("proposal-*.md")) + + self.assertEqual(exit_code, 0) + self.assertEqual(memory["observation_count"], 2) + self.assertEqual(len(memory["proposals"]), 1) + self.assertEqual(len(proposal_files), 1) + validate_memory(memory) + + +if __name__ == "__main__": + unittest.main() diff --git a/standards/lotus-family/conformance/test_ci_observation.py b/standards/lotus-family/conformance/test_ci_observation.py new file mode 100644 index 00000000..acd31bdc --- /dev/null +++ b/standards/lotus-family/conformance/test_ci_observation.py @@ -0,0 +1,184 @@ +"""Regression coverage for immutable CI causal observation artifacts.""" + +from __future__ import annotations + +import json +import tempfile +import unittest +from pathlib import Path + +from lotus_family_ci_observation import ( + AUTHORITY_GRANTS, + SCHEMA_VERSION, + build_observation, + main, + validate_observation, +) + +HERE = Path(__file__).resolve().parent +SCHEMA_PATH = HERE.parent / "ci-memory" / "ci-causal-observation-v0.1.schema.json" +SHA_A = "a" * 40 +SHA_B = "b" * 40 +OBSERVED_AT = "2026-07-18T20:30:00Z" +COMMAND = ( + "python -m unittest discover -s standards/lotus-family/conformance " + "-p 'test_*.py' -v" +) + + +def observation(**overrides: object) -> dict: + values: dict[str, object] = { + "repository": "safal207/pythiaLabs", + "ref": "refs/pull/233/merge", + "commit_sha": SHA_A, + "workflow": "Lotus Family conformance", + "workflow_run_id": "29660000000", + "workflow_run_attempt": 1, + "job": "conformance", + "step": "Run Lotus Family conformance suite", + "command": COMMAND, + "test_target": "standards/lotus-family/conformance/test_*.py", + "conclusion": "success", + "reason_code": None, + "changed_paths": [ + "standards/lotus-family/conformance/test_ci_observation.py", + "standards/lotus-family/ci-memory/SCHEMA.md", + ], + "runner_os": "Linux", + "runner_arch": "X64", + "python_version": "3.12.4", + "observed_at": OBSERVED_AT, + "predecessor_observation_id": None, + } + values.update(overrides) + return build_observation(**values) # type: ignore[arg-type] + + +class CiObservationTest(unittest.TestCase): + """Protect deterministic identity, causal uncertainty, and authority limits.""" + + def test_schema_declares_closed_required_sections(self) -> None: + schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) + self.assertEqual( + schema["properties"]["schema_version"]["const"], + SCHEMA_VERSION, + ) + self.assertFalse(schema["additionalProperties"]) + self.assertEqual( + set(schema["required"]), + { + "schema_version", + "observation_id", + "spatial", + "temporal", + "causal", + "evidence", + "learning", + "limitations", + "authority", + }, + ) + + def test_success_observation_is_deterministic_and_advisory(self) -> None: + first = observation() + second = observation(changed_paths=reversed(first["spatial"]["changed_paths"])) + + self.assertEqual(first, second) + self.assertIsNone(first["causal"]["failure_signature"]) + self.assertEqual(first["causal"]["cause_state"], "unconfirmed") + self.assertEqual(first["learning"]["confidence"], "observed_once") + self.assertFalse(first["learning"]["automatic_mutation_allowed"]) + self.assertEqual(first["authority"]["mode"], "advisory_only") + for grant in AUTHORITY_GRANTS: + self.assertFalse(first["authority"][grant]) + validate_observation(first) + + def test_failure_signature_repeats_across_run_and_commit_identity(self) -> None: + first = observation( + conclusion="failure", + reason_code="UNITTEST_FAILURE", + ) + second = observation( + commit_sha=SHA_B, + workflow_run_id="29660000001", + workflow_run_attempt=2, + observed_at="2026-07-18T20:35:00Z", + conclusion="failure", + reason_code="UNITTEST_FAILURE", + ) + + self.assertNotEqual(first["observation_id"], second["observation_id"]) + self.assertEqual( + first["causal"]["failure_signature"], + second["causal"]["failure_signature"], + ) + self.assertEqual(first["causal"]["cause_state"], "unconfirmed") + self.assertEqual(second["causal"]["cause_state"], "unconfirmed") + + def test_failure_signature_changes_with_reason_code(self) -> None: + first = observation(conclusion="failure", reason_code="IMPORT_ERROR") + second = observation(conclusion="failure", reason_code="ASSERTION_FAILURE") + self.assertNotEqual( + first["causal"]["failure_signature"]["digest"], + second["causal"]["failure_signature"]["digest"], + ) + + def test_rejects_unsafe_changed_paths_and_invalid_commit(self) -> None: + with self.assertRaisesRegex(ValueError, "escapes repository"): + observation(changed_paths=["../secret.txt"]) + with self.assertRaisesRegex(ValueError, "40 hexadecimal"): + observation(commit_sha="main") + + def test_cli_writes_one_valid_observation(self) -> None: + with tempfile.TemporaryDirectory() as directory: + output = Path(directory) / "observation.json" + exit_code = main( + [ + "--repository", + "safal207/pythiaLabs", + "--ref", + "refs/pull/233/merge", + "--commit-sha", + SHA_A, + "--workflow", + "Lotus Family conformance", + "--workflow-run-id", + "29660000000", + "--workflow-run-attempt", + "1", + "--job", + "conformance", + "--step", + "Run Lotus Family conformance suite", + "--command", + COMMAND, + "--test-target", + "standards/lotus-family/conformance/test_*.py", + "--conclusion", + "failure", + "--reason-code", + "UNITTEST_FAILURE", + "--runner-os", + "Linux", + "--runner-arch", + "X64", + "--python-version", + "3.12.4", + "--observed-at", + OBSERVED_AT, + "--output", + str(output), + ] + ) + payload = json.loads(output.read_text(encoding="utf-8")) + + self.assertEqual(exit_code, 0) + self.assertEqual(payload["schema_version"], SCHEMA_VERSION) + self.assertEqual(payload["causal"]["conclusion"], "failure") + self.assertIsNotNone(payload["causal"]["failure_signature"]) + self.assertIn("changed_paths_unavailable", payload["limitations"]) + validate_observation(payload) + + +if __name__ == "__main__": + unittest.main()