"""Pure deterministic state transitions for Phase 5.5 convergence.""" from __future__ import annotations from copy import deepcopy import hashlib import json from typing import Any, Mapping from .execution_identity import ExecutionManifest WORKING_SCHEMA_VERSION = "1.0" V2_WORKING_SCHEMA_VERSION = "2.0" EXECUTION_IDENTITY_VERSION = 2 FINAL_SCHEMA_VERSION = "1.3" _AUDIENCES = {"analysis", "lead", "report-writer"} # 발견을 낼 수 있는 audience. 리드는 계약상 Phase 6 자체 리뷰를 수행하므로 # 출처가 될 수 있지만, 교차검증 표는 아니다 — 합의 수는 아래에서 분석 워커만 # 센다. 리드를 분석 워커로 위장 선언하는 것을 계약이 금지하는 이상, 리드의 # 발견을 담을 자리가 없으면 그 리뷰 결과는 어디에도 못 간다. _FINDING_SOURCE_AUDIENCES = {"analysis", "lead"} _AUDIENCE_SOURCE_ROLES = { "analysis": frozenset({"analyser", "designer", "planner", "verifier"}), "lead": frozenset({"leader"}), "report-writer": frozenset({"report-writer"}), } _VERIFICATION_MODES = {"lightweight", "full-reanalysis"} _CLASSIFICATION_COUNT_KEYS = { "full-consensus": "fullConsensus", "partial-consensus": "partialConsensus", "contested": "contested", "worker-unique": "workerUnique", } _VERDICTS = {"agree", "disagree", "supplement", "verification-error"} _INPUT_VERDICTS = _VERDICTS | {"unverifiable"} _DISAGREE_BASES = {"counter-evidence", "burden-not-met"} _DISPATCH_STATUSES = {"completed", "timeout", "error", "not-run"} _CRITIC_CLASSIFICATIONS = set(_CLASSIFICATION_COUNT_KEYS) | {"unverified"} _FINAL_V13_KEYS = { "schemaVersion", "taskKey", "config", "findings", "roundHistory", "round2SkippedReason", "finalState", "totalRounds", "finalClassificationCounts", "criticVerification", "unverifiedGaps", } class ConvergenceContractError(ValueError): """Raised when convergence input violates the persisted-state contract.""" def grouped_input_digest(grouped_input: Mapping[str, Any]) -> str: """Return SHA-256 over canonical compact JSON.""" encoded = json.dumps( grouped_input, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def seed_working_state( grouped_input: Mapping[str, Any], *, execution_manifest: ExecutionManifest | None = None, ) -> dict[str, Any]: """Validate grouped findings, classify Round 0, and create the queue.""" source = _object(grouped_input, "grouped input") identity_version = _grouped_execution_identity_version(source) task_key = _required_string(source, "taskKey", "grouped input") config = _parse_config(source.get("config")) workers = _parse_workers(source.get("workers"), identity_version) _validate_worker_execution_identity( workers, execution_manifest, identity_version, ) analysis_workers = [ worker["workerId"] for worker in workers if worker["audience"] == "analysis" ] finding_sources = [ worker["workerId"] for worker in workers if worker["audience"] in _FINDING_SOURCE_AUDIENCES ] state = { "schemaVersion": source["schemaVersion"], "taskKey": task_key, "groupsDigest": grouped_input_digest(source), "config": config, "workers": workers, "findings": [], "queueFindingIds": [], "roundHistory": [], "stopReason": None, } if identity_version == EXECUTION_IDENTITY_VERSION: state.update({ "executionIdentityVersion": EXECUTION_IDENTITY_VERSION, "runManifestPath": _required_string( source, "runManifestPath", "grouped input", ), }) if not config["enabled"] or len(analysis_workers) < 2: state["stopReason"] = "auto-disabled" return state findings, queue = _parse_groups( source.get("groups"), analysis_workers, finding_sources, adversarial=config["adversarial"], ) state["findings"] = findings state["queueFindingIds"] = queue return state def classification_counts_from_findings( findings: list[Mapping[str, Any]], ) -> dict[str, int]: """Return the four v1.2 finalClassificationCounts keys.""" counts = {value: 0 for value in _CLASSIFICATION_COUNT_KEYS.values()} for finding in findings: key = _CLASSIFICATION_COUNT_KEYS.get(finding.get("classification")) if key: counts[key] += 1 return counts def plan_next_round(state: Mapping[str, Any]) -> dict[str, Any]: """Return the next immutable roster-aware dispatch plan or final gate.""" current = _object(state, "working state") queue, findings, history = _validate_plannable_state(current) config = _object(current.get("config"), "working state.config") effective_max = config.get("effectiveMaxRounds") if not isinstance(effective_max, int): raise ConvergenceContractError( "working state.config.effectiveMaxRounds must be an integer" ) stop_reason = current.get("stopReason") reason = None if stop_reason == "auto-disabled" or not config.get("enabled"): reason = "auto-disabled" elif stop_reason == "all-reverify-non-result": reason = "all-reverify-non-result" elif effective_max == 1 and history: reason = "max-rounds-1" elif not queue: reason = "queue-empty" elif len(history) >= effective_max: reason = "max-rounds-reached" round_number = len(history) + 1 if reason is not None: return { "schemaVersion": WORKING_SCHEMA_VERSION, "action": "finalize", "round": round_number, "inputQueueSize": len(queue), "reason": reason, "dispatches": [], "skippedWorkers": [], } workers = current.get("workers") if not isinstance(workers, list): raise ConvergenceContractError("working state.workers must be an array") dispatches: list[dict[str, Any]] = [] skipped: list[dict[str, str]] = [] for worker in workers: if not isinstance(worker, Mapping) or worker.get("audience") != "analysis": continue worker_id = _required_string(worker, "workerId", "working state worker") worker_participant_ref = _worker_participant_ref(worker) eligible = [ finding_id for finding_id in queue if _worker_is_independent_from_finding( worker_id, worker_participant_ref, findings[finding_id], workers, ) ] if eligible: dispatches.append({ **_worker_execution_refs(worker), "worker": worker_id, "findingIds": eligible, }) else: skipped.append( { **_worker_execution_refs(worker), "worker": worker_id, "reason": "no items to verify", } ) return { "schemaVersion": WORKING_SCHEMA_VERSION, "action": "dispatch", "round": round_number, "inputQueueSize": len(queue), "dispatches": dispatches, "skippedWorkers": skipped, } def _worker_participant_ref(worker: Mapping[str, Any]) -> str | None: value = worker.get("participantRef") return value if isinstance(value, str) and value else None def _worker_is_independent_from_finding( worker_id: str, participant_ref: str | None, finding: Mapping[str, Any], workers: list[Any], ) -> bool: origin_worker = finding.get("originWorker") if participant_ref is None: return origin_worker != worker_id origin = next( ( row for row in workers if isinstance(row, Mapping) and row.get("workerId") == origin_worker ), None, ) origin_participant_ref = ( _worker_participant_ref(origin) if origin is not None else None ) return origin_participant_ref != participant_ref def _worker_execution_refs(worker: Mapping[str, Any]) -> dict[str, str]: participant_ref = _worker_participant_ref(worker) source_ref = worker.get("sourceRoleExecutionRef") if participant_ref is None or not isinstance(source_ref, str) or not source_ref: return {} return { "participantRef": participant_ref, "sourceRoleExecutionRef": source_ref, } def classify_collaborative_round( votes: Mapping[str, Mapping[str, Any]], ) -> str | None: """Classify one collaborative round using non-error votes only.""" parsed = _parsed_votes(votes, adversarial=False) usable = [vote for vote in parsed.values() if vote["verdict"] != "verification-error"] if not usable: return None agreeing = sum(vote["verdict"] in {"agree", "supplement"} for vote in usable) disagreeing = sum(vote["verdict"] == "disagree" for vote in usable) if agreeing == len(usable): return "full-consensus" if disagreeing == len(usable): return "worker-unique" if agreeing > len(usable) / 2: return "partial-consensus" return None def classify_adversarial_round( votes: Mapping[str, Mapping[str, Any]], ) -> str | None: """Classify one adversarial round using documented refutation bases.""" parsed = _parsed_votes(votes, adversarial=True) usable = [vote for vote in parsed.values() if vote["verdict"] != "verification-error"] if not usable: return None disagrees = [vote for vote in usable if vote["verdict"] == "disagree"] if not disagrees: caveats = sum(vote["verdict"] == "supplement" for vote in usable) return ( "partial-consensus" if caveats > len(usable) / 2 else "full-consensus" ) if len(disagrees) == len(usable): return "worker-unique" if any(vote["disagreeBasis"] == "counter-evidence" for vote in disagrees): return None burden_count = sum( vote["disagreeBasis"] == "burden-not-met" for vote in disagrees ) if burden_count > len(usable) / 2: return None return "partial-consensus" def _round_has_counter_evidence( votes: Mapping[str, Mapping[str, Any]], ) -> bool: return any( vote.get("verdict") == "disagree" and vote.get("disagreeBasis") == "counter-evidence" for vote in votes.values() if isinstance(vote, Mapping) ) def _classify_adversarial_history( rounds: list[Mapping[str, Any]], ) -> str | None: """Classify adversarial rounds without erasing earlier counter-evidence.""" counter_evidence_seen = False for row in rounds: votes = row.get("votes") if isinstance(row, Mapping) else None if not isinstance(votes, Mapping): continue counter_evidence_seen = ( counter_evidence_seen or _round_has_counter_evidence(votes) ) classification = classify_adversarial_round(votes) if classification == "worker-unique": return classification if classification is not None and not counter_evidence_seen: return classification return None def apply_round_results( state: Mapping[str, Any], plan: Mapping[str, Any], results: Mapping[str, Any], ) -> dict[str, Any]: """Validate and reduce one complete re-verification round.""" current = _object(state, "working state") expected_plan = plan_next_round(current) supplied_plan = _object(plan, "round plan") if expected_plan.get("action") != "dispatch": raise ConvergenceContractError("working state has no dispatch round to apply") if supplied_plan != expected_plan: raise ConvergenceContractError("round plan does not match current working state") payload = _object(results, "round results") if payload.get("schemaVersion") != WORKING_SCHEMA_VERSION: raise ConvergenceContractError("round results schemaVersion must be 1.0") if payload.get("round") != expected_plan["round"]: raise ConvergenceContractError("round results round does not match plan round") statuses = _parse_dispatch_results(payload.get("dispatches"), expected_plan) raw_votes = _object(payload.get("votesByFinding"), "votesByFinding") planned_by_finding = _planned_workers_by_finding(expected_plan) votes_by_finding = _validate_round_votes( raw_votes, planned_by_finding, statuses, ) updated = deepcopy(dict(current)) queue = list(updated["queueFindingIds"]) findings = {row["findingId"]: row for row in updated["findings"]} adversarial = bool(updated["config"].get("adversarial")) resolved: list[str] = [] for finding_id in queue: finding = findings[finding_id] round_votes = _votes_for_finding( finding_id, planned_by_finding, statuses, votes_by_finding, adversarial, ) finding["rounds"].append( {"round": expected_plan["round"], "votes": round_votes} ) classification = ( _classify_adversarial_history(finding["rounds"]) if adversarial else classify_collaborative_round(round_votes) ) if classification is not None: finding["classification"] = classification resolved.append(finding_id) _recompute_worker_positions(finding, updated["workers"]) updated["queueFindingIds"] = [ finding_id for finding_id in queue if finding_id not in set(resolved) ] dispatch_history = [ { "worker": row["worker"], "status": statuses[row["worker"]]["status"], "durationMs": statuses[row["worker"]]["durationMs"], } for row in expected_plan["dispatches"] ] skipped = [ {"worker": row["worker"], "reason": row["reason"]} for row in expected_plan["skippedWorkers"] ] for row in dispatch_history: if row["status"] != "completed": skipped.append( { "worker": row["worker"], "reason": "dispatch-non-result", "terminalStatus": row["status"], } ) updated["roundHistory"].append( { "round": expected_plan["round"], "inputQueueSize": len(queue), "resolvedCount": len(resolved), "carriedForwardCount": len(updated["queueFindingIds"]), "dispatches": dispatch_history, "skippedWorkers": skipped, } ) if dispatch_history and all( row["status"] != "completed" for row in dispatch_history ): updated["stopReason"] = "all-reverify-non-result" return updated def apply_critic_gap_results( state: Mapping[str, Any], results: Mapping[str, Any], ) -> dict[str, Any]: """Reduce one coverage-critic verification batch into working state.""" current = _object(state, "working state") errors = validate_working_state(current) if errors: raise ConvergenceContractError("invalid working state: " + "; ".join(errors)) current_critic = current.get("config", {}).get("critic") if ( isinstance(current_critic, Mapping) and current_critic.get("mode") == "acceptance-devils-advocate" ): raise ConvergenceContractError( "acceptance critic candidates must use confirm-or-downgrade" ) if current.get("criticVerification") is not None: raise ConvergenceContractError("critic gap batch was already applied") if plan_next_round(current).get("action") != "finalize": raise ConvergenceContractError("main finding queue must be terminal") payload = _object(results, "critic gap results") provider, model, dispatches, gaps = _parse_critic_gap_results(current, payload) updated = deepcopy(dict(current)) ledger_gaps, unverified = _reduce_critic_gaps(updated, provider, gaps) updated["criticVerification"] = { "schemaVersion": WORKING_SCHEMA_VERSION, "provider": provider, "modelExecutionValue": model, "analyserRoster": [ worker["workerId"] for worker in updated["workers"] if worker["audience"] == "analysis" ], "dispatches": dispatches, "gaps": ledger_gaps, } merged = sum(gap["mergedFindingId"] is not None for gap in ledger_gaps) rejected = sum( gap["classification"] in {"contested", "worker-unique"} for gap in ledger_gaps ) updated["config"]["critic"] = { "provider": provider, "modelExecutionValue": model, "gapsProposed": len(ledger_gaps), "gapsMerged": merged, "gapsRejected": rejected, "gapsUnverified": len(unverified), } updated["unverifiedGaps"] = unverified updated_errors = validate_working_state(updated) if updated_errors: raise ConvergenceContractError( "invalid updated working state: " + "; ".join(updated_errors) ) return updated def finalize_working_state(state: Mapping[str, Any]) -> dict[str, Any]: """Return the public schema-v1.3 convergence artifact.""" errors = validate_working_state(state) if errors: raise ConvergenceContractError("invalid working state: " + "; ".join(errors)) current = deepcopy(dict(state)) config = deepcopy(current["config"]) findings = deepcopy(current["findings"]) queue = set(current["queueFindingIds"]) adversarial = bool(config.get("adversarial")) for finding in findings: if finding["findingId"] not in queue: continue finding["classification"] = ( "contested" if adversarial else _final_collaborative_classification(finding) ) stop_reason = current.get("stopReason") history = deepcopy(current["roundHistory"]) if stop_reason == "auto-disabled": config["autoDisabled"] = "fewer-than-two-analysers" skip_reason = "auto-disabled" final_state = "converged" elif config["effectiveMaxRounds"] == 1 and history: skip_reason = "max-rounds-1" if stop_reason == "all-reverify-non-result": final_state = "aborted-non-result" elif not queue: final_state = "converged" else: final_state = "max-rounds-reached" elif stop_reason == "all-reverify-non-result": skip_reason = "all-reverify-non-result" final_state = "aborted-non-result" elif len(history) >= 2: skip_reason = "not-skipped" final_state = "converged" if not queue else "max-rounds-reached" elif not queue: skip_reason = "queue-empty" final_state = "converged" else: skip_reason = "not-skipped" final_state = "max-rounds-reached" final = { "schemaVersion": FINAL_SCHEMA_VERSION, "taskKey": current["taskKey"], "config": config, "findings": findings, "roundHistory": history, "round2SkippedReason": skip_reason, "finalState": final_state, "totalRounds": len(history), "finalClassificationCounts": classification_counts_from_findings(findings), } if "criticVerification" in current: final["criticVerification"] = deepcopy(current["criticVerification"]) final["unverifiedGaps"] = deepcopy(current.get("unverifiedGaps", [])) final_errors = validate_final_state(final) if final_errors: raise ConvergenceContractError( "finalized convergence state is invalid: " + "; ".join(final_errors) ) return final def _working_execution_identity_version( state: Mapping[str, Any], errors: list[str], ) -> int | None: schema_version = state.get("schemaVersion") identity_version = state.get("executionIdentityVersion") run_manifest_path = state.get("runManifestPath") if schema_version == WORKING_SCHEMA_VERSION: if ( "executionIdentityVersion" in state or "runManifestPath" in state ): errors.append( "working schemaVersion 1.0 v1 state forbids v2 identity fields" ) return 1 if schema_version == V2_WORKING_SCHEMA_VERSION: if identity_version != EXECUTION_IDENTITY_VERSION: errors.append( "working execution identity version must be 2 for schemaVersion 2.0" ) if not _nonempty_string(run_manifest_path): errors.append("working runManifestPath must be a non-empty string") return EXECUTION_IDENTITY_VERSION errors.append(f"unsupported working schemaVersion: {schema_version!r}") return None def validate_working_state(state: Mapping[str, Any]) -> list[str]: """Return structural and transition errors for a working-state object.""" errors: list[str] = [] if not isinstance(state, Mapping): return ["working state must be an object"] identity_version = _working_execution_identity_version(state, errors) if not _nonempty_string(state.get("taskKey")): errors.append("taskKey must be a non-empty string") digest = state.get("groupsDigest") if not isinstance(digest, str) or len(digest) != 64: errors.append("groupsDigest must be a SHA-256 hex digest") config = state.get("config") if not isinstance(config, Mapping): errors.append("config must be an object") workers = state.get("workers") if not isinstance(workers, list): errors.append("workers must be an array") workers = [] else: seen_workers: set[str] = set() for index, worker in enumerate(workers): if not isinstance(worker, Mapping): errors.append(f"workers[{index}] must be an object") continue worker_id = worker.get("workerId") audience = worker.get("audience") if not _nonempty_string(worker_id): errors.append(f"workers[{index}].workerId must be a non-empty string") elif worker_id in seen_workers: errors.append(f"duplicate workerId: {worker_id}") else: seen_workers.add(worker_id) if not _nonempty_string(audience) or audience not in _AUDIENCES: errors.append( f"workers[{index}].audience is unsupported: {audience!r}. " "Finding-producing workers (implementation verifiers included) " "use 'analysis'; the okstra lead's own review uses 'lead'; " "only the report author uses 'report-writer'. " f"Allowed: {sorted(_AUDIENCES)}." ) participant_ref = worker.get("participantRef") source_ref = worker.get("sourceRoleExecutionRef") has_execution_fields = ( "participantRef" in worker or "sourceRoleExecutionRef" in worker ) if identity_version == 1 and has_execution_fields: errors.append( f"workers[{index}] v1 row forbids execution identity references" ) elif identity_version == EXECUTION_IDENTITY_VERSION and ( not _nonempty_string(participant_ref) or not _nonempty_string(source_ref) ): errors.append( f"workers[{index}] must carry participantRef and " "sourceRoleExecutionRef together" ) findings = state.get("findings") if not isinstance(findings, list): errors.append("findings must be an array") findings = [] queue = state.get("queueFindingIds") if not isinstance(queue, list): errors.append("queueFindingIds must be an array") queue = [] elif len(queue) != len(set(queue)): errors.append("duplicate queue finding ID") history = state.get("roundHistory") if not isinstance(history, list): errors.append("roundHistory must be an array") history = [] errors.extend(_validate_round_history(history)) by_id: dict[str, Mapping[str, Any]] = {} for finding in findings: if not isinstance(finding, Mapping): errors.append("finding must be an object") continue finding_id = finding.get("findingId") if not _nonempty_string(finding_id): errors.append("findingId must be a non-empty string") continue if finding_id in by_id: errors.append(f"duplicate findingId: {finding_id}") by_id[str(finding_id)] = finding errors.extend(_validate_finding_ledger(finding, len(history), bool(config and config.get("adversarial")))) for finding_id in queue: if finding_id not in by_id: errors.append(f"queue references missing finding: {finding_id}") elif by_id[finding_id].get("classification") is not None: errors.append(f"queued finding is already classified: {finding_id}") for finding_id, finding in by_id.items(): if finding.get("classification") is None and finding_id not in queue: errors.append(f"unclassified finding is absent from queue: {finding_id}") if history and history[-1].get("carriedForwardCount") != len(queue): errors.append("last round carriedForwardCount disagrees with queue length") if state.get("stopReason") not in {None, "auto-disabled", "all-reverify-non-result"}: errors.append("stopReason is unsupported") errors.extend(_validate_critic_contract(state, findings)) return errors def validate_final_state(state: Mapping[str, Any]) -> list[str]: """Validate historical v1.0-v1.2 and strict current v1.3 artifacts.""" errors: list[str] = [] if not isinstance(state, Mapping): return ["final state must be an object"] version = state.get("schemaVersion") if version not in {"1.0", "1.1", "1.2", FINAL_SCHEMA_VERSION}: errors.append("unsupported final schemaVersion") config = state.get("config") if not isinstance(config, Mapping): errors.append("config must be an object") config = {} effective = config.get("effectiveMaxRounds") if not isinstance(effective, int) or isinstance(effective, bool) or not 1 <= effective <= 3: errors.append("config.effectiveMaxRounds must be an integer from 1 to 3") findings = state.get("findings") if not isinstance(findings, list): errors.append("findings must be an array") findings = [] history = state.get("roundHistory") if not isinstance(history, list): errors.append("roundHistory must be an array") history = [] errors.extend(_validate_round_history(history)) adversarial = bool(config.get("adversarial")) seen_ids: set[str] = set() for finding in findings: if not isinstance(finding, Mapping): errors.append("finding must be an object") continue finding_id = finding.get("findingId") if not _nonempty_string(finding_id): errors.append("findingId must be a non-empty string") elif finding_id in seen_ids: errors.append(f"duplicate findingId: {finding_id}") else: seen_ids.add(finding_id) classification = finding.get("classification") if classification not in _CLASSIFICATION_COUNT_KEYS: errors.append(f"finding {finding_id} has invalid classification") errors.extend(_validate_finding_ledger(finding, len(history), adversarial)) rounds = finding.get("rounds") if isinstance(finding.get("rounds"), list) else [] if classification == "contested" and rounds: if rounds[-1].get("round") != len(history): errors.append(f"finding {finding_id} contested before the final round") if classification != "contested": errors.extend(_validate_no_reappearance_after_resolution(finding, adversarial)) expected = _expected_final_classification(finding, adversarial) if expected is not None and classification != expected: errors.append( f"finding {finding_id} classification {classification} " f"does not match replayed classification {expected}" ) errors.extend(_validate_round_ledger_counts(findings, history, adversarial)) total_rounds = state.get("totalRounds") if total_rounds != len(history): errors.append("totalRounds does not match roundHistory length") declared = state.get("finalClassificationCounts") if not isinstance(declared, Mapping): errors.append("finalClassificationCounts must be an object") else: actual = classification_counts_from_findings(findings) if dict(declared) != actual: errors.append( f"finalClassificationCounts {dict(declared)} does not match findings {actual}" ) reason = state.get("round2SkippedReason") errors.extend( _validate_final_reason( reason, state.get("finalState"), history, effective, config, ) ) if version == FINAL_SCHEMA_VERSION: unexpected = set(state) - _FINAL_V13_KEYS if unexpected: errors.append( "final v1.3 has unsupported top-level fields: " + ", ".join(sorted(unexpected)) ) errors.extend(_validate_critic_contract(state, findings)) return errors def _parse_critic_gap_results( state: Mapping[str, Any], payload: Mapping[str, Any], ) -> tuple[str, str, list[dict[str, Any]], list[dict[str, Any]]]: if payload.get("schemaVersion") != WORKING_SCHEMA_VERSION: raise ConvergenceContractError("critic gap results schemaVersion must be 1.0") if payload.get("taskKey") != state.get("taskKey"): raise ConvergenceContractError("critic gap results taskKey does not match") if payload.get("mode") != "coverage": raise ConvergenceContractError( "apply-critic-gaps accepts coverage mode only; acceptance candidates " "use confirm-or-downgrade" ) provider = _required_string(payload, "provider", "critic gap results") model = _required_string(payload, "modelExecutionValue", "critic gap results") roster = { worker["workerId"] for worker in state.get("workers", []) if isinstance(worker, Mapping) and worker.get("audience") == "analysis" } dispatches, completed = _parse_critic_dispatches( payload.get("dispatches"), roster, provider ) gaps = _parse_critic_gaps(payload.get("gaps"), roster, provider, completed) return provider, model, dispatches, gaps def _parse_critic_dispatches( value: Any, roster: set[str], provider: str, ) -> tuple[list[dict[str, Any]], set[str]]: if not isinstance(value, list): raise ConvergenceContractError("critic gap results dispatches must be an array") dispatches: list[dict[str, Any]] = [] seen: set[str] = set() for index, raw in enumerate(value): row = _object(raw, f"critic dispatches[{index}]") worker = _required_string(row, "worker", f"critic dispatches[{index}]") if worker in seen: raise ConvergenceContractError(f"duplicate critic voter dispatch: {worker}") if worker not in roster: raise ConvergenceContractError( f"critic voter must be a non-critic analyser: {worker}" ) status = _required_string(row, "status", f"critic dispatch {worker}") duration = row.get("durationMs") if status not in _DISPATCH_STATUSES: raise ConvergenceContractError(f"critic dispatch {worker} has invalid status") if not isinstance(duration, int) or isinstance(duration, bool) or duration < 0: raise ConvergenceContractError( f"critic dispatch {worker}.durationMs must be a non-negative integer" ) seen.add(worker) dispatches.append({"worker": worker, "status": status, "durationMs": duration}) expected = set(roster) if seen != expected: missing = sorted(expected - seen) raise ConvergenceContractError( "critic dispatches must account for every non-critic analyser exactly " f"once; missing: {', '.join(missing) or 'none'}" ) completed = {row["worker"] for row in dispatches if row["status"] == "completed"} return dispatches, completed def _parse_critic_gaps( value: Any, roster: set[str], provider: str, completed: set[str], ) -> list[dict[str, Any]]: if not isinstance(value, list): raise ConvergenceContractError("critic gap results gaps must be an array") gaps: list[dict[str, Any]] = [] seen: set[str] = set() for index, raw in enumerate(value): gap = _object(raw, f"critic gaps[{index}]") gap_id = _required_string(gap, "gapId", f"critic gaps[{index}]") if gap_id in seen: raise ConvergenceContractError(f"duplicate critic gapId: {gap_id}") seen.add(gap_id) parsed = _parse_critic_gap(gap, index, roster, provider, completed) parsed["gapId"] = gap_id gaps.append(parsed) return gaps def _parse_critic_gap( gap: Mapping[str, Any], index: int, roster: set[str], provider: str, completed: set[str], ) -> dict[str, Any]: label = f"critic gaps[{index}]" votes_value = _object(gap.get("votes"), f"{label}.votes") votes: dict[str, dict[str, Any]] = {} for worker, raw_vote in votes_value.items(): if worker not in roster: raise ConvergenceContractError( f"critic voter must be a non-critic analyser: {worker}" ) if worker not in completed: raise ConvergenceContractError( f"critic voter has no completed dispatch: {worker}" ) votes[worker] = _parse_vote( raw_vote, adversarial=True, allow_unverifiable=True, ) parsed = { "summary": _required_string(gap, "summary", label), "category": _required_string(gap, "category", label), "ticketIds": _string_array_allow_empty( gap.get("ticketIds"), f"{label}.ticketIds" ), "originEvidence": _required_string(gap, "originEvidence", label), "votes": votes, } if "evidenceArtifacts" in gap: parsed["evidenceArtifacts"] = _parse_evidence_artifacts( gap.get("evidenceArtifacts"), label ) return parsed def _reduce_critic_gaps( state: dict[str, Any], provider: str, gaps: list[dict[str, Any]], ) -> tuple[list[dict[str, Any]], list[dict[str, str]]]: ledger: list[dict[str, Any]] = [] unverified: list[dict[str, str]] = [] next_number = _next_finding_number(state["findings"]) for gap in gaps: classification = _critic_gap_classification(gap["votes"]) merged_id = None if classification in {"full-consensus", "partial-consensus"}: merged_id = f"F-{next_number:03d}" next_number += 1 state["findings"].append( _critic_finding(gap, provider, merged_id, classification, state["workers"]) ) elif classification == "unverified": unverified.append( { "gapId": gap["gapId"], "summary": gap["summary"], "reason": "no usable analyser vote", } ) ledger.append( { "gapId": gap["gapId"], "summary": gap["summary"], "category": gap["category"], "ticketIds": deepcopy(gap["ticketIds"]), "originEvidence": gap["originEvidence"], "classification": classification, "mergedFindingId": merged_id, "votes": deepcopy(gap["votes"]), **( {"evidenceArtifacts": deepcopy(gap["evidenceArtifacts"])} if "evidenceArtifacts" in gap else {} ), } ) return ledger, unverified def _critic_gap_classification(votes: Mapping[str, Mapping[str, Any]]) -> str: usable = [vote for vote in votes.values() if vote["verdict"] != "verification-error"] if not usable: return "unverified" return classify_adversarial_round(votes) or "contested" def _next_finding_number(findings: list[Mapping[str, Any]]) -> int: numbers = [] for finding in findings: finding_id = finding.get("findingId") if isinstance(finding_id, str) and finding_id.startswith("F-"): suffix = finding_id[2:] if suffix.isdigit(): numbers.append(int(suffix)) return max(numbers, default=0) + 1 def _critic_finding( gap: Mapping[str, Any], provider: str, finding_id: str, classification: str, workers: list[Mapping[str, Any]], ) -> dict[str, Any]: agreeing = { worker for worker, vote in gap["votes"].items() if vote["verdict"] in {"agree", "supplement"} } dissenting = { worker for worker, vote in gap["votes"].items() if vote["verdict"] == "disagree" } roster = [row["workerId"] for row in workers if row.get("audience") == "analysis"] finding = { "findingId": finding_id, "summary": gap["summary"], "category": gap["category"], "ticketIds": deepcopy(gap["ticketIds"]), "originWorker": f"{provider}-critic", "originEvidence": gap["originEvidence"], "discoveredBy": {}, "sourceItems": [], "source": "critic", "classification": classification, "rounds": [], "consensusWorkers": [worker for worker in roster if worker in agreeing], "dissentingWorkers": [worker for worker in roster if worker in dissenting], } if "evidenceArtifacts" in gap: finding["evidenceArtifacts"] = deepcopy(gap["evidenceArtifacts"]) return finding def _validate_critic_contract( state: Mapping[str, Any], findings: list[Any], ) -> list[str]: config = state.get("config") critic_config = config.get("critic") if isinstance(config, Mapping) else None ledger = state.get("criticVerification") unverified = state.get("unverifiedGaps") critic_findings: dict[str, Mapping[str, Any]] = {} errors: list[str] = [] for finding in findings: if not isinstance(finding, Mapping) or finding.get("source") != "critic": continue finding_id = finding.get("findingId") if not _nonempty_string(finding_id): errors.append("findingId must be a non-empty string") continue critic_findings[finding_id] = finding if ( isinstance(critic_config, Mapping) and critic_config.get("mode") == "acceptance-devils-advocate" ): errors.extend( _validate_acceptance_critic_summary( critic_config, ledger, unverified, critic_findings ) ) return errors if ledger is None and critic_config is None and unverified is None: if critic_findings: errors.append("critic-origin finding requires criticVerification ledger") return errors if not isinstance(ledger, Mapping): errors.append("criticVerification must be an object") return errors if not isinstance(critic_config, Mapping): errors.append("config.critic must be an object") critic_config = {} if not isinstance(unverified, list): errors.append("unverifiedGaps must be an array") unverified = [] errors.extend(_validate_critic_ledger_shape(ledger)) errors.extend(_validate_critic_roster(state, ledger)) gaps = ledger.get("gaps") if isinstance(ledger.get("gaps"), list) else [] errors.extend( _validate_critic_gap_links( gaps, critic_findings, unverified, critic_config, ledger, ) ) if critic_config.get("provider") != ledger.get("provider"): errors.append("config.critic.provider does not match criticVerification") if critic_config.get("modelExecutionValue") != ledger.get("modelExecutionValue"): errors.append( "config.critic.modelExecutionValue does not match criticVerification" ) return errors def _validate_acceptance_critic_summary( config: Mapping[str, Any], ledger: Any, unverified: Any, critic_findings: Mapping[Any, Mapping[str, Any]], ) -> list[str]: allowed = { "mode", "provider", "modelExecutionValue", "candidatesProposed", "confirmedBlockers", "downgradedToResidual", } errors = [] if set(config) == allowed else [ "acceptance critic config has unsupported fields" ] counts = ( config.get("candidatesProposed"), config.get("confirmedBlockers"), config.get("downgradedToResidual"), ) valid_counts = all( isinstance(value, int) and not isinstance(value, bool) and value >= 0 for value in counts ) if not valid_counts: errors.append("acceptance critic counts must be non-negative integers") elif counts[0] != counts[1] + counts[2]: errors.append( "acceptance critic candidatesProposed must equal confirmedBlockers + " "downgradedToResidual" ) if ledger is not None or unverified is not None or critic_findings: errors.append("acceptance critic must not use coverage critic merge semantics") return errors def _validate_critic_ledger_shape(ledger: Mapping[str, Any]) -> list[str]: errors: list[str] = [] allowed = { "schemaVersion", "provider", "modelExecutionValue", "analyserRoster", "dispatches", "gaps", } if set(ledger) != allowed: errors.append("criticVerification has unsupported fields") if ledger.get("schemaVersion") != WORKING_SCHEMA_VERSION: errors.append("criticVerification.schemaVersion must be 1.0") for key in ("provider", "modelExecutionValue"): if not _nonempty_string(ledger.get(key)): errors.append(f"criticVerification.{key} must be a non-empty string") roster = ledger.get("analyserRoster") if not isinstance(roster, list): errors.append("criticVerification.analyserRoster must be a unique string array") elif not all(_nonempty_string(worker) for worker in roster): errors.append("criticVerification.analyserRoster must be a unique string array") elif len(roster) != len(set(roster)): errors.append("criticVerification.analyserRoster must be a unique string array") dispatches = ledger.get("dispatches") if not isinstance(dispatches, list): errors.append("criticVerification.dispatches must be an array") else: errors.extend(_validate_critic_ledger_dispatches(dispatches)) gaps = ledger.get("gaps") if not isinstance(gaps, list): errors.append("criticVerification.gaps must be an array") return errors def _validate_critic_ledger_dispatches(dispatches: list[Any]) -> list[str]: errors: list[str] = [] seen: set[Any] = set() for index, row in enumerate(dispatches): if not isinstance(row, Mapping) or set(row) != {"worker", "status", "durationMs"}: errors.append(f"criticVerification.dispatches[{index}] is invalid") continue worker = row.get("worker") if not _nonempty_string(worker): errors.append("criticVerification has invalid or duplicate dispatch worker") elif worker in seen: errors.append("criticVerification has invalid or duplicate dispatch worker") else: seen.add(worker) status = row.get("status") if not _nonempty_string(status) or status not in _DISPATCH_STATUSES: errors.append("criticVerification has invalid dispatch status") duration = row.get("durationMs") if not isinstance(duration, int) or isinstance(duration, bool) or duration < 0: errors.append("criticVerification has invalid dispatch durationMs") return errors def _validate_critic_gap_links( gaps: list[Any], critic_findings: Mapping[Any, Mapping[str, Any]], unverified: list[Any], config: Mapping[str, Any], ledger: Mapping[str, Any], ) -> list[str]: errors: list[str] = [] seen_gaps: set[str] = set() merged_ids: set[str] = set() unverified_ids: set[str] = set() classifications: list[Any] = [] for finding_id, finding in critic_findings.items(): if finding.get("rounds") != []: errors.append( f"critic-origin finding rounds must be empty: {finding_id}" ) for index, gap in enumerate(gaps): gap_errors, gap_id, merged_id, classification = _validate_critic_gap_row( gap, index ) errors.extend(gap_errors) if _nonempty_string(gap_id): if gap_id in seen_gaps: errors.append(f"duplicate critic gapId: {gap_id}") seen_gaps.add(gap_id) classifications.append(classification) if isinstance(merged_id, str): if merged_id in merged_ids: errors.append(f"duplicate mergedFindingId: {merged_id}") merged_ids.add(merged_id) finding = critic_findings.get(merged_id) if finding is None: errors.append(f"critic mergedFindingId {merged_id} is not a critic-origin finding") elif finding.get("classification") != classification: errors.append(f"critic mergedFindingId {merged_id} classification mismatch") elif not gap_errors and finding != _replay_critic_finding( gap, merged_id, ledger, ): errors.append( f"critic mergedFindingId {merged_id} content does not match " "criticVerification ledger" ) if classification == "unverified" and isinstance(gap_id, str): unverified_ids.add(gap_id) if set(critic_findings) != merged_ids: errors.append( "critic-origin findings do not match criticVerification " "mergedFindingId links" ) errors.extend(_validate_unverified_gap_rows(unverified, unverified_ids)) errors.extend(_validate_critic_counts(config, classifications)) return errors def _validate_critic_gap_row( gap: Any, index: int, ) -> tuple[list[str], Any, Any, Any]: if not isinstance(gap, Mapping): return ([f"criticVerification.gaps[{index}] must be an object"], None, None, None) errors: list[str] = [] required = { "gapId", "summary", "category", "ticketIds", "originEvidence", "classification", "mergedFindingId", "votes", } allowed = required | {"evidenceArtifacts"} if set(gap) - allowed: errors.append(f"criticVerification.gaps[{index}] has unsupported fields") if required - set(gap): errors.append(f"criticVerification.gaps[{index}] has missing required fields") gap_id = gap.get("gapId") classification = gap.get("classification") merged_id = gap.get("mergedFindingId") if not _nonempty_string(gap_id): errors.append(f"criticVerification.gaps[{index}].gapId is invalid") classification_valid = ( _nonempty_string(classification) and classification in _CRITIC_CLASSIFICATIONS ) if not classification_valid: errors.append(f"criticVerification gap {gap_id} has invalid classification") if classification_valid and classification in {"full-consensus", "partial-consensus"}: if not _nonempty_string(merged_id): errors.append(f"criticVerification gap {gap_id} requires mergedFindingId") elif classification_valid and merged_id is not None: errors.append(f"criticVerification gap {gap_id} must not be merged") errors.extend(_validate_critic_gap_content(gap, gap_id)) errors.extend(_validate_critic_gap_votes(gap_id, gap.get("votes"), classification)) return errors, gap_id, merged_id, classification def _validate_critic_gap_content( gap: Mapping[str, Any], gap_id: Any, ) -> list[str]: errors: list[str] = [] for key in ("summary", "category", "originEvidence"): if not _nonempty_string(gap.get(key)): errors.append(f"criticVerification gap {gap_id} {key} is invalid") ticket_ids = gap.get("ticketIds") if not isinstance(ticket_ids, list) or not all( _nonempty_string(ticket_id) for ticket_id in ticket_ids ): errors.append(f"criticVerification gap {gap_id} ticketIds is invalid") artifacts = gap.get("evidenceArtifacts") if "evidenceArtifacts" not in gap: return errors if isinstance(artifacts, list): expected = {"path", "sha256", "command", "environment"} for index, artifact in enumerate(artifacts): if isinstance(artifact, Mapping) and set(artifact) != expected: errors.append( f"criticVerification gap {gap_id}.evidenceArtifacts[{index}] " "has unsupported fields" ) try: _parse_evidence_artifacts( artifacts, f"criticVerification gap {gap_id}", ) except ConvergenceContractError as exc: errors.append(str(exc)) return errors def _replay_critic_finding( gap: Mapping[str, Any], finding_id: str, ledger: Mapping[str, Any], ) -> dict[str, Any]: roster = ledger.get("analyserRoster") workers = [ {"workerId": worker, "audience": "analysis"} for worker in roster if _nonempty_string(worker) ] if isinstance(roster, list) else [] return _critic_finding( gap, str(ledger.get("provider")), finding_id, gap["classification"], workers, ) def _validate_critic_gap_votes( gap_id: Any, votes: Any, classification: Any, ) -> list[str]: if not isinstance(votes, Mapping): return [f"criticVerification gap {gap_id} votes must be an object"] errors: list[str] = [] parsed: dict[str, dict[str, Any]] = {} for worker, vote in votes.items(): basis = vote.get("disagreeBasis") if isinstance(vote, Mapping) else None if basis is not None and not _nonempty_string(basis): errors.append( f"criticVerification gap {gap_id} vote {worker}: " "disagreeBasis must be a non-empty string" ) continue try: parsed[worker] = _parse_vote(vote, adversarial=True) except ConvergenceContractError as exc: errors.append(f"criticVerification gap {gap_id} vote {worker}: {exc}") if ( not errors and _nonempty_string(classification) and classification in _CRITIC_CLASSIFICATIONS ): expected = _critic_gap_classification(parsed) if classification != expected: errors.append( f"criticVerification gap {gap_id} classification {classification} " f"does not match replayed classification {expected}" ) return errors def _validate_unverified_gap_rows( rows: list[Any], expected_ids: set[str], ) -> list[str]: errors: list[str] = [] actual_ids: list[str] = [] for index, row in enumerate(rows): if not isinstance(row, Mapping) or set(row) != {"gapId", "summary", "reason"}: errors.append(f"unverifiedGaps[{index}] is invalid") continue for key in ("gapId", "summary", "reason"): if not _nonempty_string(row.get(key)): errors.append(f"unverifiedGaps[{index}].{key} is invalid") if isinstance(row.get("gapId"), str): actual_ids.append(row["gapId"]) if len(actual_ids) != len(set(actual_ids)) or set(actual_ids) != expected_ids: errors.append("unverifiedGaps do not match unverified criticVerification gaps") return errors def _validate_critic_counts( config: Mapping[str, Any], classifications: list[Any], ) -> list[str]: allowed = { "provider", "modelExecutionValue", "gapsProposed", "gapsMerged", "gapsRejected", "gapsUnverified", } errors: list[str] = [] if set(config) != allowed: errors.append("config.critic has unsupported fields") expected = { "gapsProposed": len(classifications), "gapsMerged": sum( _nonempty_string(value) and value in {"full-consensus", "partial-consensus"} for value in classifications ), "gapsRejected": sum( _nonempty_string(value) and value in {"contested", "worker-unique"} for value in classifications ), "gapsUnverified": sum(value == "unverified" for value in classifications), } for key, value in expected.items(): declared = config.get(key) if not isinstance(declared, int) or isinstance(declared, bool) or declared < 0: errors.append(f"config.critic.{key} must be a non-negative integer") elif declared != value: errors.append(f"config.critic.{key} {declared} does not match ledger {value}") proposed = config.get("gapsProposed") merged = config.get("gapsMerged") rejected = config.get("gapsRejected") unverified = config.get("gapsUnverified") counts = (proposed, merged, rejected, unverified) if all( isinstance(value, int) and not isinstance(value, bool) for value in counts ): if proposed != merged + rejected + unverified: errors.append( "config.critic.gapsProposed must equal gapsMerged + gapsRejected + gapsUnverified" ) return errors def _validate_critic_roster( state: Mapping[str, Any], ledger: Mapping[str, Any], ) -> list[str]: declared = ledger.get("analyserRoster") roster = { worker for worker in declared if _nonempty_string(worker) } if isinstance(declared, list) else set() workers = state.get("workers") provider = ledger.get("provider") dispatches = ledger.get("dispatches") errors: list[str] = [] if isinstance(workers, list): actual = [ worker.get("workerId") for worker in workers if isinstance(worker, Mapping) and worker.get("audience") == "analysis" ] if declared != actual: errors.append("criticVerification.analyserRoster does not match workers") for row in dispatches if isinstance(dispatches, list) else []: worker = row.get("worker") if isinstance(row, Mapping) else None if not _nonempty_string(worker) or worker not in roster: errors.append(f"critic voter must be a non-critic analyser: {worker}") expected_dispatches = set(roster) actual_dispatches = { row.get("worker") for row in dispatches if isinstance(row, Mapping) and _nonempty_string(row.get("worker")) } if isinstance(dispatches, list) else set() if actual_dispatches != expected_dispatches: errors.append( "criticVerification.dispatches must account for every non-critic " "analyser exactly once" ) completed = { row.get("worker") for row in dispatches if ( isinstance(row, Mapping) and _nonempty_string(row.get("worker")) and row.get("status") == "completed" ) } if isinstance(dispatches, list) else set() gaps = ledger.get("gaps") for gap in gaps if isinstance(gaps, list) else []: votes = gap.get("votes") if isinstance(gap, Mapping) else None for voter in votes if isinstance(votes, Mapping) else {}: if voter not in roster: errors.append(f"critic voter must be a non-critic analyser: {voter}") elif voter not in completed: errors.append(f"critic voter has no completed dispatch: {voter}") return errors def _final_collaborative_classification(finding: Mapping[str, Any]) -> str: usable: list[Mapping[str, Any]] = [] for round_row in finding.get("rounds", []): for vote in round_row.get("votes", {}).values(): if vote.get("verdict") != "verification-error": usable.append(vote) agreeing = sum(vote.get("verdict") in {"agree", "supplement"} for vote in usable) return "partial-consensus" if usable and agreeing > len(usable) / 2 else "contested" def _expected_final_classification( finding: Mapping[str, Any], adversarial: bool, ) -> str | None: rounds = finding.get("rounds") if not isinstance(rounds, list) or not rounds: return None if adversarial: try: return _classify_adversarial_history(rounds) or "contested" except ConvergenceContractError: return "contested" for row in rounds: if not isinstance(row, Mapping) or not isinstance(row.get("votes"), Mapping): continue try: resolved = classify_collaborative_round(row["votes"]) except ConvergenceContractError: continue if resolved is not None: return resolved return _final_collaborative_classification(finding) def _validate_round_ledger_counts( findings: list[Any], history: list[Any], adversarial: bool, ) -> list[str]: errors: list[str] = [] for round_number, history_row in enumerate(history, start=1): if not isinstance(history_row, Mapping): continue ledgers = _round_ledgers(findings, round_number) resolved = _resolved_ledger_count(findings, round_number, adversarial) expected = (len(ledgers), resolved, len(ledgers) - resolved) actual = ( history_row.get("inputQueueSize"), history_row.get("resolvedCount"), history_row.get("carriedForwardCount"), ) if actual != expected: errors.append( f"round {round_number} counters {actual} do not match finding ledgers {expected}" ) return errors def _round_ledgers( findings: list[Any], round_number: int, ) -> list[Mapping[str, Any]]: ledgers: list[Mapping[str, Any]] = [] for finding in findings: if not isinstance(finding, Mapping): continue rounds = finding.get("rounds") if not isinstance(rounds, list): continue ledgers.extend( row for row in rounds if isinstance(row, Mapping) and row.get("round") == round_number ) return ledgers def _resolved_ledger_count( findings: list[Any], round_number: int, adversarial: bool, ) -> int: resolved = 0 for finding in findings: rounds = finding.get("rounds") if isinstance(finding, Mapping) else None if not isinstance(rounds, list): continue current_rounds = [ row for row in rounds if isinstance(row, Mapping) and isinstance(row.get("round"), int) and row["round"] <= round_number ] current = next( (row for row in current_rounds if row.get("round") == round_number), None, ) votes = current.get("votes") if isinstance(current, Mapping) else None if not isinstance(votes, Mapping): continue try: classification = ( _classify_adversarial_history(current_rounds) if adversarial else classify_collaborative_round(votes) ) except ConvergenceContractError: continue resolved += classification is not None return resolved def _validate_round_history(history: list[Any]) -> list[str]: errors: list[str] = [] previous_carry = None for index, row in enumerate(history, start=1): if not isinstance(row, Mapping): errors.append(f"roundHistory[{index - 1}] must be an object") continue if row.get("round") != index: errors.append("roundHistory has duplicate or gapped round numbers") input_size = row.get("inputQueueSize") resolved = row.get("resolvedCount") carried = row.get("carriedForwardCount") if not all(isinstance(value, int) for value in (input_size, resolved, carried)): errors.append(f"round {index} counters must be integers") elif input_size != resolved + carried: errors.append(f"round arithmetic mismatch at round {index}") if previous_carry is not None and input_size != previous_carry: errors.append(f"next-round input mismatch at round {index}") previous_carry = carried dispatches = row.get("dispatches") if not isinstance(dispatches, list): errors.append(f"round {index} dispatches must be an array") else: for dispatch in dispatches: if not isinstance(dispatch, Mapping) or dispatch.get("status") not in _DISPATCH_STATUSES: errors.append(f"round {index} has invalid dispatch status") return errors def _validate_finding_ledger( finding: Mapping[str, Any], total_rounds: int, adversarial: bool, ) -> list[str]: errors: list[str] = [] finding_id = finding.get("findingId") origin = finding.get("originWorker") if "evidenceArtifacts" in finding: try: _parse_evidence_artifacts( finding.get("evidenceArtifacts"), f"finding {finding_id}" ) except ConvergenceContractError as exc: errors.append(str(exc)) rounds = finding.get("rounds") if not isinstance(rounds, list): return [f"finding {finding_id} rounds must be an array"] expected_round = 1 for row in rounds: if not isinstance(row, Mapping): errors.append(f"finding {finding_id} round ledger must be an object") continue round_number = row.get("round") if round_number != expected_round: errors.append(f"finding {finding_id} has duplicate or gapped round ledger") expected_round += 1 if not isinstance(round_number, int) or round_number > total_rounds: errors.append(f"finding {finding_id} references a non-executed round") votes = row.get("votes") if not isinstance(votes, Mapping): errors.append(f"finding {finding_id} votes must be an object") continue if origin in votes: errors.append(f"finding {finding_id} contains an origin worker vote") for worker, vote in votes.items(): try: _parse_vote(vote, adversarial=adversarial) except ConvergenceContractError as exc: errors.append(f"finding {finding_id} vote {worker}: {exc}") return errors def _validate_no_reappearance_after_resolution( finding: Mapping[str, Any], adversarial: bool, ) -> list[str]: rounds = finding.get("rounds") if not isinstance(rounds, list): return [] for index, row in enumerate(rounds[:-1]): votes = row.get("votes") if isinstance(row, Mapping) else None if not isinstance(votes, Mapping): continue try: classification = ( _classify_adversarial_history(rounds[: index + 1]) if adversarial else classify_collaborative_round(votes) ) except ConvergenceContractError: continue if classification is not None: return [ f"finding {finding.get('findingId')} reappears after resolution in round {index + 1}" ] return [] def _validate_final_reason( reason: Any, final_state: Any, history: list[Any], effective_max: Any, config: Mapping[str, Any], ) -> list[str]: errors: list[str] = [] allowed = { "auto-disabled", "queue-empty", "max-rounds-1", "all-reverify-non-result", "not-skipped", } if reason not in allowed: return ["round2SkippedReason is unsupported"] last_dispatches = history[-1].get("dispatches", []) if history else [] all_non_result = bool(last_dispatches) and all( isinstance(row, Mapping) and row.get("status") != "completed" for row in last_dispatches ) auto_disabled = ( config.get("enabled") is False or config.get("autoDisabled") == "fewer-than-two-analysers" ) expected_reason = None if auto_disabled: expected_reason = "auto-disabled" elif effective_max == 1 and history: expected_reason = "max-rounds-1" elif all_non_result: expected_reason = "all-reverify-non-result" elif len(history) >= 2: expected_reason = "not-skipped" elif not history or history[-1].get("carriedForwardCount") == 0: expected_reason = "queue-empty" if expected_reason is not None and reason != expected_reason: errors.append( f"round2SkippedReason {reason} does not match replayed reason {expected_reason}" ) queue_drained = bool(history) and history[-1].get("carriedForwardCount") == 0 if all_non_result: expected_state = "aborted-non-result" elif reason in {"auto-disabled", "queue-empty"} or queue_drained: expected_state = "converged" else: expected_state = "max-rounds-reached" if reason == "all-reverify-non-result" and not all_non_result: errors.append("all-reverify-non-result requires only non-result dispatches") if reason == "max-rounds-1" and effective_max != 1: errors.append("max-rounds-1 requires effectiveMaxRounds=1") if reason == "queue-empty" and history and history[-1].get("carriedForwardCount") != 0: errors.append("queue-empty requires zero carriedForwardCount") if reason == "not-skipped" and len(history) < 2: errors.append("not-skipped requires at least two rounds") if final_state != expected_state: errors.append( f"finalState {final_state} is inconsistent with round2SkippedReason {reason}" ) return errors def _nonempty_string(value: Any) -> bool: return isinstance(value, str) and bool(value.strip()) def _parse_dispatch_results( value: Any, plan: Mapping[str, Any], ) -> dict[str, dict[str, Any]]: if not isinstance(value, list): raise ConvergenceContractError("round results dispatches must be an array") planned_workers = [row["worker"] for row in plan["dispatches"]] statuses: dict[str, dict[str, Any]] = {} for index, raw in enumerate(value): row = _object(raw, f"round results dispatches[{index}]") worker = _required_string(row, "worker", f"round results dispatches[{index}]") if worker in statuses: raise ConvergenceContractError(f"duplicate dispatch result: {worker}") if worker not in planned_workers: raise ConvergenceContractError(f"unplanned worker in dispatch results: {worker}") status = _required_string(row, "status", f"dispatch {worker}") if status not in _DISPATCH_STATUSES: raise ConvergenceContractError(f"dispatch {worker} has invalid status") duration = row.get("durationMs") if not isinstance(duration, int) or isinstance(duration, bool) or duration < 0: raise ConvergenceContractError( f"dispatch {worker}.durationMs must be a non-negative integer" ) statuses[worker] = {"status": status, "durationMs": duration} missing = [worker for worker in planned_workers if worker not in statuses] if missing: raise ConvergenceContractError( "missing status for planned dispatch: " + ", ".join(missing) ) return statuses def _planned_workers_by_finding( plan: Mapping[str, Any], ) -> dict[str, list[str]]: planned: dict[str, list[str]] = {} for row in plan["dispatches"]: for finding_id in row["findingIds"]: planned.setdefault(finding_id, []).append(row["worker"]) return planned def _validate_round_votes( raw_votes: Mapping[str, Any], planned_by_finding: Mapping[str, list[str]], statuses: Mapping[str, Mapping[str, Any]], ) -> dict[str, dict[str, Mapping[str, Any]]]: parsed: dict[str, dict[str, Mapping[str, Any]]] = {} for finding_id, worker_votes_value in raw_votes.items(): if finding_id not in planned_by_finding: raise ConvergenceContractError(f"unplanned finding in votes: {finding_id}") worker_votes = _object(worker_votes_value, f"votesByFinding.{finding_id}") parsed[finding_id] = {} for worker, vote_value in worker_votes.items(): if worker not in planned_by_finding[finding_id]: raise ConvergenceContractError( f"origin worker or unplanned worker voted on {finding_id}: {worker}" ) if statuses[worker]["status"] != "completed": raise ConvergenceContractError( f"non-result dispatch supplied a vote: {worker} {finding_id}" ) parsed[finding_id][worker] = _object( vote_value, f"votesByFinding.{finding_id}.{worker}" ) for finding_id, workers in planned_by_finding.items(): for worker in workers: if statuses[worker]["status"] == "completed" and worker not in parsed.get( finding_id, {} ): raise ConvergenceContractError( f"missing vote for completed worker {worker} on {finding_id}" ) return parsed def _votes_for_finding( finding_id: str, planned_by_finding: Mapping[str, list[str]], statuses: Mapping[str, Mapping[str, Any]], votes_by_finding: Mapping[str, Mapping[str, Mapping[str, Any]]], adversarial: bool, ) -> dict[str, dict[str, Any]]: votes: dict[str, dict[str, Any]] = {} for worker in planned_by_finding.get(finding_id, []): status = statuses[worker]["status"] if status == "completed": votes[worker] = _parse_vote( votes_by_finding[finding_id][worker], adversarial=adversarial, allow_unverifiable=True, ) else: votes[worker] = { "verdict": "verification-error", "disagreeBasis": None, "explanation": f"dispatch ended with terminal status {status}", } return votes def _recompute_worker_positions( finding: dict[str, Any], workers: list[Mapping[str, Any]], ) -> None: source_workers = set(finding.get("discoveredBy", {}).keys()) consensus = set(source_workers) dissent = set() for round_row in finding.get("rounds", []): for worker, vote in round_row.get("votes", {}).items(): if vote.get("verdict") in {"agree", "supplement"}: consensus.add(worker) dissent.discard(worker) elif vote.get("verdict") == "disagree": dissent.add(worker) consensus.discard(worker) roster = [ worker.get("workerId") for worker in workers if worker.get("audience") == "analysis" ] finding["consensusWorkers"] = [worker for worker in roster if worker in consensus] finding["dissentingWorkers"] = [worker for worker in roster if worker in dissent] def _parsed_votes( votes: Mapping[str, Mapping[str, Any]], *, adversarial: bool, ) -> dict[str, dict[str, Any]]: if not isinstance(votes, Mapping): raise ConvergenceContractError("votes must be an object") return { worker: _parse_vote(vote, adversarial=adversarial) for worker, vote in votes.items() } def _parse_vote( value: Any, *, adversarial: bool, allow_unverifiable: bool = False, ) -> dict[str, Any]: vote = _object(value, "vote") verdict = _required_string(vote, "verdict", "vote") allowed = _INPUT_VERDICTS if allow_unverifiable else _VERDICTS if verdict not in allowed: raise ConvergenceContractError(f"unsupported vote verdict: {verdict}") if verdict == "unverifiable": verdict = "verification-error" explanation = _required_string(vote, "explanation", "vote") basis = vote.get("disagreeBasis") if verdict == "disagree" and adversarial: if basis not in _DISAGREE_BASES: raise ConvergenceContractError( "adversarial disagree vote requires disagreeBasis" ) elif basis is not None: raise ConvergenceContractError( "disagreeBasis is allowed only for adversarial disagree votes" ) return { "verdict": verdict, "disagreeBasis": basis, "explanation": explanation, } def _validate_plannable_state( state: Mapping[str, Any], ) -> tuple[list[str], dict[str, Mapping[str, Any]], list[Mapping[str, Any]]]: version_errors: list[str] = [] _working_execution_identity_version(state, version_errors) if version_errors: raise ConvergenceContractError("; ".join(version_errors)) queue_value = state.get("queueFindingIds") if not isinstance(queue_value, list) or not all( isinstance(item, str) and item for item in queue_value ): raise ConvergenceContractError( "working state.queueFindingIds must be a string array" ) queue = list(queue_value) if len(queue) != len(set(queue)): raise ConvergenceContractError("duplicate queue finding ID") finding_rows = state.get("findings") if not isinstance(finding_rows, list): raise ConvergenceContractError("working state.findings must be an array") findings: dict[str, Mapping[str, Any]] = {} for row in finding_rows: finding = _object(row, "working state finding") finding_id = _required_string(finding, "findingId", "working state finding") if finding_id in findings: raise ConvergenceContractError(f"duplicate findingId: {finding_id}") findings[finding_id] = finding for finding_id in queue: finding = findings.get(finding_id) if finding is None: raise ConvergenceContractError( f"queue references missing finding: {finding_id}" ) if finding.get("classification") is not None: raise ConvergenceContractError( f"queued finding is already classified: {finding_id}" ) history_value = state.get("roundHistory") if not isinstance(history_value, list): raise ConvergenceContractError("working state.roundHistory must be an array") history = [_object(row, "round history row") for row in history_value] if history: carried = history[-1].get("carriedForwardCount") if carried != len(queue): raise ConvergenceContractError( "last round carriedForwardCount disagrees with current queue length" ) return queue, findings, history def _grouped_execution_identity_version(source: Mapping[str, Any]) -> int: schema_version = source.get("schemaVersion") identity_version = source.get("executionIdentityVersion") workers = source.get("workers") worker_rows = workers if isinstance(workers, list) else [] has_worker_identity = any( isinstance(worker, Mapping) and ( "participantRef" in worker or "sourceRoleExecutionRef" in worker ) for worker in worker_rows ) if schema_version == WORKING_SCHEMA_VERSION: if ( "executionIdentityVersion" in source or "runManifestPath" in source or has_worker_identity ): raise ConvergenceContractError( "v1 convergence groups forbid v2 execution identity fields" ) return 1 if schema_version == V2_WORKING_SCHEMA_VERSION: if identity_version != EXECUTION_IDENTITY_VERSION: raise ConvergenceContractError( "v2 convergence groups require executionIdentityVersion 2" ) _required_string(source, "runManifestPath", "grouped input") return EXECUTION_IDENTITY_VERSION raise ConvergenceContractError( f"unsupported grouped input schemaVersion: {schema_version!r}" ) def _parse_config(value: Any) -> dict[str, Any]: config = _object(value, "config") enabled = config.get("enabled") adversarial = config.get("adversarial") if not isinstance(enabled, bool): raise ConvergenceContractError("config.enabled must be boolean") if not isinstance(adversarial, bool): raise ConvergenceContractError("config.adversarial must be boolean") max_rounds = _round_limit(config.get("maxRounds"), "maxRounds") effective = _round_limit( config.get("effectiveMaxRounds"), "effectiveMaxRounds" ) if effective > max_rounds: raise ConvergenceContractError( "config.effectiveMaxRounds cannot exceed maxRounds" ) mode = _required_string(config, "verificationMode", "config") if mode not in _VERIFICATION_MODES: raise ConvergenceContractError("config.verificationMode is unsupported") if adversarial and mode != "full-reanalysis": raise ConvergenceContractError( "adversarial convergence requires full-reanalysis verificationMode" ) return { "enabled": enabled, "adversarial": adversarial, "maxRounds": max_rounds, "effectiveMaxRounds": effective, "verificationMode": mode, } def _parse_workers(value: Any, identity_version: int) -> list[dict[str, str]]: if not isinstance(value, list): raise ConvergenceContractError("workers must be an array") workers: list[dict[str, str]] = [] seen: set[str] = set() for index, raw in enumerate(value): worker = _object(raw, f"workers[{index}]") worker_id = _required_string(worker, "workerId", f"workers[{index}]") audience = _required_string(worker, "audience", f"workers[{index}]") if audience not in _AUDIENCES: raise ConvergenceContractError( f"workers[{index}] has unsupported audience: {audience}. " "Convergence audience is a functional role, not a phase label: " "every finding-producing worker uses 'analysis' (an " "implementation run's verifiers included), the okstra lead's own " "review uses 'lead', and only the report author uses " f"'report-writer'. Allowed: {sorted(_AUDIENCES)}." ) if worker_id in seen: raise ConvergenceContractError(f"duplicate workerId: {worker_id}") seen.add(worker_id) parsed = {"workerId": worker_id, "audience": audience} participant_ref = worker.get("participantRef") source_ref = worker.get("sourceRoleExecutionRef") has_execution_refs = ( "participantRef" in worker or "sourceRoleExecutionRef" in worker ) if identity_version == 1 and has_execution_refs: raise ConvergenceContractError( "v1 convergence workers forbid execution identity references" ) if identity_version == EXECUTION_IDENTITY_VERSION and ( not _nonempty_string(participant_ref) or not _nonempty_string(source_ref) ): raise ConvergenceContractError( "v2 convergence workers require execution identity references: " f"workers[{index}] must carry participantRef and " "sourceRoleExecutionRef" ) if identity_version == EXECUTION_IDENTITY_VERSION: parsed.update({ "participantRef": str(participant_ref), "sourceRoleExecutionRef": str(source_ref), }) workers.append(parsed) return workers def _validate_worker_execution_identity( workers: list[dict[str, str]], execution_manifest: ExecutionManifest | None, identity_version: int, ) -> None: if identity_version == 1: if execution_manifest is not None: raise ConvergenceContractError( "v1 convergence forbids execution manifest authority; " "omit --run-manifest" ) return if execution_manifest is None: raise ConvergenceContractError( "v2 convergence requires execution manifest authority; " "pass --run-manifest" ) if execution_manifest.legacy: raise ConvergenceContractError( "v2 convergence workers require v2 execution manifest" ) roles = { role.role_execution_ref: role for role in execution_manifest.role_executions } for worker in workers: source_ref = worker["sourceRoleExecutionRef"] source = roles.get(source_ref) if source is None: raise ConvergenceContractError( f"unknown sourceRoleExecutionRef: {source_ref}" ) if source.participant_ref != worker["participantRef"]: raise ConvergenceContractError( f"worker {worker['workerId']} participantRef does not match " f"sourceRoleExecutionRef: {source_ref}" ) allowed_roles = _AUDIENCE_SOURCE_ROLES[worker["audience"]] if source.role not in allowed_roles: raise ConvergenceContractError( f"{worker['audience']} audience source role is not allowed: " f"{source.role}" ) source_refs = [worker["sourceRoleExecutionRef"] for worker in workers] if len(source_refs) != len(set(source_refs)): raise ConvergenceContractError("duplicate sourceRoleExecutionRef") def _parse_groups( value: Any, analysis_workers: list[str], finding_sources: list[str], *, adversarial: bool, ) -> tuple[list[dict[str, Any]], list[str]]: if not isinstance(value, list): raise ConvergenceContractError("groups must be an array") findings: list[dict[str, Any]] = [] queue: list[str] = [] seen_ids: set[str] = set() for index, raw in enumerate(value): group = _object(raw, f"groups[{index}]") finding_id = _required_string(group, "findingId", f"groups[{index}]") if finding_id in seen_ids: raise ConvergenceContractError(f"duplicate findingId: {finding_id}") seen_ids.add(finding_id) finding, source_workers = _parse_group( group, index, analysis_workers, finding_sources ) if len(source_workers) >= 2 and not adversarial: finding["classification"] = "full-consensus" else: queue.append(finding_id) findings.append(finding) return findings, queue def _parse_group( group: Mapping[str, Any], index: int, analysis_workers: list[str], finding_sources: list[str], ) -> tuple[dict[str, Any], list[str]]: """그룹 하나를 읽는다. 출처는 발견을 낼 수 있는 모두, 합의는 분석 워커만. 두 목록을 따로 받는 이유가 이 함수의 전부다. 리드도 발견을 낼 수 있어야 하지만 그 발견이 교차검증 표로 세어지면, 독립 검증자 없이 해소된 것처럼 보이는 findings 가 생긴다. """ label = f"groups[{index}]" origin = _required_string(group, "originWorker", label) if origin not in finding_sources: raise ConvergenceContractError( f"{label}.originWorker cannot produce findings in this run: {origin}" ) origin_evidence = _required_string(group, "originEvidence", label) discovered_by = _parse_discovered_by( group.get("discoveredBy"), label, finding_sources ) if origin not in discovered_by: raise ConvergenceContractError( f"{label}.originWorker must appear in discoveredBy" ) source_items = _parse_source_items( group.get("sourceItems"), label, finding_sources ) source_workers = [ worker_id for worker_id in analysis_workers if worker_id in discovered_by or any(item["worker"] == worker_id for item in source_items) ] finding = { "findingId": _required_string(group, "findingId", label), "summary": _required_string(group, "summary", label), "category": _required_string(group, "category", label), "ticketIds": _string_array_allow_empty( group.get("ticketIds"), f"{label}.ticketIds" ), "originWorker": origin, "originEvidence": origin_evidence, "discoveredBy": discovered_by, "sourceItems": source_items, "classification": None, "rounds": [], "consensusWorkers": source_workers, "dissentingWorkers": [], } if "evidenceArtifacts" in group: finding["evidenceArtifacts"] = _parse_evidence_artifacts( group.get("evidenceArtifacts"), label ) return finding, source_workers def _parse_evidence_artifacts(value: Any, label: str) -> list[dict[str, str]]: if not isinstance(value, list) or not value: raise ConvergenceContractError( f"{label}.evidenceArtifacts must be a non-empty array" ) artifacts: list[dict[str, str]] = [] for index, raw in enumerate(value): item_label = f"{label}.evidenceArtifacts[{index}]" artifact = _object(raw, item_label) path = _required_string(artifact, "path", item_label) if not _is_normalized_okstra_path(path): raise ConvergenceContractError( f"{item_label}.path must be a normalized project-relative .okstra path" ) sha256 = _required_string(artifact, "sha256", item_label) if len(sha256) != 64 or any( char not in "0123456789abcdef" for char in sha256 ): raise ConvergenceContractError( f"{item_label}.sha256 must be a lowercase SHA-256 hex digest" ) artifacts.append( { "path": path, "sha256": sha256, "command": _required_string(artifact, "command", item_label), "environment": _required_string( artifact, "environment", item_label ), } ) return artifacts def _is_normalized_okstra_path(path: str) -> bool: if not path.startswith(".okstra/") or "\\" in path: return False segments = path.split("/") return all(segment not in {"", ".", ".."} for segment in segments) def _parse_discovered_by( value: Any, label: str, finding_sources: list[str], ) -> dict[str, dict[str, str]]: raw = _object(value, f"{label}.discoveredBy") parsed: dict[str, dict[str, str]] = {} for worker_id, details_value in raw.items(): if worker_id not in finding_sources: raise ConvergenceContractError( f"{label}.discoveredBy contains a worker that produces no " f"findings in this run: {worker_id}" ) details = _object(details_value, f"{label}.discoveredBy.{worker_id}") parsed[worker_id] = { "itemId": _required_string( details, "itemId", f"{label}.discoveredBy.{worker_id}" ), "evidence": _required_string( details, "evidence", f"{label}.discoveredBy.{worker_id}" ), } if not parsed: raise ConvergenceContractError(f"{label}.discoveredBy must not be empty") return parsed def _parse_source_items( value: Any, label: str, finding_sources: list[str], ) -> list[dict[str, str]]: if not isinstance(value, list) or not value: raise ConvergenceContractError(f"{label}.sourceItems must be a non-empty array") items: list[dict[str, str]] = [] for index, raw in enumerate(value): item = _object(raw, f"{label}.sourceItems[{index}]") worker = _required_string(item, "worker", f"{label}.sourceItems[{index}]") if worker not in finding_sources: raise ConvergenceContractError( f"{label}.sourceItems contains {worker}, which produces no " "findings in this run; report-writer never votes" ) items.append( { "worker": worker, "itemId": _required_string( item, "itemId", f"{label}.sourceItems[{index}]" ), } ) return items def _round_limit(value: Any, name: str) -> int: if not isinstance(value, int) or isinstance(value, bool) or not 1 <= value <= 3: raise ConvergenceContractError(f"config.{name} must be an integer from 1 to 3") return value def _string_array(value: Any, label: str) -> list[str]: if not isinstance(value, list) or not value: raise ConvergenceContractError(f"{label} must be a non-empty string array") result = [] for item in value: if not isinstance(item, str) or not item.strip(): raise ConvergenceContractError(f"{label} must contain non-empty strings") result.append(item.strip()) return result def _string_array_allow_empty(value: Any, label: str) -> list[str]: if not isinstance(value, list): raise ConvergenceContractError(f"{label} must be a string array") result = [] for item in value: if not isinstance(item, str) or not item.strip(): raise ConvergenceContractError(f"{label} must contain non-empty strings") result.append(item.strip()) return result def _object(value: Any, label: str) -> Mapping[str, Any]: if not isinstance(value, Mapping): raise ConvergenceContractError(f"{label} must be an object") return value def _required_string( payload: Mapping[str, Any], key: str, label: str, ) -> str: value = payload.get(key) if not isinstance(value, str) or not value.strip(): raise ConvergenceContractError(f"{label}.{key} must be a non-empty string") return value.strip()