"""Fingerprint and materialize implementation design-preparation requests.""" from __future__ import annotations import argparse import fcntl import hashlib import json import os import re import stat import sys import unicodedata import uuid from collections.abc import Mapping from contextlib import contextmanager from copy import deepcopy from dataclasses import dataclass from datetime import datetime, timezone from fnmatch import fnmatchcase from pathlib import Path from typing import Any, Iterator from .final_report_schema import load_schema, validate as validate_schema from .json_boundary import JsonBoundaryError, load_owned_object from okstra_project import ResolverError, resolve_project_root ASSESSMENT_FIELDS = ( "id", "kind", "title", "stageRefs", "status", "need", "knownFacts", "openQuestions", "aiProposal", "humanConfirmation", "workingAssumption", "guardrails", "reviewAt", "ifStillOpen", "replanTriggerFields", "blockReason", "requiredDecision", "notApplicableReason", ) ASSESSMENT_SCHEMA_VERSION = 1 REQUEST_TEMPLATE_VERSION = 1 _PLANNING_DATA_RE = re.compile( r"^final-report-implementation-planning-(?P\d+)\.data\.json$" ) _REQUEST_STATUSES = {"provisional", "blocked"} _INPUT_DECISIONS = {"accept-draft", "modify-draft", "reject-draft", "defer"} _INPUT_FILENAME_RE = re.compile( r"^design-prep-input-(?P\d+)-(?PPREP-\d{3})-" r"r(?P\d{3,})-(?P[0-9a-f-]{36})\.md$" ) _ASSESSMENT_FINGERPRINT_RE = re.compile(r"^sha256:[0-9a-f]{64}$") _CAPTURED_BY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") _INPUT_FRONTMATTER_FIELDS = ( "schema-version", "source-report", "assessment-fingerprint", "item-id", "input-id", "revision", "created-at", "created-by", "captured-by", ) _INPUT_BODY_RE = re.compile( r"^# Design Preparation Input\n\n" r"## Decision\n\n(?P[^\n]+)\n\n" r"## Overrides\n\n```json\n(?P[^\n]*)\n```\n\n" r"## Notes\n\n(?P.*)\n$", re.DOTALL, ) class DesignPrepError(RuntimeError): """A design-preparation snapshot or sidecar violates its artifact contract.""" class _NonRegularInputError(DesignPrepError): pass @dataclass(frozen=True) class ParsedDesignPrepInput: path: Path item_id: str assessment_fingerprint: str revision: int input_id: str decision: str overrides: dict[str, Any] notes: str @dataclass(frozen=True) class DesignPrepDecision: outcome: str effective_items: tuple[dict[str, Any], ...] assumptions_to_inject: tuple[str, ...] warnings: tuple[str, ...] request_paths: tuple[str, ...] reason: str def as_prompt_markdown(self) -> str: lines = ["## Effective Design Preparation", f"- Outcome: `{self.outcome}`"] for item in self.effective_items: lines.extend( [ f"### {item['id']} — {item['kind']}", f"- Assessment fingerprint: `{item['assessmentFingerprint']}`", f"- Decision: `{item['decision']}`", "- Effective proposal: `" f"{json.dumps(item['effectiveProposal'], ensure_ascii=False, sort_keys=True)}`", "- User overrides: `" f"{json.dumps(item['overrides'], ensure_ascii=False, sort_keys=True)}`", ] ) lines.extend(f"- Guardrail: {value}" for value in item.get("guardrails") or []) lines.extend( "- Dependency carry evidence: " f"{json.dumps(value, ensure_ascii=False, sort_keys=True)}" for value in item.get("dependencyCarryEvidence") or [] ) lines.extend(f"- Assumption: {value}" for value in self.assumptions_to_inject) lines.extend(f"- Warning: {value}" for value in self.warnings) return "\n".join(lines) + "\n" def _normalise_json(value: Any) -> Any: if isinstance(value, str): newlines = value.replace("\r\n", "\n").replace("\r", "\n") return unicodedata.normalize("NFC", newlines) if isinstance(value, list): return [_normalise_json(item) for item in value] if isinstance(value, Mapping): return {key: _normalise_json(value[key]) for key in sorted(value)} return value def _sha256_json(payload: Mapping[str, Any]) -> str: canonical = json.dumps( _normalise_json(payload), ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return "sha256:" + hashlib.sha256(canonical).hexdigest() def assessment_fingerprint(item: Mapping[str, Any]) -> str: """Hash only the versioned semantic fields of a design-preparation item.""" if not isinstance(item, Mapping): raise DesignPrepError("design preparation item must be an object") payload: dict[str, Any] = { "assessmentSchemaVersion": ASSESSMENT_SCHEMA_VERSION } payload.update({field: item[field] for field in ASSESSMENT_FIELDS if field in item}) return _sha256_json(payload) def _load_plan_data(data_path: Path) -> dict[str, Any]: try: data = load_owned_object(data_path, artifact="planning report record") except JsonBoundaryError as exc: if exc.reason == "top-level JSON value must be an object": raise DesignPrepError(f"planning data must be an object: {data_path}") from exc raise DesignPrepError(f"cannot read planning data {data_path}: {exc}") from exc if not isinstance(data, dict): raise DesignPrepError(f"planning data must be an object: {data_path}") return data def _planning_seq(data_path: Path) -> str: match = _PLANNING_DATA_RE.match(data_path.name) if match is None: raise DesignPrepError(f"not an implementation-planning data path: {data_path}") return match.group("seq") def _design_prep_item_schema() -> dict[str, Any]: full_schema = load_schema() return { "$defs": full_schema["$defs"], "$ref": "#/$defs/DesignPrepItem", } def _validate_design_prep_item( item: Mapping[str, Any], item_schema: dict[str, Any], ) -> None: errors = validate_schema(item, item_schema) if errors: item_id = item.get("id", "") raise DesignPrepError( f"design preparation item {item_id}: {errors[0]}" ) def _design_preparation_items(data: Mapping[str, Any]) -> list[Mapping[str, Any]]: planning = data.get("implementationPlanning") if planning is None: return [] if not isinstance(planning, Mapping): raise DesignPrepError("implementationPlanning must be an object") preparation = planning.get("designPreparation") if preparation is None: return [] if not isinstance(preparation, Mapping): raise DesignPrepError("designPreparation must be an object") items = preparation.get("items", []) if not isinstance(items, list): raise DesignPrepError("designPreparation.items must be an array") item_schema = _design_prep_item_schema() for item in items: if not isinstance(item, Mapping): raise DesignPrepError("design preparation item must be an object") _validate_design_prep_item(item, item_schema) return items def _report_language(data: Mapping[str, Any]) -> str: meta = data.get("meta") or {} if not isinstance(meta, Mapping): raise DesignPrepError("planning data meta must be an object") return str(meta.get("reportLanguage") or "en") def _pretty(value: Any) -> str: return json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) def _render_request_body_without_fingerprints( *, data_path: Path, item: Mapping[str, Any], ) -> str: source = f"reports/{data_path.name}" proposal = item.get("aiProposal") or {} default_when_unanswered = ( item["workingAssumption"] if item["status"] == "provisional" else item["blockReason"] ) return ( "---\n" "schema-version: 1\n" f"source-report: {source}\n" f"item-id: {item['id']}\n" "---\n\n" f"# Design Preparation Request — {item['id']}\n\n" f"## 1. Why this is needed\n\n{_pretty(item['need'])}\n\n" "## 2. Known constraints and evidence\n\n" f"{_pretty(item.get('knownFacts') or [])}\n\n" f"## 3. AI-prepared proposal\n\n{_pretty(proposal)}\n\n" "## 4. Assumptions and confidence\n\n" f"{_pretty(proposal.get('assumptions') or [])}\n\n" "## 5. Open facts or decisions\n\n" f"{_pretty(item.get('openQuestions') or [])}\n\n" "## 6. Editable fields and replan triggers\n\n" f"{_pretty(item.get('replanTriggerFields') or [])}\n\n" "## 7. Default when unanswered\n\n" f"{_pretty(default_when_unanswered)}\n\n" "## 8. Guardrails and review point\n\n" f"{_pretty({'guardrails': item.get('guardrails') or [], 'reviewAt': item.get('reviewAt')})}\n\n" "## 9. Suggested evidence\n\n" f"{_pretty(proposal.get('evidence') or [])}\n\n" "## 10. Response examples\n\n" "- Accept: `okstra design-prep write --decision accept-draft --confirmed`\n" "- Modify: pass a JSON object with `--overrides`.\n" ) def _request_content_fingerprint( assessment: str, report_language: str, body_without_fingerprints: str, ) -> str: return _sha256_json( { "requestTemplateVersion": REQUEST_TEMPLATE_VERSION, "assessmentFingerprint": assessment, "reportLanguage": report_language, "renderedRequestBodyWithoutFingerprintFields": body_without_fingerprints, } ) def _complete_request( body_without_fingerprints: str, assessment: str, request_content: str, ) -> str: lines = body_without_fingerprints.splitlines() insert_at = next( index for index, line in enumerate(lines) if line.startswith("item-id:") ) + 1 lines[insert_at:insert_at] = [ f"assessment-fingerprint: {assessment}", f"request-content-fingerprint: {request_content}", ] return "\n".join(lines) + "\n" def _request_target( *, data_path: Path, planning_seq: str, item: Mapping[str, Any], ) -> Path: item_id = item.get("id") if not isinstance(item_id, str) or not item_id: raise DesignPrepError("design preparation item id must be a non-empty string") expected_relative = ( "design-prep-requests/" f"design-prep-request-{planning_seq}-{item_id}.md" ) if item.get("requestPath") != expected_relative: raise DesignPrepError( f"item {item_id} must use canonical requestPath {expected_relative}" ) request_dir = data_path.parent.parent / "design-prep-requests" target = (data_path.parent.parent / expected_relative).resolve() if target.parent != request_dir: raise DesignPrepError(f"requestPath escapes design-prep-requests: {target}") return target def _render_request( *, data_path: Path, planning_seq: str, report_language: str, item: Mapping[str, Any], ) -> tuple[Path, bytes]: target = _request_target( data_path=data_path, planning_seq=planning_seq, item=item, ) assessment = assessment_fingerprint(item) body = _render_request_body_without_fingerprints( data_path=data_path, item=item, ) request_content = _request_content_fingerprint( assessment, report_language, body, ) return target, _complete_request(body, assessment, request_content).encode("utf-8") def _require_matching_request(target: Path, expected: bytes, conflict: str) -> None: try: existing = target.read_bytes() except OSError as exc: raise DesignPrepError(f"cannot read existing request {target}: {exc}") from exc if existing != expected: raise DesignPrepError(f"refuses to overwrite {conflict}: {target}") def _write_request_no_clobber(target: Path, encoded: bytes) -> None: target.parent.mkdir(parents=True, exist_ok=True) if target.exists(): _require_matching_request(target, encoded, "different request") return temp = target.with_name(f".{target.name}.tmp-{os.getpid()}-{uuid.uuid4()}") try: with temp.open("x", encoding="utf-8") as handle: handle.write(encoded.decode("utf-8")) handle.flush() os.fsync(handle.fileno()) try: os.link(temp, target) except FileExistsError: _require_matching_request(target, encoded, "raced request") except DesignPrepError: raise except OSError as exc: raise DesignPrepError(f"cannot materialize request {target}: {exc}") from exc finally: temp.unlink(missing_ok=True) def materialize_design_prep_requests(plan_data_path: Path) -> list[Path]: """Create deterministic request sidecars without overwriting other content.""" data_path = Path(plan_data_path).resolve() data = _load_plan_data(data_path) planning_seq = _planning_seq(data_path) report_language = _report_language(data) items = _design_preparation_items(data) written: list[Path] = [] for item in items: if item.get("status") not in _REQUEST_STATUSES: continue target, encoded = _render_request( data_path=data_path, planning_seq=planning_seq, report_language=report_language, item=item, ) _write_request_no_clobber(target, encoded) written.append(target) return written def load_design_prep_items(report_path: Path) -> tuple[dict[str, Any], ...]: """Return validated snapshot items without exposing mutable report data.""" data_path = _canonical_data_path(report_path) data = _load_plan_data(data_path) return tuple(deepcopy(dict(item)) for item in _design_preparation_items(data)) def _canonical_data_path(report_path: Path) -> Path: path = Path(report_path) if _PLANNING_DATA_RE.match(path.name): candidate = path elif re.match(r"^final-report-implementation-planning-\d+\.md$", path.name): candidate = path.with_name(path.name.removesuffix(".md") + ".data.json") else: raise DesignPrepError(f"not an implementation-planning report path: {path}") try: resolved = candidate.resolve() except (OSError, RuntimeError) as exc: raise DesignPrepError(f"cannot resolve planning data path {candidate}: {exc}") from exc _planning_seq(resolved) return resolved def _legacy_decision() -> DesignPrepDecision: return DesignPrepDecision( outcome="proceed", effective_items=(), assumptions_to_inject=(), warnings=("legacy-unassessed",), request_paths=(), reason="legacy plan has no design preparation assessment", ) def _require_json_object_keys(value: Any) -> None: if isinstance(value, Mapping): for key, nested in value.items(): if not isinstance(key, str): raise TypeError("JSON object keys must be strings") _require_json_object_keys(nested) elif isinstance(value, list): for nested in value: _require_json_object_keys(nested) def _require_input_contract( decision: Any, overrides: Any, notes: Any, captured_by: Any, ) -> None: if not isinstance(decision, str) or decision not in _INPUT_DECISIONS: raise DesignPrepError(f"unsupported design preparation decision: {decision}") if not isinstance(overrides, Mapping): raise DesignPrepError("design preparation overrides must be an object") try: _require_json_object_keys(overrides) encoded_overrides = json.dumps( _normalise_json(dict(overrides)), ensure_ascii=False, sort_keys=True, allow_nan=False, ) encoded_overrides.encode("utf-8") except (TypeError, ValueError, RecursionError) as exc: raise DesignPrepError( "design preparation overrides must be a JSON object" ) from exc if decision != "modify-draft" and overrides: raise DesignPrepError("only modify-draft may include overrides") if not isinstance(notes, str): raise DesignPrepError("design preparation notes must be a string") try: notes.encode("utf-8") except UnicodeError as exc: raise DesignPrepError("design preparation notes must be valid UTF-8") from exc if decision == "reject-draft" and not notes.strip(): raise DesignPrepError("reject-draft notes must be non-empty") if not isinstance(captured_by, str) or not _CAPTURED_BY_RE.fullmatch(captured_by): raise DesignPrepError("captured_by must be a non-empty safe identifier") def _find_design_prep_item( data: Mapping[str, Any], item_id: str, ) -> Mapping[str, Any]: if not isinstance(item_id, str) or not re.fullmatch(r"PREP-\d{3}", item_id): raise DesignPrepError(f"invalid design preparation item id: {item_id}") matches = [item for item in _design_preparation_items(data) if item.get("id") == item_id] if len(matches) != 1: raise DesignPrepError(f"expected one design preparation item {item_id}") return matches[0] def _plan_input_dir(data_path: Path) -> Path: if data_path.parent.name != "reports": raise DesignPrepError(f"planning data must be under a reports directory: {data_path}") input_dir = data_path.parent.parent / "design-prep-inputs" try: plan_run_root = data_path.parent.parent.resolve() resolved_input_dir = input_dir.resolve() except (OSError, RuntimeError) as exc: raise DesignPrepError(f"cannot resolve plan run input path: {exc}") from exc if resolved_input_dir.parent != plan_run_root: raise DesignPrepError(f"design preparation input path escapes plan run root: {input_dir}") return input_dir def _open_input_dir_fd(input_dir: Path) -> int: return os.open( input_dir, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, ) def _require_input_dir_identity(input_dir: Path, input_dir_fd: int) -> None: opened = os.fstat(input_dir_fd) try: current = os.stat(input_dir, follow_symlinks=False) except OSError as exc: raise DesignPrepError( f"cannot verify design preparation input directory {input_dir}: {exc}" ) from exc if ( not stat.S_ISDIR(current.st_mode) or (current.st_dev, current.st_ino) != (opened.st_dev, opened.st_ino) ): raise DesignPrepError( f"design preparation input directory changed during operation: {input_dir}" ) @contextmanager def _input_dir_flock(input_dir_fd: int) -> Iterator[None]: flags = os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW try: lock_fd = os.open( ".design-prep-input.lock", flags, 0o600, dir_fd=input_dir_fd, ) except FileNotFoundError: # APFS can report ENOENT to one opener racing the initial lock creation. lock_fd = os.open( ".design-prep-input.lock", flags, 0o600, dir_fd=input_dir_fd, ) try: fcntl.flock(lock_fd, fcntl.LOCK_EX) try: yield finally: fcntl.flock(lock_fd, fcntl.LOCK_UN) finally: os.close(lock_fd) def _serialize_input( *, data_path: Path, item: Mapping[str, Any], decision: str, overrides: dict[str, Any], notes: str, revision: int, input_id: str, created_at: str, captured_by: str, ) -> str: encoded_overrides = json.dumps( _normalise_json(overrides), ensure_ascii=False, sort_keys=True, separators=(",", ":"), ) return ( "---\n" "schema-version: 1\n" f"source-report: reports/{data_path.name}\n" f"assessment-fingerprint: {assessment_fingerprint(item)}\n" f"item-id: {item['id']}\n" f"input-id: {input_id}\n" f"revision: {revision}\n" f"created-at: {created_at}\n" "created-by: user\n" f"captured-by: {captured_by}\n" "---\n\n" "# Design Preparation Input\n\n" f"## Decision\n\n{decision}\n\n" "## Overrides\n\n" f"```json\n{encoded_overrides}\n```\n\n" f"## Notes\n\n{notes.rstrip()}\n" ) def _parse_flat_frontmatter(text: str, path: Path) -> tuple[dict[str, str], str]: if not text.startswith("---\n"): raise DesignPrepError(f"malformed input {path}: missing frontmatter") closing = text.find("\n---\n", 4) if closing < 0: raise DesignPrepError(f"malformed input {path}: unterminated frontmatter") fields: dict[str, str] = {} ordered_fields: list[str] = [] for line in text[4:closing].splitlines(): if ": " not in line: raise DesignPrepError(f"malformed input {path}: invalid frontmatter line") key, value = line.split(": ", 1) if key in fields or not value: raise DesignPrepError(f"malformed input {path}: invalid frontmatter field {key}") fields[key] = value ordered_fields.append(key) if tuple(ordered_fields) != _INPUT_FRONTMATTER_FIELDS: raise DesignPrepError(f"malformed input {path}: unexpected frontmatter fields") body = text[closing + 5 :] if not body.startswith("\n"): raise DesignPrepError(f"malformed input {path}: missing body separator") return fields, body[1:] def _require_uuid4(value: str, field: str, path: Path) -> None: try: parsed = uuid.UUID(value) except (ValueError, AttributeError) as exc: raise DesignPrepError(f"malformed input {path}: invalid {field}") from exc if str(parsed) != value or parsed.version != 4: raise DesignPrepError(f"malformed input {path}: invalid {field}") def _parse_input_body(body: str, path: Path) -> tuple[str, dict[str, Any], str]: match = _INPUT_BODY_RE.fullmatch(body) if match is None: raise DesignPrepError(f"malformed input {path}: invalid markdown body") decision = match.group("decision") if decision not in _INPUT_DECISIONS: raise DesignPrepError(f"malformed input {path}: invalid decision") try: overrides = json.loads(match.group("overrides")) except json.JSONDecodeError as exc: raise DesignPrepError(f"malformed input {path}: invalid overrides JSON") from exc notes = match.group("notes") try: _require_input_contract(decision, overrides, notes, "parsed-sidecar") except DesignPrepError as exc: raise DesignPrepError(f"malformed input {path}: {exc}") from exc return decision, dict(overrides), notes def _require_regular_input(metadata: os.stat_result, path: Path) -> None: if not stat.S_ISREG(metadata.st_mode): raise _NonRegularInputError(f"malformed input {path}: not a regular file") def _parse_input_path( path: Path, input_dir_fd: int, planning_seq: str, item_id: str, ) -> ParsedDesignPrepInput: filename = _INPUT_FILENAME_RE.fullmatch(path.name) if filename is None: raise DesignPrepError(f"malformed input {path}: invalid filename") try: filename_revision = int(filename.group("revision")) except ValueError as exc: raise DesignPrepError(f"malformed input {path}: invalid filename revision") from exc if filename_revision <= 0: raise DesignPrepError(f"malformed input {path}: filename revision must be positive") filename_input_id = filename.group("input_id") if filename.group("seq") != planning_seq or filename.group("item") != item_id: raise DesignPrepError(f"malformed input {path}: filename identity mismatch") _require_uuid4(filename_input_id, "filename UUID", path) try: _require_regular_input( os.stat(path.name, dir_fd=input_dir_fd, follow_symlinks=False), path, ) candidate_fd = os.open( path.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=input_dir_fd, ) try: _require_regular_input(os.fstat(candidate_fd), path) with os.fdopen( candidate_fd, "r", encoding="utf-8", closefd=False, ) as handle: text = handle.read() finally: os.close(candidate_fd) except (OSError, UnicodeError) as exc: raise DesignPrepError(f"malformed input {path}: cannot read: {exc}") from exc fields, body = _parse_flat_frontmatter(text, path) expected_source = ( "reports/" f"final-report-implementation-planning-{planning_seq}.data.json" ) if fields["schema-version"] != "1" or fields["source-report"] != expected_source: raise DesignPrepError(f"malformed input {path}: source contract mismatch") if fields["item-id"] != item_id or fields["input-id"] != filename_input_id: raise DesignPrepError(f"malformed input {path}: frontmatter identity mismatch") try: revision = int(fields["revision"]) except ValueError as exc: raise DesignPrepError(f"malformed input {path}: invalid revision") from exc if revision <= 0: raise DesignPrepError(f"malformed input {path}: revision must be positive") if revision != filename_revision or filename.group("revision") != f"{revision:03d}": raise DesignPrepError(f"malformed input {path}: revision mismatch") _require_uuid4(fields["input-id"], "input-id", path) if not _ASSESSMENT_FINGERPRINT_RE.fullmatch(fields["assessment-fingerprint"]): raise DesignPrepError(f"malformed input {path}: invalid assessment fingerprint") if fields["created-by"] != "user" or not _CAPTURED_BY_RE.fullmatch(fields["captured-by"]): raise DesignPrepError(f"malformed input {path}: invalid input provenance") try: datetime.strptime(fields["created-at"], "%Y-%m-%dT%H:%M:%SZ") except ValueError as exc: raise DesignPrepError(f"malformed input {path}: invalid created-at") from exc decision, overrides, notes = _parse_input_body(body, path) return ParsedDesignPrepInput( path=path, item_id=item_id, assessment_fingerprint=fields["assessment-fingerprint"], revision=revision, input_id=fields["input-id"], decision=decision, overrides=overrides, notes=notes, ) def _revision_from_candidate_name(path: Path) -> int | None: match = re.search(r"-r(\d+)-", path.name) try: return int(match.group(1)) if match else None except ValueError: return None def _parse_inputs_for_item( input_dir: Path, input_dir_fd: int, planning_seq: str, item_id: str, *, warnings: list[str] | None = None, invalid_revisions: set[int] | None = None, ) -> list[ParsedDesignPrepInput]: try: pattern = f"design-prep-input-{planning_seq}-{item_id}-r*-*.md" candidates = sorted( input_dir / name for name in os.listdir(input_dir_fd) if fnmatchcase(name, pattern) ) except OSError as exc: error = DesignPrepError( f"cannot scan design preparation inputs {input_dir}: {exc}" ) if warnings is None: raise error from exc warnings.append(str(error)) return [] parsed: list[ParsedDesignPrepInput] = [] for path in candidates: try: parsed.append( _parse_input_path(path, input_dir_fd, planning_seq, item_id) ) except DesignPrepError as exc: if warnings is None and not isinstance(exc, _NonRegularInputError): raise if warnings is not None: warnings.append(str(exc)) revision = _revision_from_candidate_name(path) if revision is not None and invalid_revisions is not None: invalid_revisions.add(revision) return parsed def _stat_identity(metadata: os.stat_result) -> tuple[int, int]: return metadata.st_dev, metadata.st_ino def _unlink_input_name( input_dir_fd: int, name: str, expected_identity: tuple[int, int], ) -> None: try: current = os.stat(name, dir_fd=input_dir_fd, follow_symlinks=False) except FileNotFoundError: return if _stat_identity(current) == expected_identity: os.unlink(name, dir_fd=input_dir_fd) def _write_input_temp( input_dir_fd: int, temp_name: str, body: str, ) -> tuple[int, int]: temp_fd = os.open( temp_name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=input_dir_fd, ) try: identity = _stat_identity(os.fstat(temp_fd)) try: with os.fdopen(temp_fd, "w", encoding="utf-8", closefd=False) as handle: handle.write(body) handle.flush() os.fsync(temp_fd) except BaseException: _unlink_input_name(input_dir_fd, temp_name, identity) raise return identity finally: os.close(temp_fd) def _publish_input( input_dir_fd: int, temp_name: str, target_name: str, body: str, ) -> tuple[int, int]: temp_identity = _write_input_temp(input_dir_fd, temp_name, body) try: os.link( temp_name, target_name, src_dir_fd=input_dir_fd, dst_dir_fd=input_dir_fd, follow_symlinks=False, ) return temp_identity finally: _unlink_input_name(input_dir_fd, temp_name, temp_identity) def write_design_prep_input( report_path: Path, item_id: str, decision: str, overrides: Mapping[str, Any], notes: str, *, captured_by: str, ) -> Path: """Append a uniquely-revisioned user input sidecar for one assessment item.""" _require_input_contract(decision, overrides, notes, captured_by) data_path = _canonical_data_path(report_path) data = _load_plan_data(data_path) item = _find_design_prep_item(data, item_id) planning_seq = _planning_seq(data_path) input_dir = _plan_input_dir(data_path) input_dir_fd: int | None = None try: input_dir.mkdir(parents=True, exist_ok=True) input_dir_fd = _open_input_dir_fd(input_dir) with _input_dir_flock(input_dir_fd): invalid_revisions: set[int] = set() existing = _parse_inputs_for_item( input_dir, input_dir_fd, planning_seq, item_id, invalid_revisions=invalid_revisions, ) revisions = {row.revision for row in existing}.union(invalid_revisions) revision = max(revisions, default=0) + 1 input_id = str(uuid.uuid4()) created_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") body = _serialize_input( data_path=data_path, item=item, decision=decision, overrides=dict(overrides), notes=notes, revision=revision, input_id=input_id, created_at=created_at, captured_by=captured_by, ) target_name = ( f"design-prep-input-{planning_seq}-{item_id}-" f"r{revision:03d}-{input_id}.md" ) temp_name = f".{target_name}.tmp-{os.getpid()}" target_identity = _publish_input( input_dir_fd, temp_name, target_name, body, ) try: _require_input_dir_identity(input_dir, input_dir_fd) except DesignPrepError: _unlink_input_name(input_dir_fd, target_name, target_identity) raise return input_dir / target_name except DesignPrepError: raise except OSError as exc: raise DesignPrepError( f"cannot write design preparation input for {item_id}: {exc}" ) from exc finally: if input_dir_fd is not None: os.close(input_dir_fd) def _override_paths(value: Mapping[str, Any], prefix: str = "") -> set[str]: paths = set() for key, nested in value.items(): path = f"{prefix}.{key}" if prefix else str(key) paths.add(path) if isinstance(nested, Mapping): paths.update(_override_paths(nested, path)) return paths def _effective_outcome( item: Mapping[str, Any], selected: ParsedDesignPrepInput | None, ) -> str: status = str(item.get("status") or "") if status in {"ready", "not-applicable"}: return "proceed" if selected is None or selected.decision == "defer": return "proceed" if status == "provisional" else "wait_for_input" if selected.decision == "reject-draft": return "wait_for_input" if status == "blocked": return "replan" trigger_fields = set(item.get("replanTriggerFields") or []) if selected.decision == "modify-draft" and trigger_fields.intersection( _override_paths(selected.overrides) ): return "replan" return "proceed" def _select_latest_input( rows: list[ParsedDesignPrepInput], expected_fingerprint: str, item_id: str, warnings: list[str], invalid_revisions: set[int], ) -> ParsedDesignPrepInput | None: by_revision: dict[int, list[ParsedDesignPrepInput]] = {} for row in rows: by_revision.setdefault(row.revision, []).append(row) revisions = set(by_revision).union(invalid_revisions) for revision in sorted(revisions, reverse=True): candidates = by_revision.get(revision, []) if revision in invalid_revisions: continue if len(candidates) != 1: warnings.append(f"duplicate revision {revision} for {item_id}; skipped") continue selected = candidates[0] if selected.assessment_fingerprint != expected_fingerprint: warnings.append(f"stale revision {revision} for {item_id}; skipped") continue return selected return None def _recursive_merge( proposal: Mapping[str, Any], overrides: Mapping[str, Any], ) -> dict[str, Any]: merged = deepcopy(dict(proposal)) for key, override in overrides.items(): current = merged.get(key) if isinstance(current, Mapping) and isinstance(override, Mapping): merged[key] = _recursive_merge(current, override) else: merged[key] = deepcopy(override) return merged def _parse_depends_on(value: Any) -> set[int]: if value in (None, "", "(none)"): return set() if not isinstance(value, str): raise ValueError("dependsOn must be a string") parts = [part.strip() for part in value.split(",")] if any(not part.isdigit() for part in parts): raise ValueError("dependsOn must contain stage numbers") return {int(part) for part in parts} def _dependency_stages( planning: Mapping[str, Any], stage_number: int | None, warnings: list[str], ) -> set[int]: if stage_number is None: return set() stages = planning.get("stages") or [] if not isinstance(stages, list): warnings.append("malformed implementationPlanning.stages; carry ignored") return set() graph: dict[int, set[int]] = {} for row in stages: try: if not isinstance(row, Mapping) or not isinstance(row.get("stage"), int): raise ValueError("stage row must contain an integer stage") graph[row["stage"]] = _parse_depends_on(row.get("dependsOn")) except ValueError as exc: warnings.append(f"malformed stage dependency; carry ignored: {exc}") dependencies: set[int] = set() pending = list(graph.get(stage_number, set())) while pending: dependency = pending.pop() if dependency in dependencies: continue dependencies.add(dependency) pending.extend(graph.get(dependency, set()) - dependencies) return dependencies def _read_carry(path: Path, stage: int, warnings: list[str]) -> Mapping[str, Any] | None: if not path.is_file(): return None try: payload = load_owned_object(path, artifact="design preparation carry") except JsonBoundaryError as exc: warnings.append(f"malformed carry stage {stage}; ignored: {exc}") return None if not isinstance(payload, Mapping): warnings.append(f"malformed carry stage {stage}; ignored") return None if payload.get("stageNumber") != stage: warnings.append(f"stale carry stage {stage}; ignored") return None if payload.get("status") != "completed": return None return payload def _carry_evidence_for_item( *, data_path: Path, dependency_stages: set[int], item_id: str, fingerprint: str, warnings: list[str], ) -> tuple[dict[str, Any], ...]: carry_dir = data_path.parent.parent.parent / "implementation/carry" attached: list[dict[str, Any]] = [] seen: set[str] = set() for stage in sorted(dependency_stages): carry = _read_carry(carry_dir / f"stage-{stage}.json", stage, warnings) if carry is None: continue rows = carry.get("designPrepEvidence") or [] if not isinstance(rows, list): warnings.append(f"malformed carry evidence stage {stage}; ignored") continue for row in rows: if not isinstance(row, Mapping): warnings.append(f"malformed carry evidence stage {stage}; ignored") continue if row.get("itemId") != item_id: continue if row.get("assessmentFingerprint") != fingerprint: warnings.append(f"stale carry evidence stage {stage} for {item_id}; ignored") continue resolution = row.get("resolution") evidence_values = row.get("evidence") if ( not isinstance(resolution, str) or not resolution.strip() or not isinstance(evidence_values, list) or not evidence_values or any( not isinstance(value, str) or not value.strip() for value in evidence_values ) ): warnings.append( f"malformed carry evidence stage {stage} for {item_id}; ignored" ) continue evidence = deepcopy(dict(row)) identity = json.dumps(evidence, ensure_ascii=False, sort_keys=True) if identity in seen: warnings.append(f"duplicate carry evidence stage {stage} for {item_id}; ignored") continue seen.add(identity) evidence["sourceStage"] = stage attached.append(evidence) return tuple(attached) def _effective_item( item: Mapping[str, Any], selected: ParsedDesignPrepInput | None, outcome: str, carry: tuple[dict[str, Any], ...], ) -> dict[str, Any]: overrides = ( selected.overrides if selected is not None and selected.decision == "modify-draft" else {} ) proposal = deepcopy(dict(item.get("aiProposal") or {})) if selected is not None and selected.decision == "modify-draft" and outcome == "proceed": proposal = _recursive_merge(proposal, overrides) return { "id": item["id"], "kind": item["kind"], "assessmentFingerprint": assessment_fingerprint(item), "decision": selected.decision if selected is not None else "unanswered", "effectiveProposal": proposal, "overrides": deepcopy(overrides), "guardrails": tuple(deepcopy(item.get("guardrails") or [])), "reviewAt": deepcopy(item.get("reviewAt")), "ifStillOpen": item.get("ifStillOpen"), "sourceInput": str(selected.path) if selected is not None else None, "dependencyCarryEvidence": carry, } def _provisional_assumptions( item: Mapping[str, Any], effective: Mapping[str, Any], outcome: str, ) -> list[str]: if item.get("status") != "provisional" or outcome != "proceed": return [] assumptions = [str(item["workingAssumption"])] if effective["overrides"]: assumptions.append( f"{item['id']} user overrides: " + json.dumps(effective["overrides"], ensure_ascii=False, sort_keys=True) ) return assumptions def _aggregate_outcome(outcomes: list[str]) -> str: if "wait_for_input" in outcomes: return "wait_for_input" if "replan" in outcomes: return "replan" return "proceed" def _decision_reason(outcome: str) -> str: if outcome == "wait_for_input": return "design preparation requires user input" if outcome == "replan": return "design preparation changes require replanning" return "design preparation is effective" def _resolve_effective_entry( *, item: Mapping[str, Any], input_dir: Path | None, input_dir_fd: int | None, planning_seq: str, data_path: Path, dependency_stages: set[int], warnings: list[str], ) -> tuple[dict[str, Any], str, list[str]]: item_id = str(item["id"]) fingerprint = assessment_fingerprint(item) invalid_revisions: set[int] = set() rows = ( _parse_inputs_for_item( input_dir, input_dir_fd, planning_seq, item_id, warnings=warnings, invalid_revisions=invalid_revisions, ) if input_dir is not None and input_dir_fd is not None else [] ) selected = _select_latest_input( rows, fingerprint, item_id, warnings, invalid_revisions, ) outcome = _effective_outcome(item, selected) carry = _carry_evidence_for_item( data_path=data_path, dependency_stages=dependency_stages, item_id=item_id, fingerprint=fingerprint, warnings=warnings, ) effective = _effective_item(item, selected, outcome, carry) return effective, outcome, _provisional_assumptions(item, effective, outcome) @contextmanager def _resolver_input_dir( data_path: Path, warnings: list[str], ) -> Iterator[tuple[Path | None, int | None]]: try: input_dir = _plan_input_dir(data_path) except DesignPrepError as exc: warnings.append(str(exc)) yield None, None return try: input_dir_fd = _open_input_dir_fd(input_dir) except FileNotFoundError: yield input_dir, None return except OSError as exc: warnings.append(f"cannot scan design preparation inputs {input_dir}: {exc}") yield input_dir, None return try: yield input_dir, input_dir_fd finally: os.close(input_dir_fd) def _resolve_design_prep_items( *, items: list[Mapping[str, Any]], input_dir: Path | None, input_dir_fd: int | None, planning_seq: str, data_path: Path, dependency_stages: set[int], warnings: list[str], ) -> DesignPrepDecision: effective_items: list[dict[str, Any]] = [] assumptions: list[str] = [] outcomes: list[str] = [] for item in items: effective, item_outcome, item_assumptions = _resolve_effective_entry( item=item, input_dir=input_dir, input_dir_fd=input_dir_fd, planning_seq=planning_seq, data_path=data_path, dependency_stages=dependency_stages, warnings=warnings, ) effective_items.append(effective) assumptions.extend(item_assumptions) outcomes.append(item_outcome) outcome = _aggregate_outcome(outcomes) request_paths = tuple( str(item["requestPath"]) for item in items if isinstance(item.get("requestPath"), str) ) return DesignPrepDecision( outcome=outcome, effective_items=tuple(effective_items), assumptions_to_inject=tuple(assumptions), warnings=tuple(warnings), request_paths=request_paths, reason=_decision_reason(outcome), ) def resolve_design_prep( plan_data_path: Path, stage_number: int | None = None, ) -> DesignPrepDecision: """Resolve effective design preparation without changing the approved snapshot.""" data_path = _canonical_data_path(plan_data_path) if not data_path.is_file(): return _legacy_decision() data = _load_plan_data(data_path) planning = data.get("implementationPlanning") if planning is None: return _legacy_decision() if not isinstance(planning, Mapping): raise DesignPrepError("implementationPlanning must be an object") if planning.get("designPreparation") is None: return _legacy_decision() items = _design_preparation_items(data) if stage_number is not None: items = [item for item in items if stage_number in item.get("stageRefs", [])] planning_seq = _planning_seq(data_path) warnings: list[str] = [] dependency_stages = _dependency_stages(planning, stage_number, warnings) with _resolver_input_dir(data_path, warnings) as (input_dir, input_dir_fd): decision = _resolve_design_prep_items( items=items, input_dir=input_dir, input_dir_fd=input_dir_fd, planning_seq=planning_seq, data_path=data_path, dependency_stages=dependency_stages, warnings=warnings, ) if input_dir is not None and input_dir_fd is not None: _require_input_dir_identity(input_dir, input_dir_fd) return decision def _design_prep_item_view( item: Mapping[str, Any], effective_by_id: Mapping[str, dict[str, Any]], ) -> dict[str, Any]: item_id = str(item["id"]) return { "item": deepcopy(dict(item)), "effectiveInput": deepcopy(effective_by_id.get(item_id)), "assessmentFingerprint": assessment_fingerprint(item), "requestPath": item.get("requestPath"), } def _snapshot_summary(report_path: Path) -> dict[str, Any]: data_path = _canonical_data_path(report_path) items = load_design_prep_items(data_path) decision = resolve_design_prep(data_path) effective_by_id = { str(item["id"]): item for item in decision.effective_items } return { "report": str(data_path), "outcome": decision.outcome, "reason": decision.reason, "warnings": list(decision.warnings), "items": [ _design_prep_item_view(item, effective_by_id) for item in items ], } def _discover_planning_reports(project_root: Path) -> list[Path]: pattern = ( ".okstra/tasks/*/*/runs/implementation-planning/reports/" "final-report-implementation-planning-*.data.json" ) return sorted(project_root.glob(pattern)) def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="design-prep") subparsers = parser.add_subparsers(dest="command", required=True) list_parser = subparsers.add_parser("list") list_parser.add_argument("--project-root", default="") list_parser.add_argument("--report", default="") show_parser = subparsers.add_parser("show") show_parser.add_argument("--report", required=True) show_parser.add_argument("--item", default="") write_parser = subparsers.add_parser("write") write_parser.add_argument("--report", required=True) write_parser.add_argument("--item", required=True) write_parser.add_argument( "--decision", choices=sorted(_INPUT_DECISIONS), required=True, ) write_parser.add_argument("--overrides", default="{}") write_parser.add_argument("--notes", default="") write_parser.add_argument("--confirmed", action="store_true") return parser def _run_list_command(args: argparse.Namespace) -> dict[str, Any]: if args.report: reports = [Path(args.report)] else: project_root = resolve_project_root( explicit_root=args.project_root or "", cwd=str(Path.cwd()), ) reports = _discover_planning_reports(project_root) return {"reports": [_snapshot_summary(report) for report in reports]} def _run_show_command(args: argparse.Namespace) -> dict[str, Any]: summary = _snapshot_summary(Path(args.report)) if not args.item: return summary matches = [row for row in summary["items"] if row["item"]["id"] == args.item] if len(matches) != 1: raise DesignPrepError(f"expected one design preparation item {args.item}") return {"report": summary["report"], **matches[0]} def _run_write_command(args: argparse.Namespace) -> dict[str, Any]: if not args.confirmed: raise DesignPrepError("write requires --confirmed") overrides = json.loads(args.overrides) if not isinstance(overrides, dict): raise DesignPrepError("--overrides must be a JSON object") items = load_design_prep_items(Path(args.report)) item = next((row for row in items if row["id"] == args.item), None) if item is None: raise DesignPrepError(f"expected one design preparation item {args.item}") path = write_design_prep_input( Path(args.report), args.item, args.decision, overrides, args.notes, captured_by="okstra-cli", ) match = _INPUT_FILENAME_RE.fullmatch(path.name) if match is None: raise DesignPrepError(f"writer returned a non-canonical input path: {path}") return { "path": str(path), "revision": int(match.group("revision")), "inputId": match.group("input_id"), "assessmentFingerprint": assessment_fingerprint(item), } def _run_cli_command(args: argparse.Namespace) -> dict[str, Any]: if args.command == "list": return _run_list_command(args) if args.command == "show": return _run_show_command(args) return _run_write_command(args) def main(argv: list[str]) -> int: parser = _build_parser() args = parser.parse_args(argv) try: payload = _run_cli_command(args) except (DesignPrepError, ResolverError, json.JSONDecodeError) as exc: print(f"design-prep: {exc}", file=sys.stderr) return 1 print(json.dumps(payload, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))