diff --git a/docs/README.md b/docs/README.md index 8f66f04427..be6af8fc73 100644 --- a/docs/README.md +++ b/docs/README.md @@ -2,7 +2,7 @@ AutoSkillit is a Claude Code plugin that runs YAML recipes through a multi-level orchestrator. The bundled recipes implement issue → plan → worktree -→ tests → PR → merge pipelines using 61 MCP tools and 142 bundled skills. +→ tests → PR → merge pipelines using 62 MCP tools and 142 bundled skills. ## Start here diff --git a/docs/execution/architecture.md b/docs/execution/architecture.md index 7eaa2414f7..8fadfaa8a8 100644 --- a/docs/execution/architecture.md +++ b/docs/execution/architecture.md @@ -4,7 +4,7 @@ How AutoSkillit runs a recipe end to end: orchestrator, kitchen gating, clone an ## Overview -AutoSkillit is a Claude Code plugin that orchestrates automated workflows using headless sessions. It provides 61 MCP tools and 142 bundled skills, organized into a gated visibility system. +AutoSkillit is a Claude Code plugin that orchestrates automated workflows using headless sessions. It provides 62 MCP tools and 142 bundled skills, organized into a gated visibility system. ## Core Concepts @@ -68,7 +68,7 @@ AutoSkillit supports four session modes with different tool and skill visibility `$ claude`); `/open-kitchen` reveals kitchen tools. - **`$ autoskillit order`**: Pipeline orchestrator session. Kitchen is pre-opened at startup — - all 61 MCP tools are available immediately. All skill tiers are accessible. The orchestrator + all 62 MCP tools are available immediately. All skill tiers are accessible. The orchestrator delegates work through `run_skill` (headless sessions) and `run_cmd` (shell commands). - **`run_skill` (headless)**: Worker sessions launched by the orchestrator. Sees 4 Free Range diff --git a/docs/execution/tool-access.md b/docs/execution/tool-access.md index e94157d7b0..8f042b6eab 100644 --- a/docs/execution/tool-access.md +++ b/docs/execution/tool-access.md @@ -1,6 +1,6 @@ # MCP Tool Access Control -AutoSkillit provides 61 MCP tools organized into three access levels that control which +AutoSkillit provides 62 MCP tools organized into three access levels that control which session types can see each tool. ## Three Access Levels @@ -92,7 +92,7 @@ missing kitchen visibility. ## Complete MCP Tool Access Control Map -All 61 tools with their access level, tags, source file, and functional category. +All 62 tools with their access level, tags, source file, and functional category. **Tag abbreviations**: AS = `autoskillit`, K = `kitchen`, HL = `headless`, GH = `github`, CI = `ci`, CL = `clone`, TL = `telemetry`, FL = `fleet` diff --git a/src/autoskillit/config/ingredient_defaults.py b/src/autoskillit/config/ingredient_defaults.py index 6755a5da5e..36c408bd47 100644 --- a/src/autoskillit/config/ingredient_defaults.py +++ b/src/autoskillit/config/ingredient_defaults.py @@ -42,6 +42,7 @@ "validate_recipe", "get_recipe_section", "complete_recipe_initialization", + "write_audit_cycle_artifact", ), ), ("Agents", ("unlock_agent_pack",)), diff --git a/src/autoskillit/core/__init__.py b/src/autoskillit/core/__init__.py index 22b1e75b61..1b7a218903 100644 --- a/src/autoskillit/core/__init__.py +++ b/src/autoskillit/core/__init__.py @@ -18,6 +18,10 @@ "_collect_disabled_feature_tags", "_AUTOSKILLIT_GITIGNORE_ENTRIES", "_COMMITTED_BY_DESIGN", + "_MAX_ASSOCIATION_FILES", + "_MAX_REFERENCED_ARTIFACTS_PER_CALL", + "_PLAN_ASSOCIATION_DOMAIN", + "_PLAN_ASSOCIATION_KEYS", } ) __all__ = [n for n in __all__ if n not in _PRIVATE_REEXPORTS] diff --git a/src/autoskillit/core/__init__.pyi b/src/autoskillit/core/__init__.pyi index 160886b809..d70db2bfba 100644 --- a/src/autoskillit/core/__init__.pyi +++ b/src/autoskillit/core/__init__.pyi @@ -205,6 +205,10 @@ from .tool_sequence_analysis import ( from .tool_sequence_analysis import render_adjacency_table as render_adjacency_table from .tool_sequence_analysis import render_dot as render_dot from .tool_sequence_analysis import render_mermaid as render_mermaid +from .types import _MAX_ASSOCIATION_FILES as _MAX_ASSOCIATION_FILES +from .types import _MAX_REFERENCED_ARTIFACTS_PER_CALL as _MAX_REFERENCED_ARTIFACTS_PER_CALL +from .types import _PLAN_ASSOCIATION_DOMAIN as _PLAN_ASSOCIATION_DOMAIN +from .types import _PLAN_ASSOCIATION_KEYS as _PLAN_ASSOCIATION_KEYS from .types import ABSENT_BOUND_VALUE as ABSENT_BOUND_VALUE from .types import ADMIRAL_DISPATCH_SECTIONS as ADMIRAL_DISPATCH_SECTIONS from .types import AGENT_BACKEND_CLAUDE_CODE as AGENT_BACKEND_CLAUDE_CODE diff --git a/src/autoskillit/core/io.py b/src/autoskillit/core/io.py index 4b0dc50443..54bcd8c057 100644 --- a/src/autoskillit/core/io.py +++ b/src/autoskillit/core/io.py @@ -3,9 +3,13 @@ Zero autoskillit imports. Provides atomic filesystem writes, project temp directory management, and YAML load/dump helpers. -All NEW on-disk JSON artifacts SHOULD use ``write_versioned_json`` so schema drift -is detectable. Existing artifacts are tracked in -``tests/infra/test_schema_version_convention.py`` (landed in a later phase). +New on-disk JSON artifacts fall into two families. Default to ``write_versioned_json`` +so schema drift is detectable; existing sites are tracked in +``tests/infra/test_schema_version_convention.py``. Use ``write_canonical_versioned_json`` +instead when the artifact's reader will call +``decode_versioned_json_bytes(..., require_canonical=True)`` for tamper-evident, +hash-bound content addressing — every such producer/consumer pairing must be +registered in ``tests/infra/test_canonical_json_producer_convention.py``. """ from __future__ import annotations @@ -199,27 +203,42 @@ def atomic_write( content: str, *, strict_durability: bool = False, + exclusive: bool = False, ) -> None: """Crash-safe write: write to a temp file then os.replace. Includes data fsync and directory fsync for durability on ext4/xfs. The directory fsync is skipped on Windows (no O_RDONLY semantics). + + When ``exclusive`` is True, atomically claims ``path`` via + ``os.O_CREAT | os.O_EXCL`` before writing, raising ``FileExistsError`` + with no bytes written if the destination already exists. Closes the + TOCTOU window between a separate existence check and the write. """ import sys as _sys path.parent.mkdir(parents=True, exist_ok=True) - fd, tmp = tempfile.mkstemp(dir=path.parent, suffix=".tmp") + if exclusive: + os.close(os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)) + tmp: str | None = None try: + fd, tmp = tempfile.mkstemp(dir=path.parent, suffix=".tmp") with os.fdopen(fd, "w", encoding="utf-8") as f: f.write(content) f.flush() os.fsync(f.fileno()) # durable data write os.replace(tmp, path) except Exception: - try: - os.unlink(tmp) - except OSError: - pass + if tmp is not None: + try: + os.unlink(tmp) + except OSError: + pass + if exclusive: + try: + os.unlink(path) + except OSError: + pass raise # Durable rename: fsync the parent directory on POSIX. # Default callers retain best-effort parent durability. Identity-bearing @@ -390,7 +409,11 @@ def write_versioned_json( def write_canonical_versioned_json( - path: Path, payload: dict[str, Any], schema_version: int + path: Path, + payload: dict[str, Any], + schema_version: int, + *, + exclusive: bool = False, ) -> None: """Atomically write versioned canonical JSON for hash-bound artifacts.""" from .closure_hashing import canonical_json_bytes @@ -398,7 +421,7 @@ def write_canonical_versioned_json( if not isinstance(payload, dict): raise TypeError("write_canonical_versioned_json requires a dict payload") enriched = {**payload, "schema_version": schema_version} - atomic_write(path, canonical_json_bytes(enriched).decode("utf-8")) + atomic_write(path, canonical_json_bytes(enriched).decode("utf-8"), exclusive=exclusive) def decode_versioned_json_bytes( diff --git a/src/autoskillit/core/tool_registry.py b/src/autoskillit/core/tool_registry.py index 9be00a77c2..c539d7fe53 100644 --- a/src/autoskillit/core/tool_registry.py +++ b/src/autoskillit/core/tool_registry.py @@ -90,6 +90,7 @@ "unlock_agent_pack", "wait_for_ci", "wait_for_merge_queue", + "write_audit_cycle_artifact", "write_telemetry_files", } ) @@ -393,6 +394,12 @@ def _run_skill() -> ToolDef: ("paths", "message", "cwd", "step_name"), required=("paths", "message", "cwd"), ), + _tool( + "write_audit_cycle_artifact", + ("kind", "path", "fields", "cwd", "step_name"), + required=("kind", "path", "fields", "cwd"), + wire_types={"fields": ToolWireType.OBJECT}, + ), _tool( "fetch_github_issue", ("issue_url", "include_comments"), diff --git a/src/autoskillit/core/types/_type_audit_cycle.py b/src/autoskillit/core/types/_type_audit_cycle.py index 47eed52e7d..c4af25971f 100644 --- a/src/autoskillit/core/types/_type_audit_cycle.py +++ b/src/autoskillit/core/types/_type_audit_cycle.py @@ -24,6 +24,10 @@ "PlanDispositionReport", "PlanDispositionRow", "compute_findings_digest", + "_MAX_ASSOCIATION_FILES", + "_MAX_REFERENCED_ARTIFACTS_PER_CALL", + "_PLAN_ASSOCIATION_DOMAIN", + "_PLAN_ASSOCIATION_KEYS", ] AUDIT_CYCLE_SCHEMA_VERSION = 1 @@ -36,6 +40,28 @@ _SATISFIED_RE = re.compile(r"^satisfied-by-round-([1-9][0-9]*)$") _T = TypeVar("_T") +# Shared by the write-side write_audit_cycle_artifact MCP tool (server/tools/ +# tools_audit_cycle.py) and the read-side _resolve_plan_disposition +# (recipe/_cmd_rpc_guards.py) — a single IL-0 definition so the two sides of +# the plan-association contract cannot drift into DUAL-COPY CONSTANTS. +_PLAN_ASSOCIATION_DOMAIN = "autoskillit:audit-cycle:plan-association:v1:sha256" +_PLAN_ASSOCIATION_KEYS = frozenset( + { + "schema_version", + "plan_ref", + "disposition_ref", + "parent_authority_digest", + "association_digest", + } +) +_MAX_ASSOCIATION_FILES = 256 + +# Bounds the number of referenced-artifact entries (audited_plan_refs, +# assessments, dispositions, etc.) accepted in a single write_audit_cycle_artifact +# request payload. Distinct from _MAX_ASSOCIATION_FILES, which bounds the count +# of association files already on disk under an associations/ directory glob. +_MAX_REFERENCED_ARTIFACTS_PER_CALL = 256 + def _immutable_typed_tuple( name: str, diff --git a/src/autoskillit/core/types/_type_constants_registries.py b/src/autoskillit/core/types/_type_constants_registries.py index 28b359dfbd..1f43f09700 100644 --- a/src/autoskillit/core/types/_type_constants_registries.py +++ b/src/autoskillit/core/types/_type_constants_registries.py @@ -142,7 +142,9 @@ } ) -HEADLESS_TOOLS: frozenset[str] = frozenset({"test_check", "unlock_agent_pack", "commit_files"}) +HEADLESS_TOOLS: frozenset[str] = frozenset( + {"test_check", "unlock_agent_pack", "commit_files", "write_audit_cycle_artifact"} +) FLEET_TOOLS: frozenset[str] = frozenset( { @@ -743,6 +745,7 @@ class AgentPackDef(NamedTuple): "reset_workspace": frozenset({"kitchen-core"}), "classify_fix": frozenset({"kitchen-core"}), "commit_files": frozenset({"kitchen-core"}), + "write_audit_cycle_artifact": frozenset({"kitchen-core"}), "list_recipes": frozenset({"kitchen-core", "fleet-dispatch"}), "load_recipe": frozenset({"kitchen-core", "fleet-dispatch"}), "validate_recipe": frozenset({"kitchen-core"}), @@ -870,6 +873,14 @@ class HardCapabilityMismatch(NamedTuple): codex_status="works-as-is", allowed_execution_roles=_ALL_SKILL_EXECUTION_ROLES, ), + "write_audit_cycle_artifact": SkillCapabilityDef( + description=( + "write_audit_cycle_artifact MCP tool — server-side construction, digest " + "computation, and canonical write for hash-bound audit-cycle artifacts" + ), + codex_status="works-as-is", + allowed_execution_roles=_ALL_SKILL_EXECUTION_ROLES, + ), "git_metadata_write": SkillCapabilityDef( description=( "Requires .git/ metadata write access (git commit, git rebase, " diff --git a/src/autoskillit/recipe/_cmd_rpc_guards.py b/src/autoskillit/recipe/_cmd_rpc_guards.py index 7653c8bc85..4e6e70424d 100644 --- a/src/autoskillit/recipe/_cmd_rpc_guards.py +++ b/src/autoskillit/recipe/_cmd_rpc_guards.py @@ -11,6 +11,9 @@ import regex as re from autoskillit.core import ( + _MAX_ASSOCIATION_FILES, + _PLAN_ASSOCIATION_DOMAIN, + _PLAN_ASSOCIATION_KEYS, AUDIT_CYCLE_SCHEMA_VERSION, ArtifactRef, AuditCycleVerifier, @@ -214,19 +217,6 @@ def _normalize_plan_parts(plan_parts: str) -> list[str] | None: return items -_PLAN_ASSOCIATION_DOMAIN = "autoskillit:audit-cycle:plan-association:v1:sha256" -_PLAN_ASSOCIATION_KEYS = frozenset( - { - "schema_version", - "plan_ref", - "disposition_ref", - "parent_authority_digest", - "association_digest", - } -) -_MAX_ASSOCIATION_FILES = 256 - - def _log_plan_disposition_rejection( reason: str, *, diff --git a/src/autoskillit/server/AGENTS.md b/src/autoskillit/server/AGENTS.md index 649cede82f..f2bbec3a5c 100644 --- a/src/autoskillit/server/AGENTS.md +++ b/src/autoskillit/server/AGENTS.md @@ -84,7 +84,7 @@ Controls whether the tool succeeds when called (independent of visibility): | Standard kitchen | `kitchen` | Yes | Yes (`_require_enabled`) | `run_cmd`, `run_skill`, `report_bug` | | Fleet tool | `fleet`, `kitchen-core` | Yes (via `ALL_VISIBILITY_TAGS` loop) | Yes (`_require_fleet` or `_require_enabled`) | `dispatch_food_truck`, `record_gate_dispatch` | | Fleet-dispatch tool | `fleet-dispatch` (± `kitchen-core`) | Yes (via `ALL_VISIBILITY_TAGS` loop) | Yes (`_require_enabled`) | `fetch_github_issue`, `list_recipes` | -| Headless-exempt | `kitchen`, `headless` | Yes | No | `test_check` | +| Headless-exempt | `kitchen`, `headless` | Yes | No | `test_check`, `commit_files`, `unlock_agent_pack`, `write_audit_cycle_artifact` | | Free-range | _(none of the above)_ | No | No | `open_kitchen`, `close_kitchen` | ### Registry Constants @@ -93,7 +93,7 @@ The canonical tool sets are in `core/types/_type_constants_registries.py`: - `GATED_TOOLS` — all tools that call `_require_enabled()` (validated by arch test) - `UNGATED_TOOLS` = `FREE_RANGE_TOOLS` — tools with no gating at all -- `HEADLESS_TOOLS` — `{"test_check"}` — kitchen-tagged but not application-gated +- `HEADLESS_TOOLS` — `{"test_check", "unlock_agent_pack", "commit_files", "write_audit_cycle_artifact"}` — kitchen-tagged but not application-gated - `FLEET_TOOLS` — fleet-session-only tools - `FLEET_DISPATCH_TOOLS` — fleet-dispatch-mode tools (hidden at startup, application-gated) - `ALL_VISIBILITY_TAGS` — `{"kitchen", "headless", "fleet", "fleet-dispatch", "kitchen-core", "plan-review"}` diff --git a/src/autoskillit/server/__init__.py b/src/autoskillit/server/__init__.py index e4d3810664..2e2f3dfd9f 100644 --- a/src/autoskillit/server/__init__.py +++ b/src/autoskillit/server/__init__.py @@ -86,6 +86,9 @@ from autoskillit.server.tools import ( # noqa: E402, F401 tools_agents as _tools_agents, ) +from autoskillit.server.tools import ( # noqa: E402, F401 + tools_audit_cycle as _tools_audit_cycle, +) from autoskillit.server.tools import ( # noqa: E402, F401 tools_ci as _tools_ci, ) diff --git a/src/autoskillit/server/tools/AGENTS.md b/src/autoskillit/server/tools/AGENTS.md index 600e0f4d7c..29df5089df 100644 --- a/src/autoskillit/server/tools/AGENTS.md +++ b/src/autoskillit/server/tools/AGENTS.md @@ -1,6 +1,6 @@ # tools/ -MCP `@mcp.tool()` handlers registered on import (20 tool modules). +MCP `@mcp.tool()` handlers registered on import (21 tool modules). ## Files @@ -28,7 +28,8 @@ MCP `@mcp.tool()` handlers registered on import (20 tool modules). | `tools_execution.py` | `run_cmd`, `run_python`, `run_skill` | | `tools_fleet_dispatch.py` | `dispatch_food_truck`, `record_gate_dispatch` | | `tools_fleet_reset.py` | `reset_dispatch` (full dispatch artifact cleanup) | -| `tools_git.py` | `merge_worktree`, `classify_fix`, `create_unique_branch`, `create_and_publish_branch`, `check_pr_mergeable` | +| `tools_git.py` | `merge_worktree`, `classify_fix`, `create_unique_branch`, `create_and_publish_branch`, `check_pr_mergeable`, `commit_files` | +| `tools_audit_cycle.py` | `write_audit_cycle_artifact` — server-side construction, digest computation, and canonical write for hash-bound audit-cycle artifacts (`authority`, `inventory`, `disposition_report`, `plan_association`) | | `tools_github.py` | `fetch_github_issue`, `get_issue_title`, `report_bug` | | `tools_issue_headless.py` | `prepare_issue`, `enrich_issues` (headless session tools) | | `tools_issue_labels.py` | `claim_issue`, `release_issue` (GitHub label management) | diff --git a/src/autoskillit/server/tools/tools_audit_cycle.py b/src/autoskillit/server/tools/tools_audit_cycle.py new file mode 100644 index 0000000000..0c3cddd64f --- /dev/null +++ b/src/autoskillit/server/tools/tools_audit_cycle.py @@ -0,0 +1,437 @@ +"""MCP tool handler: write_audit_cycle_artifact. + +Server-side construction, digest computation, dataclass validation, and canonical +serialization for the four hash-bound audit-cycle artifact kinds (``authority``, +``inventory``, ``disposition_report``, ``plan_association``) consumed with +``require_canonical=True``. Removes both JSON-canonicalization and digest-arithmetic +from the LLM producer's token-generation path, the way ``commit_files`` +(``tools_git.py``) removes git staging/committing from the same sessions. +""" + +from __future__ import annotations + +import json +import os +import time +from collections.abc import Callable +from pathlib import Path +from typing import Any + +import structlog +from fastmcp import Context +from fastmcp.dependencies import CurrentContext + +from autoskillit.core import ( + _MAX_REFERENCED_ARTIFACTS_PER_CALL, + _PLAN_ASSOCIATION_DOMAIN, + _PLAN_ASSOCIATION_KEYS, + AUDIT_CYCLE_SCHEMA_VERSION, + ArtifactRef, + AuditAssessment, + AuditAssessmentRow, + AuditCycleAuthority, + AuditVerdict, + ContainmentError, + PlanDispositionReport, + PlanDispositionRow, + compute_bytes_hash, + compute_canonical_hash, + get_logger, + resolve_contained_path, + write_canonical_versioned_json, +) +from autoskillit.server import mcp +from autoskillit.server._notify import track_response_size +from autoskillit.server.tools._cancellation_shield import _cancellation_shield + +logger = get_logger(__name__) + + +class _FieldError(ValueError): + """Raised when the semantic ``fields`` payload fails structural validation.""" + + +def _require_str(fields: dict[str, Any], key: str) -> str: + value = fields.get(key) + if not isinstance(value, str) or not value: + raise _FieldError(f"fields[{key!r}] must be a non-empty string") + return value + + +def _require_optional_str(fields: dict[str, Any], key: str) -> str | None: + value = fields.get(key) + if value is not None and not isinstance(value, str): + raise _FieldError(f"fields[{key!r}] must be a string or null") + return value + + +def _require_int(fields: dict[str, Any], key: str) -> int: + value = fields.get(key) + if isinstance(value, bool) or not isinstance(value, int): + raise _FieldError(f"fields[{key!r}] must be an integer") + return value + + +def _artifact_ref_spec(entry: object, label: str) -> tuple[str, str, int]: + """Validate a caller-declared ``{locator, media_type, schema_version}`` shape. + + Cheap, no file I/O — the referenced file is read later, only after every + structural check in the request has passed. + """ + if not isinstance(entry, dict): + raise _FieldError(f"{label} must be an object") + locator = entry.get("locator") + media_type = entry.get("media_type") + schema_version = entry.get("schema_version") + if not isinstance(locator, str) or not locator: + raise _FieldError(f"{label}.locator must be a non-empty string") + if not isinstance(media_type, str) or not media_type: + raise _FieldError(f"{label}.media_type must be a non-empty string") + if isinstance(schema_version, bool) or not isinstance(schema_version, int): + raise _FieldError(f"{label}.schema_version must be an integer") + return locator, media_type, schema_version + + +def _bound_referenced_artifacts(count: int) -> None: + if count > _MAX_REFERENCED_ARTIFACTS_PER_CALL: + raise _FieldError( + f"referenced-artifact count {count} exceeds " + f"_MAX_REFERENCED_ARTIFACTS_PER_CALL={_MAX_REFERENCED_ARTIFACTS_PER_CALL}" + ) + + +def _resolve_ref(spec: tuple[str, str, int], cwd: str) -> ArtifactRef: + """Containment-checked read of an already-existing referenced artifact. + + The server computes ``byte_size``/``content_digest`` from the actual bytes — + the caller never hand-computes a hash. + """ + locator, media_type, schema_version = spec + resolved = resolve_contained_path(locator, cwd) + data = resolved.read_bytes() + return ArtifactRef( + locator=locator, + media_type=media_type, + schema_version=schema_version, + byte_size=len(data), + content_digest=compute_bytes_hash(data), + ) + + +def _build_authority(fields: dict[str, Any], cwd: str) -> dict[str, Any]: + execution_generation = _require_str(fields, "execution_generation") + cycle_id = _require_str(fields, "cycle_id") + plan_set_id = _require_str(fields, "plan_set_id") + scope_id = _require_str(fields, "scope_id") + part_id = _require_str(fields, "part_id") + audit_round = _require_int(fields, "audit_round") + parent_authority_digest = _require_optional_str(fields, "parent_authority_digest") + generated_at = _require_str(fields, "generated_at") + + audited_plan_refs_raw = fields.get("audited_plan_refs") + if not isinstance(audited_plan_refs_raw, list) or not audited_plan_refs_raw: + raise _FieldError("fields['audited_plan_refs'] must be a non-empty list") + audited_plan_specs = [ + _artifact_ref_spec(entry, f"fields['audited_plan_refs'][{index}]") + for index, entry in enumerate(audited_plan_refs_raw) + ] + inventory_spec = _artifact_ref_spec(fields.get("inventory_ref"), "fields['inventory_ref']") + remediation_raw = fields.get("remediation_ref") + remediation_spec = ( + _artifact_ref_spec(remediation_raw, "fields['remediation_ref']") + if remediation_raw is not None + else None + ) + + referenced_count = len(audited_plan_specs) + 1 + (1 if remediation_spec is not None else 0) + _bound_referenced_artifacts(referenced_count) + + assessments_raw = fields.get("assessments") + if not isinstance(assessments_raw, list) or not assessments_raw: + raise _FieldError("fields['assessments'] must be a non-empty list") + assessment_rows: list[AuditAssessmentRow] = [] + for index, row in enumerate(assessments_raw): + if not isinstance(row, dict): + raise _FieldError(f"fields['assessments'][{index}] must be an object") + assessment_raw = row.get("assessment") + requirement_id = row.get("requirement_id") + requirement_text = row.get("requirement_text") + evidence_summary = row.get("evidence_summary") + if ( + not isinstance(assessment_raw, str) + or not isinstance(requirement_id, str) + or not isinstance(requirement_text, str) + or not isinstance(evidence_summary, str) + ): + raise _FieldError( + f"fields['assessments'][{index}] requires string assessment, requirement_id, " + "requirement_text, and evidence_summary" + ) + try: + assessment_value = AuditAssessment(assessment_raw) + except (TypeError, ValueError) as exc: + raise _FieldError(f"fields['assessments'][{index}].assessment invalid: {exc}") from exc + try: + assessment_rows.append( + AuditAssessmentRow.create( + requirement_id=requirement_id, + requirement_text=requirement_text, + assessment=assessment_value, + evidence_summary=evidence_summary, + ) + ) + except (TypeError, ValueError) as exc: + raise _FieldError(f"fields['assessments'][{index}] invalid: {exc}") from exc + + verdict_raw = fields.get("verdict") + if not isinstance(verdict_raw, str): + raise _FieldError("fields['verdict'] must be a string") + try: + verdict = AuditVerdict(verdict_raw) + except ValueError as exc: + raise _FieldError(f"fields['verdict'] invalid: {exc}") from exc + + # Every structural check above (assessment values, verdict, required keys, + # referenced-artifact count bound) has passed — only now do we touch disk. + audited_plan_refs = tuple(_resolve_ref(spec, cwd) for spec in audited_plan_specs) + inventory_ref = _resolve_ref(inventory_spec, cwd) + remediation_ref = _resolve_ref(remediation_spec, cwd) if remediation_spec is not None else None + + authority = AuditCycleAuthority.create( + execution_generation=execution_generation, + cycle_id=cycle_id, + plan_set_id=plan_set_id, + scope_id=scope_id, + part_id=part_id, + audit_round=audit_round, + parent_authority_digest=parent_authority_digest, + audited_plan_refs=audited_plan_refs, + inventory_ref=inventory_ref, + assessments=tuple(assessment_rows), + verdict=verdict, + remediation_ref=remediation_ref, + generated_at=generated_at, + ) + return authority.to_dict() + + +def _build_disposition_report(fields: dict[str, Any], cwd: str) -> dict[str, Any]: + execution_generation = _require_str(fields, "execution_generation") + cycle_id = _require_str(fields, "cycle_id") + plan_set_id = _require_str(fields, "plan_set_id") + scope_id = _require_str(fields, "scope_id") + part_id = _require_str(fields, "part_id") + audit_round = _require_int(fields, "audit_round") + parent_authority_digest = _require_str(fields, "parent_authority_digest") + inventory_digest = _require_str(fields, "inventory_digest") + findings_digest = _require_str(fields, "findings_digest") + generated_at = _require_str(fields, "generated_at") + + current_plan_spec = _artifact_ref_spec( + fields.get("current_plan_ref"), "fields['current_plan_ref']" + ) + _bound_referenced_artifacts(1) + + dispositions_raw = fields.get("dispositions") + if not isinstance(dispositions_raw, list) or not dispositions_raw: + raise _FieldError("fields['dispositions'] must be a non-empty list") + disposition_rows: list[PlanDispositionRow] = [] + for index, row in enumerate(dispositions_raw): + if not isinstance(row, dict): + raise _FieldError(f"fields['dispositions'][{index}] must be an object") + requirement_id = row.get("requirement_id") + disposition = row.get("disposition") + implementation_step = row.get("implementation_step") + if not isinstance(requirement_id, str) or not isinstance(disposition, str): + raise _FieldError( + f"fields['dispositions'][{index}] requires string requirement_id and disposition" + ) + if implementation_step is not None and not isinstance(implementation_step, str): + raise _FieldError( + f"fields['dispositions'][{index}].implementation_step must be a string or null" + ) + try: + disposition_rows.append( + PlanDispositionRow.create( + requirement_id=requirement_id, + disposition=disposition, + implementation_step=implementation_step, + ) + ) + except (TypeError, ValueError) as exc: + raise _FieldError(f"fields['dispositions'][{index}] invalid: {exc}") from exc + + current_plan_ref = _resolve_ref(current_plan_spec, cwd) + + report = PlanDispositionReport.create( + execution_generation=execution_generation, + cycle_id=cycle_id, + plan_set_id=plan_set_id, + scope_id=scope_id, + part_id=part_id, + audit_round=audit_round, + parent_authority_digest=parent_authority_digest, + inventory_digest=inventory_digest, + findings_digest=findings_digest, + current_plan_ref=current_plan_ref, + dispositions=tuple(disposition_rows), + generated_at=generated_at, + ) + return report.to_dict() + + +def _build_plan_association(fields: dict[str, Any], cwd: str) -> dict[str, Any]: + parent_authority_digest = _require_str(fields, "parent_authority_digest") + plan_spec = _artifact_ref_spec(fields.get("plan_ref"), "fields['plan_ref']") + disposition_spec = _artifact_ref_spec( + fields.get("disposition_ref"), "fields['disposition_ref']" + ) + _bound_referenced_artifacts(2) + + plan_ref = _resolve_ref(plan_spec, cwd) + disposition_ref = _resolve_ref(disposition_spec, cwd) + + payload: dict[str, Any] = { + "schema_version": AUDIT_CYCLE_SCHEMA_VERSION, + "plan_ref": plan_ref.to_dict(), + "disposition_ref": disposition_ref.to_dict(), + "parent_authority_digest": parent_authority_digest, + } + payload["association_digest"] = compute_canonical_hash( + payload, domain=_PLAN_ASSOCIATION_DOMAIN + ) + if frozenset(payload) != _PLAN_ASSOCIATION_KEYS: + raise _FieldError("plan_association payload key set does not match _PLAN_ASSOCIATION_KEYS") + return payload + + +def _build_inventory(fields: dict[str, Any], cwd: str) -> dict[str, Any]: + del cwd # inventory has no referenced-artifact fields + requirement_ids = fields.get("requirement_ids") + requirements = fields.get("requirements") + if not isinstance(requirement_ids, list) or not isinstance(requirements, list): + raise _FieldError("fields['requirement_ids'] and fields['requirements'] must be arrays") + if not all(isinstance(item, dict) for item in requirements): + raise _FieldError("fields['requirements'] entries must be objects") + row_ids = [item.get("id") for item in requirements] + if list(requirement_ids) != row_ids: + raise _FieldError("requirement_ids and requirements order/content differ") + if any(not isinstance(item, str) or not item for item in requirement_ids): + raise _FieldError("requirement IDs must be non-empty strings") + return dict(fields) + + +_KIND_BUILDERS: dict[str, Callable[[dict[str, Any], str], dict[str, Any]]] = { + "authority": _build_authority, + "inventory": _build_inventory, + "disposition_report": _build_disposition_report, + "plan_association": _build_plan_association, +} + + +def write_audit_cycle_artifact_sync( + *, + kind: str, + path: str, + fields: dict[str, Any], + cwd: str, +) -> dict[str, Any]: + """Validate, construct, digest, and canonically write one audit-cycle artifact. + + Pure synchronous implementation — no MCP context required. Never raises; + every failure mode returns a structured ``{"success": False, "error": ...}`` + envelope, and no bytes are written until every structural and containment + check has passed. + """ + try: + builder = _KIND_BUILDERS.get(kind) + if builder is None: + return {"success": False, "error": f"unknown kind: {kind!r}"} + + if not cwd or not os.path.isdir(cwd): + return {"success": False, "error": f"cwd does not exist or is not a directory: {cwd}"} + + from autoskillit.server.git import validate_commit_paths # circular-break + + if (containment_error := validate_commit_paths(cwd, [path])) is not None: + return {"success": False, "error": containment_error} + + if not isinstance(fields, dict): + return {"success": False, "error": "fields must be an object"} + + try: + final_dict = builder(fields, cwd) + except _FieldError as exc: + return {"success": False, "error": str(exc)} + except (ContainmentError, OSError) as exc: + return {"success": False, "error": f"referenced artifact read failed: {exc}"} + except ValueError as exc: + return {"success": False, "error": str(exc)} + + try: + write_canonical_versioned_json( + Path(path), final_dict, AUDIT_CYCLE_SCHEMA_VERSION, exclusive=True + ) + except FileExistsError: + return {"success": False, "error": f"artifact already exists at {path}"} + except OSError as exc: + return {"success": False, "error": f"write failed: {type(exc).__name__}: {exc}"} + + content_digest = compute_bytes_hash(Path(path).read_bytes()) + return {"success": True, "path": path, "content_digest": content_digest} + except Exception as exc: + logger.error("write_audit_cycle_artifact_sync unhandled exception", exc_info=True) + return {"success": False, "error": f"{type(exc).__name__}: {exc}"} + + +@mcp.tool( + tags={"autoskillit", "kitchen", "kitchen-core", "headless"}, + annotations={"readOnlyHint": True}, +) +@_cancellation_shield() +@track_response_size("write_audit_cycle_artifact") +async def write_audit_cycle_artifact( + kind: str, + path: str, + fields: dict[str, Any], + cwd: str, + step_name: str = "", + ctx: Context = CurrentContext(), +) -> str: + """Server-side construct, digest, validate, and canonically write one audit-cycle artifact. + + Removes both JSON-canonicalization and digest-computation from the calling + session's token-generation path for the four hash-bound audit-cycle artifact + kinds consumed with ``require_canonical=True`` by the audit-cycle verifier: + ``authority``, ``inventory``, ``disposition_report``, ``plan_association``. + + Args: + kind: One of "authority", "inventory", "disposition_report", "plan_association". + path: Absolute destination path for the new artifact — must not already exist. + fields: Kind-specific semantic fields only — no pre-computed digests, no + pre-serialized bytes. Every digest is computed server-side. + cwd: Absolute containment root for both the destination and every + referenced artifact path inside ``fields``. + step_name: Optional YAML step key for wall-clock timing accumulation. + + Never raises. + """ + try: + with structlog.contextvars.bound_contextvars(tool="write_audit_cycle_artifact", cwd=cwd): + logger.info("write_audit_cycle_artifact", kind=kind, cwd=cwd) + + from autoskillit.server import _get_ctx # circular-break + + tool_ctx = _get_ctx() + _start = time.monotonic() + try: + result = write_audit_cycle_artifact_sync( + kind=kind, path=path, fields=fields, cwd=cwd + ) + finally: + if step_name: + tool_ctx.timing_log.record(step_name, time.monotonic() - _start) + return json.dumps(result) + except Exception as exc: + logger.error("write_audit_cycle_artifact unhandled exception", exc_info=True) + return json.dumps({"success": False, "error": f"{type(exc).__name__}: {exc}"}) diff --git a/src/autoskillit/skills_extended/audit-impl/SKILL.md b/src/autoskillit/skills_extended/audit-impl/SKILL.md index 4b4800ee47..bbe592d8df 100644 --- a/src/autoskillit/skills_extended/audit-impl/SKILL.md +++ b/src/autoskillit/skills_extended/audit-impl/SKILL.md @@ -1,7 +1,7 @@ --- name: audit-impl categories: [audit] -uses_capabilities: [agent_model, agent_subagent] +uses_capabilities: [agent_model, agent_subagent, write_audit_cycle_artifact] description: Audit a completed implementation against its originating plan(s). Returns GO (merge approved) or NO GO (generates remediation file for retry). Final gate before merge in any implementation pipeline. hooks: PreToolUse: @@ -74,7 +74,11 @@ requirements, scope creep, and unexpected changes. Produces a GO or NO GO verdic - Spawn all subagents via `Agent(model="sonnet")` - Resolve all plan files before starting (abort early if any are missing) - Issue all Task calls in a single message to maximize parallelism -- On every verdict, write one immutable `AuditCycleAuthority` and emit its **absolute path** +- On every verdict, call `write_audit_cycle_artifact(kind="authority", path=..., fields={...}, + cwd=...)` with the full field list (execution_generation, cycle_id, plan_set_id, scope_id, + part_id, audit_round, parent_authority_digest, audited_plan_refs, inventory_ref, + assessments, verdict, remediation_ref, generated_at) — the tool computes every digest and + writes byte-exact canonical JSON server-side. Emit the returned `path` as **absolute path** as `audit_cycle_path`. On `NO GO`, also emit the remediation path: ``` verdict = NO GO @@ -228,6 +232,7 @@ Each Explore subagent returns: "schema_version": 1, "generated_at": "ISO-8601", "plan_set_id": "identity derived from the explicit ordered audited plan refs", + "requirement_ids": ["REQ-001"], "requirements": [ { "id": "REQ-001", @@ -240,17 +245,32 @@ Each Explore subagent returns: } ``` -Write inventory, remediation (when any), closure report (when any), and authority under: +`requirement_ids` must be the ordered list of every `requirements[*].id`, in the same order. + +Write remediation (when any) and closure report (when any) as plain files, and the +inventory and authority via `write_audit_cycle_artifact(...)`, under: ``` {{AUTOSKILLIT_TEMP}}/audit-impl/cycles/{execution_generation}/{plan_set_id}/{scope_id}/{part_id}/round-{N}/ ``` -The `AuditCycleAuthority` must include the explicit execution generation, cycle ID, -plan-set/scope/part IDs, round, parent authority digest, audited plan refs, inventory ref, -ordered assessment rows, findings digest, verdict, generated timestamp, and authority -digest. `NO GO` requires a remediation `ArtifactRef`; `GO` requires `remediation_ref=null`. -Never rewrite an authority after hashing it. +In this order (the authority's `inventory_ref` must reference an already-written inventory +file so the tool can hash it): + +1. Call `write_audit_cycle_artifact(kind="inventory", path=".../round-{N}/inventory.json", + fields={...}, cwd=...)` with `fields` containing `schema_version`, `generated_at`, + `plan_set_id`, `requirement_ids`, and `requirements` per the schema above. +2. Call `write_audit_cycle_artifact(kind="authority", path=".../round-{N}/authority.json", + fields={...}, cwd=...)` with `fields` containing the explicit execution generation, cycle + ID, plan-set/scope/part IDs, round, parent authority digest, audited plan refs, + `inventory_ref` (referencing the inventory file just written in step 1), ordered + assessment rows, verdict, remediation ref, and generated timestamp. `NO GO` requires a + non-null `remediation_ref`; `GO` requires `remediation_ref=null`. + +The tool computes `findings_digest` and `authority_digest` internally from the +`assessments`/`verdict` fields and rejects any semantically inconsistent payload before +writing — never hand-assemble these digests. Never rewrite an authority after it has +been written. ### Step 2 — Load Implementation Diff diff --git a/src/autoskillit/skills_extended/make-plan/SKILL.md b/src/autoskillit/skills_extended/make-plan/SKILL.md index b7b00fb9f7..40f90f0b42 100644 --- a/src/autoskillit/skills_extended/make-plan/SKILL.md +++ b/src/autoskillit/skills_extended/make-plan/SKILL.md @@ -1,6 +1,6 @@ --- name: make-plan -uses_capabilities: [agent_model, agent_subagent] +uses_capabilities: [agent_model, agent_subagent, write_audit_cycle_artifact] activate_deps: [write-recipe] description: Planning executor. ALWAYS invoke this skill when instructed to create, devise, or write an implementation plan. Do not explore the codebase or draft a plan directly — use this skill first to load the planning workflow. hooks: @@ -336,15 +336,28 @@ orchestrators can capture the plan and, in remediation mode, its verified dispos `carried@step` must cite the concrete current `Step N`/`Step N.M` that implements the same REQ ID. `satisfied-by-round-N` must name the verified prior audit round. No other vocabulary, duplicate IDs, omitted rows, or invented padding is allowed. -5. After the final plan bytes are stable, create a canonical immutable - `PlanDispositionReport` bound to the parent authority digest, full cycle identity, - verified plan `ArtifactRef`, exact ordered disposition rows, timestamp, and report digest. - Verify the report against the plan with the production inventory-admission evaluator. -6. In the current cycle directory, write exactly one immutable association at - `associations/{verified_plan_content_digest}.json`. It contains exactly the verified - plan ref, disposition ref, parent authority digest, schema version, and association - digest. Refuse an existing different record; never search for or synthesize a latest - report. +5. After the final plan bytes are stable, call + `write_audit_cycle_artifact(kind="disposition_report", path=..., fields={...}, cwd=...)` — + this constructs and canonically writes the immutable `PlanDispositionReport` server-side — + with `fields` containing `execution_generation`, `cycle_id`, `plan_set_id`, `scope_id`, + `part_id`, `audit_round`, `current_plan_ref` (the verified plan `ArtifactRef` fields), the + exact ordered `dispositions` rows, and `generated_at` — plus `parent_authority_digest`, + `inventory_digest`, and `findings_digest` sourced directly from the parent + `AuditCycleAuthority` object verified in Step 2 above (respectively + `authority.authority_digest`, `authority.inventory_ref.content_digest`, and + `authority.findings_digest`). These three digests are not derivable from + `current_plan_ref`/`dispositions` alone — the tool requires them as explicit mandatory + fields, and rejects a payload that misvalues any of them. Verify the returned report + against the plan with the production inventory-admission evaluator. +6. Call + `write_audit_cycle_artifact(kind="plan_association", path="associations/{verified_plan_content_digest}.json", fields={...}, cwd=...)` + in the current cycle directory, with `fields` containing `plan_ref` (the verified plan + `ArtifactRef` fields), + `disposition_ref` (referencing the disposition report just written in step 5), and + `parent_authority_digest`. The tool computes the association digest server-side and + writes exactly the five canonical keys (`schema_version`, `plan_ref`, `disposition_ref`, + `parent_authority_digest`, `association_digest`). Refuse an existing different record; + never search for or synthesize a latest report. 7. Absence, duplication, evaluator rejection, or Markdown/report drift is an output-contract failure. Do not emit successful structured tokens. diff --git a/src/autoskillit/workspace/skill_capabilities.py b/src/autoskillit/workspace/skill_capabilities.py index 7be2a369ed..723a79555b 100644 --- a/src/autoskillit/workspace/skill_capabilities.py +++ b/src/autoskillit/workspace/skill_capabilities.py @@ -273,6 +273,7 @@ class _SourceLine: "agent_model": (re.compile(r"Agent\(\s*model\s*="),), "claude_dir": (re.compile(r"\.claude/"),), "commit_files": (re.compile(r"\bcommit_files\s*\("),), + "write_audit_cycle_artifact": (re.compile(r"\bwrite_audit_cycle_artifact\s*\("),), "git_metadata_write": ( re.compile(r"create_impl_worktree\.sh|git worktree add\b[ \t]+\S|git checkout -b"), re.compile(r"\bgit\s+(?:-C\s+\S+\s+)?commit(?:\s|$)"), @@ -490,6 +491,7 @@ def _classify_context( "Agent(", ".claude/", "commit_files", + "write_audit_cycle_artifact", "git ", "gh ", "open_kitchen", diff --git a/tests/_test_filter.py b/tests/_test_filter.py index 182672b39c..3a6e5905b2 100644 --- a/tests/_test_filter.py +++ b/tests/_test_filter.py @@ -292,8 +292,8 @@ class ImportContext(enum.StrEnum): "context_admission": frozenset({"core", "pipeline"}), "audit_cycle_verifier": frozenset({"core", "recipe", "server"}), "tool_registry": frozenset({"core", "recipe", "server"}), - "closure_hashing": frozenset({"core", "recipe"}), - "path_containment": frozenset({"core", "recipe"}), + "closure_hashing": frozenset({"core", "recipe", "server"}), + "path_containment": frozenset({"core", "recipe", "server"}), "closure_verifier": frozenset({"core", "execution"}), } @@ -849,6 +849,7 @@ class ImportContext(enum.StrEnum): "server/test_pipeline_deps_derivation.py", "server/test_pipeline_tracker.py", "server/test_audit_cycle_delivery_integration.py", + "server/test_tools_audit_cycle.py", # CLI file-level entries (6 of 38 import autoskillit.recipe): "cli/test_cli_prompts.py", "cli/test_l3_orchestrator_prompt.py", diff --git a/tests/arch/test_layer_enforcement.py b/tests/arch/test_layer_enforcement.py index 83cc8c0260..352a11e75a 100644 --- a/tests/arch/test_layer_enforcement.py +++ b/tests/arch/test_layer_enforcement.py @@ -1630,6 +1630,10 @@ def test_default_classes_only_instantiated_inside_factory_or_allowlist() -> None # core tests — protocol conformance checks require concrete implementations "tests/core/test_core_terminal_table.py": frozenset({"autoskillit.cli"}), "tests/core/test_types.py": frozenset({"autoskillit.execution"}), + # audit-cycle attack test's legitimate-bytes fixture is produced by the sanctioned + # server-side producer (write_audit_cycle_artifact_sync) rather than hand-assembled + # via the dataclass property — this is the exact gap #4406 Part A closes + "tests/core/test_audit_cycle_attacks.py": frozenset({"autoskillit.server"}), # execution tests — clone_guard/headless/commands use sibling layers "tests/execution/test_clone_guard.py": frozenset({"autoskillit.pipeline"}), "tests/execution/test_commands.py": frozenset({"autoskillit.cli"}), @@ -1844,6 +1848,7 @@ def test_tools_with_path_params_validate_existence(): "ci_watcher", "_close_issues_sequentially", "tool_ctx.executor", + "write_audit_cycle_artifact_sync", ) ) if not has_guard: diff --git a/tests/arch/test_subpackage_isolation.py b/tests/arch/test_subpackage_isolation.py index d05bf57650..c4b143a580 100644 --- a/tests/arch/test_subpackage_isolation.py +++ b/tests/arch/test_subpackage_isolation.py @@ -950,10 +950,11 @@ def test_no_subpackage_exceeds_10_files() -> None: "pipeline": 14, # +context admission ledger +recipe initialization reducer "fleet": 23, # +_issue_url_helpers.py # noqa: E501 "recipe/rules": 55, # +commit_guard_regression_route +rules_model +rules_gitignored_deliverable +rules_issue_scope_threading +rules_inventory_gate_bilateral +rules_verdict_context +rules_contract_recovery # noqa: E501 - "server/tools": 32, # +_pipeline_deps.py +_ordering_telemetry.py (open_kitchen + "server/tools": 33, # +_pipeline_deps.py +_ordering_telemetry.py (open_kitchen # auto-init dependency tracker + REVIEW_BEFORE_PLAN ordering telemetry) # +_backend_compat.py (shared target-resolution + fail-closed compatibility gate # for direct headless executor callers — report_bug, prepare_issue, enrich_issues) + # +tools_audit_cycle.py (server-side write_audit_cycle_artifact MCP tool, #4406) "hooks/guards": 32, # -output_budget_guard (#4286) # Three private Codex ownership modules keep lock, prelaunch transaction, # and per-attempt storage concerns out of the public backend gateway: diff --git a/tests/core/test_audit_cycle_attacks.py b/tests/core/test_audit_cycle_attacks.py index 5e89aee9dc..5174ecc5ab 100644 --- a/tests/core/test_audit_cycle_attacks.py +++ b/tests/core/test_audit_cycle_attacks.py @@ -122,11 +122,64 @@ def test_artifact_reference_rejects_size_digest_and_post_reference_mutation( def test_authority_rejects_forged_digest_and_noncanonical_bytes(tmp_path: Path) -> None: - authority = _authority(tmp_path) + """The legitimate-bytes half is produced by the sanctioned production + write path (write_audit_cycle_artifact_sync), not by AuditCycleAuthority + constructed and serialized directly in the test — closing the gap where + this attack test's own "legitimate" fixture bypassed the real producer. + """ + from autoskillit.server.tools.tools_audit_cycle import write_audit_cycle_artifact_sync + + (tmp_path / "plan.md").write_bytes(b"plan") + (tmp_path / "inventory.json").write_bytes(b"inventory") + (tmp_path / "remediation.md").write_bytes(b"remediation") + path = tmp_path / "authority.json" - path.write_bytes(authority.canonical_bytes) + result = write_audit_cycle_artifact_sync( + kind="authority", + path=str(path), + fields={ + "execution_generation": "generation-1", + "cycle_id": "cycle-1", + "plan_set_id": "plans-1", + "scope_id": "scope-1", + "part_id": "part-a", + "audit_round": 1, + "parent_authority_digest": None, + "audited_plan_refs": [ + { + "locator": str(tmp_path / "plan.md"), + "media_type": "application/json", + "schema_version": 1, + }, + ], + "inventory_ref": { + "locator": str(tmp_path / "inventory.json"), + "media_type": "application/json", + "schema_version": 1, + }, + "remediation_ref": { + "locator": str(tmp_path / "remediation.md"), + "media_type": "application/json", + "schema_version": 1, + }, + "assessments": [ + { + "requirement_id": "REQ-001", + "requirement_text": "requirement", + "assessment": "MISSING", + "evidence_summary": "missing", + } + ], + "verdict": "NO GO", + "generated_at": "2026-07-23T00:01:00Z", + }, + cwd=str(tmp_path), + ) + assert result["success"] is True, result + verifier = AuditCycleVerifier(tmp_path) - assert verifier.load_authority(path) == authority + authority = verifier.load_authority(path) + assert authority.cycle_id == "cycle-1" forged = authority.to_dict() forged["cycle_id"] = "forged" import json diff --git a/tests/core/test_io.py b/tests/core/test_io.py index f50c986c14..6ca6a2d586 100644 --- a/tests/core/test_io.py +++ b/tests/core/test_io.py @@ -288,6 +288,52 @@ def fail_parent_fsync(fd): assert target.read_text() == "{}" +def test_atomic_write_exclusive_raises_file_exists_error_without_overwriting(tmp_path): + from autoskillit.core.io import atomic_write + + target = tmp_path / "claimed.json" + target.write_text("original content") + + with pytest.raises(FileExistsError): + atomic_write(target, "new content", exclusive=True) + + assert target.read_text() == "original content" + + +def test_atomic_write_default_exclusive_false_preserves_overwrite_behavior(tmp_path): + from autoskillit.core.io import atomic_write + + target = tmp_path / "overwritable.json" + target.write_text("original content") + + atomic_write(target, "new content") + + assert target.read_text() == "new content" + + +def test_write_canonical_versioned_json_writes_canonical_bytes(tmp_path): + from autoskillit.core.closure_hashing import parse_canonical_json_bytes + from autoskillit.core.io import write_canonical_versioned_json + + target = tmp_path / "canonical.json" + write_canonical_versioned_json(target, {"b": 2, "a": 1}, schema_version=1) + + parsed = parse_canonical_json_bytes(target.read_bytes()) + assert parsed == {"a": 1, "b": 2, "schema_version": 1} + + +def test_write_canonical_versioned_json_forwards_exclusive(tmp_path): + from autoskillit.core.io import write_canonical_versioned_json + + target = tmp_path / "canonical.json" + target.write_bytes(b"pre-existing") + + with pytest.raises(FileExistsError): + write_canonical_versioned_json(target, {"a": 1}, schema_version=1, exclusive=True) + + assert target.read_bytes() == b"pre-existing" + + def test_write_versioned_json_forwards_strict_durability(tmp_path, monkeypatch): from autoskillit.core import io as io_mod diff --git a/tests/core/test_type_constants.py b/tests/core/test_type_constants.py index d879f6e2ed..875ce3e00a 100644 --- a/tests/core/test_type_constants.py +++ b/tests/core/test_type_constants.py @@ -405,7 +405,12 @@ def test_provider_profile_in_private_env_vars() -> None: def test_headless_tools_contains_expected_names(): from autoskillit.core.types import HEADLESS_TOOLS - assert HEADLESS_TOOLS == {"test_check", "unlock_agent_pack", "commit_files"} + assert HEADLESS_TOOLS == { + "test_check", + "unlock_agent_pack", + "commit_files", + "write_audit_cycle_artifact", + } def test_free_range_tools_contains_expected_names(): diff --git a/tests/docs/test_doc_counts.py b/tests/docs/test_doc_counts.py index 954846b392..6e9276b812 100644 --- a/tests/docs/test_doc_counts.py +++ b/tests/docs/test_doc_counts.py @@ -207,9 +207,9 @@ def _count_semantic_rule_files() -> int: # ----- tests ------------------------------------------------------------------ -def test_kitchen_tagged_tool_count_is_42() -> None: +def test_kitchen_tagged_tool_count_is_43() -> None: count = _count_kitchen_tools() - assert count == 42, f"Expected 42 kitchen-tagged tools; found {count}" + assert count == 43, f"Expected 43 kitchen-tagged tools; found {count}" def test_free_range_tool_count_is_17() -> None: @@ -218,9 +218,9 @@ def test_free_range_tool_count_is_17() -> None: ) -def test_headless_tool_count_is_3() -> None: - assert _count_headless_tools() == 3, ( - f"Expected 3 headless-tagged tools; found {_count_headless_tools()}" +def test_headless_tool_count_is_4() -> None: + assert _count_headless_tools() == 4, ( + f"Expected 4 headless-tagged tools; found {_count_headless_tools()}" ) @@ -302,8 +302,8 @@ def _assert_doc_states_number(doc: Path, label: str, expected: int) -> None: DOCS_DIR / "execution" / "tool-access.md", ], ) -def test_docs_state_61_mcp_tools(doc_path: Path) -> None: - _assert_doc_states_number(doc_path, "MCP tools", 61) +def test_docs_state_62_mcp_tools(doc_path: Path) -> None: + _assert_doc_states_number(doc_path, "MCP tools", 62) @pytest.mark.parametrize( diff --git a/tests/infra/AGENTS.md b/tests/infra/AGENTS.md index d12ce5bb2c..7f9ce6ee71 100644 --- a/tests/infra/AGENTS.md +++ b/tests/infra/AGENTS.md @@ -16,6 +16,7 @@ CI/CD configuration, security, guard coverage, and release sanity tests. | `test_ask_user_question_guard.py` | Tests for the ask_user_question_guard PreToolUse hook | | `test_background_exec_guard.py` | Tests for background_exec_guard.py PreToolUse hook — blocks run_in_background=true in skill sessions | | `test_branch_protection_guard.py` | Tests for hooks/branch_protection_guard.py — PreToolUse branch protection | +| `test_canonical_json_producer_convention.py` | Producer ratchet: every `require_canonical=True` JSON consumer site must have a registered, verified server-side producer and SKILL.md reference | | `test_check_pyi_stub_format.py` | Unit tests for scripts/check_pyi_stub_format.py pre-commit hook — validates FunctionDef, ClassDef, and non-relative import rejection | | `test_check_pyi_stub_symbols.py` | Unit tests for scripts/check_pyi_stub_symbols.py pre-commit hook — validates missing symbol detection, completeness acceptance, underscore skipping, and __all__ usage | | `test_ci_dev_config.py` | Structural enforcement: CI workflow and pre-commit configuration must contain required quality gates | diff --git a/tests/infra/test_canonical_json_producer_convention.py b/tests/infra/test_canonical_json_producer_convention.py new file mode 100644 index 0000000000..d533deac2d --- /dev/null +++ b/tests/infra/test_canonical_json_producer_convention.py @@ -0,0 +1,236 @@ +"""Producer ratchet: enforce that every strict-canonical JSON consumer has a +registered, verified server-side producer. + +Scans src/autoskillit/ for decode_versioned_json_bytes(..., require_canonical=True) +call sites. Each such site must be registered in _CANONICAL_JSON_ARTIFACT_REGISTRY, +pointing at a producer function that itself calls write_canonical_versioned_json, and +at a SKILL.md section that names the producer's MCP tool by symbol. This closes the +gap #4406 exhibited: a Python consumer demanding canonical bytes with no mechanical +guarantee that its LLM-agent producer emits them. +""" + +from __future__ import annotations + +import ast +from pathlib import Path +from typing import NamedTuple + +import pytest + +pytestmark = [pytest.mark.layer("infra"), pytest.mark.medium] + + +class CanonicalArtifactDef(NamedTuple): + consumer_site: tuple[str, int] + producer_symbol: str + producer_site: tuple[str, int] + skill_md_refs: tuple[tuple[str, int, int], ...] + + +def _is_literal_true(node: ast.expr) -> bool: + return isinstance(node, ast.Constant) and node.value is True + + +def _scan_require_canonical_consumer_sites() -> set[tuple[str, int]]: + """AST-scan src/autoskillit/ for decode_versioned_json_bytes(require_canonical=True). + + Returns set of (relative_path, line_number) for call sites that pass + require_canonical=True as a keyword argument. + """ + src_root = Path(__file__).resolve().parents[2] / "src" / "autoskillit" + sites: set[tuple[str, int]] = set() + + for py_file in src_root.rglob("*.py"): + try: + tree = ast.parse(py_file.read_text(), filename=str(py_file)) + except SyntaxError: + continue + + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + func = node.func + is_decode_call = ( + isinstance(func, ast.Name) and func.id == "decode_versioned_json_bytes" + ) or (isinstance(func, ast.Attribute) and func.attr == "decode_versioned_json_bytes") + if not is_decode_call: + continue + for kw in node.keywords: + if kw.arg == "require_canonical" and _is_literal_true(kw.value): + rel = str(py_file.relative_to(src_root.parent.parent)) + sites.add((rel, node.lineno)) + break + + return sites + + +def _find_function_at_line( + tree: ast.Module, lineno: int +) -> ast.FunctionDef | ast.AsyncFunctionDef | None: + for node in ast.walk(tree): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.lineno == lineno: + return node + return None + + +def _find_call_at_line(tree: ast.Module, lineno: int, func_name: str) -> ast.Call | None: + for node in ast.walk(tree): + if not isinstance(node, ast.Call) or node.lineno != lineno: + continue + func = node.func + matches = (isinstance(func, ast.Name) and func.id == func_name) or ( + isinstance(func, ast.Attribute) and func.attr == func_name + ) + if matches: + return node + return None + + +# All four canonical audit-cycle artifact kinds converge on this one function before +# their bytes are written — see server/tools/tools_audit_cycle.py:332-384. +_AUDIT_CYCLE_ARTIFACT_PRODUCER_SITE: tuple[str, int] = ( + "src/autoskillit/server/tools/tools_audit_cycle.py", + 332, +) + +_CANONICAL_JSON_ARTIFACT_REGISTRY: dict[str, CanonicalArtifactDef] = { + "authority": CanonicalArtifactDef( + consumer_site=("src/autoskillit/core/audit_cycle_verifier.py", 424), + producer_symbol="write_audit_cycle_artifact", + producer_site=_AUDIT_CYCLE_ARTIFACT_PRODUCER_SITE, + skill_md_refs=(("src/autoskillit/skills_extended/audit-impl/SKILL.md", 77, 81),), + ), + "disposition_report": CanonicalArtifactDef( + consumer_site=("src/autoskillit/core/audit_cycle_verifier.py", 444), + producer_symbol="write_audit_cycle_artifact", + producer_site=_AUDIT_CYCLE_ARTIFACT_PRODUCER_SITE, + skill_md_refs=(("src/autoskillit/skills_extended/make-plan/SKILL.md", 339, 341),), + ), + "inventory": CanonicalArtifactDef( + consumer_site=("src/autoskillit/core/audit_cycle_verifier.py", 572), + producer_symbol="write_audit_cycle_artifact", + producer_site=_AUDIT_CYCLE_ARTIFACT_PRODUCER_SITE, + skill_md_refs=(("src/autoskillit/skills_extended/audit-impl/SKILL.md", 260, 262),), + ), + "plan_association": CanonicalArtifactDef( + consumer_site=("src/autoskillit/recipe/_cmd_rpc_guards.py", 274), + producer_symbol="write_audit_cycle_artifact", + producer_site=_AUDIT_CYCLE_ARTIFACT_PRODUCER_SITE, + skill_md_refs=(("src/autoskillit/skills_extended/make-plan/SKILL.md", 352, 354),), + ), +} + + +# Documented exception: consumer sites that intentionally decode non-canonical +# versioned JSON. Paired with the writer that produces them so a future accidental +# flip to require_canonical=True is caught by test_non_canonical_exceptions_below. +_NON_CANONICAL_JSON_EXCEPTIONS: dict[str, tuple[tuple[str, int], str]] = { + "closure_report.json": ( + ("src/autoskillit/core/closure_verifier.py", 59), + "Closure reports are written by write_versioned_json (see the " + "_write_report fixture in tests/core/test_closure_verifier.py), not " + "write_canonical_versioned_json — no content-addressed tamper-evidence " + "chain requires byte-exact canonical bytes for this artifact.", + ), +} + + +class TestCanonicalJsonProducerConvention: + def test_require_canonical_consumers_have_registered_producers(self): + """Every require_canonical=True consumer site must have a registry entry.""" + current = _scan_require_canonical_consumer_sites() + registered = {entry.consumer_site for entry in _CANONICAL_JSON_ARTIFACT_REGISTRY.values()} + added = current - registered + removed = registered - current + + msg_parts = [] + if added: + msg_parts.append( + "New require_canonical=True consumer sites found (register a " + "CanonicalArtifactDef in _CANONICAL_JSON_ARTIFACT_REGISTRY):\n" + + "\n".join(f" + {f}:{ln}" for f, ln in sorted(added)) + ) + if removed: + msg_parts.append( + "Registered consumer sites no longer found (remove from " + "_CANONICAL_JSON_ARTIFACT_REGISTRY):\n" + + "\n".join(f" - {f}:{ln}" for f, ln in sorted(removed)) + ) + assert current == registered, "\n\n".join(msg_parts) + + def test_registered_producers_are_the_sanctioned_writer(self): + """Every registered producer must itself call write_canonical_versioned_json.""" + repo_root = Path(__file__).resolve().parents[2] + checked: set[tuple[str, int]] = set() + for kind, entry in _CANONICAL_JSON_ARTIFACT_REGISTRY.items(): + site = entry.producer_site + if site in checked: + continue + checked.add(site) + relative_path, lineno = site + source_path = repo_root / relative_path + tree = ast.parse(source_path.read_text(), filename=str(source_path)) + function = _find_function_at_line(tree, lineno) + assert function is not None, f"{kind}: no function defined at {relative_path}:{lineno}" + call_names = { + node.func.id if isinstance(node.func, ast.Name) else node.func.attr + for node in ast.walk(function) + if isinstance(node, ast.Call) and isinstance(node.func, (ast.Name, ast.Attribute)) + } + assert "write_canonical_versioned_json" in call_names, ( + f"{kind}: producer at {relative_path}:{lineno} no longer calls " + "write_canonical_versioned_json" + ) + + def test_registered_skill_md_refs_name_the_producer(self): + """Every registered skill_md_ref section must mention the producer's symbol.""" + repo_root = Path(__file__).resolve().parents[2] + for kind, entry in _CANONICAL_JSON_ARTIFACT_REGISTRY.items(): + for relative_path, start_line, end_line in entry.skill_md_refs: + skill_md_path = repo_root / relative_path + lines = skill_md_path.read_text().splitlines() + section = "\n".join(lines[start_line - 1 : end_line]) + assert entry.producer_symbol in section, ( + f"{kind}: {relative_path}:{start_line}-{end_line} does not mention " + f"{entry.producer_symbol!r}" + ) + + def test_new_require_canonical_consumer_without_registered_producer_fails(self, monkeypatch): + """Meta-test: a fake extra consumer site should cause the ratchet to fail.""" + original_scan = _scan_require_canonical_consumer_sites + + def patched_scan(): + sites = original_scan() + sites.add(("src/autoskillit/fake_canonical_module.py", 999)) + return sites + + monkeypatch.setattr( + "tests.infra.test_canonical_json_producer_convention." + "_scan_require_canonical_consumer_sites", + patched_scan, + ) + with pytest.raises(AssertionError, match="fake_canonical_module"): + self.test_require_canonical_consumers_have_registered_producers() + + def test_non_canonical_exceptions_still_resolve_to_non_canonical_reads(self): + """Documented non-canonical exceptions must not pass require_canonical=True.""" + repo_root = Path(__file__).resolve().parents[2] + for artifact_name, (site, reason) in _NON_CANONICAL_JSON_EXCEPTIONS.items(): + assert reason, f"{artifact_name} has an empty exception reason" + relative_path, lineno = site + source_path = repo_root / relative_path + tree = ast.parse(source_path.read_text(), filename=str(source_path)) + call = _find_call_at_line(tree, lineno, "decode_versioned_json_bytes") + assert call is not None, ( + f"{artifact_name}: no decode_versioned_json_bytes call found at " + f"{relative_path}:{lineno}" + ) + for kw in call.keywords: + if kw.arg == "require_canonical": + assert not _is_literal_true(kw.value), ( + f"{artifact_name}: {relative_path}:{lineno} now passes " + "require_canonical=True — this pairing is documented as " + "non-canonical in _NON_CANONICAL_JSON_EXCEPTIONS; either " + "register it in _CANONICAL_JSON_ARTIFACT_REGISTRY instead or " + "revert the flag" + ) diff --git a/tests/recipe/test_rules_tools.py b/tests/recipe/test_rules_tools.py index ea35a93f31..2bc9066827 100644 --- a/tests/recipe/test_rules_tools.py +++ b/tests/recipe/test_rules_tools.py @@ -355,6 +355,7 @@ def test_rebase_then_push_with_force_true_passes_validation() -> None: "autoskillit.server.tools.tools_pr_ops", "autoskillit.server.tools.tools_workspace", "autoskillit.server.tools.tools_agents", + "autoskillit.server.tools.tools_audit_cycle", "autoskillit.server.tools.tools_config", "autoskillit.server.tools.tools_kitchen", "autoskillit.server.tools.tools_pipeline_tracker", diff --git a/tests/server/AGENTS.md b/tests/server/AGENTS.md index c4927dfde8..3e560419d0 100644 --- a/tests/server/AGENTS.md +++ b/tests/server/AGENTS.md @@ -141,6 +141,7 @@ Server tool handler unit tests — kitchen, execution, CI, clone, workspace tool | `test_tools_recipe_pull.py` | Tests for the `get_recipe_section` pull tool and bounded envelope architecture (Part B #4304) | | `test_tool_registry_parity.py` | Bidirectional AST parity between canonical IL-0 tool metadata and live MCP handler signatures | | `test_audit_cycle_delivery_integration.py` | Attested payload installation, exact runtime binding, trusted-head CAS, and zero-read preflight | +| `test_tools_audit_cycle.py` | Tests for `write_audit_cycle_artifact`: producer/consumer round-trip, containment, structural validation, write-once, and read-before-validate ordering | | `test_tools_report_bug.py` | Tests for report_bug MCP tool handler and supporting helpers (_parse_fingerprint, _extract_block, _parse_prepare_result, _parse_enrich_result) | | `test_tools_run_cmd.py` | Tests for run_cmd and run_python MCP tool handlers | | `test_tools_run_cmd_invariants.py` | Server-side invariant tests for run_cmd: recipe-read prohibition and write-target boundary | diff --git a/tests/server/test_server_tool_registration.py b/tests/server/test_server_tool_registration.py index f1bf396364..71bd2450b9 100644 --- a/tests/server/test_server_tool_registration.py +++ b/tests/server/test_server_tool_registration.py @@ -110,6 +110,7 @@ async def test_all_tools_exist(self, kitchen_enabled): "reset_dispatch", "get_recipe_section", "complete_recipe_initialization", + "write_audit_cycle_artifact", } assert expected == tool_names diff --git a/tests/server/test_tool_registry_parity.py b/tests/server/test_tool_registry_parity.py index 8fdef4adf3..ac197242df 100644 --- a/tests/server/test_tool_registry_parity.py +++ b/tests/server/test_tool_registry_parity.py @@ -215,6 +215,7 @@ def test_every_tool_has_an_explicit_initialization_operation() -> None: "unlock_agent_pack", "wait_for_ci", "wait_for_merge_queue", + "write_audit_cycle_artifact", "write_telemetry_files", }, } diff --git a/tests/server/test_tools_audit_cycle.py b/tests/server/test_tools_audit_cycle.py new file mode 100644 index 0000000000..d8d7ecebdd --- /dev/null +++ b/tests/server/test_tools_audit_cycle.py @@ -0,0 +1,436 @@ +"""Tests for write_audit_cycle_artifact: producer/consumer round-trip (#4406), +containment, structural validation, write-once, and read-before-validate ordering. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from autoskillit.core import ( + AUDIT_CYCLE_SCHEMA_VERSION, + AuditCycleVerifier, + compute_bytes_hash, + decode_versioned_json_bytes, + parse_canonical_json_bytes, +) +from autoskillit.recipe._cmd_rpc_guards import _resolve_plan_disposition +from autoskillit.server.tools.tools_audit_cycle import write_audit_cycle_artifact_sync + +pytestmark = [pytest.mark.layer("server"), pytest.mark.small] + +_GENERATED_AT = "2026-07-23T00:00:00Z" + + +def _authority_fields( + *, + inventory_path: Path, + plan_path: Path, + remediation_path: Path, + parent_authority_digest: str | None = None, + audit_round: int = 1, +) -> dict: + return { + "execution_generation": "generation-1", + "cycle_id": "cycle-1", + "plan_set_id": "plans-1", + "scope_id": "scope-1", + "part_id": "part-a", + "audit_round": audit_round, + "parent_authority_digest": parent_authority_digest, + "audited_plan_refs": [ + {"locator": str(plan_path), "media_type": "text/markdown", "schema_version": 1}, + ], + "inventory_ref": { + "locator": str(inventory_path), + "media_type": "application/json", + "schema_version": AUDIT_CYCLE_SCHEMA_VERSION, + }, + "remediation_ref": { + "locator": str(remediation_path), + "media_type": "text/markdown", + "schema_version": 1, + }, + "assessments": [ + { + "requirement_id": "REQ-001", + "requirement_text": "Fix the issue", + "assessment": "MISSING", + "evidence_summary": "not present", + } + ], + "verdict": "NO GO", + "generated_at": _GENERATED_AT, + } + + +def _inventory_fields() -> dict: + return { + "schema_version": AUDIT_CYCLE_SCHEMA_VERSION, + "generated_at": _GENERATED_AT, + "plan_set_id": "plans-1", + "requirement_ids": ["REQ-001"], + "requirements": [{"id": "REQ-001", "text": "Fix the issue"}], + } + + +def _write_fixture_files(tmp_path: Path) -> tuple[Path, Path]: + """Write the plan and remediation source files a round-trip test references.""" + plan_path = tmp_path / "plan.md" + plan_path.write_text("## Implementation Steps\n\n### Step 1: Fix\n") + remediation_path = tmp_path / "remediation.md" + remediation_path.write_text("REQ-001 remains") + return plan_path, remediation_path + + +class TestProducerConsumerRoundTrip: + """Step 1 items 1a-1c: the sanctioned producer against the real consumer.""" + + def test_full_cycle_round_trip_accepted_by_real_consumers(self, tmp_path: Path) -> None: + cwd = str(tmp_path) + cycle_dir = tmp_path / "cycle" + plan_path, remediation_path = _write_fixture_files(tmp_path) + + inventory_path = cycle_dir / "inventory.json" + inventory_result = write_audit_cycle_artifact_sync( + kind="inventory", + path=str(inventory_path), + fields=_inventory_fields(), + cwd=cwd, + ) + assert inventory_result["success"] is True, inventory_result + parse_canonical_json_bytes(inventory_path.read_bytes()) # byte-exact canonical + + authority_path = cycle_dir / "authority.json" + authority_result = write_audit_cycle_artifact_sync( + kind="authority", + path=str(authority_path), + fields=_authority_fields( + inventory_path=inventory_path, + plan_path=plan_path, + remediation_path=remediation_path, + ), + cwd=cwd, + ) + assert authority_result["success"] is True, authority_result + parse_canonical_json_bytes(authority_path.read_bytes()) + + verifier = AuditCycleVerifier(tmp_path) + authority = verifier.load_authority(authority_path) + assert authority.verdict.value == "NO GO" + assert authority.assessments[0].requirement_id == "REQ-001" + + report_path = cycle_dir / "disposition.json" + report_result = write_audit_cycle_artifact_sync( + kind="disposition_report", + path=str(report_path), + fields={ + "execution_generation": authority.execution_generation, + "cycle_id": authority.cycle_id, + "plan_set_id": authority.plan_set_id, + "scope_id": authority.scope_id, + "part_id": authority.part_id, + "audit_round": authority.audit_round, + "parent_authority_digest": authority.authority_digest, + "inventory_digest": authority.inventory_ref.content_digest, + "findings_digest": authority.findings_digest, + "current_plan_ref": { + "locator": str(plan_path), + "media_type": "text/markdown", + "schema_version": 1, + }, + "dispositions": [ + { + "requirement_id": "REQ-001", + "disposition": "carried@step", + "implementation_step": "Step 1", + } + ], + "generated_at": _GENERATED_AT, + }, + cwd=cwd, + ) + assert report_result["success"] is True, report_result + parse_canonical_json_bytes(report_path.read_bytes()) + + report = verifier.load_report(report_path) + assert report.parent_authority_digest == authority.authority_digest + + plan_digest = compute_bytes_hash(plan_path.read_bytes()) + association_path = cycle_dir / "associations" / f"{plan_digest}.json" + association_result = write_audit_cycle_artifact_sync( + kind="plan_association", + path=str(association_path), + fields={ + "plan_ref": { + "locator": str(plan_path), + "media_type": "text/markdown", + "schema_version": 1, + }, + "disposition_ref": { + "locator": str(report_path), + "media_type": "application/json", + "schema_version": 1, + }, + "parent_authority_digest": authority.authority_digest, + }, + cwd=cwd, + ) + assert association_result["success"] is True, association_result + parse_canonical_json_bytes(association_path.read_bytes()) + + resolved = _resolve_plan_disposition( + audit_cycle_path=str(authority_path), + current_plan_path=plan_path, + ) + assert resolved == str(report_path) + + def test_inventory_shape_matches_the_inline_consumer_check(self, tmp_path: Path) -> None: + """Mirrors the exact inline check inside AuditCycleVerifier._verify_active_tuple.""" + inventory_path = tmp_path / "inventory.json" + result = write_audit_cycle_artifact_sync( + kind="inventory", + path=str(inventory_path), + fields=_inventory_fields(), + cwd=str(tmp_path), + ) + assert result["success"] is True + + raw = decode_versioned_json_bytes( + inventory_path.read_bytes(), + expected_version=AUDIT_CYCLE_SCHEMA_VERSION, + require_canonical=True, + ) + assert raw is not None + requirement_ids = tuple(raw["requirement_ids"]) + row_ids = tuple(item["id"] for item in raw["requirements"]) + assert requirement_ids == row_ids + assert all(isinstance(item, str) and item for item in requirement_ids) + + +class TestDestinationContainment: + def test_destination_escaping_cwd_is_rejected(self, tmp_path: Path) -> None: + outside = tmp_path.parent / "outside-dest" / "authority.json" + result = write_audit_cycle_artifact_sync( + kind="inventory", + path=str(outside), + fields=_inventory_fields(), + cwd=str(tmp_path), + ) + assert result["success"] is False + assert "escapes cwd" in result["error"] + assert not outside.exists() + + +class TestReferencedArtifactContainment: + def test_referenced_artifact_escaping_cwd_is_rejected(self, tmp_path: Path) -> None: + outside_dir = tmp_path.parent / "outside-plan" + outside_dir.mkdir(exist_ok=True) + outside_plan = outside_dir / "plan.md" + outside_plan.write_text("plan content") + _, remediation_path = _write_fixture_files(tmp_path) + inventory_path = tmp_path / "inventory.json" + write_audit_cycle_artifact_sync( + kind="inventory", + path=str(inventory_path), + fields=_inventory_fields(), + cwd=str(tmp_path), + ) + + authority_path = tmp_path / "authority.json" + result = write_audit_cycle_artifact_sync( + kind="authority", + path=str(authority_path), + fields=_authority_fields( + inventory_path=inventory_path, + plan_path=outside_plan, + remediation_path=remediation_path, + ), + cwd=str(tmp_path), + ) + assert result["success"] is False + assert "referenced artifact read failed" in result["error"] + assert not authority_path.exists() + + +class TestMalformedFieldsRejected: + def test_invalid_assessment_value_rejected_without_raising_or_partial_write( + self, tmp_path: Path + ) -> None: + plan_path, remediation_path = _write_fixture_files(tmp_path) + inventory_path = tmp_path / "inventory.json" + write_audit_cycle_artifact_sync( + kind="inventory", + path=str(inventory_path), + fields=_inventory_fields(), + cwd=str(tmp_path), + ) + + fields = _authority_fields( + inventory_path=inventory_path, plan_path=plan_path, remediation_path=remediation_path + ) + fields["assessments"][0]["assessment"] = "NOT_A_REAL_ASSESSMENT" + + authority_path = tmp_path / "authority.json" + result = write_audit_cycle_artifact_sync( + kind="authority", path=str(authority_path), fields=fields, cwd=str(tmp_path) + ) + assert result["success"] is False + assert "assessment" in result["error"] + assert not authority_path.exists() + + def test_missing_required_field_rejected(self, tmp_path: Path) -> None: + fields = _inventory_fields() + del fields["requirement_ids"] + path = tmp_path / "inventory.json" + result = write_audit_cycle_artifact_sync( + kind="inventory", path=str(path), fields=fields, cwd=str(tmp_path) + ) + assert result["success"] is False + assert not path.exists() + + +class TestUnknownKindRejected: + def test_unknown_kind_returns_structured_error_with_no_io(self, tmp_path: Path) -> None: + path = tmp_path / "authroity.json" + result = write_audit_cycle_artifact_sync( + kind="authroity", + path=str(path), + fields={ + "audited_plan_refs": [ + { + "locator": str(tmp_path / "does-not-exist.md"), + "media_type": "text/markdown", + "schema_version": 1, + } + ] + }, + cwd=str(tmp_path), + ) + assert result == {"success": False, "error": "unknown kind: 'authroity'"} + assert not path.exists() + + +class TestWriteOnceGuard: + def test_existing_destination_is_rejected_and_unchanged(self, tmp_path: Path) -> None: + path = tmp_path / "inventory.json" + path.write_bytes(b"pre-existing content") + + result = write_audit_cycle_artifact_sync( + kind="inventory", path=str(path), fields=_inventory_fields(), cwd=str(tmp_path) + ) + assert result["success"] is False + assert "already exists" in result["error"] + assert path.read_bytes() == b"pre-existing content" + + +class TestReadBeforeValidateOrdering: + def test_structural_failure_precedes_referenced_artifact_reads(self, tmp_path: Path) -> None: + """Regression test: a bad assessment value must fail before ANY referenced + artifact is read, even when many referenced-artifact paths are present. + + Points every referenced-artifact locator at a directory — resolve_contained_path + raises ContainmentError("Regular file required") if it is ever invoked on one. + A failure surfacing as an assessment-validation error (not a containment error) + proves no referenced-artifact read occurred before structural validation. + """ + bad_dir = tmp_path / "not-a-file" + bad_dir.mkdir() + bad_ref = { + "locator": str(bad_dir), + "media_type": "text/markdown", + "schema_version": 1, + } + + fields = { + "execution_generation": "generation-1", + "cycle_id": "cycle-1", + "plan_set_id": "plans-1", + "scope_id": "scope-1", + "part_id": "part-a", + "audit_round": 1, + "parent_authority_digest": None, + "audited_plan_refs": [bad_ref] * 50, + "inventory_ref": bad_ref, + "remediation_ref": bad_ref, + "assessments": [ + { + "requirement_id": "REQ-001", + "requirement_text": "Fix the issue", + "assessment": "NOT_A_REAL_ASSESSMENT", + "evidence_summary": "not present", + } + ], + "verdict": "NO GO", + "generated_at": _GENERATED_AT, + } + + authority_path = tmp_path / "authority.json" + result = write_audit_cycle_artifact_sync( + kind="authority", path=str(authority_path), fields=fields, cwd=str(tmp_path) + ) + assert result["success"] is False + assert "assessment" in result["error"] + assert "Regular file" not in result["error"] + assert "Containment" not in result["error"] + assert not authority_path.exists() + + def test_too_many_referenced_artifacts_rejected_before_reads(self, tmp_path: Path) -> None: + bad_dir = tmp_path / "not-a-file" + bad_dir.mkdir() + bad_ref = { + "locator": str(bad_dir), + "media_type": "text/markdown", + "schema_version": 1, + } + fields = { + "execution_generation": "generation-1", + "cycle_id": "cycle-1", + "plan_set_id": "plans-1", + "scope_id": "scope-1", + "part_id": "part-a", + "audit_round": 1, + "parent_authority_digest": None, + "audited_plan_refs": [bad_ref] * 1000, + "inventory_ref": bad_ref, + "remediation_ref": None, + "assessments": [ + { + "requirement_id": "REQ-001", + "requirement_text": "Fix the issue", + "assessment": "MISSING", + "evidence_summary": "not present", + } + ], + "verdict": "NO GO", + "generated_at": _GENERATED_AT, + } + authority_path = tmp_path / "authority.json" + result = write_audit_cycle_artifact_sync( + kind="authority", path=str(authority_path), fields=fields, cwd=str(tmp_path) + ) + assert result["success"] is False + assert "exceeds" in result["error"] + assert not authority_path.exists() + + +class TestNeverRaises: + def test_non_dict_fields_returns_structured_error(self, tmp_path: Path) -> None: + result = write_audit_cycle_artifact_sync( + kind="inventory", + path=str(tmp_path / "inventory.json"), + fields=None, # type: ignore[arg-type] + cwd=str(tmp_path), + ) + assert result["success"] is False + + def test_missing_cwd_returns_structured_error(self, tmp_path: Path) -> None: + result = write_audit_cycle_artifact_sync( + kind="inventory", + path=str(tmp_path / "inventory.json"), + fields=_inventory_fields(), + cwd=str(tmp_path / "does-not-exist"), + ) + assert result["success"] is False + assert "cwd" in result["error"] diff --git a/tests/skills/test_make_plan_capability_accuracy.py b/tests/skills/test_make_plan_capability_accuracy.py index 7ea7d45bd1..e882fc8408 100644 --- a/tests/skills/test_make_plan_capability_accuracy.py +++ b/tests/skills/test_make_plan_capability_accuracy.py @@ -14,4 +14,4 @@ def test_make_plan_declares_exact_worker_capabilities() -> None: parsed = read_skill_frontmatter(pkg_root() / "skills_extended" / "make-plan" / "SKILL.md") assert parsed.data is not None caps = set(parsed.data.get("uses_capabilities", [])) - assert caps == {"agent_model", "agent_subagent"} + assert caps == {"agent_model", "agent_subagent", "write_audit_cycle_artifact"}