OpenSource/ECC
1
0 Fork 0
mirror of https://github.com/affaan-m/ECC.git synced 2026-11-07 08:35:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: validate skill contribution observations and split compiler

Browse Source
This commit is contained in:
djpjronline-netizen committed 2026-10-16 10:30:32 -04:00
1 parent 786f5fb654
commit d8ac98144f
22 files changed
+1055 -684
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.agents/skills/ecc-contribution-review/SKILL.md
+4 -3
View File
@@ -1,13 +1,14 @@
---
name: ecc-contribution-review
description: Prepare an ECC repository contribution for review by organizing repository, install-surface, cross-harness, validation, security, and commit observations into a local proposal. Use when preparing a skill, agent, command, hook, rule, or script contribution for review.
description: Prepare an ECC skill contribution for review by organizing repository, install-surface, cross-harness, validation, security, and commit observations into a local proposal. Use when preparing a skill contribution for review.
metadata:
origin: ECC
---
# ECC Contribution Review Packet
# ECC Skill Contribution Review Packet
Use this workflow to prepare contribution observations for maintainer review.
Use this workflow to prepare skill contribution observations for maintainer review.
This version supports `changeKind: "skill"` with canonical source under `skills/`.
It compiles supplied observations; it does not independently inspect Git,
files, CI, test results, coverage, or referenced evidence. Treat every imported
claim as unverified until a reviewer checks its evidence.
.agents/skills/ecc-contribution-review/agents/openai.yaml
+3 -3
View File
@@ -1,7 +1,7 @@
interface:
display_name: "ECC Contribution Review"
short_description: "Prepare evidence-aware ECC contribution review packets"
display_name: "ECC Skill Contribution Review"
short_description: "Prepare evidence-aware ECC skill review packets"
brand_color: "#0EA5E9"
default_prompt: "Use $ecc-contribution-review to prepare an ECC contribution review packet from current observations and evidence."
default_prompt: "Use $ecc-contribution-review to prepare an ECC skill contribution review packet from current observations and evidence."
policy:
allow_implicit_invocation: true
.agents/skills/ecc-conventions/SKILL.md
+6 -5
View File
@@ -442,13 +442,14 @@ Based on analysis of the codebase, follow these practices:
*This skill was auto-generated by [ECC Tools](https://ecc.tools). Review and customize as needed for your team.*
## Cross-harness contribution review workflow
## Cross-harness skill contribution review workflow
For changes that touch ECC skills, agents, commands, hooks, rules, scripts,
catalogs, or install surfaces, follow the canonical
For ECC skill contributions, including their related tests, catalogs, and
install surfaces, follow the canonical
[`ecc-contribution-review` skill](../../../skills/ecc-contribution-review/SKILL.md).
Its compiler checks the structure and consistency of supplied observations; it
does not independently verify Git state, files, CI, tests, coverage, or
This version supports only `changeKind: "skill"`. Its compiler checks the
structure and consistency of supplied observations; it does not independently
verify Git state, files, CI, tests, coverage, or
referenced evidence. Treat its output as a review packet, not proof of
contribution readiness.
.cursor/skills/ecc-contribution-review/SKILL.md
+4 -3
View File
@@ -1,13 +1,14 @@
---
name: ecc-contribution-review
description: Prepare an ECC repository contribution for review by organizing repository, install-surface, cross-harness, validation, security, and commit observations into a local proposal. Use when preparing a skill, agent, command, hook, rule, or script contribution for review.
description: Prepare an ECC skill contribution for review by organizing repository, install-surface, cross-harness, validation, security, and commit observations into a local proposal. Use when preparing a skill contribution for review.
metadata:
origin: ECC
---
# ECC Contribution Review Packet
# ECC Skill Contribution Review Packet
Use this workflow to prepare contribution observations for maintainer review.
Use this workflow to prepare skill contribution observations for maintainer review.
This version supports `changeKind: "skill"` with canonical source under `skills/`.
It compiles supplied observations; it does not independently inspect Git,
files, CI, test results, coverage, or referenced evidence. Treat every imported
claim as unverified until a reviewer checks its evidence.
docs/COMMAND-AGENT-MAP.md
+1 -1
View File
@@ -80,7 +80,7 @@ This document lists each slash command and the primary agent(s) or skills it inv
| Skill | Purpose | Entry point |
|-------|---------|-------------|
| [ecc-contribution-review](../skills/ecc-contribution-review/SKILL.md) | Prepare a review packet for an ECC contribution from current repository observations | Invoke the skill, then verify every imported claim and evidence reference independently. |
| [ecc-contribution-review](../skills/ecc-contribution-review/SKILL.md) | Prepare a review packet for an ECC skill contribution from current repository observations | Invoke the skill, then verify every imported claim and evidence reference independently. |
| [osint-investigation](../skills/osint-investigation/SKILL.md) | Public-source investigation methodology: competing hypotheses, falsification, visual verification, and evidence tracking | Invoke the self-contained skill directly using available host capabilities. |
## How to use this map
manifests/context-packs/skill-triggers@1.json
+7 -7
View File
@@ -3,14 +3,14 @@
"skills": 312,
"withTriggers": 34
},
"generatedAt": "2026-10-15T21:40:37.005Z",
"generatedAt": "2026-10-16T14:28:25.401Z",
"id": "skill-triggers@1",
"model": {
"effort": null,
"id": "hand-seeded",
"source": "manual-curation-pending-regeneration"
},
"registryDigest": "d38a54e7c1a1b22075c20e8935d0d5bf527e6e188335c405f31a1261a2c7f2cd",
"registryDigest": "6562670f1b4bfe1ab99b91c0511f633401278130b517e4e12fb085d17ac103c0",
"schemaVersion": 1,
"triggers": {
"skill:agent-runtime-gateway-smoke-test": [
@@ -137,10 +137,10 @@
"multi service orchestration"
],
"skill:ecc-contribution-review": [
"prepare ecc contribution for review",
"organize contribution security evidence",
"review cross-harness contribution observations",
"compile ecc contribution proposal"
"prepare ecc skill contribution for review",
"organize skill contribution security evidence",
"review cross-harness skill contribution observations",
"compile ecc skill contribution proposal"
],
"skill:error-handling": [
"error types",
@@ -297,5 +297,5 @@
"before creating a pr"
]
},
"triggersDigest": "21715491ba8c845552ff2e9a60da98948beb63fcf86d7e6a9ce6f8782976e942"
"triggersDigest": "47baf9408ee9f662289036228151a1aebcee296dd14ab12bad41168d1ccdce87"
}
manifests/install-components.json
+1 -1
View File
@@ -505,7 +505,7 @@
{
"id": "skill:ecc-contribution-review",
"family": "skill",
"description": "Optional contribution review packet workflow for ECC repositories.",
"description": "Optional skill contribution review packet workflow for ECC repositories.",
"modules": [
"ecc-contribution-review"
]
manifests/install-modules.json
+1 -1
View File
@@ -1423,7 +1423,7 @@
{
"id": "ecc-contribution-review",
"kind": "skills",
"description": "Optional contribution review packet workflow for ECC repositories.",
"description": "Optional skill contribution review packet workflow for ECC repositories.",
"paths": [
"skills/ecc-contribution-review"
],
skills/ecc-contribution-review/SKILL.md
+4 -3
View File
@@ -1,13 +1,14 @@
---
name: ecc-contribution-review
description: Prepare an ECC repository contribution for review by organizing repository, install-surface, cross-harness, validation, security, and commit observations into a local proposal. Use when preparing a skill, agent, command, hook, rule, or script contribution for review.
description: Prepare an ECC skill contribution for review by organizing repository, install-surface, cross-harness, validation, security, and commit observations into a local proposal. Use when preparing a skill contribution for review.
metadata:
origin: ECC
---
# ECC Contribution Review Packet
# ECC Skill Contribution Review Packet
Use this workflow to prepare contribution observations for maintainer review.
Use this workflow to prepare skill contribution observations for maintainer review.
This version supports `changeKind: "skill"` with canonical source under `skills/`.
It compiles supplied observations; it does not independently inspect Git,
files, CI, test results, coverage, or referenced evidence. Treat every imported
claim as unverified until a reviewer checks its evidence.
skills/ecc-contribution-review/agents/openai.yaml
+3 -3
View File
@@ -1,7 +1,7 @@
interface:
display_name: "ECC Contribution Review"
short_description: "Prepare evidence-aware ECC contribution review packets"
display_name: "ECC Skill Contribution Review"
short_description: "Prepare evidence-aware ECC skill review packets"
brand_color: "#0EA5E9"
default_prompt: "Use $ecc-contribution-review to prepare an ECC contribution review packet from current observations and evidence."
default_prompt: "Use $ecc-contribution-review to prepare an ECC skill contribution review packet from current observations and evidence."
policy:
allow_implicit_invocation: true
skills/ecc-contribution-review/docs/DESIGN.md
+5 -4
View File
@@ -1,8 +1,9 @@
# ECC cross-harness contribution review
# ECC cross-harness skill contribution review
This compiler converts supplied observations into a structured contribution
review packet. It checks the declared source ownership, generated-root
exclusion, install coverage, cross-harness identity, catalog reconciliation,
This compiler converts supplied skill observations into a structured contribution
review packet. It supports only `changeKind: "skill"`, with canonical source
under `skills/` and the required skill artifacts and harness mirrors. It checks
the declared source ownership, generated-root exclusion, install coverage, cross-harness identity, catalog reconciliation,
TDD claims, security claims, validation gates, and commit proposal for internal
consistency. It does not inspect the repository or independently verify any
claim or referenced evidence, so it cannot establish contribution readiness.
skills/ecc-contribution-review/docs/SECURITY-EVIDENCE.md
+10 -7
View File
@@ -1,25 +1,28 @@
# Contribution review security evidence
Recorded 2026-10-15 against the source hashes in
Recorded 2026-10-16 against the source hashes in
`../tests/security-evidence/source-sha256.json`.
The scanner output is `../tests/security-evidence/bandit.json`.
These are observed local results, not an upstream CI or maintainer approval.
## Scope and results
- Bandit 1.9.4: compiler source scanned, zero findings, zero skipped tests,
- Bandit 1.9.4: both compiler modules scanned, zero findings, zero skipped tests,
no `nosec` suppressions, exit 0.
- Python 3.14.3, pytest 9.1.1, pytest-cov 7.1.0: 30 tests passed,
including 12 focused security tests; compiler branch-enabled combined
coverage 93.64%, above the 80% gate.
- Python 3.14.3, pytest 9.1.1, pytest-cov 7.1.0: 38 tests passed,
including 12 focused security tests and 8 review regression tests; compiler branch-enabled combined
coverage 95.78%, above the 80% gate.
- Ruff 0.17.1: changed Python source and tests passed.
- Mypy 2.4.0: compiler passed.
- Workflow security validator: 13 workflows passed.
Security regressions exercise rejected authority escalation, malformed input,
non-finite JSON numbers, invalid UTF-8, existing-file and symlink preservation,
malformed failure probes, non-skill labels, envelope values, impossible numeric
claims, function/module size policy, non-finite JSON numbers, invalid UTF-8, existing-file and symlink preservation,
output receipt accuracy, canonical/parity record agreement, and non-execution
of supplied runner argv. Source hashes are verified by a security regression in CI; stale evidence fails the job.
of supplied runner argv. Source hashes for both compiler modules, all three test modules, and canonical
agent metadata are verified by a security regression in CI; stale evidence fails
the job.
Both new agent metadata files are checked for identical
content and absence of extra tool, dependency, or permission configuration.
The metadata retains ECC's existing implicit-invocation convention.
skills/ecc-contribution-review/docs/WORKFLOW.md
+3 -2
View File
@@ -1,8 +1,9 @@
# Workflow
# Skill contribution workflow
1. Resolve the active ECC checkout, non-default branch, base/head, package lock,
manifests, catalog script, placement policy, and checkpoint.
2. Classify the change and name its canonical surface before editing a mirror.
2. Use `changeKind: "skill"` and name the canonical source under `skills/`
before editing a mirror. This version supports skill contributions only.
3. Observe RED with named failing behaviors.
4. Implement GREEN in the canonical surface, then REFACTOR.
5. Update only required install manifests, documentation, and harness adapters.
skills/ecc-contribution-review/references/contribution-review-contract.md
+13 -6
View File
@@ -1,23 +1,30 @@
# ECC contribution review contract
# ECC skill contribution review contract
The compiler can only prepare a packet for review. It validates the shape and
internal consistency of supplied observations; it does not independently
The compiler prepares packets for skill contributions only
(`changeKind: "skill"`). It validates the shape and internal consistency of
supplied observations; it does not independently
inspect repository files, Git state, CI, tests, coverage, or evidence refs.
Every structurally valid packet returns `require_review`. Contradictions
return `reject`. Neither result approves or qualifies a contribution.
The envelope requires `schemaVersion: "1.0.0"`, an ISO 8601 `observedAt`
timestamp with a timezone, and a nonempty string `subjectRef`. Unsupported
versions and malformed identifiers produce validation faults.
The local observation contract checks for:
- exact active repository, non-default branch, base/head, package lock, install
manifests, catalog script, placement policy, and checkpoint identity;
- a bounded, non-breaking, skills-first change with no unnecessary command shim;
- a bounded, non-breaking skill change under `skills/` with no unnecessary
command shim;
- one canonical curated source, explicit non-generated artifacts, manifest
coverage, tests, and required harness metadata;
- normalized content parity across Claude, Codex, and Cursor with zero drift;
- filesystem/documented counts, manifest coverage, intentional unshipped
exceptions, catalog check, and command registry reconciled;
exceptions, catalog check, and command registry reconciled, with nonnegative
curated and intentionally unshipped skill counts;
- observed RED/GREEN/REFACTOR, named failing tests, passing behavior tests, and
line/function/branch/statement coverage at or above 80%;
line/function/branch/statement coverage between 80% and 100% inclusive;
- eight green validators, security and supply-chain review, bounded change/test
budgets, twelve hostile probes, and seven typed runners;
- a concise conventional commit proposal, while commit/push/PR effects remain
skills/ecc-contribution-review/scripts/compile_contribution_review.py
+183 -627
View File
@@ -6,67 +6,21 @@ from __future__ import annotations
import argparse
import hashlib
import json
import math
import re
import sys
from pathlib import Path, PurePosixPath
from pathlib import Path
from typing import Any
ROLES = {
"canonical-skill", "install-manifest",
"agents-mirror", "cursor-mirror",
"codex-metadata", "unit-test",
}
HARNESSES = {"claude", "codex", "cursor"}
GATES = {
"strict-skill-frontmatter", "install-manifests",
"catalog-sync", "command-registry",
"harness-adapters", "unicode-safety",
"no-personal-paths", "test-and-coverage",
}
FAILURES = {
"generated-root-as-canonical", "missing-manifest-entry",
"harness-content-drift", "catalog-count-drift",
"stale-command-registry", "missing-red-test",
"coverage-regression", "secret-in-diff",
"personal-path-in-diff", "unsafe-hook-permission",
"breaking-change-unmarked", "commit-before-gates",
}
FAILURE_OUTCOMES: dict[Any, str] = {
"generated-root-as-canonical": "reject", "missing-manifest-entry": "hold",
"harness-content-drift": "hold", "catalog-count-drift": "hold",
"stale-command-registry": "hold", "missing-red-test": "hold",
"coverage-regression": "hold", "secret-in-diff": "reject",
"personal-path-in-diff": "reject", "unsafe-hook-permission": "reject",
"breaking-change-unmarked": "hold", "commit-before-gates": "reject",
}
RUNNERS = {
"resolve-canonical-surface", "compile-change-plan",
"run-red-green-refactor", "reconcile-harness-parity",
"reconcile-catalog-and-manifests", "run-security-and-validation-gates",
"prepare-commit-proposal",
}
AUTHORITY_FIELDS = {
"mayWriteSourceFiles", "mayExecuteTests",
"mayUpdateCatalog",
"mayCommit",
"mayPush",
"mayOpenPullRequest",
"mayWriteControlStore",
"mayWriteCanonicalMemory",
"maySemanticCommit",
}
EFFECT_FIELDS = {
"sourceFileWriteRequested",
"testExecutionRequested",
"catalogWriteRequested",
"commitRequested",
"pushRequested",
"pullRequestRequested",
"controlStoreWriteRequested",
"canonicalMemoryWriteRequested",
"semanticCommitRequested",
}
from contribution_review_validation import (
AUTHORITY_FIELDS,
is_sha256,
validate,
)
from contribution_review_validation import (
RUNNERS as RUNNERS,
)
from contribution_review_validation import (
safe_repo_path as safe_repo_path,
)
def canonical(value: Any) -> Any:
@@ -89,538 +43,25 @@ def checksum(value: Any) -> str:
return hashlib.sha256(encoded).hexdigest()
def is_sha256(value: Any) -> bool:
return isinstance(value, str) and re.fullmatch(r"[0-9a-f]{64}", value) is not None
def is_finite_number(value: Any) -> bool:
"""Accept finite JSON numbers while excluding booleans."""
if isinstance(value, bool):
return False
if isinstance(value, int):
return True
return isinstance(value, float) and math.isfinite(value)
def reject_nonstandard_json_number(value: str) -> None:
"""Reject Python json's non-standard NaN and Infinity extensions."""
raise ValueError(f"non-standard JSON number: {value}")
def safe_repo_path(value: Any) -> bool:
"""Accept normalized relative POSIX paths, rejecting home/drive/traversal refs."""
if not isinstance(value, str) or not value or "\\" in value or "\x00" in value:
return False
if value.startswith("~") or re.match(r"^[A-Za-z]:", value):
return False
if any(part in {"", ".", ".."} for part in value.split("/")):
return False
path = PurePosixPath(value)
return not path.is_absolute() and path.as_posix() == value
def mapping(data: dict[str, Any], field: str) -> dict[str, Any]:
value = data.get(field, {})
return value if isinstance(value, dict) else {}
def add_issue(issues: list[dict[str, Any]], code: str, plane: int, detail: str) -> None:
if code not in {issue["code"] for issue in issues}:
issues.append({"code": code, "plane": plane, "detail": detail})
def sequence(data: dict[str, Any], field: str) -> list[Any]:
value = data.get(field, [])
return value if isinstance(value, list) else []
def _validate(observation: Any) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
faults: list[dict[str, Any]] = []
contradictions: list[dict[str, Any]] = []
required = {
"schemaVersion", "observedAt", "subjectRef", "repoIdentity", "changeIntent",
"surfaceContract", "artifactRecords", "harnessParity", "catalogContract",
"validationMatrix", "tddContract", "securityContract", "budgetContract",
"commitContract", "failureProbes", "checkpointContract", "runnerProposals",
"claimsSourceFilesWrittenByController", "claimsTestsExecutedByController",
"claimsCatalogUpdatedByController", "claimsCommitCreated", "claimsPushPerformed",
"claimsPullRequestOpened", "requestedEffects", "claimedAuthority",
}
if not isinstance(observation, dict) or not required.issubset(observation):
add_issue(faults, "observation_envelope_incomplete", 1,
"Required ECC contribution observations are missing.")
return faults, contradictions
map_fields = (
"repoIdentity", "changeIntent", "surfaceContract", "harnessParity",
"catalogContract", "tddContract", "securityContract", "budgetContract",
"commitContract", "checkpointContract", "requestedEffects", "claimedAuthority",
)
list_fields = ("artifactRecords", "validationMatrix", "failureProbes", "runnerProposals")
if any(not isinstance(observation[field], dict) for field in map_fields) or any(
not isinstance(observation[field], list) for field in list_fields
):
add_issue(faults, "observation_shape_invalid", 1,
"Declared maps and arrays must retain their documented shapes.")
return faults, contradictions
repo = observation["repoIdentity"]
repo_fields = {
"rootName", "rootPathRef", "branch", "baseSha256", "headSha256",
"packageLockSha256", "installModulesSha256", "installProfilesSha256",
"installComponentsSha256", "catalogScriptSha256", "skillPolicySha256",
"checkpointSha256",
}
if (
set(repo) != repo_fields
or repo.get("rootName") != "everything-claude-code"
or repo.get("rootPathRef") != "active-checkout:everything-claude-code"
or not isinstance(repo.get("branch"), str)
or not repo.get("branch")
or repo.get("branch") in {"main", "master"}
or any(not is_sha256(repo.get(key)) for key in repo_fields if key.endswith("Sha256"))
or repo.get("baseSha256") == repo.get("headSha256")
):
add_issue(faults, "repo_identity_failed", 2,
"Active ECC root, non-default branch, distinct base/head, manifests, catalog, policy, and checkpoint hashes are required.")
intent = observation["changeIntent"]
intent_fields = {
"changeId", "changeKind", "canonicalSurface", "canonicalPath",
"legacyCommandShimRequired", "breakingChange", "owner", "scopes",
"evidenceRefs", "counterevidenceRefs",
}
scopes = intent.get("scopes")
if (
set(intent) != intent_fields
or not isinstance(intent.get("changeId"), str)
or not intent.get("changeId")
or intent.get("changeKind") not in {"skill", "agent", "command", "hook", "rule", "script"}
or intent.get("canonicalSurface") != "skills"
or not safe_repo_path(intent.get("canonicalPath"))
or not intent.get("canonicalPath", "").startswith("skills/")
or intent.get("legacyCommandShimRequired") is not False
or intent.get("breakingChange") is not False
or intent.get("owner") != "ECC"
or not isinstance(scopes, list)
or set(scopes) != {"skills", "tests", "manifests", "cross-harness"}
or not isinstance(intent.get("evidenceRefs"), list)
or not intent.get("evidenceRefs")
or not isinstance(intent.get("counterevidenceRefs"), list)
):
add_issue(faults, "change_intent_failed", 1,
"A bounded non-breaking skills-first change with explicit scopes and evidence refs is required.")
surface = observation["surfaceContract"]
surface_fields = {
"canonicalFirst", "generatedRootsExcluded", "curatedSkill",
"installManifestUpdated", "commandShimAdded", "provenanceRequired",
"mirrorTargets", "docsUpdateRequired", "evidenceRefs", "counterevidenceRefs",
}
mirror_targets = surface.get("mirrorTargets")
if (
set(surface) != surface_fields
or surface.get("canonicalFirst") is not True
or surface.get("generatedRootsExcluded") is not True
or surface.get("curatedSkill") is not True
or surface.get("installManifestUpdated") is not True
or surface.get("commandShimAdded") is not False
or surface.get("provenanceRequired") is not False
or not isinstance(mirror_targets, list)
or set(mirror_targets) != {"agents", "cursor"}
or surface.get("docsUpdateRequired") is not True
or not isinstance(surface.get("evidenceRefs"), list)
or not surface.get("evidenceRefs")
):
add_issue(faults, "canonical_surface_failed", 6,
"Curated skills/ must lead; generated roots must be excluded; manifest, docs, and intentional mirrors are required.")
artifacts = observation["artifactRecords"]
artifact_fields = {
"path", "role", "sha256", "exists", "sourceOfTruth", "generated",
"contentIdentitySha256", "evidenceRefs",
}
artifact_paths: list[Any] = []
artifact_bad = False
for artifact in artifacts:
if not isinstance(artifact, dict):
artifact_bad = True
continue
artifact_paths.append(artifact.get("path"))
source_of_truth = artifact.get("role") == "canonical-skill"
if (
set(artifact) != artifact_fields
or not safe_repo_path(artifact.get("path"))
or artifact.get("role") not in ROLES
or not is_sha256(artifact.get("sha256"))
or artifact.get("exists") is not True
or artifact.get("sourceOfTruth") is not source_of_truth
or artifact.get("generated") is not False
or not is_sha256(artifact.get("contentIdentitySha256"))
or not isinstance(artifact.get("evidenceRefs"), list)
or not artifact.get("evidenceRefs")
):
artifact_bad = True
canonical_artifacts = [
item for item in artifacts
if isinstance(item, dict) and item.get("role") == "canonical-skill"
]
if (
len(artifacts) != 6
or {item.get("role") for item in artifacts if isinstance(item, dict)} != ROLES
or len(artifact_paths) != len(set(artifact_paths))
or len(canonical_artifacts) != 1
or canonical_artifacts[0].get("path") != intent.get("canonicalPath")
or artifact_bad
):
add_issue(faults, "artifact_graph_failed", 2,
"Six unique, extant, hash-bound artifacts with one exact canonical source are required.")
parity = observation["harnessParity"]
parity_fields = {
"canonicalContentSha256", "targets", "parityPassed", "driftIds",
"evidenceRefs", "counterevidenceRefs",
}
target_fields = {
"harness", "path", "contentIdentitySha256", "frontmatterAdapted",
"metadataPresent", "evidenceRefs",
}
targets = parity.get("targets", [])
canonical_content = parity.get("canonicalContentSha256")
artifact_by_path = {
item.get("path"): item
for item in artifacts
if isinstance(item, dict)
}
expected_roles: dict[Any, str] = {
"claude": "canonical-skill",
"codex": "agents-mirror",
"cursor": "cursor-mirror",
}
parity_artifacts_bad = any(
not isinstance(item, dict)
or item.get("path") not in artifact_by_path
or artifact_by_path[item.get("path")].get("role") != expected_roles.get(item.get("harness"))
or artifact_by_path[item.get("path")].get("contentIdentitySha256")
!= item.get("contentIdentitySha256")
for item in targets
) if isinstance(targets, list) else True
if (
set(parity) != parity_fields
or not is_sha256(canonical_content)
or not isinstance(targets, list)
or len(targets) != 3
or {item.get("harness") for item in targets if isinstance(item, dict)} != HARNESSES
or any(
not isinstance(item, dict)
or set(item) != target_fields
or not safe_repo_path(item.get("path"))
or item.get("contentIdentitySha256") != canonical_content
or not isinstance(item.get("frontmatterAdapted"), bool)
or item.get("metadataPresent") is not True
or not isinstance(item.get("evidenceRefs"), list)
or not item.get("evidenceRefs")
for item in targets
)
or parity.get("parityPassed") is not True
or parity.get("driftIds") != []
or not isinstance(parity.get("evidenceRefs"), list)
or not parity.get("evidenceRefs")
or parity_artifacts_bad
):
add_issue(faults, "harness_parity_failed", 3,
"Claude, Codex, and Cursor surfaces must retain normalized content identity with explicit adaptations and no drift.")
catalog = observation["catalogContract"]
catalog_fields = {
"filesystemAgents", "documentedAgents", "filesystemSkills", "documentedSkills",
"filesystemCommands", "documentedCommands", "installModuleCount",
"curatedSkillsReferenced", "intentionallyUnshippedSkills", "catalogCheckPassed",
"commandRegistryCheckPassed", "manifestValidationPassed", "countsReconciled",
"evidenceRefs", "counterevidenceRefs",
}
count_pairs = (
("filesystemAgents", "documentedAgents"),
("filesystemSkills", "documentedSkills"),
("filesystemCommands", "documentedCommands"),
)
counts_valid = all(
isinstance(catalog.get(left), int)
and not isinstance(catalog.get(left), bool)
and catalog.get(left) > 0
and catalog.get(left) == catalog.get(right)
for left, right in count_pairs
)
skills_valid = all(
isinstance(catalog.get(key), int) and not isinstance(catalog.get(key), bool)
for key in ("curatedSkillsReferenced", "intentionallyUnshippedSkills", "filesystemSkills")
)
if (
set(catalog) != catalog_fields
or not counts_valid
or not isinstance(catalog.get("installModuleCount"), int)
or isinstance(catalog.get("installModuleCount"), bool)
or catalog.get("installModuleCount", 0) <= 0
or not skills_valid
or catalog.get("curatedSkillsReferenced", 0) + catalog.get("intentionallyUnshippedSkills", 0) != catalog.get("filesystemSkills")
or any(catalog.get(key) is not True for key in (
"catalogCheckPassed", "commandRegistryCheckPassed", "manifestValidationPassed", "countsReconciled"
))
or not isinstance(catalog.get("evidenceRefs"), list)
or not catalog.get("evidenceRefs")
):
add_issue(faults, "catalog_reconciliation_failed", 3,
"Filesystem/documented counts, manifest coverage, intentional exceptions, catalog, and registry checks must reconcile.")
validation = observation["validationMatrix"]
validation_fields = {"gateId", "commandArgv", "expectedExit", "observedExit", "evidenceRefs"}
if (
len(validation) != 8
or {item.get("gateId") for item in validation if isinstance(item, dict)} != GATES
or any(
not isinstance(item, dict)
or set(item) != validation_fields
or not isinstance(item.get("commandArgv"), list)
or not item.get("commandArgv")
or not all(isinstance(arg, str) and arg for arg in item.get("commandArgv", []))
or item.get("expectedExit") != 0
or item.get("observedExit") != 0
or not isinstance(item.get("evidenceRefs"), list)
or not item.get("evidenceRefs")
for item in validation
)
):
add_issue(faults, "validation_matrix_failed", 5,
"Eight exact argv gates must be recorded green with evidence references.")
tdd = observation["tddContract"]
tdd_fields = {
"redObserved", "failingTestIds", "greenObserved", "refactorObserved",
"coverageThresholdPercent", "lineCoveragePercent", "functionCoveragePercent",
"branchCoveragePercent", "statementCoveragePercent", "newBehaviorTestCount",
"allTestsPass", "evidenceRefs", "counterevidenceRefs",
}
coverage_fields = (
"lineCoveragePercent", "functionCoveragePercent", "branchCoveragePercent",
"statementCoveragePercent",
)
if (
set(tdd) != tdd_fields
or any(tdd.get(key) is not True for key in ("redObserved", "greenObserved", "refactorObserved", "allTestsPass"))
or not isinstance(tdd.get("failingTestIds"), list)
or not tdd.get("failingTestIds")
or tdd.get("coverageThresholdPercent") != 80
or any(not is_finite_number(tdd.get(key)) or tdd.get(key) < 80 for key in coverage_fields)
or not isinstance(tdd.get("newBehaviorTestCount"), int)
or isinstance(tdd.get("newBehaviorTestCount"), bool)
or tdd.get("newBehaviorTestCount", 0) < 3
or not isinstance(tdd.get("evidenceRefs"), list)
or not tdd.get("evidenceRefs")
):
add_issue(faults, "tdd_contract_failed", 4,
"Observed RED/GREEN/REFACTOR, named failures, passing behavior tests, and at least 80 percent coverage are required.")
security = observation["securityContract"]
security_fields = {
"secretsDetected", "personalPathsDetected", "unsafePermissionsDetected",
"hookChangesReviewed", "promptInjectionBoundaryReviewed", "supplyChainChangesReviewed",
"untrustedContentCannotGrantAuthority", "evidenceRefs", "counterevidenceRefs",
}
if (
set(security) != security_fields
or any(security.get(key) is not False for key in (
"secretsDetected", "personalPathsDetected", "unsafePermissionsDetected"
))
or any(security.get(key) is not True for key in (
"hookChangesReviewed", "promptInjectionBoundaryReviewed",
"supplyChainChangesReviewed", "untrustedContentCannotGrantAuthority"
))
or not isinstance(security.get("evidenceRefs"), list)
or not security.get("evidenceRefs")
):
add_issue(faults, "security_contract_failed", 6,
"Secret, path, permission, hook, prompt-injection, supply-chain, and authority observations are required.")
budget = observation["budgetContract"]
budget_fields = {
"maximumChangedFiles", "observedChangedFiles", "maximumChangedLines",
"observedChangedLines", "maximumTestRuntimeSeconds", "observedTestRuntimeSeconds",
"evidenceRefs",
}
budget_pairs = (
("observedChangedFiles", "maximumChangedFiles"),
("observedChangedLines", "maximumChangedLines"),
("observedTestRuntimeSeconds", "maximumTestRuntimeSeconds"),
)
if (
set(budget) != budget_fields
or any(
not is_finite_number(budget.get(actual))
or not is_finite_number(budget.get(limit))
or budget.get(actual) < 0
or budget.get(actual) > budget.get(limit)
for actual, limit in budget_pairs
)
or not isinstance(budget.get("evidenceRefs"), list)
or not budget.get("evidenceRefs")
):
add_issue(faults, "change_budget_failed", 6,
"Changed-file, changed-line, and test-runtime budgets must be nonnegative and bounded.")
commit = observation["commitContract"]
commit_fields = {
"type", "scope", "subject", "subjectLength", "conventional", "commitCreated",
"pushPerformed", "pullRequestOpened", "evidenceRefs", "counterevidenceRefs",
}
subject = commit.get("subject")
if (
set(commit) != commit_fields
or commit.get("type") not in {"feat", "fix", "refactor", "docs", "test", "chore", "perf", "ci"}
or commit.get("scope") != "skills"
or not isinstance(subject, str)
or not subject
or commit.get("subjectLength") != len(subject)
or len(subject) > 72
or commit.get("conventional") is not True
or any(commit.get(key) is not False for key in ("commitCreated", "pushPerformed", "pullRequestOpened"))
or not isinstance(commit.get("evidenceRefs"), list)
or not commit.get("evidenceRefs")
):
add_issue(faults, "commit_contract_failed", 6,
"A concise conventional commit proposal is required; commit, push, and PR effects must remain false.")
probes = observation["failureProbes"]
probe_fields = {"probeId", "failureMode", "expectedOutcome", "observedOutcome", "evidenceRefs"}
if (
len(probes) != 12
or {item.get("failureMode") for item in probes if isinstance(item, dict)} != FAILURES
or len({item.get("probeId") for item in probes if isinstance(item, dict)}) != 12
or any(
not isinstance(item, dict)
or set(item) != probe_fields
or item.get("expectedOutcome") != FAILURE_OUTCOMES.get(item.get("failureMode"))
or item.get("observedOutcome") != item.get("expectedOutcome")
or not isinstance(item.get("evidenceRefs"), list)
or not item.get("evidenceRefs")
for item in probes
)
):
add_issue(faults, "failure_probe_coverage_failed", 5,
"All twelve ECC contribution probes must match expected outcomes and cite evidence.")
checkpoint = observation["checkpointContract"]
checkpoint_fields = {
"checkpointSha256", "repoHeadSha256", "policySha256", "lastDecisionId",
"plannedRunnerCount", "statePathRef", "remainingUnknowns",
}
if (
set(checkpoint) != checkpoint_fields
or checkpoint.get("checkpointSha256") != repo.get("checkpointSha256")
or checkpoint.get("repoHeadSha256") != repo.get("headSha256")
or checkpoint.get("policySha256") != repo.get("skillPolicySha256")
or not checkpoint.get("lastDecisionId")
or checkpoint.get("plannedRunnerCount") != 7
or not isinstance(checkpoint.get("statePathRef"), str)
or not checkpoint.get("statePathRef", "").startswith(".memory-kit/")
or not isinstance(checkpoint.get("remainingUnknowns"), list)
or not checkpoint.get("remainingUnknowns")
):
add_issue(faults, "checkpoint_contract_failed", 5,
"Exact repository, policy, checkpoint, runner, state, and unknown identities are required.")
runners = observation["runnerProposals"]
runner_fields = {"role", "argv"}
if (
len(runners) != 7
or {item.get("role") for item in runners if isinstance(item, dict)} != RUNNERS
or any(
not isinstance(item, dict)
or set(item) != runner_fields
or not isinstance(item.get("argv"), list)
or not item.get("argv")
or not all(isinstance(arg, str) and arg for arg in item.get("argv", []))
for item in runners
)
):
add_issue(faults, "runner_proposals_invalid", 5,
"Seven typed non-shell argv proposals are required.")
forbidden_claims = (
("claimsSourceFilesWrittenByController", "file_write_claim_forbidden"),
("claimsTestsExecutedByController", "test_execution_claim_forbidden"),
("claimsCatalogUpdatedByController", "catalog_write_claim_forbidden"),
("claimsCommitCreated", "commit_claim_forbidden"),
("claimsPushPerformed", "push_claim_forbidden"),
("claimsPullRequestOpened", "pull_request_claim_forbidden"),
)
for field, code in forbidden_claims:
if observation.get(field) is not False:
contradictions.append({
"code": code,
"plane": 2,
"detail": "The compiler cannot claim repository or external effects.",
})
requested_effects = observation["requestedEffects"]
if set(requested_effects) != EFFECT_FIELDS or any(
value is not False for value in requested_effects.values()
):
contradictions.append({
"code": "external_action_requested",
"plane": 6,
"detail": "Compilation cannot request source, test, catalog, Git, PR, or memory effects.",
})
claimed_authority = observation["claimedAuthority"]
if set(claimed_authority) != AUTHORITY_FIELDS or any(
value is not False for value in claimed_authority.values()
):
contradictions.append({
"code": "unsafe_generated_authority",
"plane": 6,
"detail": "Generated authority must remain false.",
})
return faults, contradictions
def validate(observation: Any) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
"""Validate an untrusted JSON value without letting malformed shapes crash the CLI."""
try:
return _validate(observation)
except (AttributeError, TypeError, ValueError):
fault = {
"code": "observation_shape_invalid",
"plane": 1,
"detail": "Observation values must match the documented JSON types.",
}
return [fault], []
def compile_contract(
observation: Any,
source_path: Path,
capsule_file_written: bool = False,
) -> dict[str, Any]:
faults, contradictions = validate(observation)
reasons = sorted({issue["code"] for issue in faults + contradictions})
decision = "reject" if contradictions else "require_review"
outcome = "reject" if decision == "reject" else "review_packet_compiled"
data = observation if isinstance(observation, dict) else {}
def mapping(field: str) -> dict[str, Any]:
value = data.get(field, {})
return value if isinstance(value, dict) else {}
def sequence(field: str) -> list[Any]:
value = data.get(field, [])
return value if isinstance(value, list) else []
artifacts = sequence("artifactRecords")
validation = sequence("validationMatrix")
change_intent = mapping("changeIntent")
repo_identity = mapping("repoIdentity")
checkpoint = mapping("checkpointContract")
output_receipt: dict[str, Any] = {
"sourceFilesWritten": False,
"capsuleFileWritten": capsule_file_written,
"testsExecuted": False,
"catalogUpdated": False,
"commitCreated": False,
"pushPerformed": False,
"pullRequestOpened": False,
"controlStoreWritten": False,
"canonicalMemoryWritten": False,
"semanticCommitted": False,
}
artifact: dict[str, Any] = {
def _packet_identity(data: dict[str, Any]) -> dict[str, Any]:
change_intent = mapping(data, "changeIntent")
repo_identity = mapping(data, "repoIdentity")
return {
"artifactType": "ecc-cross-harness-contribution-review-packet",
"schemaVersion": "1.0.0",
"artifactId": "ecc-contribution-review-packet",
@@ -636,12 +77,27 @@ def compile_contract(
"viability-and-policy",
],
},
"observationEnvelopes": [{
"sourcePathRef": "observation-envelope",
"repoIdentity": repo_identity,
"changeIntent": change_intent,
"surfaceContract": mapping("surfaceContract"),
}],
"observationEnvelopes": [
{
"sourcePathRef": "observation-envelope",
"repoIdentity": repo_identity,
"changeIntent": change_intent,
"surfaceContract": mapping(data, "surfaceContract"),
}
],
}
def _packet_review(
data: dict[str, Any],
faults: list[dict[str, Any]],
contradictions: list[dict[str, Any]],
) -> dict[str, Any]:
change_intent = mapping(data, "changeIntent")
checkpoint = mapping(data, "checkpointContract")
reasons = sorted({issue["code"] for issue in faults + contradictions})
decision = "reject" if contradictions else "require_review"
return {
"authorityEnvelope": {key: False for key in sorted(AUTHORITY_FIELDS)},
"policyEnvelope": {
"proposalOnly": True,
@@ -653,79 +109,130 @@ def compile_contract(
},
"controllerDecision": {
"decision": decision,
"reasonCodes": reasons or ["imported_observations_require_independent_review"],
"reasonCodes": reasons
or ["imported_observations_require_independent_review"],
},
"contradictionScan": "computed",
"contradictions": contradictions,
"semanticPageFaults": faults,
"counterevidence": (
list(change_intent.get("counterevidenceRefs", []))
if isinstance(change_intent.get("counterevidenceRefs"), list)
else []
),
"counterevidence": list(change_intent.get("counterevidenceRefs", []))
if isinstance(change_intent.get("counterevidenceRefs"), list)
else [],
"omissions": [
"The compiler validates supplied claims but does not inspect the repository or referenced evidence.",
"The compiler does not execute tests, update catalogs, commit, push, or open a pull request.",
"A capsule file is written only to a new path explicitly supplied by the caller.",
],
"residualUncertainty": (
list(checkpoint.get("remainingUnknowns", []))
if isinstance(checkpoint.get("remainingUnknowns"), list)
else []
),
"residualUncertainty": list(checkpoint.get("remainingUnknowns", []))
if isinstance(checkpoint.get("remainingUnknowns"), list)
else [],
}
def _packet_payload(
data: dict[str, Any], decision: str, output_receipt: dict[str, Any]
) -> dict[str, Any]:
artifacts = sequence(data, "artifactRecords")
validation = sequence(data, "validationMatrix")
change_intent = mapping(data, "changeIntent")
repo_identity = mapping(data, "repoIdentity")
outcome = "reject" if decision == "reject" else "review_packet_compiled"
return {
"domainPayload": {
"contributionOutcome": outcome,
"assessmentBasis": "unverified imported claims; internal consistency only",
"repoClaims": repo_identity,
"changeClaims": change_intent,
"surfaceClaims": mapping("surfaceContract"),
"surfaceClaims": mapping(data, "surfaceContract"),
"artifactClaims": artifacts,
"harnessParityClaims": mapping("harnessParity"),
"catalogClaims": mapping("catalogContract"),
"harnessParityClaims": mapping(data, "harnessParity"),
"catalogClaims": mapping(data, "catalogContract"),
"validationClaims": validation,
"tddClaims": mapping("tddContract"),
"securityClaims": mapping("securityContract"),
"budgetClaims": mapping("budgetContract"),
"commitClaims": mapping("commitContract"),
"tddClaims": mapping(data, "tddContract"),
"securityClaims": mapping(data, "securityContract"),
"budgetClaims": mapping(data, "budgetContract"),
"commitClaims": mapping(data, "commitContract"),
"failureProbeCoverage": {
item.get("failureMode"): item.get("expectedOutcome") == item.get("observedOutcome")
for item in sequence("failureProbes")
if isinstance(item, dict)
item.get("failureMode"): item.get("expectedOutcome")
== item.get("observedOutcome")
for item in sequence(data, "failureProbes")
if isinstance(item, dict) and isinstance(item.get("failureMode"), str)
},
"runnerProposals": [
{**item, "executed": False, "requiresExternalAuthorization": True}
for item in sequence("runnerProposals")
for item in sequence(data, "runnerProposals")
if isinstance(item, dict)
],
"contributionReceiptProposal": {
"observationImported": True,
**output_receipt,
},
},
}
}
def _packet_evidence(data: dict[str, Any], decision: str) -> dict[str, Any]:
artifacts = sequence(data, "artifactRecords")
validation = sequence(data, "validationMatrix")
repo_identity = mapping(data, "repoIdentity")
return {
"unverifiedEvidenceClaims": {
"verified": False,
"basis": "imported observation envelope",
"repoHeadSha256": repo_identity.get("headSha256"),
"artifactPaths": [item.get("path") for item in artifacts if isinstance(item, dict)],
"artifactSha256": [item.get("sha256") for item in artifacts if isinstance(item, dict)],
"gateIds": [item.get("gateId") for item in validation if isinstance(item, dict)],
"artifactPaths": [
item.get("path") for item in artifacts if isinstance(item, dict)
],
"artifactSha256": [
item.get("sha256") for item in artifacts if isinstance(item, dict)
],
"gateIds": [
item.get("gateId") for item in validation if isinstance(item, dict)
],
},
"epistemicLedger": [
{"event": "repository_and_change_observation_imported", "status": "observed"},
{"event": "canonical_surface_parity_catalog_and_tdd_evaluated", "status": "computed_from_claims"},
{"event": "security_budget_and_commit_contract_evaluated", "status": "computed_from_claims"},
{
"event": "repository_and_change_observation_imported",
"status": "observed",
},
{
"event": "canonical_surface_parity_catalog_and_tdd_evaluated",
"status": "computed_from_claims",
},
{
"event": "security_budget_and_commit_contract_evaluated",
"status": "computed_from_claims",
},
{"event": "contribution_review_packet_compiled", "status": decision},
],
}
def _packet_planes(
data: dict[str, Any],
decision: str,
faults: list[dict[str, Any]],
contradictions: list[dict[str, Any]],
) -> dict[str, Any]:
artifacts = sequence(data, "artifactRecords")
validation = sequence(data, "validationMatrix")
return {
"pccaPlanes": {
"observation-and-data": {"artifactCount": len(artifacts)},
"epistemic-and-lineage": {
"sha256ShapedArtifactClaims": sum(
is_sha256(item.get("sha256")) for item in artifacts if isinstance(item, dict)
is_sha256(item.get("sha256"))
for item in artifacts
if isinstance(item, dict)
)
},
"state-and-relationship": {
"harnessTargetCount": len(mapping("harnessParity").get("targets", []))
if isinstance(mapping("harnessParity").get("targets", []), list)
"harnessTargetCount": len(
mapping(data, "harnessParity").get("targets", [])
)
if isinstance(mapping(data, "harnessParity").get("targets", []), list)
else 0
},
"cognitive-control": {"validationGateCount": len(validation)},
@@ -734,7 +241,14 @@ def compile_contract(
"faultCount": len(faults),
"contradictionCount": len(contradictions),
},
},
}
}
def _packet_memory(data: dict[str, Any]) -> dict[str, Any]:
repo_identity = mapping(data, "repoIdentity")
checkpoint = mapping(data, "checkpointContract")
return {
"memoryMetabolism": {
"checkpointSha256": checkpoint.get("checkpointSha256"),
"repoHeadSha256": repo_identity.get("headSha256"),
@@ -748,7 +262,39 @@ def compile_contract(
"review feedback",
],
"supersessionRequired": True,
},
}
}
def _output_receipt(capsule_file_written: bool) -> dict[str, Any]:
return {
"sourceFilesWritten": False,
"capsuleFileWritten": capsule_file_written,
"testsExecuted": False,
"catalogUpdated": False,
"commitCreated": False,
"pushPerformed": False,
"pullRequestOpened": False,
"controlStoreWritten": False,
"canonicalMemoryWritten": False,
"semanticCommitted": False,
}
def compile_contract(
observation: Any, source_path: Path, capsule_file_written: bool = False
) -> dict[str, Any]:
faults, contradictions = validate(observation)
data = observation if isinstance(observation, dict) else {}
decision = "reject" if contradictions else "require_review"
output_receipt = _output_receipt(capsule_file_written)
artifact = {
**_packet_identity(data),
**_packet_review(data, faults, contradictions),
**_packet_payload(data, decision, output_receipt),
**_packet_evidence(data, decision),
**_packet_planes(data, decision, faults, contradictions),
**_packet_memory(data),
"returnReceipt": output_receipt,
}
artifact["semanticChecksum"] = checksum(artifact)
@@ -759,7 +305,9 @@ def compile_contract(
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("observation", type=Path, help="JSON observation envelope")
parser.add_argument("--output", type=Path, help="write the generated capsule to this path")
parser.add_argument(
"--output", type=Path, help="write the generated capsule to this path"
)
args = parser.parse_args()
if args.output is not None and args.observation.resolve() == args.output.resolve():
print("error: output path must not overwrite the observation envelope", file=sys.stderr)
@@ -772,11 +320,14 @@ def main() -> int:
except (OSError, UnicodeError, ValueError, RecursionError) as error:
print(f"error: cannot read observation JSON: {error}", file=sys.stderr)
return 2
capsule_written = args.output is not None
try:
packet = compile_contract(observation, args.observation, capsule_file_written=capsule_written)
rendered = json.dumps(packet, indent=2, sort_keys=True, ensure_ascii=False) + "\n"
packet = compile_contract(
observation, args.observation, capsule_file_written=capsule_written
)
rendered = (
json.dumps(packet, indent=2, sort_keys=True, ensure_ascii=False) + "\n"
)
rendered.encode("utf-8")
except (UnicodeError, ValueError, RecursionError) as error:
print(f"error: cannot compile observation JSON: {error}", file=sys.stderr)
@@ -788,12 +339,17 @@ def main() -> int:
with args.output.open("x", encoding="utf-8") as output_file:
output_file.write(rendered)
except FileExistsError:
print("error: output path already exists; refusing to overwrite", file=sys.stderr)
print(
"error: output path already exists; refusing to overwrite",
file=sys.stderr,
)
return 2
except OSError as error:
print(f"error: cannot write capsule output: {error}", file=sys.stderr)
return 2
return 0 if not packet["semanticPageFaults"] and not packet["contradictions"] else 2
return (
0 if not packet["semanticPageFaults"] and (not packet["contradictions"]) else 2
)
if __name__ == "__main__":
skills/ecc-contribution-review/scripts/contribution_review_validation.py
+636
View File
@@ -0,0 +1,636 @@
"""Section validators for the skill contribution observation contract."""
from __future__ import annotations
import math
import re
from datetime import datetime
from pathlib import PurePosixPath
from typing import Any
ROLES = {
"canonical-skill", "install-manifest",
"agents-mirror", "cursor-mirror",
"codex-metadata", "unit-test",
}
HARNESSES = {"claude", "codex", "cursor"}
GATES = {
"strict-skill-frontmatter", "install-manifests",
"catalog-sync", "command-registry",
"harness-adapters", "unicode-safety",
"no-personal-paths", "test-and-coverage",
}
FAILURES = {
"generated-root-as-canonical", "missing-manifest-entry",
"harness-content-drift", "catalog-count-drift",
"stale-command-registry", "missing-red-test",
"coverage-regression", "secret-in-diff",
"personal-path-in-diff", "unsafe-hook-permission",
"breaking-change-unmarked", "commit-before-gates",
}
FAILURE_OUTCOMES: dict[Any, str] = {
"generated-root-as-canonical": "reject", "missing-manifest-entry": "hold",
"harness-content-drift": "hold", "catalog-count-drift": "hold",
"stale-command-registry": "hold", "missing-red-test": "hold",
"coverage-regression": "hold", "secret-in-diff": "reject",
"personal-path-in-diff": "reject", "unsafe-hook-permission": "reject",
"breaking-change-unmarked": "hold", "commit-before-gates": "reject",
}
RUNNERS = {
"resolve-canonical-surface", "compile-change-plan",
"run-red-green-refactor", "reconcile-harness-parity",
"reconcile-catalog-and-manifests", "run-security-and-validation-gates",
"prepare-commit-proposal",
}
AUTHORITY_FIELDS = {
"mayWriteSourceFiles", "mayExecuteTests",
"mayUpdateCatalog",
"mayCommit",
"mayPush",
"mayOpenPullRequest",
"mayWriteControlStore",
"mayWriteCanonicalMemory",
"maySemanticCommit",
}
EFFECT_FIELDS = {
"sourceFileWriteRequested",
"testExecutionRequested",
"catalogWriteRequested",
"commitRequested",
"pushRequested",
"pullRequestRequested",
"controlStoreWriteRequested",
"canonicalMemoryWriteRequested",
"semanticCommitRequested",
}
def is_sha256(value: Any) -> bool:
return isinstance(value, str) and re.fullmatch(r"[0-9a-f]{64}", value) is not None
def is_finite_number(value: Any) -> bool:
"""Accept finite JSON numbers while excluding booleans."""
if isinstance(value, bool):
return False
if isinstance(value, int):
return True
return isinstance(value, float) and math.isfinite(value)
def safe_repo_path(value: Any) -> bool:
"""Accept normalized relative POSIX paths, rejecting home/drive/traversal refs."""
if not isinstance(value, str) or not value or "\\" in value or "\x00" in value:
return False
if value.startswith("~") or re.match(r"^[A-Za-z]:", value):
return False
if any(part in {"", ".", ".."} for part in value.split("/")):
return False
path = PurePosixPath(value)
return not path.is_absolute() and path.as_posix() == value
def add_issue(issues: list[dict[str, Any]], code: str, plane: int, detail: str) -> None:
if code not in {issue["code"] for issue in issues}:
issues.append({"code": code, "plane": plane, "detail": detail})
def _validate_repo(observation: dict[str, Any], faults: list[dict[str, Any]]) -> None:
repo = observation["repoIdentity"]
repo_fields = {
"rootName", "rootPathRef", "branch", "baseSha256", "headSha256",
"packageLockSha256", "installModulesSha256", "installProfilesSha256",
"installComponentsSha256", "catalogScriptSha256", "skillPolicySha256",
"checkpointSha256",
}
if (
set(repo) != repo_fields
or repo.get("rootName") != "everything-claude-code"
or repo.get("rootPathRef") != "active-checkout:everything-claude-code"
or not isinstance(repo.get("branch"), str)
or not repo.get("branch")
or repo.get("branch") in {"main", "master"}
or any(not is_sha256(repo.get(key)) for key in repo_fields if key.endswith("Sha256"))
or repo.get("baseSha256") == repo.get("headSha256")
):
add_issue(faults, "repo_identity_failed", 2,
"Active ECC root, non-default branch, distinct base/head, manifests, catalog, policy, and checkpoint hashes are required.")
def _validate_intent(observation: dict[str, Any], faults: list[dict[str, Any]]) -> None:
intent = observation["changeIntent"]
intent_fields = {
"changeId", "changeKind", "canonicalSurface", "canonicalPath",
"legacyCommandShimRequired", "breakingChange", "owner", "scopes",
"evidenceRefs", "counterevidenceRefs",
}
scopes = intent.get("scopes")
if (
set(intent) != intent_fields
or not isinstance(intent.get("changeId"), str)
or not intent.get("changeId")
or intent.get("changeKind") != "skill"
or intent.get("canonicalSurface") != "skills"
or not safe_repo_path(intent.get("canonicalPath"))
or not intent.get("canonicalPath", "").startswith("skills/")
or intent.get("legacyCommandShimRequired") is not False
or intent.get("breakingChange") is not False
or intent.get("owner") != "ECC"
or not isinstance(scopes, list)
or set(scopes) != {"skills", "tests", "manifests", "cross-harness"}
or not isinstance(intent.get("evidenceRefs"), list)
or not intent.get("evidenceRefs")
or not isinstance(intent.get("counterevidenceRefs"), list)
):
add_issue(faults, "change_intent_failed", 1,
"A bounded non-breaking skills-first change with explicit scopes and evidence refs is required.")
def _validate_surface(observation: dict[str, Any], faults: list[dict[str, Any]]) -> None:
surface = observation["surfaceContract"]
surface_fields = {
"canonicalFirst", "generatedRootsExcluded", "curatedSkill",
"installManifestUpdated", "commandShimAdded", "provenanceRequired",
"mirrorTargets", "docsUpdateRequired", "evidenceRefs", "counterevidenceRefs",
}
mirror_targets = surface.get("mirrorTargets")
if (
set(surface) != surface_fields
or surface.get("canonicalFirst") is not True
or surface.get("generatedRootsExcluded") is not True
or surface.get("curatedSkill") is not True
or surface.get("installManifestUpdated") is not True
or surface.get("commandShimAdded") is not False
or surface.get("provenanceRequired") is not False
or not isinstance(mirror_targets, list)
or set(mirror_targets) != {"agents", "cursor"}
or surface.get("docsUpdateRequired") is not True
or not isinstance(surface.get("evidenceRefs"), list)
or not surface.get("evidenceRefs")
):
add_issue(faults, "canonical_surface_failed", 6,
"Curated skills/ must lead; generated roots must be excluded; manifest, docs, and intentional mirrors are required.")
def _validate_artifacts(observation: dict[str, Any], faults: list[dict[str, Any]]) -> None:
intent = observation["changeIntent"]
artifacts = observation["artifactRecords"]
artifact_fields = {
"path", "role", "sha256", "exists", "sourceOfTruth", "generated",
"contentIdentitySha256", "evidenceRefs",
}
artifact_paths: list[Any] = []
artifact_bad = False
for artifact in artifacts:
if not isinstance(artifact, dict):
artifact_bad = True
continue
artifact_paths.append(artifact.get("path"))
source_of_truth = artifact.get("role") == "canonical-skill"
if (
set(artifact) != artifact_fields
or not safe_repo_path(artifact.get("path"))
or artifact.get("role") not in ROLES
or not is_sha256(artifact.get("sha256"))
or artifact.get("exists") is not True
or artifact.get("sourceOfTruth") is not source_of_truth
or artifact.get("generated") is not False
or not is_sha256(artifact.get("contentIdentitySha256"))
or not isinstance(artifact.get("evidenceRefs"), list)
or not artifact.get("evidenceRefs")
):
artifact_bad = True
canonical_artifacts = [
item for item in artifacts
if isinstance(item, dict) and item.get("role") == "canonical-skill"
]
if (
len(artifacts) != 6
or {item.get("role") for item in artifacts if isinstance(item, dict)} != ROLES
or len(artifact_paths) != len(set(artifact_paths))
or len(canonical_artifacts) != 1
or canonical_artifacts[0].get("path") != intent.get("canonicalPath")
or artifact_bad
):
add_issue(faults, "artifact_graph_failed", 2,
"Six unique, extant, hash-bound artifacts with one exact canonical source are required.")
def _parity_targets_valid(targets: list[Any], canonical_content: Any) -> bool:
target_fields = {
"harness", "path", "contentIdentitySha256", "frontmatterAdapted",
"metadataPresent", "evidenceRefs",
}
return not any(
not isinstance(item, dict)
or set(item) != target_fields
or not safe_repo_path(item.get("path"))
or item.get("contentIdentitySha256") != canonical_content
or not isinstance(item.get("frontmatterAdapted"), bool)
or item.get("metadataPresent") is not True
or not isinstance(item.get("evidenceRefs"), list)
or not item.get("evidenceRefs")
for item in targets
)
def _validate_parity(observation: dict[str, Any], faults: list[dict[str, Any]]) -> None:
artifacts = observation["artifactRecords"]
parity = observation["harnessParity"]
parity_fields = {
"canonicalContentSha256", "targets", "parityPassed", "driftIds",
"evidenceRefs", "counterevidenceRefs",
}
targets = parity.get("targets", [])
canonical_content = parity.get("canonicalContentSha256")
artifact_by_path = {
item.get("path"): item
for item in artifacts
if isinstance(item, dict)
}
expected_roles: dict[Any, str] = {
"claude": "canonical-skill",
"codex": "agents-mirror",
"cursor": "cursor-mirror",
}
parity_artifacts_bad = any(
not isinstance(item, dict)
or item.get("path") not in artifact_by_path
or artifact_by_path[item.get("path")].get("role") != expected_roles.get(item.get("harness"))
or artifact_by_path[item.get("path")].get("contentIdentitySha256")
!= item.get("contentIdentitySha256")
for item in targets
) if isinstance(targets, list) else True
if (
set(parity) != parity_fields
or not is_sha256(canonical_content)
or not isinstance(targets, list)
or len(targets) != 3
or {item.get("harness") for item in targets if isinstance(item, dict)} != HARNESSES
or not _parity_targets_valid(targets, canonical_content)
or parity.get("parityPassed") is not True
or parity.get("driftIds") != []
or not isinstance(parity.get("evidenceRefs"), list)
or not parity.get("evidenceRefs")
or parity_artifacts_bad
):
add_issue(faults, "harness_parity_failed", 3,
"Claude, Codex, and Cursor surfaces must retain normalized content identity with explicit adaptations and no drift.")
def _validate_catalog(observation: dict[str, Any], faults: list[dict[str, Any]]) -> None:
catalog = observation["catalogContract"]
catalog_fields = {
"filesystemAgents", "documentedAgents", "filesystemSkills", "documentedSkills",
"filesystemCommands", "documentedCommands", "installModuleCount",
"curatedSkillsReferenced", "intentionallyUnshippedSkills", "catalogCheckPassed",
"commandRegistryCheckPassed", "manifestValidationPassed", "countsReconciled",
"evidenceRefs", "counterevidenceRefs",
}
count_pairs = (
("filesystemAgents", "documentedAgents"),
("filesystemSkills", "documentedSkills"),
("filesystemCommands", "documentedCommands"),
)
counts_valid = all(
isinstance(catalog.get(left), int)
and not isinstance(catalog.get(left), bool)
and catalog.get(left) > 0
and catalog.get(left) == catalog.get(right)
for left, right in count_pairs
)
skills_valid = all(
isinstance(catalog.get(key), int) and not isinstance(catalog.get(key), bool)
and catalog.get(key) >= 0
for key in ("curatedSkillsReferenced", "intentionallyUnshippedSkills", "filesystemSkills")
)
if (
set(catalog) != catalog_fields
or not counts_valid
or not isinstance(catalog.get("installModuleCount"), int)
or isinstance(catalog.get("installModuleCount"), bool)
or catalog.get("installModuleCount", 0) <= 0
or not skills_valid
or catalog.get("curatedSkillsReferenced", 0) + catalog.get("intentionallyUnshippedSkills", 0) != catalog.get("filesystemSkills")
or any(catalog.get(key) is not True for key in (
"catalogCheckPassed", "commandRegistryCheckPassed", "manifestValidationPassed", "countsReconciled"
))
or not isinstance(catalog.get("evidenceRefs"), list)
or not catalog.get("evidenceRefs")
):
add_issue(faults, "catalog_reconciliation_failed", 3,
"Filesystem/documented counts, manifest coverage, intentional exceptions, catalog, and registry checks must reconcile.")
def _validate_gates(observation: dict[str, Any], faults: list[dict[str, Any]]) -> None:
validation = observation["validationMatrix"]
validation_fields = {"gateId", "commandArgv", "expectedExit", "observedExit", "evidenceRefs"}
if (
len(validation) != 8
or {item.get("gateId") for item in validation if isinstance(item, dict)} != GATES
or any(
not isinstance(item, dict)
or set(item) != validation_fields
or not isinstance(item.get("commandArgv"), list)
or not item.get("commandArgv")
or not all(isinstance(arg, str) and arg for arg in item.get("commandArgv", []))
or item.get("expectedExit") != 0
or item.get("observedExit") != 0
or not isinstance(item.get("evidenceRefs"), list)
or not item.get("evidenceRefs")
for item in validation
)
):
add_issue(faults, "validation_matrix_failed", 5,
"Eight exact argv gates must be recorded green with evidence references.")
def _validate_tdd(observation: dict[str, Any], faults: list[dict[str, Any]]) -> None:
tdd = observation["tddContract"]
tdd_fields = {
"redObserved", "failingTestIds", "greenObserved", "refactorObserved",
"coverageThresholdPercent", "lineCoveragePercent", "functionCoveragePercent",
"branchCoveragePercent", "statementCoveragePercent", "newBehaviorTestCount",
"allTestsPass", "evidenceRefs", "counterevidenceRefs",
}
coverage_fields = (
"lineCoveragePercent", "functionCoveragePercent", "branchCoveragePercent",
"statementCoveragePercent",
)
if (
set(tdd) != tdd_fields
or any(tdd.get(key) is not True for key in ("redObserved", "greenObserved", "refactorObserved", "allTestsPass"))
or not isinstance(tdd.get("failingTestIds"), list)
or not tdd.get("failingTestIds")
or tdd.get("coverageThresholdPercent") != 80
or any(not is_finite_number(tdd.get(key)) or not 80 <= tdd.get(key) <= 100 for key in coverage_fields)
or not isinstance(tdd.get("newBehaviorTestCount"), int)
or isinstance(tdd.get("newBehaviorTestCount"), bool)
or tdd.get("newBehaviorTestCount", 0) < 3
or not isinstance(tdd.get("evidenceRefs"), list)
or not tdd.get("evidenceRefs")
):
add_issue(faults, "tdd_contract_failed", 4,
"Observed RED/GREEN/REFACTOR, named failures, passing behavior tests, and at least 80 percent coverage are required.")
def _validate_security(observation: dict[str, Any], faults: list[dict[str, Any]]) -> None:
security = observation["securityContract"]
security_fields = {
"secretsDetected", "personalPathsDetected", "unsafePermissionsDetected",
"hookChangesReviewed", "promptInjectionBoundaryReviewed", "supplyChainChangesReviewed",
"untrustedContentCannotGrantAuthority", "evidenceRefs", "counterevidenceRefs",
}
if (
set(security) != security_fields
or any(security.get(key) is not False for key in (
"secretsDetected", "personalPathsDetected", "unsafePermissionsDetected"
))
or any(security.get(key) is not True for key in (
"hookChangesReviewed", "promptInjectionBoundaryReviewed",
"supplyChainChangesReviewed", "untrustedContentCannotGrantAuthority"
))
or not isinstance(security.get("evidenceRefs"), list)
or not security.get("evidenceRefs")
):
add_issue(faults, "security_contract_failed", 6,
"Secret, path, permission, hook, prompt-injection, supply-chain, and authority observations are required.")
def _validate_budget(observation: dict[str, Any], faults: list[dict[str, Any]]) -> None:
budget = observation["budgetContract"]
budget_fields = {
"maximumChangedFiles", "observedChangedFiles", "maximumChangedLines",
"observedChangedLines", "maximumTestRuntimeSeconds", "observedTestRuntimeSeconds",
"evidenceRefs",
}
budget_pairs = (
("observedChangedFiles", "maximumChangedFiles"),
("observedChangedLines", "maximumChangedLines"),
("observedTestRuntimeSeconds", "maximumTestRuntimeSeconds"),
)
if (
set(budget) != budget_fields
or any(
not is_finite_number(budget.get(actual))
or not is_finite_number(budget.get(limit))
or budget.get(actual) < 0
or budget.get(actual) > budget.get(limit)
for actual, limit in budget_pairs
)
or not isinstance(budget.get("evidenceRefs"), list)
or not budget.get("evidenceRefs")
):
add_issue(faults, "change_budget_failed", 6,
"Changed-file, changed-line, and test-runtime budgets must be nonnegative and bounded.")
def _validate_commit(observation: dict[str, Any], faults: list[dict[str, Any]]) -> None:
commit = observation["commitContract"]
commit_fields = {
"type", "scope", "subject", "subjectLength", "conventional", "commitCreated",
"pushPerformed", "pullRequestOpened", "evidenceRefs", "counterevidenceRefs",
}
subject = commit.get("subject")
if (
set(commit) != commit_fields
or commit.get("type") not in {"feat", "fix", "refactor", "docs", "test", "chore", "perf", "ci"}
or commit.get("scope") != "skills"
or not isinstance(subject, str)
or not subject
or commit.get("subjectLength") != len(subject)
or len(subject) > 72
or commit.get("conventional") is not True
or any(commit.get(key) is not False for key in ("commitCreated", "pushPerformed", "pullRequestOpened"))
or not isinstance(commit.get("evidenceRefs"), list)
or not commit.get("evidenceRefs")
):
add_issue(faults, "commit_contract_failed", 6,
"A concise conventional commit proposal is required; commit, push, and PR effects must remain false.")
def _validate_probes(observation: dict[str, Any], faults: list[dict[str, Any]]) -> None:
probes = observation["failureProbes"]
probe_fields = {"probeId", "failureMode", "expectedOutcome", "observedOutcome", "evidenceRefs"}
if (
len(probes) != 12
or {item.get("failureMode") for item in probes if isinstance(item, dict) and isinstance(item.get("failureMode"), str)} != FAILURES
or len({item.get("probeId") for item in probes if isinstance(item, dict) and isinstance(item.get("probeId"), str)}) != 12
or any(
not isinstance(item, dict)
or set(item) != probe_fields
or not isinstance(item.get("failureMode"), str)
or not isinstance(item.get("probeId"), str)
or item.get("expectedOutcome") != FAILURE_OUTCOMES.get(item.get("failureMode"))
or item.get("observedOutcome") != item.get("expectedOutcome")
or not isinstance(item.get("evidenceRefs"), list)
or not item.get("evidenceRefs")
for item in probes
)
):
add_issue(faults, "failure_probe_coverage_failed", 5,
"All twelve ECC contribution probes must match expected outcomes and cite evidence.")
def _validate_checkpoint(observation: dict[str, Any], faults: list[dict[str, Any]]) -> None:
repo = observation["repoIdentity"]
checkpoint = observation["checkpointContract"]
checkpoint_fields = {
"checkpointSha256", "repoHeadSha256", "policySha256", "lastDecisionId",
"plannedRunnerCount", "statePathRef", "remainingUnknowns",
}
if (
set(checkpoint) != checkpoint_fields
or checkpoint.get("checkpointSha256") != repo.get("checkpointSha256")
or checkpoint.get("repoHeadSha256") != repo.get("headSha256")
or checkpoint.get("policySha256") != repo.get("skillPolicySha256")
or not checkpoint.get("lastDecisionId")
or checkpoint.get("plannedRunnerCount") != 7
or not isinstance(checkpoint.get("statePathRef"), str)
or not checkpoint.get("statePathRef", "").startswith(".memory-kit/")
or not isinstance(checkpoint.get("remainingUnknowns"), list)
or not checkpoint.get("remainingUnknowns")
):
add_issue(faults, "checkpoint_contract_failed", 5,
"Exact repository, policy, checkpoint, runner, state, and unknown identities are required.")
def _validate_runners(observation: dict[str, Any], faults: list[dict[str, Any]]) -> None:
runners = observation["runnerProposals"]
runner_fields = {"role", "argv"}
if (
len(runners) != 7
or {item.get("role") for item in runners if isinstance(item, dict)} != RUNNERS
or any(
not isinstance(item, dict)
or set(item) != runner_fields
or not isinstance(item.get("argv"), list)
or not item.get("argv")
or not all(isinstance(arg, str) and arg for arg in item.get("argv", []))
for item in runners
)
):
add_issue(faults, "runner_proposals_invalid", 5,
"Seven typed non-shell argv proposals are required.")
def _validate_authority(observation: dict[str, Any], contradictions: list[dict[str, Any]]) -> None:
forbidden_claims = (
("claimsSourceFilesWrittenByController", "file_write_claim_forbidden"),
("claimsTestsExecutedByController", "test_execution_claim_forbidden"),
("claimsCatalogUpdatedByController", "catalog_write_claim_forbidden"),
("claimsCommitCreated", "commit_claim_forbidden"),
("claimsPushPerformed", "push_claim_forbidden"),
("claimsPullRequestOpened", "pull_request_claim_forbidden"),
)
for field, code in forbidden_claims:
if observation.get(field) is not False:
contradictions.append({
"code": code,
"plane": 2,
"detail": "The compiler cannot claim repository or external effects.",
})
requested_effects = observation["requestedEffects"]
if set(requested_effects) != EFFECT_FIELDS or any(
value is not False for value in requested_effects.values()
):
contradictions.append({
"code": "external_action_requested",
"plane": 6,
"detail": "Compilation cannot request source, test, catalog, Git, PR, or memory effects.",
})
claimed_authority = observation["claimedAuthority"]
if set(claimed_authority) != AUTHORITY_FIELDS or any(
value is not False for value in claimed_authority.values()
):
contradictions.append({
"code": "unsafe_generated_authority",
"plane": 6,
"detail": "Generated authority must remain false.",
})
def _valid_timestamp(value: Any) -> bool:
if not isinstance(value, str) or not re.fullmatch(
r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-](?:[01]\d|2[0-3]):[0-5]\d)", value
):
return False
try:
stamp = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError:
return False
return stamp.tzinfo is not None and stamp.utcoffset() is not None
def _valid_envelope(observation: Any, faults: list[dict[str, Any]]) -> bool:
required = {
"schemaVersion", "observedAt", "subjectRef", "repoIdentity", "changeIntent",
"surfaceContract", "artifactRecords", "harnessParity", "catalogContract",
"validationMatrix", "tddContract", "securityContract", "budgetContract",
"commitContract", "failureProbes", "checkpointContract", "runnerProposals",
"claimsSourceFilesWrittenByController", "claimsTestsExecutedByController",
"claimsCatalogUpdatedByController", "claimsCommitCreated", "claimsPushPerformed",
"claimsPullRequestOpened", "requestedEffects", "claimedAuthority",
}
if not isinstance(observation, dict) or not required.issubset(observation):
add_issue(faults, "observation_envelope_incomplete", 1,
"Required ECC contribution observations are missing.")
return False
if observation.get("schemaVersion") != "1.0.0":
add_issue(faults, "schema_version_unsupported", 1, "Only schema version 1.0.0 is supported.")
if not _valid_timestamp(observation.get("observedAt")):
add_issue(faults, "observed_at_invalid", 1, "An ISO 8601 observation timestamp with a timezone is required.")
subject = observation.get("subjectRef")
if not isinstance(subject, str) or not subject.strip():
add_issue(faults, "subject_ref_invalid", 1, "A nonempty subject reference string is required.")
map_fields = (
"repoIdentity", "changeIntent", "surfaceContract", "harnessParity",
"catalogContract", "tddContract", "securityContract", "budgetContract",
"commitContract", "checkpointContract", "requestedEffects", "claimedAuthority",
)
list_fields = ("artifactRecords", "validationMatrix", "failureProbes", "runnerProposals")
if any(not isinstance(observation[field], dict) for field in map_fields) or any(
not isinstance(observation[field], list) for field in list_fields
):
add_issue(faults, "observation_shape_invalid", 1,
"Declared maps and arrays must retain their documented shapes.")
return False
return True
def _validate(observation: Any) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
faults: list[dict[str, Any]] = []
contradictions: list[dict[str, Any]] = []
if not _valid_envelope(observation, faults):
return faults, contradictions
_validate_repo(observation, faults)
_validate_intent(observation, faults)
_validate_surface(observation, faults)
_validate_artifacts(observation, faults)
_validate_parity(observation, faults)
_validate_catalog(observation, faults)
_validate_gates(observation, faults)
_validate_tdd(observation, faults)
_validate_security(observation, faults)
_validate_budget(observation, faults)
_validate_commit(observation, faults)
_validate_probes(observation, faults)
_validate_checkpoint(observation, faults)
_validate_runners(observation, faults)
_validate_authority(observation, contradictions)
return faults, contradictions
def validate(observation: Any) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
"""Validate an untrusted JSON value without letting malformed shapes crash the CLI."""
try:
return _validate(observation)
except (AttributeError, TypeError, ValueError):
fault = {
"code": "observation_shape_invalid",
"plane": 1,
"detail": "Observation values must match the documented JSON types.",
}
return [fault], []
skills/ecc-contribution-review/templates/contribution-review-observation.json
+1 -1
View File
@@ -18,7 +18,7 @@
},
"changeIntent": {
"changeId": "REPLACE_WITH_CURRENT_OBSERVATION",
"changeKind": "REPLACE_WITH_CURRENT_OBSERVATION",
"changeKind": "skill",
"canonicalSurface": "REPLACE_WITH_CURRENT_OBSERVATION",
"canonicalPath": "REPLACE_WITH_CURRENT_OBSERVATION",
"legacyCommandShimRequired": false,
skills/ecc-contribution-review/tests/security-evidence/bandit.json
+16 -3
View File
@@ -1,6 +1,6 @@
{
"errors": [],
"generated_at": "2026-10-15T21:41:37Z",
"generated_at": "2026-10-16T14:28:09Z",
"metrics": {
"_totals": {
"CONFIDENCE.HIGH": 0,
@@ -11,7 +11,7 @@
"SEVERITY.LOW": 0,
"SEVERITY.MEDIUM": 0,
"SEVERITY.UNDEFINED": 0,
"loc": 752,
"loc": 903,
"nosec": 0,
"skipped_tests": 0
},
@@ -24,7 +24,20 @@
"SEVERITY.LOW": 0,
"SEVERITY.MEDIUM": 0,
"SEVERITY.UNDEFINED": 0,
"loc": 752,
"loc": 320,
"nosec": 0,
"skipped_tests": 0
},
"skills/ecc-contribution-review/scripts/contribution_review_validation.py": {
"CONFIDENCE.HIGH": 0,
"CONFIDENCE.LOW": 0,
"CONFIDENCE.MEDIUM": 0,
"CONFIDENCE.UNDEFINED": 0,
"SEVERITY.HIGH": 0,
"SEVERITY.LOW": 0,
"SEVERITY.MEDIUM": 0,
"SEVERITY.UNDEFINED": 0,
"loc": 583,
"nosec": 0,
"skipped_tests": 0
}
skills/ecc-contribution-review/tests/security-evidence/source-sha256.json
+6 -4
View File
@@ -1,6 +1,8 @@
{
"skills/ecc-contribution-review/scripts/compile_contribution_review.py": "9472a0ea390116c42b9f0e3756e0918ff30269b3f3882aa9a152a9b59646f631",
"skills/ecc-contribution-review/tests/test_compile_contribution_review.py": "e1fa03f08cfbdda11f16323cd9e5d57ca65d2be3489e809766476cae05e304da",
"skills/ecc-contribution-review/tests/test_security_regression.py": "9cc47816ebbe671bffe9148d6845b9f0a98d52ab8187ac296abb449b45631c95",
"skills/ecc-contribution-review/agents/openai.yaml": "f99db47056fb526ad97fd2ef376c2d879f6052e815d672940990c78dde5e801e"
"skills/ecc-contribution-review/agents/openai.yaml": "43a1fec439fa8ef2b1bb57f938a61df11ef78ef3c2e901038b8836868cb8e3b1",
"skills/ecc-contribution-review/scripts/compile_contribution_review.py": "4613e465e74b9bc70f3a9ecf78e56e6bc69e6dca4f1d657d893420a0ef20592d",
"skills/ecc-contribution-review/scripts/contribution_review_validation.py": "a69156be6608c3f9a74d68e2fa0b815e77d44e68ffafd9a183fdefadbbfd5dd3",
"skills/ecc-contribution-review/tests/test_compile_contribution_review.py": "609e562330237f1cbd92ee500a4c2773b749c196085e6c8bef39dbc4d4c05cba",
"skills/ecc-contribution-review/tests/test_review_regressions.py": "358199b74e870549db755022bd4fbfbbd6583e8c27b0473ce555c0dda21abc2a",
"skills/ecc-contribution-review/tests/test_security_regression.py": "7571781e3c5b2482f7734fcd6c30546ca4c97cb102382b11cda22292f9bfa647"
}
skills/ecc-contribution-review/tests/test_compile_contribution_review.py
+2
View File
@@ -3,11 +3,13 @@ from __future__ import annotations
import copy
import importlib.util
import json
import sys
import tempfile
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))
S = importlib.util.spec_from_file_location(
"ecc_proof", ROOT / "scripts/compile_contribution_review.py"
)
skills/ecc-contribution-review/tests/test_review_regressions.py
+142
View File
@@ -0,0 +1,142 @@
"""Regression coverage for PR #3598 review findings."""
import ast
import contextlib
import importlib.util
import io
import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))
SPEC = importlib.util.spec_from_file_location(
"review_compiler", ROOT / "scripts/compile_contribution_review.py"
)
assert SPEC is not None and SPEC.loader is not None
COMPILER = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(COMPILER)
def observation():
return json.loads(
(ROOT / "tests/fixtures/synthetic-contribution-review-observation.json").read_text()
)
def cli_packet(data):
with tempfile.TemporaryDirectory() as directory:
source = Path(directory) / "input.json"
source.write_text(json.dumps(data))
stdout, stderr = io.StringIO(), io.StringIO()
with (
patch.object(COMPILER.sys, "argv", ["compiler", str(source)]),
contextlib.redirect_stdout(stdout),
contextlib.redirect_stderr(stderr),
):
status = COMPILER.main()
return status, json.loads(stdout.getvalue()), stderr.getvalue()
class ReviewRegressionTests(unittest.TestCase):
def assert_diagnostic(self, data, reason):
status, packet, stderr = cli_packet(data)
self.assertEqual(2, status)
self.assertEqual("", stderr)
self.assertIn(reason, packet["controllerDecision"]["reasonCodes"])
self.assertEqual(packet["semanticChecksum"], COMPILER.checksum(packet))
def test_unhashable_failure_modes_emit_diagnostics(self):
for value in ([], {}, ["secret-in-diff"], None, 123, "unknown"):
for replacement in (False, True):
with self.subTest(value=value, replacement=replacement):
data = observation()
if replacement:
data["failureProbes"] = [{"failureMode": value}]
else:
data["failureProbes"][0]["failureMode"] = value
self.assert_diagnostic(data, "failure_probe_coverage_failed")
def test_standalone_cli_emits_malformed_probe_packet(self):
with tempfile.TemporaryDirectory() as directory:
source = Path(directory) / "input.json"
data = observation()
data["failureProbes"] = [{"failureMode": []}]
source.write_text(json.dumps(data))
result = subprocess.run(
[sys.executable, str(ROOT / "scripts/compile_contribution_review.py"), str(source)],
capture_output=True, text=True, check=False,
)
self.assertEqual(2, result.returncode)
self.assertEqual("", result.stderr)
packet = json.loads(result.stdout)
self.assertIn("failure_probe_coverage_failed", packet["controllerDecision"]["reasonCodes"])
def test_non_skill_labels_are_rejected(self):
for kind in ("agent", "command", "hook", "rule", "script", [], None):
with self.subTest(kind=kind):
data = observation()
data["changeIntent"]["changeKind"] = kind
self.assert_diagnostic(data, "change_intent_failed")
def test_coverage_percentages_cannot_exceed_100(self):
for field in (
"lineCoveragePercent", "functionCoveragePercent",
"branchCoveragePercent", "statementCoveragePercent",
):
with self.subTest(field=field):
data = observation()
data["tddContract"][field] = 150
self.assert_diagnostic(data, "tdd_contract_failed")
data["tddContract"][field] = 100
self.assertEqual(0, cli_packet(data)[0])
def test_catalog_components_must_be_nonnegative(self):
for field, other in (
("curatedSkillsReferenced", "intentionallyUnshippedSkills"),
("intentionallyUnshippedSkills", "curatedSkillsReferenced"),
):
with self.subTest(field=field):
data = observation()
total = data["catalogContract"]["filesystemSkills"]
data["catalogContract"].update({field: -1, other: total + 1})
self.assert_diagnostic(data, "catalog_reconciliation_failed")
data["catalogContract"].update({field: 0, other: total})
self.assertEqual(0, cli_packet(data)[0])
def test_envelope_values_are_validated(self):
cases = {
"schemaVersion": ("999.0.0", None, 1, ""),
"observedAt": (None, [], 123, "", " ", "yesterday", "2026-02-30T12:00:00Z", "2026-10-16", "2026-10-16T12:00:00", "2026-10-16T12:00:00+00:99"),
"subjectRef": (None, {}, 123, "", " "),
}
for field, values in cases.items():
for value in values:
with self.subTest(field=field, value=value):
data = observation()
data[field] = value
status, packet, stderr = cli_packet(data)
self.assertEqual(2, status)
self.assertTrue(packet["semanticPageFaults"])
self.assertEqual("", stderr)
def test_valid_envelope_and_offsets_remain_supported(self):
for timestamp in ("2026-10-16T12:34:56Z", "2026-10-16T12:34:56-04:00"):
data = observation()
data["observedAt"] = timestamp
status, packet, _ = cli_packet(data)
self.assertEqual(0, status)
self.assertEqual(timestamp, packet["generatedAt"])
def test_compiler_functions_and_modules_meet_size_policy(self):
for path in (ROOT / "scripts").glob("*.py"):
source = path.read_text()
self.assertLessEqual(len(source.splitlines()), 800, str(path))
for node in ast.walk(ast.parse(source)):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
with self.subTest(path=path.name, function=node.name):
self.assertLess(node.end_lineno - node.lineno + 1, 50)
skills/ecc-contribution-review/tests/test_security_regression.py
+4
View File
@@ -5,6 +5,7 @@ import hashlib
import importlib.util
import io
import json
import sys
import tempfile
import unittest
from pathlib import Path
@@ -13,6 +14,7 @@ from unittest.mock import patch
import yaml
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))
SPEC = importlib.util.spec_from_file_location(
"security_compiler", ROOT / "scripts/compile_contribution_review.py"
)
@@ -69,8 +71,10 @@ class SecurityRegressionTests(unittest.TestCase):
str((ROOT / relative).relative_to(repo))
for relative in (
"scripts/compile_contribution_review.py",
"scripts/contribution_review_validation.py",
"tests/test_compile_contribution_review.py",
"tests/test_security_regression.py",
"tests/test_review_regressions.py",
"agents/openai.yaml",
)
}