diff --git a/Scripts/local_open_weight_review.py b/Scripts/local_open_weight_review.py new file mode 100644 index 0000000..34823c0 --- /dev/null +++ b/Scripts/local_open_weight_review.py @@ -0,0 +1,1260 @@ +#!/usr/bin/env python3 +"""Fail-closed local open-weight review cascade. + +This module is deliberately narrow. It may be used by a *local* operator to +review a quarantined engineering corpus, but it cannot decide science, alter an +experiment, create embeddings, train a model, or control the application. On +import it reads no source data and contacts no backend. The command-line entry +point requires an explicit source path and two operator attestations. + +Only metadata-only receipts are serializable. Raw records are held in memory +for one stateless model request and are never passed to aggregation. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import re +import subprocess +import time +import urllib.error +import urllib.request +from collections import Counter, defaultdict +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Iterable, Mapping, Protocol +from urllib.parse import urlparse + + +CHUNK_SCHEMA = "nc-local-review-chunk-v0" +FINDING_SCHEMA = "nc-local-review-finding-v0" +SUMMARY_SCHEMA = "nc-local-review-summary-v0" +RUN_MANIFEST_SCHEMA = "nc-local-review-run-manifest-v0" +CRITIC_SCHEMA = "nc-local-review-critic-v0" +CONFIG_SCHEMA = "nc-local-review-cascade-config-v0" +CASCADE_VERSION = "nc-local-open-weight-review-cascade-v0" +DEFAULT_CONFIGURATION_PATH = Path(__file__).resolve().parents[1] / "configs" / "local-open-weight-review-v0.json" + +ALLOWED_FINDING_TYPES = frozenset( + { + "malformed_record", + "duplicate_index", + "non_monotonic_index", + "candidate_role_inconsistency", + "candidate_duplication", + "selection_inertia", + "repetitive_synthesis", + "state_discontinuity", + "generator_identity_change", + "possible_context_loss", + "candidate_selection_mismatch", + "unclassified", + } +) +ALLOWED_SEVERITIES = frozenset({"low", "medium", "high"}) +REQUIRED_DISPOSITION = { + "corpus_role": "engineering_replay_only", + "development_only_permanent": True, + "eligible_for_encoder_training": False, + "eligible_for_encoder_evaluation": False, + "eligible_for_policy_training": False, + "eligible_for_policy_evaluation": False, + "eligible_for_science": False, + "cloud_exposure_allowed": False, +} +_FORBIDDEN_FIELD_TERMS = ( + "eeg", + "cognitive", + "emotion", + "intention", + "agreement", + "diagnos", + "brain", + "training", + "scientific", + "policy_target", +) +_FORBIDDEN_TEXT_PATTERNS = tuple( + re.compile(pattern, flags=re.IGNORECASE) + for pattern in ( + r"\beeg\s+(?:label|feature|target)", + r"\b(?:cognitive|emotional)\s+(?:label|state)", + r"\b(?:intention|agreement|diagnosis)\b", + r"\bbrain[ -]dialogue\b", + r"\bpolicy[ -]training\b", + r"\bscientific\s+(?:conclusion|claim)\b", + r"\bmodel[ -]training\s+recommendation\b", + ) +) +_LEAK_TOKEN_PATTERN = re.compile(r"[\w]+", flags=re.UNICODE) +_FINDING_ID_PATTERN = re.compile(r"[a-zA-Z0-9][a-zA-Z0-9._:-]{0,127}\Z") +_CITATION_ID_PATTERN = re.compile(r"line:([1-9][0-9]*)\Z") +_METADATA_FORBIDDEN_KEYS = frozenset( + { + "raw_record", + "raw_records", + "record", + "records", + "raw_response", + "response", + "response_text", + "response_bytes", + "prompt", + "prompt_text", + "prompt_content", + "completion", + } +) + +FROZEN_SYSTEM_PROMPT = """You are a local engineering reviewer of a permanently quarantined corpus. + +Your only task is to report bounded replay or dialogue-runtime defects from the +provided chunk. Use source-line citations exactly as supplied. Return one JSON +array with the required finding schema. Do not reproduce source text. + +Do not infer thoughts, intentions, emotions, agreement, diagnoses, cognitive +state, brain state, or scientific conclusions. Do not propose training, EEG +labels, EEG features, policy targets, or application changes. If the evidence +is ambiguous, use finding_type \"unclassified\". Every finding needs at least +one citation and must cite at least one non-overlap record in this chunk.""" + + +class ReviewContractError(ValueError): + """A deterministic contract failure that is safe to pass between reviewers.""" + + def __init__(self, code: str, message: str | None = None) -> None: + self.code = code + super().__init__(message or code) + + +def canonical_json(value: Any) -> str: + return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")) + + +def sha256_bytes(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def sha256_json(value: Any) -> str: + return sha256_bytes(canonical_json(value).encode("utf-8")) + + +def _reject_nonfinite_constant(value: str) -> None: + raise ReviewContractError("nonfinite_number", "JSON must not contain non-finite numeric constants") + + +def strict_json_loads(value: str) -> Any: + try: + return json.loads(value, parse_constant=_reject_nonfinite_constant) + except ReviewContractError: + raise + except json.JSONDecodeError as exc: + raise ReviewContractError("invalid_json", "response must be one complete JSON value") from exc + + +def _require_nonempty_string(value: Any, *, code: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise ReviewContractError(code, "required text is absent") + return value + + +def require_quarantine_disposition(value: Any) -> dict[str, Any]: + if not isinstance(value, dict): + raise ReviewContractError("missing_disposition", "source disposition is required") + for key, expected in REQUIRED_DISPOSITION.items(): + if value.get(key) != expected: + raise ReviewContractError("invalid_disposition", f"source disposition violates {key}") + return dict(value) + + +@dataclass(frozen=True) +class ReviewRecord: + source_line: int + legacy_turn_index: int + record: dict[str, Any] + + +@dataclass(frozen=True) +class SourceEnvelope: + source_sha256: str + quarantine_report_sha256: str + disposition: dict[str, Any] + records: tuple[ReviewRecord, ...] + source_path: Path | None = None + + def verify_unchanged(self) -> None: + """Fail the run if an explicit source file changed after envelope creation.""" + if self.source_path is not None and sha256_bytes(self.source_path.read_bytes()) != self.source_sha256: + raise ReviewContractError("source_changed", "source checksum changed during review") + + +@dataclass(frozen=True) +class ChunkRecord: + citation_id: str + source_line: int + legacy_turn_index: int + is_overlap_record: bool + record: dict[str, Any] + + +@dataclass(frozen=True) +class ChunkEnvelope: + chunk_id: str + source_sha256: str + records: tuple[ChunkRecord, ...] + + @property + def allowed_citation_ids(self) -> tuple[str, ...]: + return tuple(record.citation_id for record in self.records) + + @property + def new_record_citation_ids(self) -> tuple[str, ...]: + return tuple(record.citation_id for record in self.records if not record.is_overlap_record) + + def prompt_payload(self, validation_error_codes: Iterable[str] = ()) -> dict[str, Any]: + """Construct one in-memory request. This value must never be persisted.""" + payload: dict[str, Any] = { + "schema_version": CHUNK_SCHEMA, + "chunk_id": self.chunk_id, + "source_sha256": self.source_sha256, + "records": [ + { + "citation_id": record.citation_id, + "source_line": record.source_line, + "legacy_turn_index": record.legacy_turn_index, + "is_overlap_record": record.is_overlap_record, + "record": record.record, + } + for record in self.records + ], + "allowed_citation_ids": list(self.allowed_citation_ids), + "new_record_citation_ids": list(self.new_record_citation_ids), + } + errors = sorted(set(validation_error_codes)) + if errors: + payload["previous_validation_error_codes"] = errors + return payload + + def metadata_receipt(self) -> dict[str, Any]: + return { + "schema_version": CHUNK_SCHEMA, + "chunk_id": self.chunk_id, + "source_sha256": self.source_sha256, + "allowed_citation_ids": list(self.allowed_citation_ids), + "new_record_citation_ids": list(self.new_record_citation_ids), + } + + +@dataclass(frozen=True) +class ReviewConfiguration: + raw: dict[str, Any] + valid_records_per_chunk: int + overlap_records: int + proposer_model: str + reviewer_models: tuple[str, ...] + maximum_attempts: int + temperature: float + seed: int + context_limit: int | None + max_output_tokens: int | None + critic_enabled: bool + critic_model: str + + @classmethod + def from_mapping(cls, value: Mapping[str, Any]) -> "ReviewConfiguration": + allowed = { + "schema_version", + "configuration_version", + "validator", + "proposer", + "reviewer", + "critic", + "human_gate", + "chunking", + "generation", + "retry", + } + if set(value) != allowed or value.get("schema_version") != CONFIG_SCHEMA: + raise ReviewContractError("invalid_configuration", "review configuration schema is invalid") + validator = value.get("validator") + proposer = value.get("proposer") + reviewer = value.get("reviewer") + critic = value.get("critic") + human_gate = value.get("human_gate") + chunking = value.get("chunking") + generation = value.get("generation") + retry = value.get("retry") + for section in (validator, proposer, reviewer, critic, human_gate, chunking, generation, retry): + if not isinstance(section, dict): + raise ReviewContractError("invalid_configuration", "review configuration sections must be objects") + if validator != {"id": "R0", "type": "deterministic"}: + raise ReviewContractError("invalid_configuration", "R0 must be the deterministic validator") + proposer_model = _require_nonempty_string(proposer.get("model"), code="invalid_configuration") + if proposer.get("id") != "R1" or proposer.get("purpose") != "bounded_chunk_findings" or set(proposer) != {"id", "model", "purpose"}: + raise ReviewContractError("invalid_configuration", "R1 configuration is invalid") + if reviewer.get("id") != "R2" or reviewer.get("purpose") != "citation_repair_and_aggregation" or set(reviewer) != {"id", "preferred_model", "fallback_model", "purpose"}: + raise ReviewContractError("invalid_configuration", "R2 configuration is invalid") + reviewer_values = (reviewer.get("preferred_model"), reviewer.get("fallback_model")) + reviewer_models = tuple(value for value in reviewer_values if isinstance(value, str) and value) + if not reviewer_models: + raise ReviewContractError("invalid_configuration", "R2 needs at least one configured model") + if set(critic) != {"id", "model", "purpose", "enabled_by_default"} or critic.get("id") != "R3" or critic.get("purpose") != "explicit_independent_adjudication" or critic.get("enabled_by_default") is not False or not isinstance(critic.get("model"), str) or not critic["model"]: + raise ReviewContractError("invalid_configuration", "R3 must be disabled by default") + if human_gate.get("id") != "R4" or not isinstance(human_gate.get("required_for"), list): + raise ReviewContractError("invalid_configuration", "R4 configuration is invalid") + required_human = {"policy_changes", "experiment_changes", "scientific_claims", "promotion_decisions"} + if set(human_gate["required_for"]) != required_human: + raise ReviewContractError("invalid_configuration", "R4 gate coverage is incomplete") + if set(chunking) != {"valid_records_per_chunk", "overlap_records", "conversation_memory_between_chunks"}: + raise ReviewContractError("invalid_configuration", "chunking configuration is invalid") + chunk_size = chunking["valid_records_per_chunk"] + overlap = chunking["overlap_records"] + if isinstance(chunk_size, bool) or not isinstance(chunk_size, int) or chunk_size <= 0: + raise ReviewContractError("invalid_configuration", "chunk size must be a positive integer") + if isinstance(overlap, bool) or not isinstance(overlap, int) or not 0 <= overlap < chunk_size: + raise ReviewContractError("invalid_configuration", "chunk overlap is invalid") + if chunking["conversation_memory_between_chunks"] is not False: + raise ReviewContractError("invalid_configuration", "cross-chunk conversation memory is prohibited") + if set(generation) != {"temperature", "seed", "context_limit", "max_output_tokens"}: + raise ReviewContractError("invalid_configuration", "generation configuration is invalid") + temperature = generation["temperature"] + seed = generation["seed"] + if isinstance(temperature, bool) or not isinstance(temperature, (int, float)) or not math.isfinite(float(temperature)) or float(temperature) != 0.0: + raise ReviewContractError("invalid_configuration", "temperature must be pinned to 0.0") + if isinstance(seed, bool) or not isinstance(seed, int): + raise ReviewContractError("invalid_configuration", "seed must be an integer") + for key in ("context_limit", "max_output_tokens"): + item = generation[key] + if item is not None and (isinstance(item, bool) or not isinstance(item, int) or item <= 0): + raise ReviewContractError("invalid_configuration", f"{key} must be null or a positive integer") + if set(retry) != {"maximum_attempts"} or retry["maximum_attempts"] not in {1, 2, 3}: + raise ReviewContractError("invalid_configuration", "retry count must be between one and three") + return cls( + raw=json.loads(canonical_json(dict(value))), + valid_records_per_chunk=chunk_size, + overlap_records=overlap, + proposer_model=proposer_model, + reviewer_models=reviewer_models, + maximum_attempts=retry["maximum_attempts"], + temperature=float(temperature), + seed=seed, + context_limit=generation["context_limit"], + max_output_tokens=generation["max_output_tokens"], + critic_enabled=False, + critic_model=critic["model"], + ) + + @property + def sha256(self) -> str: + return sha256_json(self.raw) + + +def default_configuration() -> dict[str, Any]: + """Load the tracked default configuration without embedding model identities in code.""" + try: + value = strict_json_loads(DEFAULT_CONFIGURATION_PATH.read_text(encoding="utf-8")) + except OSError as exc: + raise ReviewContractError("missing_configuration", "default cascade configuration is unavailable") from exc + if not isinstance(value, dict): + raise ReviewContractError("invalid_configuration", "default configuration must be a JSON object") + return value + + +def build_chunk_envelopes(source: SourceEnvelope, configuration: ReviewConfiguration) -> list[ChunkEnvelope]: + source.verify_unchanged() + records = list(source.records) + chunks: list[ChunkEnvelope] = [] + start = 0 + ordinal = 1 + while start < len(records): + selected = records[start : start + configuration.valid_records_per_chunk] + overlap_count = configuration.overlap_records if ordinal > 1 else 0 + chunk_records = tuple( + ChunkRecord( + citation_id=f"line:{record.source_line}", + source_line=record.source_line, + legacy_turn_index=record.legacy_turn_index, + is_overlap_record=position < overlap_count, + record=record.record, + ) + for position, record in enumerate(selected) + ) + chunks.append( + ChunkEnvelope( + chunk_id=f"{source.source_sha256[:16]}:chunk:{ordinal:04d}", + source_sha256=source.source_sha256, + records=chunk_records, + ) + ) + if start + configuration.valid_records_per_chunk >= len(records): + break + start += configuration.valid_records_per_chunk - configuration.overlap_records + ordinal += 1 + return chunks + + +def _iter_strings(value: Any) -> Iterable[str]: + if isinstance(value, str): + yield value + elif isinstance(value, dict): + for item in value.values(): + yield from _iter_strings(item) + elif isinstance(value, list): + for item in value: + yield from _iter_strings(item) + + +def _private_ngrams(chunk: ChunkEnvelope, width: int = 5) -> set[str]: + ngrams: set[str] = set() + for record in chunk.records: + for text in _iter_strings(record.record): + tokens = [token.casefold() for token in _LEAK_TOKEN_PATTERN.findall(text)] + for start in range(0, max(0, len(tokens) - width + 1)): + candidate = " ".join(tokens[start : start + width]) + if len(candidate) >= 24: + ngrams.add(candidate) + return ngrams + + +def contains_substantial_private_text(value: str, chunk: ChunkEnvelope) -> bool: + """Bounded n-gram detector; it is a guardrail, never a proof of privacy.""" + output_tokens = [token.casefold() for token in _LEAK_TOKEN_PATTERN.findall(value)] + source_ngrams = _private_ngrams(chunk) + for start in range(0, max(0, len(output_tokens) - 5 + 1)): + if " ".join(output_tokens[start : start + 5]) in source_ngrams: + return True + return False + + +def _contains_forbidden_text(value: str) -> bool: + return any(pattern.search(value) is not None for pattern in _FORBIDDEN_TEXT_PATTERNS) + + +def _reject_unknown_or_forbidden_fields(value: dict[str, Any], expected: set[str]) -> None: + extra = set(value) - expected + if not extra: + return + if any(any(term in field.casefold() for term in _FORBIDDEN_FIELD_TERMS) for field in extra): + raise ReviewContractError("forbidden_field", "response contains a prohibited field") + raise ReviewContractError("unknown_field", "response contains an unsupported field") + + +def validate_findings_response(response_text: str, chunk: ChunkEnvelope) -> list[dict[str, Any]]: + """R0 validates a complete model response without repairing it.""" + value = strict_json_loads(response_text) + if not isinstance(value, list): + raise ReviewContractError("response_not_array", "response must be a JSON array") + allowed = set(chunk.allowed_citation_ids) + new = set(chunk.new_record_citation_ids) + source_lines = {record.citation_id: record.source_line for record in chunk.records} + seen_ids: set[str] = set() + findings: list[dict[str, Any]] = [] + expected_finding = { + "schema_version", + "finding_id", + "finding_type", + "severity", + "confidence", + "evidence", + "observation", + "engineering_implication", + "contains_verbatim_private_text", + } + for finding in value: + if not isinstance(finding, dict): + raise ReviewContractError("invalid_finding", "finding must be an object") + _reject_unknown_or_forbidden_fields(finding, expected_finding) + if set(finding) != expected_finding: + raise ReviewContractError("missing_finding_field", "finding is missing required fields") + if finding["schema_version"] != FINDING_SCHEMA: + raise ReviewContractError("wrong_finding_schema", "finding schema version is invalid") + finding_id = _require_nonempty_string(finding["finding_id"], code="invalid_finding_id") + if not _FINDING_ID_PATTERN.fullmatch(finding_id) or finding_id in seen_ids: + raise ReviewContractError("invalid_finding_id", "finding id is malformed or repeated") + seen_ids.add(finding_id) + if finding["finding_type"] not in ALLOWED_FINDING_TYPES: + raise ReviewContractError("unknown_finding_type", "finding type is not allowed") + if finding["severity"] not in ALLOWED_SEVERITIES: + raise ReviewContractError("invalid_severity", "severity is not allowed") + confidence = finding["confidence"] + if isinstance(confidence, bool) or not isinstance(confidence, (int, float)) or not math.isfinite(float(confidence)): + raise ReviewContractError("nonfinite_number", "confidence must be finite") + if not 0.0 <= float(confidence) <= 1.0: + raise ReviewContractError("confidence_out_of_range", "confidence must be within [0, 1]") + evidence = finding["evidence"] + if not isinstance(evidence, list) or not evidence: + raise ReviewContractError("empty_evidence", "finding evidence must be nonempty") + citations: list[str] = [] + for item in evidence: + if not isinstance(item, dict): + raise ReviewContractError("invalid_evidence", "evidence must be an object") + _reject_unknown_or_forbidden_fields(item, {"citation_id", "claim"}) + if set(item) != {"citation_id", "claim"}: + raise ReviewContractError("invalid_evidence", "evidence fields are incomplete") + citation_id = _require_nonempty_string(item["citation_id"], code="invalid_citation") + match = _CITATION_ID_PATTERN.fullmatch(citation_id) + if match is None: + raise ReviewContractError("invalid_citation", "citation ids must use line:") + if citation_id not in allowed: + raise ReviewContractError("citation_not_allowed", "citation is not in this chunk") + if source_lines[citation_id] != int(match.group(1)): + raise ReviewContractError("citation_line_mismatch", "citation does not match its source line") + claim = _require_nonempty_string(item["claim"], code="invalid_evidence") + if _contains_forbidden_text(claim) or contains_substantial_private_text(claim, chunk): + raise ReviewContractError("private_or_forbidden_evidence", "evidence claim violates the content boundary") + citations.append(citation_id) + if not any(citation in new for citation in citations): + raise ReviewContractError("overlap_only_evidence", "at least one citation must be a new chunk record") + observation = _require_nonempty_string(finding["observation"], code="invalid_observation") + implication = _require_nonempty_string(finding["engineering_implication"], code="invalid_implication") + if finding["contains_verbatim_private_text"] is not False: + raise ReviewContractError("verbatim_flag", "finding must explicitly deny verbatim private text") + for text in (observation, implication): + if _contains_forbidden_text(text): + raise ReviewContractError("forbidden_semantic_content", "finding has prohibited semantic content") + if contains_substantial_private_text(text, chunk): + raise ReviewContractError("private_text_leak", "finding contains substantial source text") + findings.append( + { + "schema_version": FINDING_SCHEMA, + "finding_id": finding_id, + "finding_type": finding["finding_type"], + "severity": finding["severity"], + "confidence": float(confidence), + "evidence": [{"citation_id": item["citation_id"], "claim": item["claim"]} for item in evidence], + "observation": observation, + "engineering_implication": implication, + "contains_verbatim_private_text": False, + } + ) + return sorted(findings, key=lambda item: item["finding_id"]) + + +@dataclass(frozen=True) +class BackendResponse: + raw_text: str | None + model_identity: str + model_digest: str | None + runtime_name: str + runtime_version: str | None + elapsed_ms: float + prompt_tokens: int | None + completion_tokens: int | None + stop_reason: str | None + error_status: str | None = None + + +class LocalReviewBackend(Protocol): + backend_type: str + endpoint_class: str + explicitly_local: bool + + def invoke(self, *, model: str, system_prompt: str, user_payload: dict[str, Any], configuration: ReviewConfiguration) -> BackendResponse: + """Invoke one stateless local completion without persisting source payload.""" + + +def classify_local_endpoint(endpoint: str) -> str: + parsed = urlparse(endpoint) + if parsed.scheme == "unix" and parsed.netloc == "" and parsed.path.startswith("/"): + return "unix_socket" + if parsed.scheme not in {"http", "https"} or parsed.username is not None or parsed.password is not None: + raise ReviewContractError("remote_endpoint", "backend endpoint must be a local loopback or Unix socket") + if parsed.hostname not in {"127.0.0.1", "localhost", "::1"} or parsed.path not in {"", "/"} or parsed.query or parsed.fragment: + raise ReviewContractError("remote_endpoint", "backend endpoint must be a local loopback or Unix socket") + return "loopback" + + +class OllamaLocalBackend: + """Small Ollama-compatible local HTTP adapter. It intentionally has no cloud mode.""" + + backend_type = "ollama" + explicitly_local = True + + def __init__(self, endpoint: str) -> None: + self.endpoint_class = classify_local_endpoint(endpoint) + if self.endpoint_class != "loopback": + raise ReviewContractError("unsupported_endpoint", "Ollama HTTP requires a loopback HTTP endpoint") + self.endpoint = endpoint.rstrip("/") + + def _request(self, path: str, payload: dict[str, Any]) -> dict[str, Any]: + request = urllib.request.Request( + f"{self.endpoint}{path}", + data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), + method="POST", + headers={"Content-Type": "application/json", "Accept": "application/json"}, + ) + try: + with urllib.request.urlopen(request, timeout=120) as response: + return strict_json_loads(response.read().decode("utf-8")) + except (urllib.error.URLError, TimeoutError, UnicodeDecodeError) as exc: + raise ReviewContractError("backend_error", "local backend request failed") from exc + + def invoke(self, *, model: str, system_prompt: str, user_payload: dict[str, Any], configuration: ReviewConfiguration) -> BackendResponse: + _require_local_model_name(model) + started = time.monotonic() + response = self._request( + "/api/chat", + { + "model": model, + "stream": False, + "format": "json", + "messages": [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": canonical_json(user_payload)}, + ], + "options": { + "temperature": configuration.temperature, + "seed": configuration.seed, + **({"num_ctx": configuration.context_limit} if configuration.context_limit is not None else {}), + **({"num_predict": configuration.max_output_tokens} if configuration.max_output_tokens is not None else {}), + }, + }, + ) + if not isinstance(response, dict): + raise ReviewContractError("backend_error", "local backend returned a non-object") + message = response.get("message") + content = message.get("content") if isinstance(message, dict) else None + resolved_model = response.get("model") if isinstance(response.get("model"), str) else model + _require_local_model_name(resolved_model) + return BackendResponse( + raw_text=content if isinstance(content, str) else None, + model_identity=resolved_model, + model_digest=response.get("digest") if isinstance(response.get("digest"), str) else None, + runtime_name="ollama", + runtime_version=None, + elapsed_ms=(time.monotonic() - started) * 1000, + prompt_tokens=response.get("prompt_eval_count") if isinstance(response.get("prompt_eval_count"), int) else None, + completion_tokens=response.get("eval_count") if isinstance(response.get("eval_count"), int) else None, + stop_reason=response.get("done_reason") if isinstance(response.get("done_reason"), str) else None, + error_status=None if isinstance(content, str) else "missing_response_content", + ) + + +class MockLocalBackend: + """Deterministic test backend. It never opens a socket or writes prompts.""" + + backend_type = "mock" + endpoint_class = "mock" + explicitly_local = True + + def __init__(self, responses_by_model: Mapping[str, Iterable[BackendResponse | str]]) -> None: + self._responses = {model: list(responses) for model, responses in responses_by_model.items()} + self.invocations: list[dict[str, Any]] = [] + + def invoke(self, *, model: str, system_prompt: str, user_payload: dict[str, Any], configuration: ReviewConfiguration) -> BackendResponse: + self.invocations.append( + { + "model": model, + "system_prompt_sha256": sha256_bytes(system_prompt.encode("utf-8")), + "payload_schema": user_payload.get("schema_version"), + "previous_validation_error_codes": user_payload.get("previous_validation_error_codes", []), + } + ) + values = self._responses.get(model) + if not values: + return BackendResponse(None, model, None, "mock", "v0", 0.0, None, None, None, "no_mock_response") + response = values.pop(0) + if isinstance(response, BackendResponse): + return response + return BackendResponse(response, model, f"mock:{model}", "mock", "v0", 0.0, None, None, "stop", None) + + +@dataclass(frozen=True) +class OperatorAttestation: + network_isolation_operator_attested: bool + prompt_response_logging_confined: bool + + def require_complete(self) -> None: + if not self.network_isolation_operator_attested: + raise ReviewContractError("missing_network_attestation", "operator must attest network state; loopback alone is insufficient") + if not self.prompt_response_logging_confined: + raise ReviewContractError("unconfined_logging", "operator must attest prompt and response logging confinement") + + +@dataclass(frozen=True) +class AttemptReceipt: + attempt: int + stage: str + requested_model: str + resolved_model: str | None + model_digest: str | None + runtime_name: str + runtime_version: str | None + endpoint_class: str + elapsed_ms: float + prompt_tokens: int | None + completion_tokens: int | None + stop_reason: str | None + status: str + validation_error_codes: tuple[str, ...] + + def as_dict(self) -> dict[str, Any]: + return { + "attempt": self.attempt, + "stage": self.stage, + "requested_model": self.requested_model, + "resolved_model": self.resolved_model, + "model_digest": self.model_digest, + "runtime_name": self.runtime_name, + "runtime_version": self.runtime_version, + "endpoint_class": self.endpoint_class, + "elapsed_ms": self.elapsed_ms, + "prompt_tokens": self.prompt_tokens, + "completion_tokens": self.completion_tokens, + "stop_reason": self.stop_reason, + "status": self.status, + "validation_error_codes": list(self.validation_error_codes), + } + + +@dataclass(frozen=True) +class ChunkReviewResult: + chunk: dict[str, Any] + findings: tuple[dict[str, Any], ...] + review_status: str + attempts: tuple[AttemptReceipt, ...] + rejection_error_codes: tuple[str, ...] + + def persisted(self) -> dict[str, Any]: + return { + **self.chunk, + "findings": list(self.findings), + "review_status": self.review_status, + "attempts": [attempt.as_dict() for attempt in self.attempts], + "rejection_error_codes": list(self.rejection_error_codes), + } + + +@dataclass(frozen=True) +class CascadeRun: + source: SourceEnvelope + configuration: ReviewConfiguration + backend: LocalReviewBackend + chunk_results: tuple[ChunkReviewResult, ...] + attestation: OperatorAttestation + + +def _attempt_schedule(configuration: ReviewConfiguration) -> list[tuple[str, str]]: + schedule = [("R1", configuration.proposer_model)] + schedule.extend(("R2", model) for model in configuration.reviewer_models) + return schedule[: configuration.maximum_attempts] + + +def _require_local_model_name(model: str) -> None: + """Reject explicit cloud-routing identities without making a vendor assumption.""" + normalized = model.casefold() + if ":cloud" in normalized or "://" in normalized or normalized.startswith("http:") or normalized.startswith("https:"): + raise ReviewContractError("remote_model", "configured model identity is not explicitly local") + + +def run_review_cascade( + source: SourceEnvelope, + configuration: ReviewConfiguration, + backend: LocalReviewBackend, + attestation: OperatorAttestation, +) -> CascadeRun: + """Run R0/R1/R2. R3 is intentionally unavailable to automatic retries.""" + if not backend.explicitly_local or backend.endpoint_class not in {"loopback", "unix_socket", "mock"}: + raise ReviewContractError("backend_not_local", "review backend must be explicitly classified as local") + attestation.require_complete() + source.verify_unchanged() + results: list[ChunkReviewResult] = [] + for chunk in build_chunk_envelopes(source, configuration): + attempts: list[AttemptReceipt] = [] + prior_errors: list[str] = [] + accepted: list[dict[str, Any]] | None = None + for ordinal, (stage, model) in enumerate(_attempt_schedule(configuration), start=1): + _require_local_model_name(model) + source.verify_unchanged() + payload = chunk.prompt_payload(prior_errors if ordinal > 1 else ()) + response: BackendResponse | None = None + try: + response = backend.invoke(model=model, system_prompt=FROZEN_SYSTEM_PROMPT, user_payload=payload, configuration=configuration) + source.verify_unchanged() + if response.error_status is not None or response.raw_text is None: + raise ReviewContractError("backend_error", "backend did not return a complete response") + accepted = validate_findings_response(response.raw_text, chunk) + attempts.append( + AttemptReceipt( + ordinal, stage, model, response.model_identity, response.model_digest, response.runtime_name, + response.runtime_version, backend.endpoint_class, response.elapsed_ms, response.prompt_tokens, + response.completion_tokens, response.stop_reason, "accepted", (), + ) + ) + break + except ReviewContractError as error: + prior_errors.append(error.code) + attempts.append( + AttemptReceipt( + ordinal, + stage, + model, + response.model_identity if response is not None else None, + response.model_digest if response is not None else None, + response.runtime_name if response is not None else getattr(backend, "backend_type", "unknown"), + response.runtime_version if response is not None else None, + backend.endpoint_class, + response.elapsed_ms if response is not None else 0.0, + response.prompt_tokens if response is not None else None, + response.completion_tokens if response is not None else None, + response.stop_reason if response is not None else None, + "rejected", + (error.code,), + ) + ) + metadata = chunk.metadata_receipt() + if accepted is None: + results.append(ChunkReviewResult(metadata, (), "rejected", tuple(attempts), tuple(sorted(set(prior_errors))))) + else: + results.append(ChunkReviewResult(metadata, tuple(accepted), "accepted", tuple(attempts), ())) + source.verify_unchanged() + return CascadeRun(source, configuration, backend, tuple(results), attestation) + + +def run_explicit_critic_adjudication( + *, + chunk: ChunkEnvelope, + accepted_finding: dict[str, Any], + configuration: ReviewConfiguration, + backend: LocalReviewBackend, + attestation: OperatorAttestation, + requested_by_human: bool, +) -> dict[str, Any]: + """Run R3 only on explicit human request; its output cannot override R0. + + The critic receives the original frozen chunk and an already accepted + structured finding, never a previous model's raw response. It returns a + separate receipt for human review rather than mutating a cascade result. + """ + if not requested_by_human: + raise ReviewContractError("critic_requires_human_request", "R3 adjudication requires explicit human request") + if not backend.explicitly_local or backend.endpoint_class not in {"loopback", "unix_socket", "mock"}: + raise ReviewContractError("backend_not_local", "critic backend must be explicitly classified as local") + attestation.require_complete() + _require_local_model_name(configuration.critic_model) + validated_input = validate_findings_response(canonical_json([accepted_finding]), chunk) + if len(validated_input) != 1: + raise ReviewContractError("invalid_critic_input", "R3 accepts exactly one R0-valid finding") + payload = chunk.prompt_payload() + payload["operation"] = "independent_adjudication" + payload["accepted_finding"] = validated_input[0] + response = backend.invoke(model=configuration.critic_model, system_prompt=FROZEN_SYSTEM_PROMPT, user_payload=payload, configuration=configuration) + if response.error_status is not None or response.raw_text is None: + raise ReviewContractError("backend_error", "R3 did not return a complete response") + critic_findings = validate_findings_response(response.raw_text, chunk) + return { + "schema_version": CRITIC_SCHEMA, + "status": "advisory_only", + "critic_id": "R3", + "requested_by_human": True, + "input_finding_id": validated_input[0]["finding_id"], + "critic_findings": critic_findings, + "backend": { + "requested_model": configuration.critic_model, + "resolved_model": response.model_identity, + "model_digest": response.model_digest, + "runtime": response.runtime_name, + "runtime_version": response.runtime_version, + "endpoint_class": backend.endpoint_class, + }, + "r0_override_permitted": False, + "science_influence_allowed": False, + "promotion_status": "not_eligible", + "live_control": False, + } + + +def _assert_metadata_only(value: Any) -> None: + if isinstance(value, dict): + forbidden = set(value) & _METADATA_FORBIDDEN_KEYS + if forbidden: + raise ReviewContractError("raw_payload_in_aggregate", "aggregate received prohibited raw payload fields") + for item in value.values(): + _assert_metadata_only(item) + elif isinstance(value, list): + for item in value: + _assert_metadata_only(item) + + +def aggregate_review_results(results: Iterable[ChunkReviewResult], *, source_sha256: str, disposition: dict[str, Any]) -> dict[str, Any]: + """Aggregate accepted structured findings only. Raw chunk records cannot enter.""" + require_quarantine_disposition(disposition) + persisted = [result.persisted() for result in results] + _assert_metadata_only(persisted) + by_type: Counter[str] = Counter() + source_lines: set[int] = set() + confidence_bins = {"0.00_to_0.24": 0, "0.25_to_0.49": 0, "0.50_to_0.74": 0, "0.75_to_1.00": 0} + classifications: dict[int, set[str]] = defaultdict(set) + model_runs: list[dict[str, Any]] = [] + accepted = 0 + rejected = 0 + for result in persisted: + if result["review_status"] == "accepted": + accepted += 1 + else: + rejected += 1 + for attempt in result["attempts"]: + model_runs.append({key: attempt[key] for key in attempt if key != "validation_error_codes"}) + for finding in result["findings"]: + by_type[finding["finding_type"]] += 1 + confidence = finding["confidence"] + key = "0.00_to_0.24" if confidence < 0.25 else "0.25_to_0.49" if confidence < 0.5 else "0.50_to_0.74" if confidence < 0.75 else "0.75_to_1.00" + confidence_bins[key] += 1 + for evidence in finding["evidence"]: + line = int(_CITATION_ID_PATTERN.fullmatch(evidence["citation_id"]).group(1)) + source_lines.add(line) + classifications[line].add(finding["finding_type"]) + conflicts = [ + {"source_line": line, "finding_types": sorted(types)} + for line, types in sorted(classifications.items()) + if len(types) > 1 + ] + return { + "schema_version": SUMMARY_SCHEMA, + "source_sha256": source_sha256, + "model_runs": sorted(model_runs, key=canonical_json), + "chunk_count": len(persisted), + "accepted_finding_count": sum(len(result["findings"]) for result in persisted), + "rejected_response_count": rejected, + "accepted_chunk_count": accepted, + "finding_counts_by_type": dict(sorted(by_type.items())), + "affected_source_lines": sorted(source_lines), + "confidence_distribution": confidence_bins, + "conflicting_findings": conflicts, + "privacy_checks": { + "raw_records_received_by_aggregator": False, + "bounded_private_text_detector": "synthetic_ngram_overlap_v0", + "private_text_detector_is_not_a_proof_of_privacy": True, + "embeddings_created": False, + "weights_updated": False, + }, + "science_influence_allowed": False, + "disposition": disposition, + } + + +def _git_commit(repository_root: Path) -> str: + try: + result = subprocess.run( + ["git", "-C", str(repository_root), "rev-parse", "HEAD"], + check=True, + capture_output=True, + text=True, + ) + return result.stdout.strip() + except (OSError, subprocess.CalledProcessError): + return "not_reported" + + +def build_run_manifest(run: CascadeRun, *, repository_root: Path) -> dict[str, Any]: + attempts = [attempt for result in run.chunk_results for attempt in result.attempts] + first = attempts[0] if attempts else None + return { + "schema_version": RUN_MANIFEST_SCHEMA, + "cascade_version": CASCADE_VERSION, + "source_sha256": run.source.source_sha256, + "quarantine_report_sha256": run.source.quarantine_report_sha256, + "orchestrator_commit": _git_commit(repository_root), + "orchestrator_sha256": sha256_bytes(Path(__file__).read_bytes()), + "configuration_sha256": run.configuration.sha256, + "system_prompt_sha256": sha256_bytes(FROZEN_SYSTEM_PROMPT.encode("utf-8")), + "backend": { + "runtime": first.runtime_name if first is not None else getattr(run.backend, "backend_type", "not_reported"), + "endpoint_class": run.backend.endpoint_class, + "requested_model": first.requested_model if first is not None else "not_reported", + "resolved_model": first.resolved_model if first is not None else "not_reported", + "model_digest": first.model_digest if first is not None else "not_reported", + "network_isolation_operator_attested": run.attestation.network_isolation_operator_attested, + }, + "generation": { + "temperature": run.configuration.temperature, + "seed": run.configuration.seed, + "context_limit": run.configuration.context_limit, + "max_output_tokens": run.configuration.max_output_tokens, + }, + "privacy": { + "embeddings_created": False, + "weights_updated": False, + "persistent_memory_enabled": False, + "cloud_exposure_allowed": False, + "prompt_response_logging_confined_operator_attested": run.attestation.prompt_response_logging_confined, + }, + "disposition": run.source.disposition, + "science_status": "pipeline_only", + "decision": "insufficient_evidence", + "promotion_status": "not_eligible", + "live_control": False, + } + + +def _approved_artifact_directory(path: Path, repository_root: Path) -> str: + resolved = path.resolve() + root = repository_root.resolve() + try: + relative = resolved.relative_to(root) + except ValueError: + return "outside_repository" + result = subprocess.run( + ["git", "-C", str(root), "check-ignore", "-q", "--", str(relative)], + capture_output=True, + check=False, + ) + if result.returncode != 0: + raise ReviewContractError("unapproved_artifact_directory", "repository-local artifacts must be Git-ignored") + return "git_ignored" + + +def write_metadata_only_artifacts(run: CascadeRun, *, artifact_directory: Path, repository_root: Path) -> dict[str, Path]: + """Persist receipts only after the destination is outside Git or ignored by it.""" + _approved_artifact_directory(artifact_directory, repository_root) + artifact_directory.mkdir(parents=True, exist_ok=True) + chunks = [result.persisted() for result in run.chunk_results] + summary = aggregate_review_results(chunks_to_results(chunks), source_sha256=run.source.source_sha256, disposition=run.source.disposition) + manifest = build_run_manifest(run, repository_root=repository_root) + for artifact in (chunks, summary, manifest): + _assert_metadata_only(artifact) + paths = { + "chunks": artifact_directory / "local-review-chunks-v0.jsonl", + "summary": artifact_directory / "local-review-summary-v0.json", + "manifest": artifact_directory / "local-review-run-manifest-v0.json", + } + paths["chunks"].write_text("".join(canonical_json(chunk) + "\n" for chunk in chunks), encoding="utf-8") + paths["summary"].write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8") + paths["manifest"].write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return paths + + +def chunks_to_results(chunks: Iterable[dict[str, Any]]) -> list[ChunkReviewResult]: + """Validate persisted receipts before allowing them into aggregation.""" + results: list[ChunkReviewResult] = [] + required = { + "schema_version", + "chunk_id", + "source_sha256", + "allowed_citation_ids", + "new_record_citation_ids", + "findings", + "review_status", + "attempts", + "rejection_error_codes", + } + for chunk in chunks: + if not isinstance(chunk, dict): + raise ReviewContractError("invalid_receipt", "persisted chunk receipt must be an object") + _assert_metadata_only(chunk) + if set(chunk) != required or chunk.get("schema_version") != CHUNK_SCHEMA: + raise ReviewContractError("invalid_receipt", "persisted chunk receipt has an invalid schema") + if chunk.get("review_status") not in {"accepted", "rejected"}: + raise ReviewContractError("invalid_receipt", "persisted chunk review status is invalid") + if not isinstance(chunk.get("findings"), list) or not isinstance(chunk.get("attempts"), list): + raise ReviewContractError("invalid_receipt", "persisted chunk receipt is malformed") + if chunk["review_status"] == "rejected" and chunk["findings"]: + raise ReviewContractError("invalid_receipt", "rejected chunks cannot contain partial findings") + allowed_citations = chunk["allowed_citation_ids"] + new_citations = chunk["new_record_citation_ids"] + if ( + not isinstance(allowed_citations, list) + or not isinstance(new_citations, list) + or not all(isinstance(item, str) and _CITATION_ID_PATTERN.fullmatch(item) for item in allowed_citations) + or not set(new_citations).issubset(set(allowed_citations)) + ): + raise ReviewContractError("invalid_receipt", "persisted citation metadata is invalid") + normalized_findings: list[dict[str, Any]] = [] + finding_keys = { + "schema_version", + "finding_id", + "finding_type", + "severity", + "confidence", + "evidence", + "observation", + "engineering_implication", + "contains_verbatim_private_text", + } + for finding in chunk["findings"]: + if not isinstance(finding, dict): + raise ReviewContractError("invalid_receipt", "persisted finding must be an object") + _reject_unknown_or_forbidden_fields(finding, finding_keys) + if set(finding) != finding_keys or finding.get("schema_version") != FINDING_SCHEMA: + raise ReviewContractError("invalid_receipt", "persisted finding schema is invalid") + finding_id = _require_nonempty_string(finding.get("finding_id"), code="invalid_receipt") + if not _FINDING_ID_PATTERN.fullmatch(finding_id) or finding.get("finding_type") not in ALLOWED_FINDING_TYPES or finding.get("severity") not in ALLOWED_SEVERITIES: + raise ReviewContractError("invalid_receipt", "persisted finding identity is invalid") + confidence = finding.get("confidence") + if isinstance(confidence, bool) or not isinstance(confidence, (int, float)) or not math.isfinite(float(confidence)) or not 0.0 <= float(confidence) <= 1.0: + raise ReviewContractError("invalid_receipt", "persisted finding confidence is invalid") + if finding.get("contains_verbatim_private_text") is not False: + raise ReviewContractError("invalid_receipt", "persisted finding cannot allow verbatim text") + evidence = finding.get("evidence") + if not isinstance(evidence, list) or not evidence: + raise ReviewContractError("invalid_receipt", "persisted finding evidence is invalid") + citations: list[str] = [] + normalized_evidence: list[dict[str, str]] = [] + for item in evidence: + if not isinstance(item, dict) or set(item) != {"citation_id", "claim"}: + raise ReviewContractError("invalid_receipt", "persisted evidence is invalid") + citation = item.get("citation_id") + claim = item.get("claim") + if not isinstance(citation, str) or citation not in allowed_citations or not isinstance(claim, str) or not claim.strip() or _contains_forbidden_text(claim): + raise ReviewContractError("invalid_receipt", "persisted evidence violates the boundary") + citations.append(citation) + normalized_evidence.append({"citation_id": citation, "claim": claim}) + if not set(citations) & set(new_citations): + raise ReviewContractError("invalid_receipt", "persisted finding has overlap-only evidence") + observation = _require_nonempty_string(finding.get("observation"), code="invalid_receipt") + implication = _require_nonempty_string(finding.get("engineering_implication"), code="invalid_receipt") + if _contains_forbidden_text(observation) or _contains_forbidden_text(implication): + raise ReviewContractError("invalid_receipt", "persisted finding has prohibited semantic content") + normalized_findings.append( + { + "schema_version": FINDING_SCHEMA, + "finding_id": finding_id, + "finding_type": finding["finding_type"], + "severity": finding["severity"], + "confidence": float(confidence), + "evidence": normalized_evidence, + "observation": observation, + "engineering_implication": implication, + "contains_verbatim_private_text": False, + } + ) + attempt_keys = { + "attempt", + "stage", + "requested_model", + "resolved_model", + "model_digest", + "runtime_name", + "runtime_version", + "endpoint_class", + "elapsed_ms", + "prompt_tokens", + "completion_tokens", + "stop_reason", + "status", + "validation_error_codes", + } + for item in chunk["attempts"]: + if not isinstance(item, dict) or set(item) != attempt_keys: + raise ReviewContractError("invalid_receipt", "persisted attempt receipt is invalid") + if item["stage"] not in {"R1", "R2"} or item["status"] not in {"accepted", "rejected"}: + raise ReviewContractError("invalid_receipt", "persisted attempt status is invalid") + if isinstance(item["attempt"], bool) or not isinstance(item["attempt"], int) or item["attempt"] < 1: + raise ReviewContractError("invalid_receipt", "persisted attempt ordinal is invalid") + if not isinstance(item["requested_model"], str) or not isinstance(item["runtime_name"], str): + raise ReviewContractError("invalid_receipt", "persisted attempt identity is invalid") + if item["endpoint_class"] not in {"loopback", "unix_socket", "mock"}: + raise ReviewContractError("invalid_receipt", "persisted endpoint class is invalid") + if not isinstance(item["elapsed_ms"], (int, float)) or not math.isfinite(float(item["elapsed_ms"])): + raise ReviewContractError("invalid_receipt", "persisted elapsed time is invalid") + if not isinstance(item["validation_error_codes"], list) or not all(isinstance(code, str) for code in item["validation_error_codes"]): + raise ReviewContractError("invalid_receipt", "persisted validation errors are invalid") + # Aggregation only needs the metadata representation, so create a small + # shell rather than reconstructing the raw in-memory chunk. + results.append( + ChunkReviewResult( + {key: chunk[key] for key in ("schema_version", "chunk_id", "source_sha256", "allowed_citation_ids", "new_record_citation_ids")}, + tuple(sorted(normalized_findings, key=lambda item: item["finding_id"])), + chunk["review_status"], + tuple( + AttemptReceipt( + item["attempt"], item["stage"], item["requested_model"], item.get("resolved_model"), item.get("model_digest"), + item["runtime_name"], item.get("runtime_version"), item["endpoint_class"], float(item["elapsed_ms"]), + item.get("prompt_tokens"), item.get("completion_tokens"), item.get("stop_reason"), item["status"], + tuple(item.get("validation_error_codes", [])), + ) + for item in chunk["attempts"] + ), + tuple(chunk["rejection_error_codes"]), + ) + ) + return results + + +def load_quarantined_source(source_path: Path, quarantine_report_path: Path) -> SourceEnvelope: + """Load an explicit local source only after validating its existing quarantine report.""" + import quarantine_dialectic_corpus as quarantine # Local sibling; no data read on import. + + report_bytes = quarantine_report_path.read_bytes() + report = strict_json_loads(report_bytes.decode("utf-8")) + if not isinstance(report, dict) or report.get("schema_version") != quarantine.PARSE_REPORT_SCHEMA: + raise ReviewContractError("invalid_quarantine_report", "quarantine report schema is invalid") + disposition = require_quarantine_disposition(report.get("disposition")) + source_bytes = source_path.read_bytes() + source_sha = sha256_bytes(source_bytes) + source_metadata = report.get("source") + if not isinstance(source_metadata, dict) or source_metadata.get("sha256") != source_sha: + raise ReviewContractError("source_checksum_mismatch", "source does not match quarantine report") + records: list[ReviewRecord] = [] + for line, raw_line in enumerate(source_bytes.splitlines(), start=1): + try: + event = quarantine.load_strict_json_object(raw_line) + quarantine.validate_turn_schema(event) + except quarantine.ParseFailure: + continue + records.append(ReviewRecord(line, event["index"], event)) + expected = report.get("records", {}).get("valid_record_count") if isinstance(report.get("records"), dict) else None + if expected != len(records): + raise ReviewContractError("source_record_count_mismatch", "source no longer matches quarantine report count") + return SourceEnvelope(source_sha, sha256_bytes(report_bytes), disposition, tuple(records), source_path) + + +def dry_run_receipt(source: SourceEnvelope, configuration: ReviewConfiguration) -> dict[str, Any]: + """Build envelopes without invoking a model or serializing raw records.""" + chunks = build_chunk_envelopes(source, configuration) + return { + "schema_version": "nc-local-review-dry-run-v0", + "source_sha256": source.source_sha256, + "chunk_count": len(chunks), + "chunk_ids": [chunk.chunk_id for chunk in chunks], + "all_chunks_have_new_records": all(bool(chunk.new_record_citation_ids) for chunk in chunks), + "configuration_sha256": configuration.sha256, + "model_invoked": False, + "disposition": source.disposition, + "science_status": "pipeline_only", + "decision": "insufficient_evidence", + "promotion_status": "not_eligible", + "live_control": False, + } + + +def _load_configuration(path: Path | None) -> ReviewConfiguration: + if path is None: + return ReviewConfiguration.from_mapping(default_configuration()) + value = strict_json_loads(path.read_text(encoding="utf-8")) + if not isinstance(value, dict): + raise ReviewContractError("invalid_configuration", "configuration must be a JSON object") + return ReviewConfiguration.from_mapping(value) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--source", required=True, type=Path) + parser.add_argument("--quarantine-report", required=True, type=Path) + parser.add_argument("--configuration", type=Path) + parser.add_argument("--dry-run", action="store_true") + parser.add_argument("--backend-url") + parser.add_argument("--artifact-directory", type=Path) + parser.add_argument("--repository-root", type=Path, default=Path(__file__).resolve().parents[1]) + parser.add_argument("--attest-network-isolation", action="store_true") + parser.add_argument("--attest-prompt-response-logging-confined", action="store_true") + args = parser.parse_args(argv) + source = load_quarantined_source(args.source, args.quarantine_report) + configuration = _load_configuration(args.configuration) + if args.dry_run: + print(json.dumps(dry_run_receipt(source, configuration), sort_keys=True)) + return 0 + if not args.backend_url or args.artifact_directory is None: + parser.error("a non-dry run requires --backend-url and --artifact-directory") + attestation = OperatorAttestation(args.attest_network_isolation, args.attest_prompt_response_logging_confined) + run = run_review_cascade(source, configuration, OllamaLocalBackend(args.backend_url), attestation) + paths = write_metadata_only_artifacts(run, artifact_directory=args.artifact_directory, repository_root=args.repository_root) + print(json.dumps({key: str(value) for key, value in paths.items()}, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/Scripts/local_review_noninterference.py b/Scripts/local_review_noninterference.py new file mode 100644 index 0000000..34fbc35 --- /dev/null +++ b/Scripts/local_review_noninterference.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +"""Deterministic metadata-only audit for review/EEG noninterference.""" + +from __future__ import annotations + +import json +from typing import Any, Iterable + + +NONINTERFERENCE_SCHEMA = "nc-local-review-eeg-noninterference-v0" + + +class NoninterferenceError(ValueError): + pass + + +def _serialized(value: Any) -> str: + return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")) + + +def _require_absent(value: Any, prohibited_values: Iterable[str], label: str) -> None: + serialized = _serialized(value) + if any(item and item in serialized for item in prohibited_values): + raise NoninterferenceError(label) + + +def audit_noninterference( + *, + dialogue_source_sha256: str, + dialogue_content_hashes: Iterable[str], + review_finding_ids: Iterable[str], + eeg_dataset_artifact: Any, + eeg_state_artifact: Any, + eeg_model_input_manifest: Any, + eeg_experiment_configuration: Any, + local_review_prompt_metadata: Any, + eeg_window_hashes: Iterable[str], + shared_training_buffer: Any = None, + dialogue_embeddings_created: bool = False, + dialogue_derived_weight_updates: bool = False, +) -> dict[str, Any]: + """Prove only absence from supplied metadata artifacts, never data quality.""" + if not isinstance(dialogue_source_sha256, str) or not dialogue_source_sha256: + raise NoninterferenceError("dialogue source SHA-256 is required") + content_hashes = tuple(dialogue_content_hashes) + finding_ids = tuple(review_finding_ids) + window_hashes = tuple(eeg_window_hashes) + _require_absent([eeg_dataset_artifact, eeg_state_artifact], [dialogue_source_sha256], "dialogue source SHA present in EEG artifact") + _require_absent(eeg_model_input_manifest, content_hashes, "dialogue content hash present in EEG model input") + _require_absent(eeg_experiment_configuration, finding_ids, "review finding present in EEG experiment configuration") + _require_absent(local_review_prompt_metadata, window_hashes, "EEG window hash present in review prompt metadata") + if shared_training_buffer is not None: + raise NoninterferenceError("review and EEG tracks must not share a training buffer") + if dialogue_embeddings_created: + raise NoninterferenceError("dialogue embeddings are prohibited") + if dialogue_derived_weight_updates: + raise NoninterferenceError("dialogue-derived weight updates are prohibited") + return { + "schema_version": NONINTERFERENCE_SCHEMA, + "status": "pass", + "checks": { + "dialogue_source_sha_absent_from_eeg_artifacts": True, + "dialogue_content_hashes_absent_from_eeg_model_inputs": True, + "review_findings_absent_from_eeg_configuration": True, + "eeg_window_hashes_absent_from_review_prompts": True, + "no_shared_training_buffer": True, + "no_dialogue_embeddings": True, + "no_dialogue_derived_weight_updates": True, + }, + "science_status": "pipeline_only", + "decision": "insufficient_evidence", + "promotion_status": "not_eligible", + "live_control": False, + } diff --git a/Scripts/quarantine_dialectic_corpus.py b/Scripts/quarantine_dialectic_corpus.py index fe93a4e..2271c43 100644 --- a/Scripts/quarantine_dialectic_corpus.py +++ b/Scripts/quarantine_dialectic_corpus.py @@ -40,6 +40,7 @@ "eligible_for_encoder_evaluation": False, "eligible_for_policy_training": False, "eligible_for_policy_evaluation": False, + "eligible_for_science": False, "contains_private_dialogue": True, "cloud_exposure_allowed": False, } diff --git a/Scripts/research_decision_register.py b/Scripts/research_decision_register.py new file mode 100644 index 0000000..e9ad5f5 --- /dev/null +++ b/Scripts/research_decision_register.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 +"""Validation for governance-only research decision-register entries. + +An entry records why a method may be studied. It cannot authorize a dependency, +runtime feature, model run, experiment, or promotion decision. +""" + +from __future__ import annotations + +import json +from typing import Any, Mapping + + +DECISION_REGISTER_SCHEMA = "nc-research-decision-register-v0" +PASSES = frozenset({1, 2, 3, 4}) +OWNERS = frozenset({"science", "engineering", "computation"}) +DATA_GATES = frozenset({"D0", "D1", "D2", "D3", "post_encoder"}) +IMPLEMENTATION_STATUSES = frozenset({"deferred", "study_only", "eligible_for_experiment"}) +REQUIRED_FIELDS = frozenset( + { + "schema_version", + "topic", + "pass", + "owner", + "registered_question", + "decision_it_can_change", + "required_data_gate", + "falsification_criterion", + "implementation_status", + "runtime_dependency_authorized", + } +) + + +class DecisionRegisterError(ValueError): + pass + + +def _required_text(value: Any, name: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise DecisionRegisterError(f"{name} must be a nonempty string") + return value + + +def validate_decision_register_entry(value: Mapping[str, Any]) -> dict[str, Any]: + """Strictly validate a metadata-only entry without granting implementation.""" + if set(value) != REQUIRED_FIELDS: + raise DecisionRegisterError("decision register entry has missing or unsupported fields") + if value.get("schema_version") != DECISION_REGISTER_SCHEMA: + raise DecisionRegisterError("decision register schema version is invalid") + if isinstance(value.get("pass"), bool) or value.get("pass") not in PASSES: + raise DecisionRegisterError("pass must be an integer in [1, 4]") + if value.get("owner") not in OWNERS: + raise DecisionRegisterError("owner is invalid") + if value.get("required_data_gate") not in DATA_GATES: + raise DecisionRegisterError("required_data_gate is invalid") + if value.get("implementation_status") not in IMPLEMENTATION_STATUSES: + raise DecisionRegisterError("implementation_status is invalid") + if value.get("runtime_dependency_authorized") is not False: + raise DecisionRegisterError("decision registers never authorize runtime dependencies") + for field in ("topic", "registered_question", "decision_it_can_change", "falsification_criterion"): + _required_text(value.get(field), field) + return json.loads(json.dumps(dict(value), sort_keys=True)) diff --git a/Scripts/review_quarantined_dialectics.py b/Scripts/review_quarantined_dialectics.py new file mode 100644 index 0000000..f5c351b --- /dev/null +++ b/Scripts/review_quarantined_dialectics.py @@ -0,0 +1,605 @@ +#!/usr/bin/env python3 +"""Run a bounded, local-only semantic engineering review of private dialectics. + +This tool accepts only a quarantined dialectic corpus and a loopback Ollama +endpoint. It sends bounded raw chunks to a local Qwen model without retaining +prompts, then persists only validated, metadata-only findings. It never creates +embeddings, changes weights, or produces scientific evidence. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import urllib.error +import urllib.request +from collections import Counter, defaultdict +from pathlib import Path +from typing import Any +from urllib.parse import urlparse + +import quarantine_dialectic_corpus as quarantine + + +REVIEW_SCHEMA = "nc-dialectic-local-review-chunk-v0" +RUN_SCHEMA = "nc-dialectic-local-review-run-v0" +AGGREGATE_SCHEMA = "nc-dialectic-local-review-aggregate-v0" +REVIEWER_VERSION = "local-dialectic-review-v1" +DEFAULT_MODEL = "qwen2.5:0.5b" +DEFAULT_OLLAMA_URL = "http://127.0.0.1:11434" +CHUNK_SIZE = 16 +CHUNK_OVERLAP = 2 +REVIEW_SEED = 42 +REVIEW_TEMPERATURE = 0.0 +DEFAULT_CONTEXT_LIMIT = 8192 +MAX_OBSERVATION_LENGTH = 400 +MAX_IMPLICATION_LENGTH = 300 +FINDING_TYPES = frozenset( + { + "malformed_record", + "duplicate_index", + "non_monotonic_index", + "candidate_role_inconsistency", + "candidate_duplication", + "selection_inertia", + "repetitive_synthesis", + "state_discontinuity", + "generator_identity_change", + "possible_context_loss", + "unclassified", + } +) +SEVERITIES = frozenset({"low", "medium", "high"}) +REVIEW_DISPOSITION = { + "corpus_role": "engineering_replay_only", + "review_role": "local_semantic_engineering_review", + "development_only_permanent": True, + "eligible_for_science": False, + "eligible_for_training": False, + "cloud_exposure_allowed": False, +} +FINDINGS_JSON_SCHEMA = { + "type": "array", + "items": { + "type": "object", + "additionalProperties": False, + "required": [ + "source_lines", + "legacy_turn_indices", + "finding_type", + "severity", + "confidence", + "observation", + "engineering_implication", + "contains_verbatim_private_text", + ], + "properties": { + "source_lines": {"type": "array", "items": {"type": "integer"}}, + "legacy_turn_indices": {"type": "array", "items": {"type": "integer"}}, + "finding_type": {"type": "string", "enum": sorted(FINDING_TYPES)}, + "severity": {"type": "string", "enum": sorted(SEVERITIES)}, + "confidence": {"type": "number", "minimum": 0, "maximum": 1}, + "observation": {"type": "string", "maxLength": MAX_OBSERVATION_LENGTH}, + "engineering_implication": {"type": "string", "maxLength": MAX_IMPLICATION_LENGTH}, + "contains_verbatim_private_text": {"type": "boolean", "const": False}, + }, + }, +} +REQUIRED_QUARANTINE_DISPOSITION = { + "corpus_role": "engineering_replay_only", + "development_only_permanent": True, + "eligible_for_encoder_training": False, + "eligible_for_encoder_evaluation": False, + "eligible_for_policy_training": False, + "eligible_for_policy_evaluation": False, + "cloud_exposure_allowed": False, +} + +SYSTEM_PROMPT = """You are reviewing a private, permanently quarantined dialogue corpus for local engineering purposes. + +The corpus is development-only. It is ineligible for encoder training, encoder evaluation, policy training, policy evaluation, scientific claims, EEG labeling, or cloud exposure. + +Identify structural and conversational-runtime defects that could affect deterministic replay or dialogue-system engineering. You may inspect semantic content only for repeated/collapsed candidates, candidate-role inconsistency, selection inertia, synthesis repetition, apparent context loss, turn-state discontinuity, candidate-diversity versus selection mismatch, or generator behavior changes. + +Do not infer thoughts, intention, emotion, agreement, diagnosis, or cognitive state. Do not propose EEG labels or features. Do not infer timestamps or speaker identities. Do not reproduce private dialogue text. Do not recommend training from this corpus. + +Return only a JSON array. Each object must contain source_lines, legacy_turn_indices, finding_type, severity, confidence, observation, engineering_implication, and contains_verbatim_private_text. The final field must be false. Use source-line order as canonical; legacy turn indices may repeat or reset. Use unclassified when evidence is ambiguous.""" + + +class ReviewContractError(ValueError): + """A local review request or model response violates the quarantine contract.""" + + +def sha256_bytes(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def sha256_json(value: Any) -> str: + return sha256_bytes(json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")) + + +def _require_disposition(value: Any) -> None: + if not isinstance(value, dict): + raise ReviewContractError("parse report lacks quarantine disposition") + for key, expected in REQUIRED_QUARANTINE_DISPOSITION.items(): + if value.get(key) != expected: + raise ReviewContractError(f"parse report quarantine disposition mismatch: {key}") + + +def load_reviewable_records(source_path: Path, parse_report_path: Path) -> tuple[dict[str, Any], list[dict[str, Any]]]: + """Load only source rows that passed the prior strict quarantine parser.""" + report = json.loads(parse_report_path.read_text()) + if report.get("schema_version") != quarantine.PARSE_REPORT_SCHEMA: + raise ReviewContractError("unexpected parse report schema") + _require_disposition(report.get("disposition")) + + source_bytes = source_path.read_bytes() + source = report.get("source") + if not isinstance(source, dict) or source.get("sha256") != sha256_bytes(source_bytes): + raise ReviewContractError("raw source checksum does not match quarantine parse report") + + records: list[dict[str, Any]] = [] + for source_line, raw_line in enumerate(source_bytes.splitlines(), start=1): + try: + event = quarantine.load_strict_json_object(raw_line) + quarantine.validate_turn_schema(event) + except quarantine.ParseFailure: + continue + records.append({"source_line": source_line, "legacy_turn_index": event["index"], "raw_event": event}) + + expected_count = report.get("records", {}).get("valid_record_count") + if expected_count != len(records): + raise ReviewContractError("raw source no longer reconstructs the quarantined valid-record count") + return report, records + + +def chunk_records(records: list[dict[str, Any]], *, chunk_size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> list[list[dict[str, Any]]]: + if chunk_size <= 0 or overlap < 0 or overlap >= chunk_size: + raise ReviewContractError("chunk size must be positive and overlap must be smaller than it") + chunks: list[list[dict[str, Any]]] = [] + start = 0 + while start < len(records): + chunk = records[start : start + chunk_size] + if not chunk: + break + chunks.append(chunk) + if start + chunk_size >= len(records): + break + start += chunk_size - overlap + return chunks + + +def _sanitize_identity(event: dict[str, Any]) -> dict[str, str] | None: + return quarantine.fingerprint_identity(event) + + +def build_chunk_context(chunk_id: str, records: list[dict[str, Any]]) -> str: + """Build the only raw-content prompt payload used by one stateless request.""" + source_line_to_legacy_index = { + str(record["source_line"]): record["legacy_turn_index"] + for record in records + } + return json.dumps( + { + "disposition": REVIEW_DISPOSITION, + "chunk_id": chunk_id, + "citation_contract": { + "canonical_source_lines_allowed": [record["source_line"] for record in records], + "legacy_turn_index_by_source_line": source_line_to_legacy_index, + "rule": "Cite only listed source lines and their exact mapped legacy index. Do not invent contiguous ordinals.", + }, + "records": [ + { + "source_line": record["source_line"], + "legacy_turn_index": record["legacy_turn_index"], + "generator_fingerprint": _sanitize_identity(record["raw_event"]), + "raw_record": record["raw_event"], + } + for record in records + ], + }, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ) + + +def _require_loopback_ollama(base_url: str) -> str: + parsed = urlparse(base_url) + if ( + parsed.scheme != "http" + or parsed.hostname not in {"127.0.0.1", "localhost", "::1"} + or parsed.port != 11434 + or parsed.username is not None + or parsed.password is not None + or parsed.path not in {"", "/"} + ): + raise ReviewContractError("local review permits only http://127.0.0.1:11434 or localhost equivalent") + return f"http://{parsed.netloc}" + + +def _request_json(url: str, *, method: str = "GET", payload: dict[str, Any] | None = None) -> dict[str, Any]: + data = None if payload is None else json.dumps(payload).encode("utf-8") + request = urllib.request.Request(url, data=data, method=method) + request.add_header("Accept", "application/json") + if data is not None: + request.add_header("Content-Type", "application/json") + try: + with urllib.request.urlopen(request, timeout=120) as response: + value = json.loads(response.read().decode("utf-8")) + except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as error: + raise ReviewContractError(f"local Ollama request failed: {error}") from error + if not isinstance(value, dict): + raise ReviewContractError("local Ollama response must be a JSON object") + return value + + +def local_model_identity(base_url: str, model: str) -> dict[str, Any]: + if not model.startswith("qwen2.5:"): + raise ReviewContractError("only a local qwen2.5 model is allowed for this review") + tags = _request_json(f"{base_url}/api/tags") + models = tags.get("models") + if not isinstance(models, list): + raise ReviewContractError("local Ollama tags response lacks models") + for candidate in models: + if isinstance(candidate, dict) and candidate.get("name") == model: + if candidate.get("remote_model") or candidate.get("remote_host"): + raise ReviewContractError("remote Ollama models are prohibited for corpus review") + digest = candidate.get("digest") + if not isinstance(digest, str) or not digest: + raise ReviewContractError("local Ollama model lacks a digest") + return { + "name": model, + "digest": digest, + "details": candidate.get("details") if isinstance(candidate.get("details"), dict) else {}, + } + raise ReviewContractError(f"local Ollama model is unavailable: {model}") + + +def _json_array_from_model_content(content: str) -> list[Any]: + stripped = content.strip() + if stripped.startswith("```"): + match = re.fullmatch(r"```(?:json)?\s*(.*?)\s*```", stripped, flags=re.DOTALL | re.IGNORECASE) + if match is not None: + stripped = match.group(1) + try: + value = json.loads(stripped) + except json.JSONDecodeError as first_error: + decoder = json.JSONDecoder() + value = None + for position, character in enumerate(stripped): + if character != "[": + continue + try: + candidate, _ = decoder.raw_decode(stripped[position:]) + except json.JSONDecodeError: + continue + if isinstance(candidate, list): + value = candidate + break + if value is None: + raise ReviewContractError("local review response was not a JSON array") from first_error + if not isinstance(value, list): + raise ReviewContractError("local review response must be a JSON array") + return value + + +def _private_text_fragments(records: list[dict[str, Any]]) -> list[str]: + fragments: list[str] = [] + + def add_fragment(value: str) -> None: + normalized = " ".join(value.split()) + if len(normalized) >= 16: + fragments.append(normalized) + words = normalized.split() + for start in range(max(0, len(words) - 3)): + phrase = " ".join(words[start : start + 4]) + if len(phrase) >= 16: + fragments.append(phrase) + + for record in records: + content = quarantine.content_projection(record["raw_event"]) + for field, value in content.items(): + if isinstance(value, str) and len(value.strip()) >= 16: + add_fragment(value) + if field == "candidates" and isinstance(value, list): + for candidate in value: + if isinstance(candidate, dict): + text = candidate.get("text") + if isinstance(text, str) and len(text.strip()) >= 16: + add_fragment(text) + return fragments + + +def _contains_private_text(value: str, fragments: list[str]) -> bool: + normalized = " ".join(value.casefold().split()) + return any(" ".join(fragment.casefold().split()) in normalized for fragment in fragments) + + +def validate_findings(response_content: str, records: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Accept only bounded, chunk-local, non-verbatim engineering findings.""" + value = _json_array_from_model_content(response_content) + allowed_lines = {record["source_line"] for record in records} + line_to_index = {record["source_line"]: record["legacy_turn_index"] for record in records} + index_to_lines: dict[int, list[int]] = defaultdict(list) + for source_line, legacy_index in line_to_index.items(): + index_to_lines[legacy_index].append(source_line) + private_fragments = _private_text_fragments(records) + findings: list[dict[str, Any]] = [] + for position, finding in enumerate(value): + if not isinstance(finding, dict): + raise ReviewContractError(f"finding {position} must be an object") + source_lines = finding.get("source_lines") + legacy_indexes = finding.get("legacy_turn_indices") + if ( + not isinstance(legacy_indexes, list) + or not legacy_indexes + or any(isinstance(index, bool) or not isinstance(index, int) for index in legacy_indexes) + ): + raise ReviewContractError(f"finding {position} has invalid legacy turn indexes") + if not isinstance(source_lines, list) or not source_lines or any(isinstance(line, bool) or not isinstance(line, int) for line in source_lines): + raise ReviewContractError(f"finding {position} has invalid source lines") + citation_normalized = False + if any(line not in allowed_lines for line in source_lines): + remapped_lines = [index_to_lines.get(index, []) for index in legacy_indexes] + if any(len(lines) != 1 for lines in remapped_lines): + raise ReviewContractError(f"finding {position} cites source lines outside its chunk") + source_lines = [lines[0] for lines in remapped_lines] + citation_normalized = True + if len(set(source_lines)) != len(source_lines): + raise ReviewContractError(f"finding {position} repeats a source line") + expected_indexes = {line_to_index[line] for line in source_lines} + if set(legacy_indexes) != expected_indexes: + raise ReviewContractError(f"finding {position} cites inconsistent source lines and legacy indexes") + finding_type = finding.get("finding_type") + severity = finding.get("severity") + confidence = finding.get("confidence") + observation = finding.get("observation") + implication = finding.get("engineering_implication") + if finding_type not in FINDING_TYPES or severity not in SEVERITIES: + raise ReviewContractError(f"finding {position} uses an unsupported category") + if isinstance(confidence, bool) or not isinstance(confidence, (int, float)) or not 0.0 <= float(confidence) <= 1.0: + raise ReviewContractError(f"finding {position} has invalid confidence") + if not isinstance(observation, str) or not observation or len(observation) > MAX_OBSERVATION_LENGTH: + raise ReviewContractError(f"finding {position} has invalid observation") + if not isinstance(implication, str) or not implication or len(implication) > MAX_IMPLICATION_LENGTH: + raise ReviewContractError(f"finding {position} has invalid engineering implication") + if finding.get("contains_verbatim_private_text") is not False: + raise ReviewContractError(f"finding {position} must deny verbatim private text") + if _contains_private_text(observation, private_fragments) or _contains_private_text(implication, private_fragments): + raise ReviewContractError(f"finding {position} contains private dialogue text") + findings.append( + { + "source_lines": sorted(source_lines), + "legacy_turn_indices": sorted(set(legacy_indexes)), + "finding_type": finding_type, + "severity": severity, + "confidence": float(confidence), + "observation": observation, + "engineering_implication": implication, + "contains_verbatim_private_text": False, + "citation_normalized_from_legacy_index": citation_normalized, + } + ) + return findings + + +def review_chunk(base_url: str, model: str, context_limit: int, chunk_id: str, records: list[dict[str, Any]]) -> list[dict[str, Any]]: + response = _request_json( + f"{base_url}/api/chat", + method="POST", + payload={ + "model": model, + "stream": False, + "format": FINDINGS_JSON_SCHEMA, + "messages": [ + {"role": "system", "content": SYSTEM_PROMPT}, + {"role": "user", "content": build_chunk_context(chunk_id, records)}, + ], + "options": {"temperature": REVIEW_TEMPERATURE, "seed": REVIEW_SEED, "num_ctx": context_limit}, + }, + ) + message = response.get("message") + if not isinstance(message, dict) or not isinstance(message.get("content"), str): + raise ReviewContractError("local Ollama chat response lacks message content") + return validate_findings(message["content"], records) + + +def aggregate_review_chunks(chunks: list[dict[str, Any]]) -> dict[str, Any]: + """Aggregate structured Pass-A findings without reopening the raw corpus.""" + category_lines: dict[str, set[int]] = defaultdict(set) + category_chunks: dict[str, set[str]] = defaultdict(set) + line_types: dict[int, set[str]] = defaultdict(set) + confidences: list[float] = [] + finding_count = 0 + for chunk in chunks: + chunk_id = str(chunk["chunk_id"]) + for finding in chunk["findings"]: + finding_count += 1 + finding_type = str(finding["finding_type"]) + source_lines = [int(line) for line in finding["source_lines"]] + category_lines[finding_type].update(source_lines) + category_chunks[finding_type].add(chunk_id) + for source_line in source_lines: + line_types[source_line].add(finding_type) + confidences.append(float(finding["confidence"])) + confidence_distribution = { + "0.00_to_0.24": sum(confidence < 0.25 for confidence in confidences), + "0.25_to_0.49": sum(0.25 <= confidence < 0.5 for confidence in confidences), + "0.50_to_0.74": sum(0.5 <= confidence < 0.75 for confidence in confidences), + "0.75_to_1.00": sum(confidence >= 0.75 for confidence in confidences), + } + categories = [ + { + "finding_type": finding_type, + "finding_count": sum( + 1 for chunk in chunks for finding in chunk["findings"] if finding["finding_type"] == finding_type + ), + "affected_source_lines": sorted(category_lines[finding_type]), + "chunk_ids": sorted(category_chunks[finding_type]), + "spans_chunk_boundaries": len(category_chunks[finding_type]) > 1, + } + for finding_type in sorted(category_lines) + ] + conflicts = [ + {"source_line": source_line, "finding_types": sorted(finding_types)} + for source_line, finding_types in sorted(line_types.items()) + if len(finding_types) > 1 + ] + return { + "schema_version": AGGREGATE_SCHEMA, + "reviewer_version": REVIEWER_VERSION, + "input_chunk_count": len(chunks), + "finding_count": finding_count, + "categories": categories, + "conflicting_findings": conflicts, + "confidence_distribution": confidence_distribution, + "disposition": REVIEW_DISPOSITION, + } + + +def _load_review_chunks(path: Path) -> list[dict[str, Any]]: + chunks: list[dict[str, Any]] = [] + for line_number, line in enumerate(path.read_text().splitlines(), start=1): + try: + chunk = json.loads(line) + except json.JSONDecodeError as error: + raise ReviewContractError(f"review findings line {line_number} is invalid JSON") from error + if not isinstance(chunk, dict) or chunk.get("schema_version") != REVIEW_SCHEMA: + raise ReviewContractError(f"review findings line {line_number} has the wrong schema") + if chunk.get("disposition") != REVIEW_DISPOSITION: + raise ReviewContractError(f"review findings line {line_number} has the wrong disposition") + if not isinstance(chunk.get("chunk_id"), str) or not isinstance(chunk.get("findings"), list): + raise ReviewContractError(f"review findings line {line_number} is malformed") + chunks.append(chunk) + return chunks + + +def run_review( + source_path: Path, + parse_report_path: Path, + findings_path: Path, + run_manifest_path: Path, + *, + base_url: str, + model: str, + context_limit: int, + prompt_logging_status: str, +) -> dict[str, Any]: + if prompt_logging_status != "verified_disabled": + raise ReviewContractError("private review requires prompt_logging_status=verified_disabled") + report, records = load_reviewable_records(source_path, parse_report_path) + chunks = chunk_records(records) + model_identity = local_model_identity(base_url, model) + review_chunks: list[dict[str, Any]] = [] + review_status = "completed" + rejected_chunk_count = 0 + for ordinal, records_chunk in enumerate(chunks, start=1): + chunk_id = f"chunk-{ordinal:03d}" + chunk_result = { + "schema_version": REVIEW_SCHEMA, + "chunk_id": chunk_id, + "source_lines": [record["source_line"] for record in records_chunk], + "legacy_turn_indices": [record["legacy_turn_index"] for record in records_chunk], + "disposition": REVIEW_DISPOSITION, + } + try: + chunk_result["findings"] = review_chunk(base_url, model, context_limit, chunk_id, records_chunk) + chunk_result["review_status"] = "accepted" + except ReviewContractError as error: + # Never persist a raw model response, even when it violates the + # requested schema or privacy boundary. The safe rejection receipt + # is enough to decide whether this model can perform the review. + chunk_result["findings"] = [] + chunk_result["review_status"] = "rejected_response" + chunk_result["rejection_reason"] = str(error) + review_chunks.append(chunk_result) + review_status = "contract_rejected" + rejected_chunk_count = 1 + break + review_chunks.append(chunk_result) + run_manifest = { + "schema_version": RUN_SCHEMA, + "reviewer_version": REVIEWER_VERSION, + "review_model": model_identity["name"], + "model_digest": model_identity["digest"], + "model_details": model_identity["details"], + "runtime": "ollama_local_loopback", + "network_enabled": False, + "temperature": REVIEW_TEMPERATURE, + "seed": REVIEW_SEED, + "context_limit": context_limit, + "chunk_size_valid_records": CHUNK_SIZE, + "chunk_overlap_records": CHUNK_OVERLAP, + "conversation_memory_between_chunks": False, + "embeddings_created": False, + "weights_updated": False, + "source_sha256": report["source"]["sha256"], + "science_influence_allowed": False, + "prompt_logging_status": "verified_disabled", + "disposition": REVIEW_DISPOSITION, + "review_status": review_status, + "accepted_chunk_count": len(review_chunks) - rejected_chunk_count, + "rejected_chunk_count": rejected_chunk_count, + "review_chunk_count": len(review_chunks), + "review_findings_sha256": sha256_json(review_chunks), + } + findings_path.parent.mkdir(parents=True, exist_ok=True) + run_manifest_path.parent.mkdir(parents=True, exist_ok=True) + findings_path.write_text("".join(json.dumps(chunk, sort_keys=True) + "\n" for chunk in review_chunks)) + run_manifest_path.write_text(json.dumps(run_manifest, indent=2, sort_keys=True) + "\n") + return run_manifest + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + subcommands = parser.add_subparsers(dest="command", required=True) + + review = subcommands.add_parser("review", help="perform stateless local Qwen chunk review") + review.add_argument("--input", required=True, type=Path, help="private local dialectic JSONL") + review.add_argument("--parse-report", required=True, type=Path) + review.add_argument("--findings-output", required=True, type=Path) + review.add_argument("--run-manifest-output", required=True, type=Path) + review.add_argument("--model", default=DEFAULT_MODEL) + review.add_argument("--ollama-url", default=DEFAULT_OLLAMA_URL) + review.add_argument("--context-limit", default=DEFAULT_CONTEXT_LIMIT, type=int) + review.add_argument( + "--prompt-logging-status", + choices=("verified_disabled",), + required=True, + help="operator attestation required before private text is sent to local Ollama", + ) + + aggregate = subcommands.add_parser("aggregate", help="aggregate structured Pass-A findings without raw dialogue") + aggregate.add_argument("--findings-input", required=True, type=Path) + aggregate.add_argument("--output", required=True, type=Path) + args = parser.parse_args(argv) + + if args.command == "aggregate": + result = aggregate_review_chunks(_load_review_chunks(args.findings_input)) + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n") + print(f"wrote {args.output} ({result['finding_count']} metadata-only findings)") + return 0 + + if args.context_limit <= 0: + raise SystemExit("--context-limit must be positive") + base_url = _require_loopback_ollama(args.ollama_url) + manifest = run_review( + args.input, + args.parse_report, + args.findings_output, + args.run_manifest_output, + base_url=base_url, + model=args.model, + context_limit=args.context_limit, + prompt_logging_status=args.prompt_logging_status, + ) + print(f"wrote {args.findings_output} and {args.run_manifest_output} ({manifest['review_chunk_count']} local-only chunks)") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/Tests/eval/test_local_dialectic_review.py b/Tests/eval/test_local_dialectic_review.py new file mode 100644 index 0000000..e4ebb73 --- /dev/null +++ b/Tests/eval/test_local_dialectic_review.py @@ -0,0 +1,161 @@ +"""Regression tests for bounded local semantic review of quarantined dialogue.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +import tempfile +import unittest +from pathlib import Path + + +SCRIPTS = Path(__file__).resolve().parents[2] / "Scripts" +sys.path.insert(0, str(SCRIPTS)) + + +def _load(name: str): + path = SCRIPTS / name + spec = importlib.util.spec_from_file_location(path.stem, path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +QUARANTINE = _load("quarantine_dialectic_corpus.py") +REVIEW = _load("review_quarantined_dialectics.py") + + +def _turn(index: int, text: str) -> dict: + return { + "index": index, + "heard": text, + "candidates": [ + {"text": f"candidate one {text}", "roleID": "coherence-seeking"}, + {"text": f"candidate two {text}", "roleID": "displacement-seeking"}, + ], + "tension": 0.4, + "margin": 0.1, + "selectionTemperature": 0.2, + "glossScalar": 0.5, + "outcome": "synthesized:synthesis", + "spokenText": f"spoken {text}", + } + + +class LocalDialecticReviewTests(unittest.TestCase): + def _quarantined_source(self, root: Path) -> tuple[Path, Path]: + source = root / "dialectic-turns-2026-07-22.jsonl" + source.write_text("\n".join(json.dumps(_turn(index, f"private phrase number {index}")) for index in range(18)) + "\n") + report = root / "parse-report.json" + QUARANTINE.inspect_corpus(source, report, root / "events.jsonl") + return source, report + + def test_chunking_is_stateless_with_fixed_overlap(self) -> None: + with tempfile.TemporaryDirectory() as directory: + source, report = self._quarantined_source(Path(directory)) + _, records = REVIEW.load_reviewable_records(source, report) + chunks = REVIEW.chunk_records(records) + self.assertEqual([len(chunk) for chunk in chunks], [16, 4]) + self.assertEqual([record["source_line"] for record in chunks[0][-2:]], [15, 16]) + self.assertEqual([record["source_line"] for record in chunks[1][:2]], [15, 16]) + context = REVIEW.build_chunk_context("chunk-001", chunks[0]) + self.assertIn("private phrase number 0", context) + self.assertIn("canonical_source_lines_allowed", context) + self.assertIn('"1":0', context) + self.assertNotIn("research", REVIEW.SYSTEM_PROMPT.casefold()) + + def test_review_response_is_chunk_bound_and_non_verbatim(self) -> None: + with tempfile.TemporaryDirectory() as directory: + source, report = self._quarantined_source(Path(directory)) + _, records = REVIEW.load_reviewable_records(source, report) + safe_response = json.dumps([{ + "source_lines": [1, 2], + "legacy_turn_indices": [0, 1], + "finding_type": "selection_inertia", + "severity": "medium", + "confidence": 0.75, + "observation": "Adjacent outputs show limited structural variation.", + "engineering_implication": "Replay should preserve source-line identity for comparison.", + "contains_verbatim_private_text": False, + }]) + findings = REVIEW.validate_findings(safe_response, records[:2]) + self.assertEqual(findings[0]["finding_type"], "selection_inertia") + self.assertFalse(findings[0]["contains_verbatim_private_text"]) + self.assertFalse(findings[0]["citation_normalized_from_legacy_index"]) + + prefixed_findings = REVIEW.validate_findings("Review result:\n" + safe_response, records[:2]) + self.assertEqual(prefixed_findings, findings) + + leaking_response = safe_response.replace( + "Adjacent outputs show limited structural variation.", + "private phrase number 0", + ) + with self.assertRaisesRegex(REVIEW.ReviewContractError, "private dialogue"): + REVIEW.validate_findings(leaking_response, records[:2]) + + legacy_citation_response = safe_response.replace('"source_lines": [1, 2]', '"source_lines": [0, 1]') + normalized = REVIEW.validate_findings(legacy_citation_response, records[:2]) + self.assertEqual(normalized[0]["source_lines"], [1, 2]) + self.assertTrue(normalized[0]["citation_normalized_from_legacy_index"]) + + def test_aggregation_never_reopens_raw_dialogue(self) -> None: + chunks = [ + { + "schema_version": REVIEW.REVIEW_SCHEMA, + "chunk_id": "chunk-001", + "findings": [{"finding_type": "duplicate_index", "source_lines": [3], "confidence": 0.8}], + "disposition": REVIEW.REVIEW_DISPOSITION, + }, + { + "schema_version": REVIEW.REVIEW_SCHEMA, + "chunk_id": "chunk-002", + "findings": [ + {"finding_type": "duplicate_index", "source_lines": [3, 17], "confidence": 0.9}, + {"finding_type": "state_discontinuity", "source_lines": [3], "confidence": 0.6}, + ], + "disposition": REVIEW.REVIEW_DISPOSITION, + }, + ] + result = REVIEW.aggregate_review_chunks(chunks) + duplicate = next(category for category in result["categories"] if category["finding_type"] == "duplicate_index") + self.assertTrue(duplicate["spans_chunk_boundaries"]) + self.assertEqual(duplicate["affected_source_lines"], [3, 17]) + self.assertEqual(result["conflicting_findings"], [{"source_line": 3, "finding_types": ["duplicate_index", "state_discontinuity"]}]) + self.assertEqual(result["disposition"], REVIEW.REVIEW_DISPOSITION) + + def test_aggregation_accepts_a_safe_rejection_receipt(self) -> None: + result = REVIEW.aggregate_review_chunks([{ + "schema_version": REVIEW.REVIEW_SCHEMA, + "chunk_id": "chunk-001", + "review_status": "rejected_response", + "rejection_reason": "finding 0 cites source lines outside its chunk", + "findings": [], + "disposition": REVIEW.REVIEW_DISPOSITION, + }]) + self.assertEqual(result["finding_count"], 0) + self.assertEqual(result["categories"], []) + + def test_remote_endpoint_and_cloud_model_are_rejected(self) -> None: + with self.assertRaisesRegex(REVIEW.ReviewContractError, "only http://127.0.0.1"): + REVIEW._require_loopback_ollama("https://example.com") + with self.assertRaisesRegex(REVIEW.ReviewContractError, "qwen2.5"): + REVIEW.local_model_identity("http://127.0.0.1:11434", "deepseek-v4-flash:cloud") + + def test_private_review_requires_prompt_logging_attestation(self) -> None: + with self.assertRaisesRegex(REVIEW.ReviewContractError, "prompt_logging_status"): + REVIEW.run_review( + Path("unused-source.jsonl"), + Path("unused-report.json"), + Path("unused-findings.jsonl"), + Path("unused-manifest.json"), + base_url="http://127.0.0.1:11434", + model="qwen2.5:0.5b", + context_limit=8192, + prompt_logging_status="unverified", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/Tests/eval/test_local_open_weight_review.py b/Tests/eval/test_local_open_weight_review.py new file mode 100644 index 0000000..c456b37 --- /dev/null +++ b/Tests/eval/test_local_open_weight_review.py @@ -0,0 +1,395 @@ +"""Synthetic contract tests for the fail-closed local open-weight review cascade.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +import tempfile +import unittest +from pathlib import Path + +import numpy as np + + +ROOT = Path(__file__).resolve().parents[2] +SCRIPTS = ROOT / "Scripts" +EEG_SOURCE = ROOT / "NeuralComposeEEG" / "src" +sys.path[:0] = [str(SCRIPTS), str(EEG_SOURCE), str(ROOT)] + + +def _load(filename: str): + path = SCRIPTS / filename + spec = importlib.util.spec_from_file_location(path.stem, path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +CASCADE = _load("local_open_weight_review.py") +NONINTERFERENCE = _load("local_review_noninterference.py") +REGISTER = _load("research_decision_register.py") + + +def _disposition() -> dict: + return { + "corpus_role": "engineering_replay_only", + "development_only_permanent": True, + "eligible_for_encoder_training": False, + "eligible_for_encoder_evaluation": False, + "eligible_for_policy_training": False, + "eligible_for_policy_evaluation": False, + "eligible_for_science": False, + "cloud_exposure_allowed": False, + } + + +def _record(index: int) -> dict: + return { + "index": index, + "payload": { + "heard": f"fixture confidential dialogue material alpha {index} stable", + "candidate": f"fixture candidate material beta {index} stable", + }, + } + + +def _source(count: int = 18, *, source_path: Path | None = None) -> object: + records = tuple( + CASCADE.ReviewRecord(41 + position, [7, 7, 4][position % 3], _record(position)) + for position in range(count) + ) + source_bytes = ("fixture-source-" + str(count)).encode("utf-8") + return CASCADE.SourceEnvelope( + CASCADE.sha256_bytes(source_bytes), + CASCADE.sha256_bytes(b"fixture-quarantine-report"), + _disposition(), + records, + source_path, + ) + + +def _configuration(*, maximum_attempts: int = 3) -> object: + value = json.loads(json.dumps(CASCADE.default_configuration())) + value["retry"]["maximum_attempts"] = maximum_attempts + return CASCADE.ReviewConfiguration.from_mapping(value) + + +def _finding(chunk: object, *, citation_id: str | None = None, finding_id: str = "finding-1") -> dict: + citation = citation_id or chunk.new_record_citation_ids[0] + return { + "schema_version": CASCADE.FINDING_SCHEMA, + "finding_id": finding_id, + "finding_type": "selection_inertia", + "severity": "medium", + "confidence": 0.75, + "evidence": [{"citation_id": citation, "claim": "A bounded structural pattern is visible."}], + "observation": "Adjacent records exhibit limited structural variation.", + "engineering_implication": "Replay should preserve source-line identity.", + "contains_verbatim_private_text": False, + } + + +def _attestation() -> object: + return CASCADE.OperatorAttestation(True, True) + + +class LocalOpenWeightReviewTests(unittest.TestCase): + def test_valid_chunk_construction_overlap_duplicate_and_nonmonotonic_indices(self) -> None: + chunks = CASCADE.build_chunk_envelopes(_source(), _configuration()) + self.assertEqual([len(chunk.records) for chunk in chunks], [16, 4]) + self.assertEqual(chunks[0].chunk_id.split(":")[-1], "0001") + self.assertEqual([record.citation_id for record in chunks[0].records[-2:]], ["line:55", "line:56"]) + self.assertEqual([record.citation_id for record in chunks[1].records[:2]], ["line:55", "line:56"]) + self.assertTrue(all(record.is_overlap_record for record in chunks[1].records[:2])) + self.assertTrue(all(not record.is_overlap_record for record in chunks[1].records[2:])) + self.assertEqual([record.legacy_turn_index for record in chunks[0].records[:3]], [7, 7, 4]) + self.assertEqual(CASCADE.FROZEN_SYSTEM_PROMPT, CASCADE.FROZEN_SYSTEM_PROMPT) + + def test_finding_validator_accepts_complete_causal_response(self) -> None: + chunk = CASCADE.build_chunk_envelopes(_source(2), _configuration())[0] + findings = CASCADE.validate_findings_response(json.dumps([_finding(chunk)]), chunk) + self.assertEqual(findings[0]["finding_type"], "selection_inertia") + self.assertEqual(findings[0]["evidence"][0]["citation_id"], "line:41") + + def test_validator_rejects_invented_and_mismatched_citations(self) -> None: + chunk = CASCADE.build_chunk_envelopes(_source(2), _configuration())[0] + invented = _finding(chunk, citation_id="line:999") + with self.assertRaisesRegex(CASCADE.ReviewContractError, "citation"): + CASCADE.validate_findings_response(json.dumps([invented]), chunk) + malformed_chunk = CASCADE.ChunkEnvelope( + chunk.chunk_id, + chunk.source_sha256, + (CASCADE.ChunkRecord("line:41", 42, 7, False, _record(0)),), + ) + mismatch = _finding(malformed_chunk) + with self.assertRaisesRegex(CASCADE.ReviewContractError, "citation"): + CASCADE.validate_findings_response(json.dumps([mismatch]), malformed_chunk) + + def test_validator_rejects_overlap_only_evidence(self) -> None: + chunk = CASCADE.build_chunk_envelopes(_source(), _configuration())[1] + finding = _finding(chunk, citation_id=chunk.records[0].citation_id) + with self.assertRaisesRegex(CASCADE.ReviewContractError, "new chunk"): + CASCADE.validate_findings_response(json.dumps([finding]), chunk) + + def test_validator_rejects_invalid_json_nonfinite_and_unknown_type(self) -> None: + chunk = CASCADE.build_chunk_envelopes(_source(2), _configuration())[0] + with self.assertRaisesRegex(CASCADE.ReviewContractError, "complete JSON"): + CASCADE.validate_findings_response("not-json", chunk) + nonfinite = json.dumps([_finding(chunk)]).replace("0.75", "NaN") + with self.assertRaisesRegex(CASCADE.ReviewContractError, "non-finite"): + CASCADE.validate_findings_response(nonfinite, chunk) + infinity = json.dumps([_finding(chunk)]).replace("0.75", "Infinity") + with self.assertRaisesRegex(CASCADE.ReviewContractError, "non-finite"): + CASCADE.validate_findings_response(infinity, chunk) + unknown = _finding(chunk) + unknown["finding_type"] = "unsupported" + with self.assertRaisesRegex(CASCADE.ReviewContractError, "not allowed"): + CASCADE.validate_findings_response(json.dumps([unknown]), chunk) + + def test_validator_rejects_out_of_range_empty_forbidden_and_private_leakage(self) -> None: + chunk = CASCADE.build_chunk_envelopes(_source(2), _configuration())[0] + out_of_range = _finding(chunk) + out_of_range["confidence"] = 1.1 + with self.assertRaisesRegex(CASCADE.ReviewContractError, "within"): + CASCADE.validate_findings_response(json.dumps([out_of_range]), chunk) + empty_evidence = _finding(chunk) + empty_evidence["evidence"] = [] + with self.assertRaisesRegex(CASCADE.ReviewContractError, "nonempty"): + CASCADE.validate_findings_response(json.dumps([empty_evidence]), chunk) + forbidden = _finding(chunk) + forbidden["eeg_label"] = "not permitted" + with self.assertRaisesRegex(CASCADE.ReviewContractError, "prohibited"): + CASCADE.validate_findings_response(json.dumps([forbidden]), chunk) + leaking = _finding(chunk) + leaking["observation"] = "fixture confidential dialogue material alpha 0 stable" + with self.assertRaisesRegex(CASCADE.ReviewContractError, "source text"): + CASCADE.validate_findings_response(json.dumps([leaking]), chunk) + + def test_loopback_acceptance_and_remote_rejection(self) -> None: + for endpoint in ("http://127.0.0.1:11434", "http://localhost:11434", "http://[::1]:11434", "unix:///tmp/local-review.sock"): + self.assertIn(CASCADE.classify_local_endpoint(endpoint), {"loopback", "unix_socket"}) + for endpoint in ("https://example.com", "http://192.168.1.10:11434", "http://localhost.evil:11434"): + with self.assertRaises(CASCADE.ReviewContractError): + CASCADE.classify_local_endpoint(endpoint) + + def test_missing_disposition_and_source_checksum_change_fail_closed(self) -> None: + incomplete = _disposition() + incomplete.pop("eligible_for_science") + with self.assertRaisesRegex(CASCADE.ReviewContractError, "disposition"): + CASCADE.require_quarantine_disposition(incomplete) + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "source.jsonl" + original = b"synthetic-source" + path.write_bytes(original) + source = _source(1, source_path=path) + source = CASCADE.SourceEnvelope(CASCADE.sha256_bytes(original), source.quarantine_report_sha256, source.disposition, source.records, path) + path.write_bytes(b"changed") + with self.assertRaisesRegex(CASCADE.ReviewContractError, "changed"): + CASCADE.build_chunk_envelopes(source, _configuration()) + + def test_quarantine_report_checksum_mismatch_is_rejected(self) -> None: + import quarantine_dialectic_corpus as quarantine + + turn = { + "index": 1, + "heard": "synthetic fixture only", + "candidates": [ + {"text": "synthetic coherence", "roleID": "coherence-seeking"}, + {"text": "synthetic displacement", "roleID": "displacement-seeking"}, + ], + "tension": 0.4, + "margin": 0.1, + "selectionTemperature": 0.2, + "glossScalar": 0.5, + "outcome": "synthesized:fixture", + } + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source = root / "fixture.jsonl" + report = root / "report.json" + source.write_text(json.dumps(turn) + "\n") + quarantine.inspect_corpus(source, report, root / "events.jsonl") + source.write_text(json.dumps({**turn, "index": 2}) + "\n") + with self.assertRaisesRegex(CASCADE.ReviewContractError, "does not match"): + CASCADE.load_quarantined_source(source, report) + + def test_source_mutation_during_attempt_aborts_the_entire_run(self) -> None: + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "source.jsonl" + original = b"fixture-source-2" + path.write_bytes(original) + initial = _source(2, source_path=path) + source = CASCADE.SourceEnvelope(CASCADE.sha256_bytes(original), initial.quarantine_report_sha256, initial.disposition, initial.records, path) + chunk = CASCADE.build_chunk_envelopes(source, _configuration())[0] + + class MutatingBackend(CASCADE.MockLocalBackend): + def invoke(self, **kwargs): + path.write_bytes(b"source-mutated-during-attempt") + return super().invoke(**kwargs) + + backend = MutatingBackend({"qwen3:0.6b": [json.dumps([_finding(chunk)])]}) + with self.assertRaisesRegex(CASCADE.ReviewContractError, "changed"): + CASCADE.run_review_cascade(source, _configuration(), backend, _attestation()) + + def test_retry_r1_failure_then_valid_r2_without_raw_response_handoff(self) -> None: + source = _source(2) + configuration = _configuration() + chunk = CASCADE.build_chunk_envelopes(source, configuration)[0] + invalid = json.dumps([_finding(chunk, citation_id="line:999")]) + valid = json.dumps([_finding(chunk)]) + backend = CASCADE.MockLocalBackend({"qwen3:0.6b": [invalid], "qwen3:4b": [valid]}) + run = CASCADE.run_review_cascade(source, configuration, backend, _attestation()) + result = run.chunk_results[0] + self.assertEqual(result.review_status, "accepted") + self.assertEqual([attempt.stage for attempt in result.attempts], ["R1", "R2"]) + self.assertEqual(backend.invocations[1]["previous_validation_error_codes"], ["citation_not_allowed"]) + self.assertEqual(backend.invocations[0]["system_prompt_sha256"], backend.invocations[1]["system_prompt_sha256"]) + + def test_bounded_retry_count_and_all_attempts_rejected(self) -> None: + source = _source(2) + configuration = _configuration(maximum_attempts=2) + backend = CASCADE.MockLocalBackend({"qwen3:0.6b": ["not-json"], "qwen3:4b": ["not-json"]}) + run = CASCADE.run_review_cascade(source, configuration, backend, _attestation()) + result = run.chunk_results[0] + self.assertEqual(result.review_status, "rejected") + self.assertEqual(len(result.attempts), 2) + self.assertEqual(len(backend.invocations), 2) + self.assertEqual(result.rejection_error_codes, ("invalid_json",)) + + def test_local_attestations_and_explicit_r3_adjudication_boundary(self) -> None: + source = _source(2) + configuration = _configuration() + chunk = CASCADE.build_chunk_envelopes(source, configuration)[0] + valid = _finding(chunk) + backend = CASCADE.MockLocalBackend({"gemma-family-local": [json.dumps([_finding(chunk, finding_id="critic-1")])]}) + with self.assertRaisesRegex(CASCADE.ReviewContractError, "attest"): + CASCADE.run_review_cascade( + source, + configuration, + CASCADE.MockLocalBackend({"qwen3:0.6b": [json.dumps([valid])]}), + CASCADE.OperatorAttestation(False, True), + ) + with self.assertRaisesRegex(CASCADE.ReviewContractError, "human request"): + CASCADE.run_explicit_critic_adjudication( + chunk=chunk, + accepted_finding=valid, + configuration=configuration, + backend=backend, + attestation=_attestation(), + requested_by_human=False, + ) + receipt = CASCADE.run_explicit_critic_adjudication( + chunk=chunk, + accepted_finding=valid, + configuration=configuration, + backend=backend, + attestation=_attestation(), + requested_by_human=True, + ) + self.assertEqual(receipt["status"], "advisory_only") + self.assertFalse(receipt["r0_override_permitted"]) + + def test_aggregator_is_metadata_only_and_deterministic(self) -> None: + source = _source(2) + configuration = _configuration() + chunk = CASCADE.build_chunk_envelopes(source, configuration)[0] + backend = CASCADE.MockLocalBackend({"qwen3:0.6b": [json.dumps([_finding(chunk)])]}) + run = CASCADE.run_review_cascade(source, configuration, backend, _attestation()) + first = CASCADE.aggregate_review_results(run.chunk_results, source_sha256=source.source_sha256, disposition=source.disposition) + second = CASCADE.aggregate_review_results(run.chunk_results, source_sha256=source.source_sha256, disposition=source.disposition) + self.assertEqual(first, second) + self.assertEqual(first["affected_source_lines"], [41]) + bad = run.chunk_results[0].persisted() + bad["raw_record"] = _record(0) + with self.assertRaisesRegex(CASCADE.ReviewContractError, "raw payload"): + CASCADE.chunks_to_results([bad]) + + def test_metadata_artifacts_never_persist_synthetic_source_content(self) -> None: + source = _source(2) + configuration = _configuration() + chunk = CASCADE.build_chunk_envelopes(source, configuration)[0] + backend = CASCADE.MockLocalBackend({"qwen3:0.6b": [json.dumps([_finding(chunk)])]}) + run = CASCADE.run_review_cascade(source, configuration, backend, _attestation()) + with tempfile.TemporaryDirectory() as directory: + paths = CASCADE.write_metadata_only_artifacts(run, artifact_directory=Path(directory), repository_root=ROOT) + serialized = "".join(path.read_text() for path in paths.values()) + self.assertNotIn("fixture confidential dialogue material", serialized) + self.assertIn("pipeline_only", serialized) + + def test_cross_track_noninterference(self) -> None: + report = NONINTERFERENCE.audit_noninterference( + dialogue_source_sha256="a" * 64, + dialogue_content_hashes=["b" * 64], + review_finding_ids=["finding-1"], + eeg_dataset_artifact={"dataset": "synthetic"}, + eeg_state_artifact={"state": "shadow_only"}, + eeg_model_input_manifest={"input": "canonical"}, + eeg_experiment_configuration={"experiment": "EXP-NC-EEG-ENC-001"}, + local_review_prompt_metadata={"schema": "chunk"}, + eeg_window_hashes=["c" * 64], + ) + self.assertEqual(report["status"], "pass") + with self.assertRaisesRegex(NONINTERFERENCE.NoninterferenceError, "source SHA"): + NONINTERFERENCE.audit_noninterference( + dialogue_source_sha256="a" * 64, + dialogue_content_hashes=[], + review_finding_ids=[], + eeg_dataset_artifact={"source": "a" * 64}, + eeg_state_artifact={}, + eeg_model_input_manifest={}, + eeg_experiment_configuration={}, + local_review_prompt_metadata={}, + eeg_window_hashes=[], + ) + + def test_synthetic_structured_state_replay_remains_shadow_only(self) -> None: + from neuralcompose_eeg.dataset import build_canonical_dataset + from neuralcompose_eeg.structured_state import load_shadow_state_records, write_shadow_state_artifacts + from NeuralComposeEEG.tests.test_pipeline import _manifest, _preprocessing_path + + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + dataset, _ = build_canonical_dataset(_manifest(root), _preprocessing_path()) + probabilities = np.zeros((len(dataset.labels), len(dataset.label_order)), dtype=np.float64) + probabilities[:, 0] = 1.0 + # Publish under a subdirectory: _manifest(root) already wrote the + # *source* manifest to root/manifest.json, and the shadow bridge + # refuses to overwrite an existing artifact that differs. + states = root / "shadow" / "states.jsonl" + manifest = root / "shadow" / "manifest.json" + written = write_shadow_state_artifacts( + dataset, + probabilities, + encoder_provenance={"model_id": "fixture", "model_revision": "v0", "source_kind": "synthetic_contract_fixture"}, + states_output=states, + manifest_output=manifest, + ) + replay = load_shadow_state_records(states, manifest) + self.assertEqual(written["status"], "insufficient_evidence") + self.assertTrue(all(item["shadow_only"] for item in replay)) + self.assertFalse(any(item["live_control"] for item in replay)) + + def test_decision_register_is_governance_only(self) -> None: + entry = { + "schema_version": REGISTER.DECISION_REGISTER_SCHEMA, + "topic": "forward model foundations", + "pass": 2, + "owner": "science", + "registered_question": "Does a specified forward model change a stated decision?", + "decision_it_can_change": "Whether to register a separate forward-model experiment.", + "required_data_gate": "D1", + "falsification_criterion": "The model fails its preregistered error criterion.", + "implementation_status": "deferred", + "runtime_dependency_authorized": False, + } + self.assertEqual(REGISTER.validate_decision_register_entry(entry), entry) + entry["runtime_dependency_authorized"] = True + with self.assertRaisesRegex(REGISTER.DecisionRegisterError, "never authorize"): + REGISTER.validate_decision_register_entry(entry) + + +if __name__ == "__main__": + unittest.main() diff --git a/configs/local-open-weight-review-v0.json b/configs/local-open-weight-review-v0.json new file mode 100644 index 0000000..9c653be --- /dev/null +++ b/configs/local-open-weight-review-v0.json @@ -0,0 +1,48 @@ +{ + "schema_version": "nc-local-review-cascade-config-v0", + "configuration_version": "local-open-weight-review-v0", + "validator": { + "id": "R0", + "type": "deterministic" + }, + "proposer": { + "id": "R1", + "model": "qwen3:0.6b", + "purpose": "bounded_chunk_findings" + }, + "reviewer": { + "id": "R2", + "preferred_model": "qwen3:4b", + "fallback_model": "qwen3:1.7b", + "purpose": "citation_repair_and_aggregation" + }, + "critic": { + "id": "R3", + "model": "gemma-family-local", + "purpose": "explicit_independent_adjudication", + "enabled_by_default": false + }, + "human_gate": { + "id": "R4", + "required_for": [ + "policy_changes", + "experiment_changes", + "scientific_claims", + "promotion_decisions" + ] + }, + "chunking": { + "valid_records_per_chunk": 16, + "overlap_records": 2, + "conversation_memory_between_chunks": false + }, + "generation": { + "temperature": 0.0, + "seed": 42, + "context_limit": 8192, + "max_output_tokens": 2048 + }, + "retry": { + "maximum_attempts": 3 + } +} diff --git a/docs/architecture/dialectic-corpus-quarantine.md b/docs/architecture/dialectic-corpus-quarantine.md index 1a18a41..3c75a57 100644 --- a/docs/architecture/dialectic-corpus-quarantine.md +++ b/docs/architecture/dialectic-corpus-quarantine.md @@ -17,11 +17,16 @@ Every derived artifact carries this exact disposition: "eligible_for_encoder_evaluation": false, "eligible_for_policy_training": false, "eligible_for_policy_evaluation": false, + "eligible_for_science": false, "contains_private_dialogue": true, "cloud_exposure_allowed": false } ``` +The cascade requires every field above. A report produced before +`eligible_for_science: false` existed must be regenerated from the unchanged +local source rather than patched by hand. + Local engineering may inspect the raw file for parser recovery, chronological reconstruction, UI replay, turn reconciliation, dialogue-state debugging, or future event-manifest development. It must not use dialogue content to create @@ -71,3 +76,62 @@ structured-state schema, temporal alignment rules, whole-session splits, privacy rules, and unseen confirmation sessions. See [the EEG methods scope](../scoping/eeg-mathematics-physics-methods-scope.md) and [ADR-005](decision-log/ADR-005-local-interaction-logging.md). + +## Local Semantic Engineering Review + +New review work should use the +[Local Open-Weight Review Cascade](local-open-weight-review-cascade.md), which +keeps source-line identity canonical, never repairs citations, limits retries, +and admits only metadata-only artifacts. It still produces development-only +engineering observations, never scientific evidence. + +An already quarantined source may be inspected by a local Qwen 0.5B-class +model for bounded engineering review. This is not a scientific analysis and +does not change the disposition above. The reviewer accepts only a loopback +Ollama endpoint and a local `qwen2.5:*` model; remote or cloud-backed model +identities are rejected. + +Before a run, the operator must verify that the local runtime will not retain +prompt text outside this quarantine directory, then attest to that state on +the command line: + +```sh +python3 Scripts/review_quarantined_dialectics.py review \ + --input "$HOME/Documents/NeuralCompose/InteractionLogs/dialectic-turns-2026-07-22.jsonl" \ + --parse-report "$HOME/Documents/NeuralCompose/InteractionLogs/local-manifests/dialectic-parse-report-2026-07-22.json" \ + --findings-output "$HOME/Documents/NeuralCompose/InteractionLogs/local-manifests/dialectic-local-review-2026-07-22.jsonl" \ + --run-manifest-output "$HOME/Documents/NeuralCompose/InteractionLogs/local-manifests/dialectic-local-review-run-2026-07-22.json" \ + --prompt-logging-status verified_disabled +``` + +The review is stateless: fixed 16-valid-record chunks, two-record overlap, +temperature `0.0`, seed `42`, no embeddings, and no weight updates. The tool +itself persists neither raw prompts nor raw responses; the external runtime is +used only after the operator verifies its retention behavior. It accepts only +bounded JSON findings that cite source lines in their own chunk, use an allowed +engineering category, and do not contain verbatim private content. + +For legacy logs, a reviewer may confuse the non-unique `index` field with a +physical source line. The tool normalizes that citation only when each cited +legacy index maps to exactly one source line inside the active chunk; ambiguous +or inconsistent citations are rejected rather than guessed. + +An invalid model response creates a metadata-only rejection receipt and stops +the run before later chunks are exposed. It stores the safe contract failure, +never the raw response. A rejected run has no validated engineering findings +and is evidence that the selected local model is unsuitable for this review +configuration, not evidence about the dialogue itself. + +Aggregate the metadata-only review stream without reopening the raw corpus: + +```sh +python3 Scripts/review_quarantined_dialectics.py aggregate \ + --findings-input "$HOME/Documents/NeuralCompose/InteractionLogs/local-manifests/dialectic-local-review-2026-07-22.jsonl" \ + --output "$HOME/Documents/NeuralCompose/InteractionLogs/local-manifests/dialectic-local-review-aggregate-2026-07-22.json" +``` + +The aggregate reports issue categories, counts, affected source lines, +conflicting findings, confidence buckets, and cross-chunk recurrence. Any +future policy or prompt affected by this review remains development-only and +must be evaluated on fresh, protocol-defined sessions that have not been +inspected. diff --git a/docs/architecture/local-open-weight-review-cascade-gate-report.md b/docs/architecture/local-open-weight-review-cascade-gate-report.md new file mode 100644 index 0000000..788369a --- /dev/null +++ b/docs/architecture/local-open-weight-review-cascade-gate-report.md @@ -0,0 +1,129 @@ +# Local Open-Weight Review Cascade Gate Report + +**Date:** 2026-07-23 + +## Repository State + +- branch at start: `feat/local-dialectic-review` +- draft PR base branch: `feat/dialectic-corpus-quarantine` +- base merge point with `main`: `611b07e0b6a1030cc01f27b3cf80dfd24286931f` +- starting commit: `6ae9b96` +- prerequisite history present: quarantine `b8c1336`, capture integrity + `9726626`, encoder contract `a90a56f`, methods scope `0441c1c` and scope + audit `3e1b55b` +- both the quarantine contract and structured EEG state bridge are present on + the branch + +The worktree contained unrelated application, telemetry, scientific-document, +packaging, and soak changes before this work began. They are not staged or +modified by this gate. + +## Cascade Implementation + +`Scripts/local_open_weight_review.py` adds R0 validation, configured R1/R2 +attempts, optional manual-only R3 advisory adjudication, and R4 human-gate +documentation. R1 receives the frozen chunk; R2 receives that same chunk plus +R0 error codes only. R3 can be requested explicitly for an R0-valid finding, +but cannot override R0 or mutate a review result. + +The narrow adapter surface supports loopback Ollama-compatible HTTP and a +deterministic mock. A stable local MLX service interface was not present, so no +MLX adapter or dependency was added. The cascade rejects a remote endpoint, +explicit cloud-routing model identity, missing local classification, unconfined +logging attestation, malformed disposition, source/report checksum mismatch, +and source mutation during review. + +`configs/local-open-weight-review-v0.json` owns the model ladder, chunking, +generation settings, and retry bound. The implementation contains no model-name +selection logic. + +## Privacy + +No private corpus, local recording, or artifact beneath a local documents +directory was read for this work. All new tests use generated records and +synthetic EEG fixtures only. No model was downloaded or run. + +Raw source records and model responses remain in memory for one request. The +only serializable outputs are metadata-only chunk receipts, summary, and run +manifest. Output directories must be outside the repository or Git-ignored. +The bounded private-text detector is documented as a guardrail rather than a +proof of privacy. + +## EEG Noninterference + +`Scripts/local_review_noninterference.py` verifies the required metadata-only +separation between review and EEG tracks. It rejects dialogue source or content +hashes in EEG artifacts or model inputs, review finding identifiers in experiment +configuration, EEG window hashes in review prompt metadata, shared training +buffers, dialogue embeddings, and dialogue-derived weight updates. + +The existing structured-state bridge remains probability-only and was verified +with synthetic deterministic replay. It remains shadow-only and cannot become +physical-data evidence, speech, intervention selection, model update, or live +control through this change. + +## Research Governance + +`Scripts/research_decision_register.py` and +[`research-decision-register.schema.json`](../scoping/research-decision-register.schema.json) +validate governance metadata only. Every entry requires +`runtime_dependency_authorized: false`; it cannot authorize a dependency, +experiment, runtime feature, or promotion. + +The existing four-pass order is unchanged. No mathematics, physics, +optimization, inverse modeling, PCA/ICA, policy/control, ARC, Core ML, encoder +architecture, preprocessing, label, budget, or promotion rule changed. + +## Validation + +```text +PYTHONPATH=NeuralComposeEEG/src python3 -m unittest -v \ + Tests/eval/test_local_open_weight_review.py \ + Tests/eval/test_dialectic_corpus_quarantine.py \ + Tests/eval/test_local_dialectic_review.py \ + NeuralComposeEEG.tests.test_pipeline \ + NeuralComposeEEG.tests.test_structured_state +# 66 passed + +swift test --filter CalibrationRecorderTests +# 7 passed + +python3 -m py_compile Scripts/local_open_weight_review.py \ + Scripts/local_review_noninterference.py \ + Scripts/research_decision_register.py \ + Scripts/quarantine_dialectic_corpus.py +# passed + +python3 -m json.tool configs/local-open-weight-review-v0.json +python3 -m json.tool docs/scoping/research-decision-register.schema.json +# passed + +# deterministic validation of internal links in the two changed Markdown files +# passed + +git diff --check +# passed +``` + +The local macOS toolchain was available, so the relevant Swift test ran. The +EEG suite exercised its existing synthetic contract coverage only; no physical +recording, model download, external worker, or local open-weight model ran. + +## Commit Plan + +One implementation commit is appropriate because the R0 validator, R1/R2/R3 +orchestration, metadata-only writer, configuration, noninterference auditor, +decision-register validator, and synthetic contract test form one atomic +fail-closed boundary. A second documentation/governance commit contains the +decision-register JSON schema, quarantine migration note, architecture guide, +and this report. + +## Final Disposition + +```yaml +status: ready_for_local_open_weight_smoke_test +science_status: pipeline_only +decision: insufficient_evidence +promotion_status: not_eligible +live_control: false +``` diff --git a/docs/architecture/local-open-weight-review-cascade.md b/docs/architecture/local-open-weight-review-cascade.md new file mode 100644 index 0000000..6c5e397 --- /dev/null +++ b/docs/architecture/local-open-weight-review-cascade.md @@ -0,0 +1,122 @@ +# Local Open-Weight Review Cascade + +`Scripts/local_open_weight_review.py` is a local-only engineering-review +subsystem. It turns an already quarantined corpus into metadata-only review +receipts. It is not an experiment runner, a scientific adjudicator, a training +pipeline, or an application-control path. + +It preserves the quarantine disposition in +[Dialectic Corpus Quarantine](dialectic-corpus-quarantine.md): review material +is permanently development-only and ineligible for encoder, policy, or science +evaluation. + +## Cascade + +| Stage | Role | Boundary | +| --- | --- | --- | +| R0 | deterministic validator | accepts only complete, citation-valid, non-leaking structured findings | +| R1 | configured local proposer | makes bounded findings for one stateless chunk | +| R2 | configured local reviewer | retries the original chunk with R0 error codes only | +| R3 | configured local Gemma-family critic | manual, advisory adjudication only; never automatic and never overrides R0 | +| R4 | human and evidence gate | required for policy, experiment, scientific, and promotion decisions | + +The tracked [default configuration](../../configs/local-open-weight-review-v0.json) +names local Qwen models as values, not as code-path assumptions: `qwen3:0.6b` +for R1, then `qwen3:4b` and `qwen3:1.7b` for bounded R2 attempts. An operator +may provide a different versioned configuration. R3 is disabled by default and +can only be invoked through the explicit adjudication API with a human request. + +Only the Ollama-compatible loopback HTTP adapter and a deterministic mock +adapter exist today. An MLX adapter is intentionally deferred until the +repository has a stable, local MLX service interface; no new MLX dependency is +introduced by this subsystem. + +## Fail-Closed Contract + +Before any request, the source must have the complete quarantine disposition, +match the SHA-256 in its quarantine report, and remain unchanged for the full +review. The backend must be explicitly local and use a loopback endpoint or a +Unix socket classification. LAN, VPN, container-host, cloud, and arbitrary +remote URLs are rejected. Loopback is only backend-boundary evidence: the +operator separately attests network isolation and confinement of prompt and +response logging to an ignored local directory. + +Chunks have canonical source-line IDs, use a versioned size and overlap, and +mark overlap records. Each invocation receives the same frozen system prompt. +No model has cross-chunk conversation memory. R2 receives the original chunk +and machine-readable R0 error codes, never an earlier raw response. + +R0 accepts only a complete JSON array of known finding types with finite +confidence, nonempty evidence, exact chunk-local citations, and at least one +new-record citation. It does not repair invented citations or accept a partial +response. A bounded five-token source-overlap detector blocks substantial text +reproduction in synthetic tests. This detector is a useful guardrail, not proof +that an artifact contains no private text. + +The metadata-only summary counts accepted findings, source-line references, +confidence buckets, conflicts, and rejection receipts. It receives no raw +records and cannot infer a new semantic claim, create an EEG hypothesis, or +recommend a model or policy change. + +## Local Use + +The dry run builds and validates in-memory chunk envelopes without starting a +model or writing prompts: + +```sh +python3 Scripts/local_open_weight_review.py \ + --source \ + --quarantine-report \ + --configuration configs/local-open-weight-review-v0.json \ + --dry-run +``` + +An operator may run a bounded local smoke test only after verifying the local +runtime's retention settings and choosing an ignored local artifact directory: + +```sh +python3 Scripts/local_open_weight_review.py \ + --source \ + --quarantine-report \ + --configuration configs/local-open-weight-review-v0.json \ + --backend-url http://127.0.0.1:11434 \ + --artifact-directory \ + --attest-network-isolation \ + --attest-prompt-response-logging-confined +``` + +This command is documentation only. It is never executed by repository tests +or CI. The implementation does not download models, create embeddings, update +weights, or retain raw prompts or responses. The manifest records unavailable +backend facts as `null` or `not_reported` rather than inventing provenance. + +## EEG Noninterference + +`Scripts/local_review_noninterference.py` checks metadata-only artifacts for +the required separation: dialogue source and content hashes are absent from EEG +artifacts and model inputs; review findings are absent from the experiment +configuration; EEG window hashes are absent from review prompt metadata; and no +shared training buffer, dialogue embedding, or dialogue-derived weight update +exists. + +The review subsystem and the structured EEG state bridge may share only JSON, +hash, validation, manifest, and disposition patterns. They share no data, +embeddings, labels, model inputs, training buffers, configuration, or scientific +results. The bridge stays `shadow_only: true`, `live_control: false`, and +`promotion_status: not_eligible`. + +## Research Governance + +The JSON schema at +[Research Decision Register](../scoping/research-decision-register.schema.json) +records whether a future method is deferred, study-only, or eligible for a +separately registered experiment. It always requires +`runtime_dependency_authorized: false`; a register entry cannot add a package, +model, runtime route, app feature, or promotion decision. + +The four-pass order in the +[EEG mathematics, physics, and methods scope](../scoping/eeg-mathematics-physics-methods-scope.md) +is unchanged: Pass 1 encoder evidence, Pass 2 registered forward/inverse +foundations, Pass 3 decision-changing methods, then Pass 4 structured-state +shadow policy. The immediate EEG action remains the first frozen physical Muse +capture, not a review-model, policy, or mathematics expansion. diff --git a/docs/scoping/research-decision-register.schema.json b/docs/scoping/research-decision-register.schema.json new file mode 100644 index 0000000..4b52954 --- /dev/null +++ b/docs/scoping/research-decision-register.schema.json @@ -0,0 +1,30 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "NeuralCompose Research Decision Register", + "type": "object", + "additionalProperties": false, + "required": [ + "schema_version", + "topic", + "pass", + "owner", + "registered_question", + "decision_it_can_change", + "required_data_gate", + "falsification_criterion", + "implementation_status", + "runtime_dependency_authorized" + ], + "properties": { + "schema_version": { "const": "nc-research-decision-register-v0" }, + "topic": { "type": "string", "minLength": 1 }, + "pass": { "type": "integer", "minimum": 1, "maximum": 4 }, + "owner": { "enum": ["science", "engineering", "computation"] }, + "registered_question": { "type": "string", "minLength": 1 }, + "decision_it_can_change": { "type": "string", "minLength": 1 }, + "required_data_gate": { "enum": ["D0", "D1", "D2", "D3", "post_encoder"] }, + "falsification_criterion": { "type": "string", "minLength": 1 }, + "implementation_status": { "enum": ["deferred", "study_only", "eligible_for_experiment"] }, + "runtime_dependency_authorized": { "const": false } + } +}