"""Canonical invocation write policy and truthful runner enforcement.""" from __future__ import annotations import hashlib import json import re import subprocess from collections.abc import Mapping, Sequence from dataclasses import dataclass from pathlib import Path, PurePosixPath from typing import Any, Literal from .domain.write_policy import ( # noqa: F401 — 재노출(값 코어는 도메인 소유) BoundaryPrecision, WriteEnforcement, WritePolicy, WritePolicyError, _absolute_text, _mapping, _PRECISION_RANK, _relative_paths, _text_sequence, validate_write_enforcement_payload, validate_write_policy_payload, write_enforcement_from_payload, write_policy_digest, write_policy_from_payload, ) from .final_report_paths import final_report_data_path from .json_boundary import JsonBoundaryError, load_owned_object from .path_hints import hydrate_active_run_context from .paths import task_conformance_manifest_file, task_qa_dir def build_write_policy( invocation: Mapping[str, Any], assignment: Mapping[str, Any], worktree: Path | None, ) -> WritePolicy: """Build one provider-neutral policy from invocation and adapter authority.""" role = _required_text(invocation, "role") project_root = Path(_required_text(invocation, "projectRoot")).resolve() source_root = _trusted_root(invocation, worktree) artifact_paths = _relative_paths(invocation.get("artifactPaths", ())) if not artifact_paths: raise WritePolicyError("artifactPaths must contain standard worker artifacts") source_mode = "project-mutation" if role == "implementer" else "source-readonly" source_policy: dict[str, Any] = { "mode": source_mode, "allowedRoot": str(source_root), "plannedPaths": list(_relative_paths(invocation.get("plannedPaths", ()))), "protectedPaths": list( _relative_paths(invocation.get("protectedPaths", (".okstra", ".git"))) ), } if "plannedPathsDeclared" in invocation: source_policy["plannedPathsDeclared"] = invocation["plannedPathsDeclared"] git_policy = _git_policy(invocation, source_root, source_mode) external_roots = _validated_auxiliary_roots(invocation, assignment) auxiliary_policy = { "scratchRoots": [ str(path) for path in _absolute_roots(invocation.get("scratchRoots", ())) ], "generatedPaths": list( _relative_paths(invocation.get("generatedPaths", ())) ), "externalRoots": [str(path) for path in external_roots], } policy = WritePolicy( artifact_policy={ "allowedRoot": str(project_root), "allowedPaths": list(artifact_paths), }, source_policy=source_policy, git_policy=git_policy, auxiliary_policy=auxiliary_policy, maximum_boundary_precision=_assignment_precision(assignment), ) validate_write_policy_payload(policy.to_payload()) return policy def _role_qa_artifact_paths(role: str, task_root: Path | None) -> tuple[Path, ...]: """The task `qa/` writes a role's own contract makes mandatory. `qa/` is a sibling of `runs/` under the task root, so the orchestrator allowance — which carries the run directory tree and the task manifest FILE — never reached it, and no worker artifact path names it. In a worktree run the artifact root is the project root while the source root is the worktree, so `_artifact_policy_failures` is live and every such write read as `artifact-root change exceeds batch policy union`: a role following its own profile could not pass the audit. Split by role because ownership is split. The executor writes the Tier 3 conformance script, its `tsconfig.json` and any real-IO qa spec under `qa/scripts/`, plus the manifest entry naming them (`_implementation-executor.md` §"Stage conformance script", §"Real-IO test isolation"). Everything else under `qa/` — the self-mock sidecar and its diff, the conformance run's `result-*.json` — is written while the verifier runs its own gates. Keeping the executor's grant to those two entries is what leaves the audit able to catch an executor that runs the verifier's self-mock gate (`_implementation-executor.md` §"Verifier gates are not yours to run"); widening it to `qa/` would silence that. The verifier's grant is the tree because the filenames are not knowable here — the sidecar carries the run's stage name and a conformance script names its own `result-*.json`. `qa/self-mock-waivers.json` is inside that grant and stays forbidden to it, enforced where the acknowledgement is read rather than where the byte is written: `validate-run.py::_validate_selfmock` fails a report whose waived entries came from anywhere but the user's file, or lack a `reason` / `acknowledgedBy`. """ if task_root is None: return () if role == "implementer": return ( task_qa_dir(task_root) / "scripts", task_conformance_manifest_file(task_root), ) if role == "verifier": return (task_qa_dir(task_root),) return () def _technical_experiment_paths( artifact_paths: Sequence[Path], task_root: Path | None, ) -> tuple[Path, ...]: """정규 작업 결과 경로에서 해당 작업자의 시험 디렉터리만 파생한다.""" if task_root is None: return () run_root = task_root / "runs" / "technical-verification" paths = set() for artifact in artifact_paths: if artifact.parent != run_root / "worker-results": continue match = re.fullmatch( r"([a-z0-9][a-z0-9-]*)-worker-technical-verification-(\d{3,})\.md", artifact.name, ) if match: paths.add(run_root / "experiments" / match[2] / match[1]) return tuple(sorted(paths)) def build_invocation_write_contract( *, role: str, project_root: Path, worktree: Path | None, artifact_paths: Sequence[Path], task_root: Path | None, maximum_precision: BoundaryPrecision, planned_paths: Sequence[str] = (), planned_paths_declared: bool = True, scratch_roots: Sequence[Path] = (), generated_paths: Sequence[str] = (), auxiliary_roots: Sequence[Path] = (), validated_auxiliary_roots: Sequence[Path] = (), ) -> tuple[WritePolicy, WriteEnforcement]: """Build the single canonical contract used by every v2 dispatcher.""" root = project_root.resolve() relative_artifacts = tuple( dict.fromkeys( _relative_to_root(path, root, "artifact") for path in ( *artifact_paths, *_technical_experiment_paths(artifact_paths, task_root), *_role_qa_artifact_paths(role, task_root), ) ) ) source_root = worktree.resolve() if worktree is not None else root source_paths = _planned_source_paths( planned_paths, root, source_root, relative_artifacts ) invocation: dict[str, Any] = { "role": role, "projectRoot": str(root), "artifactPaths": list(relative_artifacts), "plannedPaths": list(source_paths), "plannedPathsDeclared": planned_paths_declared, "protectedPaths": [".okstra", ".git"], "generatedPaths": list(generated_paths), "scratchRoots": [str(path) for path in scratch_roots], "auxiliaryRoots": [str(path) for path in auxiliary_roots], } if role == "implementer": invocation.update({ "expectedBaseCommit": _git_output(source_root, "rev-parse", "HEAD"), "allowedBranchRef": _git_output( source_root, "symbolic-ref", "-q", "HEAD" ), }) assignment = { "workerWriteCapability": { "maxBoundaryPrecision": maximum_precision, }, "validatedAuxiliaryRoots": [str(path) for path in validated_auxiliary_roots], } policy = build_write_policy(invocation, assignment, source_root) return policy, derive_write_enforcement(policy, maximum_precision, role=role) def _planned_source_paths( paths: Sequence[str], project_root: Path, source_root: Path, artifact_paths: Sequence[str], ) -> tuple[str, ...]: """산출물 권한은 프로젝트에, 소스 변경 목록은 배정 작업 디렉터리에 결속한다.""" sources: set[str] = set() for value in paths: candidate = Path(value) if candidate.is_absolute(): root = ( source_root if candidate.is_relative_to(source_root) else project_root ) relative = _relative_to_root(candidate, root, "planned") else: relative = _relative_paths((value,))[0] root = ( project_root if PurePosixPath(relative).parts[0] == ".okstra" else source_root ) _relative_to_root(root / relative, root, "planned") relative = _relative_paths((relative,))[0] path = PurePosixPath(relative) if path.parts[0] == ".okstra": if root != project_root or not any( path.is_relative_to(allowed) for allowed in artifact_paths ): raise WritePolicyError( f"planned artifact path exceeds this worker's artifact permissions: {value}" ) else: if path.parts[0] == ".git": raise WritePolicyError(f"planned source path is protected: {value}") # 원래 체크아웃의 절대 소스 경로도 실제 쓰기는 배정 작업 디렉터리에 한정된다. _relative_to_root(source_root / relative, source_root, "planned source") sources.add(relative) return tuple(sorted(sources)) def task_root_from_run_manifest( project_root: Path, manifest: Mapping[str, Any] ) -> Path | None: """The task root this run belongs to, read from the manifest's own pointer. Derived from `taskManifestPath` rather than by counting directories up from the run directory: an `implementation` run carries an extra `stage-` level, so one fixed index cannot reach the task root for every task type. """ value = manifest.get("taskManifestPath") if not isinstance(value, str) or not value: return None candidate = Path(value) return (candidate if candidate.is_absolute() else project_root / candidate).parent def planned_paths_from_run_manifest( project_root: Path, manifest: Mapping[str, Any] ) -> tuple[tuple[str, ...], bool]: active_value = manifest.get("activeRunContextPath") if not isinstance(active_value, str) or not active_value: raise WritePolicyError("implementer write policy has no active run context") # `compact_active_run_context` drops `sourceArtifacts` and stores path # hints instead; the hydrator is what puts the block back. Reading the file # raw always saw `None` here and reported the run as having no approved # stage authority, which blocked every implementer dispatch. active = hydrate_active_run_context( _read_json(_rooted(project_root, active_value), "active run context") ) source = active.get("sourceArtifacts") run_inputs_value = source.get("runInputsPath") if isinstance(source, Mapping) else None run = active.get("run") stage_value = run.get("stage") if isinstance(run, Mapping) else None if not isinstance(run_inputs_value, str) or not str(stage_value).isdigit(): raise WritePolicyError("implementer write policy has no approved stage authority") inputs_payload = _read_json(_rooted(project_root, run_inputs_value), "run inputs") inputs = inputs_payload.get("inputs") approved = inputs.get("approvedPlanPath") if isinstance(inputs, Mapping) else None if not isinstance(approved, str) or not approved: raise WritePolicyError("implementer write policy has no approved plan path") return _planned_paths_from_report(Path(approved), int(stage_value)) def planned_paths_declared(project_root: Path, manifest: Mapping[str, Any]) -> bool: """Whether this run's approved plan states its paths in the declared column.""" try: return planned_paths_from_run_manifest(project_root, manifest)[1] except WritePolicyError: return False def derive_write_enforcement( policy: WritePolicy, maximum_precision: BoundaryPrecision, *, role: str | None = None, ) -> WriteEnforcement: """Lower runner capability when policy paths cannot be enforced exactly.""" if role == "leader": enforcement = WriteEnforcement( boundary_precision="none", mutation_audit="none", attribution="batch-unattributed", audited_roots=_audited_roots(policy), observation_coverage="partial", unobserved_write_surfaces=( "lead-session-host-writes", "other-stage-refs", "shared-git-object-store", ), ) validate_write_enforcement_payload(enforcement.to_payload()) return enforcement if maximum_precision not in _PRECISION_RANK: raise WritePolicyError("worker write capability is invalid") exact_policy = ( policy.source_mode == "source-readonly" and not policy.auxiliary_policy.get("externalRoots") and not policy.auxiliary_policy.get("scratchRoots") and not policy.auxiliary_policy.get("generatedPaths") ) if maximum_precision == "exact-path" and exact_policy: boundary: BoundaryPrecision = "exact-path" elif _PRECISION_RANK[maximum_precision] >= _PRECISION_RANK["directory-boundary"]: boundary = "directory-boundary" else: boundary = "none" exact = boundary == "exact-path" and exact_policy external_roots = tuple(policy.auxiliary_policy.get("externalRoots", ())) unobserved: list[str] = [] if boundary == "none": unobserved.append("provider-process-outside-audited-roots") if external_roots: unobserved.append("shared-tool-cache") if policy.source_mode == "project-mutation": unobserved.extend(("other-stage-refs", "shared-git-object-store")) coverage: Literal["complete", "scoped", "partial"] if exact: coverage = "complete" elif unobserved: coverage = "partial" else: coverage = "scoped" enforcement = WriteEnforcement( boundary_precision=boundary, mutation_audit="none" if exact else "batch", attribution="call" if exact else "batch-unattributed", audited_roots=_audited_roots(policy), observation_coverage=coverage, unobserved_write_surfaces=tuple(unobserved), ) validate_write_enforcement_payload(enforcement.to_payload()) return enforcement def _git_policy( invocation: Mapping[str, Any], root: Path, source_mode: str ) -> dict[str, Any]: if source_mode == "source-readonly": return {"mode": "disabled"} git_dir = _git_path(root, "--git-dir") common_dir = _git_path(root, "--git-common-dir") expected_base = _required_text(invocation, "expectedBaseCommit") allowed_ref = _required_text(invocation, "allowedBranchRef") _require_commit(root, expected_base) current_ref = _git_output(root, "symbolic-ref", "-q", "HEAD") if current_ref != allowed_ref or not allowed_ref.startswith("refs/heads/"): raise WritePolicyError("allowedBranchRef does not match assigned worktree") return { "mode": "fast-forward-descendant-chain", "worktreeGitDir": str(git_dir), "gitCommonDir": str(common_dir), "expectedBaseCommit": expected_base, "allowedBranchRef": allowed_ref, "commitMode": "fast-forward-descendant-chain", } def _audited_roots(policy: WritePolicy) -> tuple[Path, ...]: values = [ Path(str(policy.source_policy["allowedRoot"])), Path(str(policy.artifact_policy["allowedRoot"])), ] values.extend(Path(value) for value in policy.auxiliary_policy.get("scratchRoots", ())) git_dir = policy.git_policy.get("worktreeGitDir") if isinstance(git_dir, str) and git_dir: values.append(Path(git_dir)) seen: set[Path] = set() roots: list[Path] = [] for value in values: resolved = value.resolve() if resolved not in seen: roots.append(resolved) seen.add(resolved) return tuple(roots) def _trusted_root(invocation: Mapping[str, Any], worktree: Path | None) -> Path: raw = worktree if worktree is not None else Path(_required_text(invocation, "projectRoot")) if not raw.is_absolute() or not raw.is_dir(): raise WritePolicyError("assigned worktree must be an existing absolute path") return raw.resolve() def _validated_auxiliary_roots( invocation: Mapping[str, Any], assignment: Mapping[str, Any] ) -> tuple[Path, ...]: requested = _absolute_roots(invocation.get("auxiliaryRoots", ())) declared = set(_absolute_roots(assignment.get("validatedAuxiliaryRoots", ()))) if any(path not in declared for path in requested): raise WritePolicyError("auxiliary root is not adapter-validated") return requested def _assignment_precision(assignment: Mapping[str, Any]) -> BoundaryPrecision: capability = assignment.get("workerWriteCapability") if not isinstance(capability, Mapping): raise WritePolicyError("workerWriteCapability is required") value = capability.get("maxBoundaryPrecision") if value not in _PRECISION_RANK: raise WritePolicyError("workerWriteCapability is invalid") return value def _absolute_roots(value: object) -> tuple[Path, ...]: roots: list[Path] = [] for text in _text_sequence(value, "absolute roots"): path = Path(text) if not path.is_absolute() or not path.exists(): raise WritePolicyError("auxiliary root must be an existing absolute path") if path.absolute() != path.resolve(): raise WritePolicyError("auxiliary root contains a symlink component") roots.append(path.resolve()) if len(set(roots)) != len(roots): raise WritePolicyError("auxiliary roots must be unique") return tuple(roots) def _relative_to_root(path: Path, root: Path, label: str) -> str: absolute = path if path.is_absolute() else root / path try: relative = absolute.absolute().relative_to(root).as_posix() except ValueError as exc: raise WritePolicyError(f"{label} path is outside project root") from exc if not relative or relative == ".": raise WritePolicyError(f"{label} path must name a file or scoped subdirectory, not root: {path}") current = root for part in PurePosixPath(relative).parts: current /= part if current.is_symlink(): raise WritePolicyError(f"{label} path contains a symlink component") return relative def _planned_paths_from_report( report_path: Path, stage: int ) -> tuple[tuple[str, ...], bool]: """(paths, declared). `declared` is False for a plan written before the structured `plannedPaths` column existed.""" # `--approved-plan` 은 레코드(`.data.json`)만 받는다(`require_approved_plan_record`). # 손으로 접미사를 붙이면 `.data.json` 에 한 번 더 붙어 `.data.data.json` 이 되고, # 그 파일은 없으므로 implementer 디스패치가 전부 막혔다. 정본 헬퍼는 이미 # 레코드인 경로를 그대로 돌려주고 `.md` 만 짝으로 바꾼다. payload = _read_json(final_report_data_path(report_path), "approved plan data") errors = planned_path_declaration_errors(payload) if errors: raise WritePolicyError(f"{report_path}: {'; '.join(errors)}") planning = payload.get("implementationPlanning") stages = planning.get("stages") if isinstance(planning, Mapping) else None selected = next( ( row for row in stages if isinstance(row, Mapping) and row.get("stage") == stage ), None, ) if isinstance(stages, list) else None steps = selected.get("stepwiseExecution") if isinstance(selected, Mapping) else None # `plannedPaths`, not `files`: the latter is prose written for a reader and # splitting it on commas produced entries like `.npmrc (template only` and # `no literal token)`, while a single cell naming a glob migration counted # as one literal path. Every file the plan actually asked for then read as # an unauthorized source change, which is a failure the executor cannot # avoid by following the plan. rows = steps if isinstance(steps, list) else [] paths = { str(path).strip().strip("`") for step in rows if isinstance(step, Mapping) for path in (step.get("plannedPaths") or []) if isinstance(path, str) and path.strip() } if paths or (rows and all( isinstance(step, Mapping) and isinstance(step.get("plannedPaths"), list) for step in rows )): # 경로 분류는 산출물 권한과 배정 작업 디렉터리를 함께 아는 정책 생성기가 맡는다. return tuple(sorted(paths)), True # A plan approved before `plannedPaths` existed carries its paths only in # the prose `files` cell. That prose does not survive being split into # paths — it mis-splits globs and parentheticals — so the ledger stays # EMPTY rather than carrying values nobody can be held to. An empty ledger # on a `project-mutation` policy is what tells the mutation audit to stand # its path comparison down (`execution_mutation_audit`); carrying the # derived fragments instead is what made every file the plan asked for read # as an unauthorized source change. An in-flight task keeps running; the # next planning run produces a plan the audit can actually enforce. # # The prose is still read, for one decision only: telling a pre-column plan # apart from a malformed one whose rows name no files at all. names_files = any( str(step.get("files") or "").strip() for step in rows if isinstance(step, Mapping) ) if not names_files: raise WritePolicyError( "implementer write policy stage has no planned paths: the approved " "plan's stepwise rows carry neither `plannedPaths` nor `files`" ) return (), False def planned_path_declaration_errors(data: Mapping[str, Any]) -> list[str]: """검사 단계의 빈 선언은 보존하고 저장소 전체를 쓰기 대상으로 승인하지 않는다.""" planning = data.get("implementationPlanning") stages = planning.get("stages") if isinstance(planning, Mapping) else None errors: list[str] = [] for si, stage in enumerate(stages if isinstance(stages, list) else []): steps = stage.get("stepwiseExecution") if isinstance(stage, Mapping) else None for ti, step in enumerate(steps if isinstance(steps, list) else []): paths = step.get("plannedPaths") if isinstance(step, Mapping) else None for pi, value in enumerate(paths if isinstance(paths, list) else []): if not isinstance(value, str): continue path = Path(value.strip().strip("`")) # 절대 경로는 디스패치와 같은 실제 작업트리를 가리킨다. 새 파일은 없어도 된다. root = path == Path(".") or path == Path(path.anchor or ".") if path.is_absolute(): root = root or (path / ".git").exists() if root: field = f"implementationPlanning.stages[{si}].stepwiseExecution[{ti}].plannedPaths[{pi}]" errors.append( f"{field}: planned path must name a file or a scoped subdirectory, " f"not a repository/worktree root: {value!r}; use [] for a read-only step" ) return errors def _rooted(project_root: Path, value: str) -> Path: path = Path(value) return path if path.is_absolute() else project_root / path def _read_json(path: Path, label: str) -> Mapping[str, Any]: try: value = load_owned_object(path, artifact=label) except (OSError, UnicodeError, JsonBoundaryError) as exc: raise WritePolicyError(f"{label} is invalid: {path}") from exc if not isinstance(value, Mapping): raise WritePolicyError(f"{label} must be an object: {path}") return value def _required_text(payload: Mapping[str, Any], key: str) -> str: value = payload.get(key) if not isinstance(value, str) or not value.strip(): raise WritePolicyError(f"{key} must be a non-empty string") return value def _git_path(root: Path, flag: str) -> Path: raw = _git_output(root, "rev-parse", "--path-format=absolute", flag) path = Path(raw).resolve() if not path.exists(): raise WritePolicyError(f"Git projection path does not exist: {path}") return path def _require_commit(root: Path, commit: str) -> None: result = subprocess.run( ["git", "-C", str(root), "cat-file", "-e", f"{commit}^{{commit}}"], capture_output=True, text=True, check=False, ) if result.returncode != 0: raise WritePolicyError(result.stderr.strip() or "expected base is not a commit") def _git_output(root: Path, *args: str) -> str: result = subprocess.run( ["git", "-C", str(root), *args], capture_output=True, text=True, check=False, ) if result.returncode != 0 or not result.stdout.strip(): detail = result.stderr.strip() or "Git command returned no value" raise WritePolicyError(detail) return result.stdout.strip()