"""CLI orchestration for deterministic Phase 5.5 convergence state.""" from __future__ import annotations import argparse from collections.abc import Mapping from dataclasses import dataclass import json from pathlib import Path, PurePosixPath import sys from typing import Any from okstra_project.state import StateError, parse_task_key, slugify from .convergence_engine import ( ConvergenceContractError, apply_critic_gap_results, apply_round_results, finalize_working_state, plan_next_round, seed_working_state, validate_final_state, validate_working_state, ) from .convergence_migration import ( SeedDecision, archive_state_bytes, decide_seed_action, migration_record, ) from .convergence_provenance import ( GROUPS_BASENAME_RE, provenance_errors, worker_result_suffix, ) from .convergence_store import ( load_convergence_critic_batch, load_convergence_round_results, load_owned_json_object, write_final_state_atomic, write_json_atomic, ) from .execution_identity import ExecutionManifest from .execution_manifest import read_execution_manifest_view from .final_report_schema import load_named_schema, validate as validate_schema from .paths import RunRef from .report_finalize import task_manifest_path from .verdict_blocks import FindingVote, VerdictBlockError, parse_finding_votes from .fixed_text import line @dataclass(frozen=True) class RunArtifactAuthority: manifest_path: Path project_root: Path run_dir: Path task_type: str task_key: str manifest_ref: str state_sequence: str execution_manifest: ExecutionManifest | None payload: Mapping[str, Any] def validated_run_authority(manifest_path: Path) -> RunArtifactAuthority: supplied = Path(manifest_path) if not supplied.is_absolute() or supplied.absolute() != supplied.resolve(): raise ConvergenceContractError("run manifest path is not canonical") payload = load_owned_json_object(supplied) execution_manifest = None if payload.get("executionIdentityVersion") == 2: execution_manifest, view = read_execution_manifest_view(supplied) payload = dict(view) project_root = _manifest_project_root(payload) manifest_ref = _manifest_run_ref(payload, project_root, supplied) run_dir, task_type, state_sequence = _validated_run_artifact_authority( payload, project_root, supplied, manifest_ref ) task_key = _manifest_authority_string(payload, "taskKey") try: _project_id, task_group, task_id = parse_task_key(task_key) except StateError as exc: raise ConvergenceContractError(str(exc)) from exc recovered = RunRef.from_run_dir(run_dir) if (slugify(task_group), slugify(task_id)) != ( recovered.task_group, recovered.task_id ): raise ConvergenceContractError( "run manifest taskKey does not match run directory" ) return RunArtifactAuthority( supplied, project_root, run_dir, task_type, task_key, manifest_ref, state_sequence, execution_manifest, payload, ) def canonical_run_state_artifact( authority: RunArtifactAuthority, *, manifest_field: str, prefix: str, label: str, ) -> Path: value = _manifest_authority_string(authority.payload, manifest_field) supplied = _canonical_project_artifact_path( authority.project_root, value, label ) expected = authority.run_dir / "state" / _canonical_run_artifact_name( prefix, authority.task_type, authority.state_sequence ) if supplied != expected: raise ConvergenceContractError(f"{label} does not match run authority") return expected def canonical_run_report_artifact(authority: RunArtifactAuthority) -> Path: report_sequence = _manifest_sequence(authority.payload, "reports") try: expected = RunRef.from_run_dir( authority.run_dir, seq=int(report_sequence) ).report except ValueError as exc: raise ConvergenceContractError( "run manifest report sequence is not canonical" ) from exc value = _manifest_authority_string( authority.payload, "expectedReportRecordPath" ) supplied = _canonical_project_artifact_path( authority.project_root, value, "report path" ) if supplied != expected: raise ConvergenceContractError("report path does not match run authority") return expected _EVIDENCE_ARTIFACT = { "path": ".okstra/tasks/example/group/task/runs/error-analysis/evidence/F-001/mysql-query.txt", "sha256": "0123456789abcdef" * 4, "command": "mysql --batch --raw < query.sql", "environment": "staging read-only replica", } _EXAMPLES: dict[str, dict[str, Any]] = { "groups": { "schemaVersion": "1.0", "taskKey": "example/group/task", "config": { "enabled": True, "adversarial": False, "maxRounds": 2, "effectiveMaxRounds": 2, "verificationMode": "lightweight", }, "workers": [ {"workerId": "claude-worker", "audience": "analysis"}, {"workerId": "codex-worker", "audience": "analysis"}, {"workerId": "report-writer", "audience": "report-writer"}, ], "groups": [ { "findingId": "F-001", "summary": "A live row contradicts the migration assumption", "category": "risk", "ticketIds": ["TASK-001"], "originWorker": "claude-worker", "originEvidence": "The live row contradicts the migration assumption", "evidenceArtifacts": [_EVIDENCE_ARTIFACT], "discoveredBy": { "claude-worker": { "itemId": "F-1", "evidence": "staging query result", } }, "sourceItems": [{"worker": "claude-worker", "itemId": "F-1"}], } ], }, "round-results": { "schemaVersion": "1.0", "round": 1, "dispatches": [ {"worker": "codex-worker", "status": "completed", "durationMs": 1250} ], "votesByFinding": { "F-001": { "codex-worker": { "verdict": "unverifiable", "disagreeBasis": None, "explanation": "staging credentials were unavailable", } } }, }, "critic-results": { "schemaVersion": "1.0", "taskKey": "example/group/task", "mode": "coverage", "provider": "codex", "terminalStatus": "completed", "durationMs": 1750, "candidates": [ { "candidateId": "CG-001", "summary": "The rollback path has no live-data verification", "category": "missing", "ticketIds": ["TASK-001"], "originEvidence": "No rollback query result is present", "evidenceArtifacts": [_EVIDENCE_ARTIFACT], } ], }, "coverage-batch": { "schemaVersion": "1.0", "taskKey": "example/group/task", "mode": "coverage", # The critic's own provider never votes on its gaps, so `provider` here # is deliberately absent from `dispatches[]`. "provider": "antigravity", "modelExecutionValue": "gemini-3.1-pro", "dispatches": [ {"worker": "claude-worker", "status": "completed", "durationMs": 1250}, {"worker": "codex-worker", "status": "completed", "durationMs": 980}, ], "gaps": [ { "gapId": "G-001", "summary": "The rollback path has no live-data verification", "category": "missing", "ticketIds": ["TASK-001"], "originEvidence": "No rollback query result is present", "evidenceArtifacts": [_EVIDENCE_ARTIFACT], "votes": { "claude-worker": { "verdict": "agree", "disagreeBasis": None, "explanation": "The plan cites no rollback verification step", }, "codex-worker": { "verdict": "agree", "disagreeBasis": None, "explanation": "No rollback assertion appears in the stage map", }, }, } ], }, } def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="okstra convergence", description="Advance and validate deterministic re-verification state.", epilog=( "Working lifecycle: groups v1.0/v2.0 → work v1.0/v2.0 → " "round-plan/results v1.0 → final v1.3\n" "historical final readers: v1.0, v1.1, v1.2" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) subparsers = parser.add_subparsers(dest="operation", required=True) example = subparsers.add_parser( "example", help="print a deterministic input artifact example", description=( "Print one deterministic valid example as JSON.\n\n" "Each kind feeds one command:\n" " groups -> seed --groups\n" " round-results -> apply-round --results\n" " coverage-batch -> apply-critic-gaps --results\n\n" "`critic-results` feeds nothing: it is the critic worker's own " "result document, NOT the `apply-critic-gaps --results` input. " "Feeding it straight in is rejected, by design — that reducer takes " "the coverage batch the lead assembles from those candidates plus " "each analyser's vote, which is what `--kind coverage-batch` prints. " 'See prompts/lead/convergence.md §"Coverage critic".' ), formatter_class=argparse.RawDescriptionHelpFormatter, ) example.add_argument( "--kind", choices=("groups", "round-results", "critic-results", "coverage-batch"), required=True, ) seed = subparsers.add_parser("seed", help="create, resume, or recover working state") seed.add_argument("--groups", type=Path, required=True) seed.add_argument("--run-manifest", type=Path) seed.add_argument("--work-state", type=Path, required=True) seed.add_argument("--final-state", type=Path, required=True) seed.add_argument("--migration-dir", type=Path, required=True) seed.add_argument("--restart-from-round0", action="store_true") prepare = subparsers.add_parser( "prepare-groups", help="publish grouped findings from fixed Markdown" ) prepare.add_argument("--run-manifest", type=Path, required=True) prepare.add_argument("--input", type=Path, required=True) plan = subparsers.add_parser("plan-round", help="write the next dispatch plan") plan.add_argument("--work-state", type=Path, required=True) plan.add_argument("--plan", type=Path, required=True) collect = subparsers.add_parser( "collect-results", help="read one round's worker responses into the apply-round input", ) collect.add_argument("--plan", type=Path, required=True) collect.add_argument("--mode", choices=("adversarial", "collaborative"), required=True, help="which reverify prompt this round used; the two " "carry different verdict vocabularies") collect.add_argument("--result", action="append", default=[], metavar="=", help="one worker's reverify result file (repeatable)") collect.add_argument("--dispatch", action="append", default=[], required=True, metavar="=:", help="terminal status per dispatched worker (repeatable)") collect.add_argument("--output", type=Path, required=True) apply = subparsers.add_parser("apply-round", help="apply structured round results") apply.add_argument("--work-state", type=Path, required=True) apply.add_argument("--plan", type=Path, required=True) apply.add_argument("--results", type=Path, required=True) apply_critic = subparsers.add_parser( "apply-critic-gaps", help="apply one coverage-critic verification batch", ) apply_critic.add_argument("--work-state", type=Path, required=True) apply_critic.add_argument("--results", type=Path, required=True) finalize = subparsers.add_parser("finalize", help="write the public final state") finalize.add_argument("--work-state", type=Path, required=True) finalize.add_argument("--output", type=Path, required=True) validate = subparsers.add_parser("validate", help="validate a persisted state") validate.add_argument("--state", type=Path, required=True) validate.add_argument("--kind", choices=("working", "final"), required=True) return parser def _emit(operation: str, action: str, path: Path) -> None: print( json.dumps( { "ok": True, "operation": operation, "action": action, "path": str(path), }, ensure_ascii=False, ) ) def _archive_existing_states( *, work_state_path: Path, final_state_path: Path, migration_dir: Path, ) -> dict[str, str] | None: records: list[dict[str, str]] = [] for path in (work_state_path, final_state_path): if not path.exists(): continue archive_path, digest = archive_state_bytes( source_path=path, migration_dir=migration_dir, ) records.append(migration_record(path, archive_path, digest)) if not records: return None final_record = next( (record for record in records if record["sourcePath"] == str(final_state_path)), None, ) return final_record or records[0] def _resume_after_legacy_final( decision: SeedDecision, *, work_state_path: Path, final_state_path: Path, migration_dir: Path, ) -> None: if decision.archive_path is None: return archive_path, digest = archive_state_bytes( source_path=final_state_path, migration_dir=migration_dir, ) state = load_owned_json_object(work_state_path) state["migration"] = migration_record(final_state_path, archive_path, digest) write_json_atomic(work_state_path, state) def _seeded_groups_provenance_errors(groups_path: Path, document: dict) -> list[str]: """Provenance errors for a grouping handed to `seed`, or none when the layout does not let us resolve the worker results it cites. Checking here is what makes the failure fixable: the lead just wrote this file, and the same claims are re-checked at Phase 7 by `validate-run`, where a bad `itemId` surfaces hours later as a blocked approval. """ if GROUPS_BASENAME_RE.match(groups_path.name) is None: return [] run_dir = groups_path.parent.parent worker_results_dir = run_dir / "worker-results" if not worker_results_dir.is_dir(): return [] # 그룹 파일명이 지닌 것은 state 시퀀스다. 워커 결과는 workerResults 시퀀스로 # 이름 붙으므로 그룹 문서가 기록한 run manifest 를 거쳐 해소한다. suffix = worker_result_suffix(run_dir, document) if suffix is None: return [] return provenance_errors( document, worker_results_dir=worker_results_dir, suffix=suffix, groups_label=groups_path.name, ) def _seed_execution_manifest( groups_path: Path, grouped_input: dict[str, Any], manifest_path: Path | None, work_state_path: Path, final_state_path: Path, ) -> ExecutionManifest | None: if manifest_path is None: return None manifest, authority = read_execution_manifest_view(manifest_path) if grouped_input.get("schemaVersion") == "2.0": _validate_v2_seed_authority( groups_path, grouped_input, manifest_path, authority, work_state_path, final_state_path, execution_manifest=manifest, ) return manifest def _validate_v2_seed_authority( groups_path: Path, grouped_input: dict[str, Any], manifest_path: Path, authority: Mapping[str, Any], work_state_path: Path, final_state_path: Path, *, execution_manifest: ExecutionManifest, ) -> None: validated = _validated_v2_work_authority( work_state_path, grouped_input, manifest_snapshot=(manifest_path, execution_manifest, authority), path_label="state output", ) if validated is None: raise ConvergenceContractError("v2 convergence groups require v2 work authority") _manifest, run_dir, task_type, state_sequence = validated expected_groups_name = _canonical_run_artifact_name( "convergence-groups", task_type, state_sequence, ) if groups_path.name != expected_groups_name: raise ConvergenceContractError( "convergence groups filename does not match run authority" ) expected_state_dir = run_dir / "state" if groups_path.absolute() != groups_path.resolve(): raise ConvergenceContractError("groups artifact path must be canonical") if groups_path.resolve().parent != expected_state_dir: raise ConvergenceContractError( "groups artifact is outside the current run state directory" ) _validate_v2_state_artifact_path( final_state_path, state_dir=expected_state_dir, task_type=task_type, state_sequence=state_sequence, prefix="convergence", label="state output", ) def _validated_v2_work_authority( work_state_path: Path, state: Mapping[str, Any], *, manifest_snapshot: tuple[ Path, ExecutionManifest, Mapping[str, Any], ] | None = None, path_label: str = "working state", ) -> tuple[ExecutionManifest, Path, str, str] | None: if state.get("schemaVersion") != "2.0": return None if state.get("executionIdentityVersion") != 2: raise ConvergenceContractError( "v2 working state requires executionIdentityVersion 2" ) task_key = _manifest_authority_string(state, "taskKey") run_ref = _manifest_authority_string(state, "runManifestPath") project_root = _project_root_for_v2_work_state(work_state_path, path_label) expected_manifest_path = _canonical_project_artifact_path( project_root, run_ref, "working runManifestPath", ) if manifest_snapshot is None: execution_manifest, authority = read_execution_manifest_view( expected_manifest_path ) manifest_path = expected_manifest_path else: supplied_manifest_path, execution_manifest, authority = manifest_snapshot manifest_path = supplied_manifest_path.resolve() if ( supplied_manifest_path.absolute() != manifest_path or manifest_path != expected_manifest_path ): raise ConvergenceContractError( "--run-manifest must use the canonical run manifest path; " "runManifestPath does not match supplied path" ) manifest_project_root = _manifest_project_root(authority) if project_root != manifest_project_root: raise ConvergenceContractError( "working state path does not match run manifest projectRoot" ) canonical_run_ref = _manifest_run_ref( authority, project_root, manifest_path, ) if task_key != authority.get("taskKey"): raise ConvergenceContractError( "working state taskKey does not match run manifest" ) if run_ref != canonical_run_ref: raise ConvergenceContractError( "working state runManifestPath does not match run manifest" ) run_dir, task_type, state_sequence = _validated_run_artifact_authority( authority, project_root, manifest_path, canonical_run_ref, ) _validate_v2_state_artifact_path( work_state_path, state_dir=run_dir / "state", task_type=task_type, state_sequence=state_sequence, prefix="convergence-work", label=path_label, ) return execution_manifest, run_dir, task_type, state_sequence def _project_root_for_v2_work_state( work_state_path: Path, path_label: str, ) -> Path: supplied_path = work_state_path.absolute() resolved_path = work_state_path.resolve() if supplied_path != resolved_path: raise ConvergenceContractError( f"{path_label} path does not match run authority" ) for parent in supplied_path.parents: if parent.name == ".okstra": return parent.parent raise ConvergenceContractError( f"{path_label} path does not match run authority" ) def _validated_run_artifact_authority( authority: Mapping[str, Any], project_root: Path, manifest_path: Path, run_ref: str, ) -> tuple[Path, str, str]: run_dir = _manifest_run_directory(authority, project_root, manifest_path) task_type = _manifest_authority_string(authority, "taskType") manifest_sequence = _manifest_sequence(authority, "manifests") expected_name = _canonical_run_artifact_name( "run-manifest", task_type, manifest_sequence, ) try: recovered = RunRef.from_run_dir( run_dir, seq=int(manifest_sequence), ) except ValueError as exc: raise ConvergenceContractError( "run manifest runDirectoryPath is not a canonical run directory" ) from exc if recovered.project_root != project_root: raise ConvergenceContractError( "run manifest projectRoot does not match recovered run directory" ) if recovered.task_type != task_type: raise ConvergenceContractError( "run manifest taskType does not match run directory" ) stage = recovered.stage invalid_stage = ( (task_type == "implementation" and (stage is None or stage < 1)) or (task_type == "final-verification" and stage is not None and stage < 1) or ( task_type not in ("implementation", "final-verification") and stage is not None ) ) if invalid_stage: raise ConvergenceContractError( "run manifest stage semantics do not match taskType" ) if recovered.run_dir != run_dir: raise ConvergenceContractError( "run manifest stage does not match runDirectoryPath" ) if manifest_path.name != expected_name: raise ConvergenceContractError( "run manifest filename does not match run authority" ) if manifest_path != recovered.manifest: raise ConvergenceContractError( "run manifest path does not match runDirectoryPath authority" ) expected_ref = recovered.manifest.relative_to(project_root).as_posix() if run_ref != expected_ref: raise ConvergenceContractError( "runManifestPath does not match canonical run authority" ) return run_dir, task_type, _manifest_sequence(authority, "state") def _manifest_authority_string( authority: Mapping[str, Any], key: str, ) -> str: value = authority.get(key) if not isinstance(value, str) or not value.strip(): raise ConvergenceContractError(f"run manifest has no {key}") return value def _manifest_sequence(authority: Mapping[str, Any], category: str) -> str: sequences = authority.get("runSequencesByCategory") if not isinstance(sequences, Mapping): raise ConvergenceContractError( "run manifest has no runSequencesByCategory object" ) return _manifest_authority_string(sequences, category) def _canonical_run_artifact_name( prefix: str, task_type: str, sequence: str, ) -> str: try: sequence_number = int(sequence) except ValueError as exc: raise ConvergenceContractError( "run manifest sequence is not canonical" ) from exc if sequence_number < 1: raise ConvergenceContractError( "run manifest sequence is not canonical" ) probe = RunRef( project_root=Path("/"), task_group="group", task_id="task", task_type=task_type, seq=sequence_number, ).manifest.name if not probe.endswith(f"-{sequence}.json"): raise ConvergenceContractError( "run manifest sequence is not canonical" ) suffix = probe.removeprefix("run-manifest").removesuffix(".json") return f"{prefix}{suffix}.json" def _manifest_run_directory( authority: Mapping[str, Any], project_root: Path, manifest_path: Path, ) -> Path: value = _manifest_authority_string(authority, "runDirectoryPath") posix = PurePosixPath(value) if ( posix.is_absolute() or posix.as_posix() != value or not posix.parts or posix.parts[0] != ".okstra" or ".." in posix.parts ): raise ConvergenceContractError( "run manifest runDirectoryPath is not canonical" ) resolved = (project_root / value).resolve() if not resolved.is_relative_to(project_root): raise ConvergenceContractError( "run manifest runDirectoryPath escapes projectRoot" ) if resolved != manifest_path.parent.parent: raise ConvergenceContractError( "run manifest runDirectoryPath does not identify the supplied run" ) return resolved def _validate_v2_state_artifact_path( supplied_path: Path, *, state_dir: Path, task_type: str, state_sequence: str, prefix: str, label: str, ) -> None: expected_path = state_dir / _canonical_run_artifact_name( prefix, task_type, state_sequence, ) if ( supplied_path.absolute() != supplied_path.resolve() or supplied_path.resolve() != expected_path ): raise ConvergenceContractError( f"{label} path does not match run authority" ) def _manifest_project_root(authority: dict[str, Any]) -> Path: value = authority.get("projectRoot") if not isinstance(value, str) or not value.strip(): raise ConvergenceContractError("run manifest has no projectRoot") project_root = Path(value) if not project_root.is_absolute() or project_root.absolute() != project_root.resolve(): raise ConvergenceContractError("run manifest projectRoot must be canonical") return project_root def _manifest_run_ref( authority: dict[str, Any], project_root: Path, manifest_path: Path, ) -> str: value = _manifest_authority_string(authority, "runManifestPath") resolved = _canonical_project_artifact_path( project_root, value, "run manifest runManifestPath", ) if resolved != manifest_path: raise ConvergenceContractError( "run manifest runManifestPath does not identify the supplied manifest" ) return value def _canonical_project_artifact_path( project_root: Path, value: str, label: str, ) -> Path: posix = PurePosixPath(value) supplied = project_root.joinpath(*posix.parts) resolved = supplied.resolve() if not resolved.is_relative_to(project_root): raise ConvergenceContractError(f"{label} escapes projectRoot") if ( posix.is_absolute() or posix.as_posix() != value or not posix.parts or posix.parts[0] != ".okstra" or ".." in posix.parts ): raise ConvergenceContractError(f"{label} is not canonical") if supplied.absolute() != resolved: raise ConvergenceContractError(f"{label} escapes projectRoot") return resolved def _seed(args: argparse.Namespace) -> tuple[str, Path]: grouped_input = load_owned_json_object(args.groups) execution_manifest = _seed_execution_manifest( args.groups, grouped_input, args.run_manifest, args.work_state, args.final_state, ) provenance = _seeded_groups_provenance_errors(args.groups, grouped_input) if provenance: raise ConvergenceContractError( "grouping cites source items no worker reported: " + "; ".join(provenance) ) state = seed_working_state( grouped_input, execution_manifest=execution_manifest, ) decision = decide_seed_action( grouped_input=grouped_input, work_state_path=args.work_state, final_state_path=args.final_state, restart_from_round0=args.restart_from_round0, ) if decision.action == "reuse-final": return decision.action, args.final_state if decision.action == "resume-work": _resume_after_legacy_final( decision, work_state_path=args.work_state, final_state_path=args.final_state, migration_dir=args.migration_dir, ) return decision.action, args.work_state if decision.action == "restart-round0": migration = _archive_existing_states( work_state_path=args.work_state, final_state_path=args.final_state, migration_dir=args.migration_dir, ) if migration is None: raise ConvergenceContractError("restart-round0 has no state to archive") state["migration"] = migration write_json_atomic(args.work_state, state) return decision.action, args.work_state def _plan_round(args: argparse.Namespace) -> tuple[str, Path]: state = load_owned_json_object(args.work_state) errors = validate_working_state(state) if errors: raise ConvergenceContractError("invalid working state: " + "; ".join(errors)) plan = plan_next_round(state) write_json_atomic(args.plan, plan) return str(plan["action"]), args.plan def _apply_round(args: argparse.Namespace) -> tuple[str, Path]: state = load_owned_json_object(args.work_state) _validated_v2_work_authority(args.work_state, state) plan = load_owned_json_object(args.plan) results = load_convergence_round_results(args.results) updated = apply_round_results(state, plan, results) errors = validate_working_state(updated) if errors: raise ConvergenceContractError("invalid updated working state: " + "; ".join(errors)) write_json_atomic(args.work_state, updated) return "applied", args.work_state def _apply_critic_gaps(args: argparse.Namespace) -> tuple[str, Path]: state = load_owned_json_object(args.work_state) _validated_v2_work_authority(args.work_state, state) results = load_convergence_critic_batch(args.results) updated = apply_critic_gap_results(state, results) write_json_atomic(args.work_state, updated) return "applied", args.work_state def _finalize(args: argparse.Namespace) -> tuple[str, Path]: state = load_owned_json_object(args.work_state) authority = _validated_v2_work_authority(args.work_state, state) if authority is not None: _manifest, run_dir, task_type, state_sequence = authority _validate_v2_state_artifact_path( args.output, state_dir=run_dir / "state", task_type=task_type, state_sequence=state_sequence, prefix="convergence", label="final state", ) final = finalize_working_state(state) migration = state.get("migration") write_final_state_atomic(args.output, final, migration=migration) return "finalized", args.output def _validate(args: argparse.Namespace) -> tuple[str, Path]: state = load_owned_json_object(args.state) errors = ( validate_working_state(state) if args.kind == "working" else validate_final_state(state) ) if errors: raise ConvergenceContractError("; ".join(errors)) return "valid", args.state def _split_pair(raw: str, label: str) -> tuple[str, str]: key, separator, value = raw.partition("=") if not separator or not key.strip() or not value.strip(): raise ConvergenceContractError(f"{label} must be =, got: {raw}") return key.strip(), value.strip() def _dispatch_rows(raw_dispatches: list[str]) -> list[dict[str, Any]]: """`dispatches[]`, from what only the lead knows. Terminal status and duration are not in the worker's response; they belong to the dispatch. Deriving them from the presence of a result file would call a timed-out worker `completed` whenever it managed to write something. """ rows: list[dict[str, Any]] = [] for raw in raw_dispatches: worker, value = _split_pair(raw, "--dispatch") status, _, duration = value.partition(":") if not duration.isdigit(): raise ConvergenceContractError( f"--dispatch for `{worker}` must end in :, got: {value}" ) rows.append({ "worker": worker, "status": status, "durationMs": int(duration) }) return rows def _planned_finding_ids(plan: dict[str, Any]) -> dict[str, list[str]]: dispatches = plan.get("dispatches") if not isinstance(dispatches, list): raise ConvergenceContractError("round plan has no `dispatches` array") planned: dict[str, list[str]] = {} for row in dispatches: worker = row.get("worker") if isinstance(row, dict) else None if not isinstance(worker, str) or not worker: raise ConvergenceContractError("every round plan dispatch needs a `worker`") planned[worker] = list(row.get("findingIds") or []) return planned def _worker_votes( raw_results: list[str], planned: dict[str, list[str]], adversarial: bool ) -> dict[str, dict[str, FindingVote]]: votes: dict[str, dict[str, FindingVote]] = {} for raw in raw_results: worker, path = _split_pair(raw, "--result") if worker not in planned: raise ConvergenceContractError( f"`{worker}` returned a result but the round plan did not dispatch it" ) parsed = parse_finding_votes( Path(path).read_text(encoding="utf-8"), adversarial=adversarial ) assigned = set(planned[worker]) missing = sorted(assigned - set(parsed)) if missing: raise ConvergenceContractError( f"`{worker}` was dispatched {len(assigned)} findings but returned " f"no verdict for {missing} — an unanswered finding cannot be " f"classified, and dropping it silently makes the round look complete" ) unknown = sorted(set(parsed) - assigned) if unknown: raise ConvergenceContractError( f"`{worker}` returned verdicts for {unknown}, which this round's " f"plan did not dispatch to it" ) votes[worker] = parsed return votes def _collect_results(args: argparse.Namespace) -> tuple[str, Path]: plan = load_owned_json_object(args.plan) planned = _planned_finding_ids(plan) dispatches = _dispatch_rows(args.dispatch) undispatched = sorted(set(planned) - {row["worker"] for row in dispatches}) if undispatched: raise ConvergenceContractError( f"the round plan dispatched {undispatched} but no --dispatch status " f"was given for them; a planned worker with no recorded outcome is " f"indistinguishable from one that was never asked" ) votes = _worker_votes(args.result, planned, args.mode == "adversarial") by_finding: dict[str, dict[str, Any]] = {} for worker, parsed in votes.items(): for finding_id, vote in parsed.items(): by_finding.setdefault(finding_id, {})[worker] = { "verdict": vote.verdict, "disagreeBasis": vote.disagree_basis, "explanation": vote.explanation, } write_json_atomic(args.output, { "schemaVersion": "1.0", "round": plan.get("round"), "dispatches": dispatches, "votesByFinding": by_finding, }) return "collected", args.output def _parse_grouping_markdown(path: Path) -> list[dict[str, Any]]: groups: list[dict[str, Any]] = [] fields: dict[str, list[str]] = {} finding_id = "" for raw in [*path.read_text(encoding="utf-8").splitlines(), "## END"]: if raw.startswith("## "): if finding_id: groups.append(_group_from_fields(finding_id, fields)) finding_id = raw[3:].strip() fields = {} elif ":" in raw and finding_id: key, value = raw.split(":", 1) fields.setdefault(key.strip().lower(), []).append(value.strip()) return groups def _parse_group_source(spec: str) -> tuple[str, str, str]: identity, separator, evidence = spec.partition("|") worker, colon, item_id = identity.strip().partition(":") if not separator or not colon or not all((worker, item_id, evidence.strip())): raise ConvergenceContractError( "Source must use worker:item-id | evidence" ) return worker, item_id, evidence.strip() def _group_from_fields( finding_id: str, fields: Mapping[str, list[str]] ) -> dict[str, Any]: scalar = { key: values[0] if len(values) == 1 else "" for key, values in fields.items() if key != "source" } source_rows = fields.get("source", []) if source_rows: parsed = [_parse_group_source(spec) for spec in source_rows] else: specs = [part.strip() for part in scalar.get("sources", "").split(",") if part.strip()] if len(specs) != 1 or not scalar.get("evidence"): raise ConvergenceContractError( "multi-worker groups require one Source per worker" ) worker, colon, item_id = specs[0].partition(":") if not colon or not worker or not item_id: raise ConvergenceContractError("Sources must use worker:item-id entries") parsed = [(worker, item_id, scalar["evidence"])] discovered, source_items = _group_source_provenance(parsed) origin_worker, separator, origin_item = scalar.get("origin", "").partition(":") if not separator or discovered.get(origin_worker, {}).get("itemId") != origin_item: raise ConvergenceContractError("Origin must identify one Source item") return { "findingId": finding_id, "summary": scalar.get("summary", ""), "category": scalar.get("category", ""), "ticketIds": [item.strip() for item in scalar.get("tickets", "").split(",") if item.strip()], "originWorker": origin_worker, "originEvidence": discovered[origin_worker]["evidence"], "discoveredBy": discovered, "sourceItems": source_items, } def _group_source_provenance( parsed: list[tuple[str, str, str]], ) -> tuple[dict[str, dict[str, str]], list[dict[str, str]]]: workers = [worker for worker, _item, _evidence in parsed] if len(workers) != len(set(workers)): raise ConvergenceContractError("duplicate Source worker in one finding") discovered = { worker: {"itemId": item, "evidence": evidence} for worker, item, evidence in parsed } sources = [ {"worker": worker, "itemId": item} for worker, item, _evidence in parsed ] if any(discovered[row["worker"]]["itemId"] != row["itemId"] for row in sources): raise ConvergenceContractError("discoveredBy and sourceItems do not match") return discovered, sources def _assignment_identity(row: Mapping[str, Any]) -> tuple[str, str, str]: role = str(row.get("role") or "") provider = str(row.get("provider") or "") model = str(row.get("modelExecutionValue") or row.get("model") or "") if not all((role, provider, model)): raise ConvergenceContractError( "worker assignment requires role, provider, and model" ) return role, provider, model def _execution_model(row: Mapping[str, Any]) -> str: binding = row.get("binding") if isinstance(binding, Mapping) and binding.get("resolvedExecutionValue"): return str(binding["resolvedExecutionValue"]) return str(row.get("modelId") or "") def _group_assignment_identity( manifest: Mapping[str, Any], worker_id: str, ) -> tuple[str, str, str, int]: assignments = manifest.get("workerAssignments") if isinstance(manifest.get("workerAssignments"), list) else [] normalized = worker_id.removesuffix("-worker") if normalized == "lead": duplicate = any( isinstance(row, Mapping) and str(row.get("workerId", "")).removesuffix("-worker") == "lead" for row in assignments ) if duplicate: raise ConvergenceContractError("ambiguous lead assignment identity") lead = manifest.get("leadAssignment") if not isinstance(lead, Mapping): raise ConvergenceContractError("lead assignment identity is missing") role, provider, model = _assignment_identity(lead) if role != "lead": raise ConvergenceContractError("leadAssignment role must be lead") return "leader", provider, model, 1 matches = [(index, row) for index, row in enumerate(assignments) if isinstance(row, Mapping) and str(row.get("workerId", "")).removesuffix("-worker") == normalized] if len(matches) != 1: raise ConvergenceContractError( f"worker assignment identity is missing or duplicate for {worker_id}" ) index, assignment = matches[0] role, provider, model = _assignment_identity(assignment) ordinal = sum( 1 for row in assignments[:index + 1] if isinstance(row, Mapping) and str(row.get("role") or "") == role ) return role, provider, model, ordinal def _group_worker(manifest: Mapping[str, Any], worker_id: str) -> dict[str, str]: role, provider, model, ordinal = _group_assignment_identity( manifest, worker_id ) executions = manifest.get("roleExecutions") if isinstance(manifest.get("roleExecutions"), list) else [] candidates = [row for row in executions if isinstance(row, Mapping) and row.get("role") == role and row.get("provider") == provider and _execution_model(row) == model and row.get("ordinal") == ordinal] if len(candidates) != 1: raise ConvergenceContractError(f"no canonical role execution for {worker_id}") execution = candidates[0] if not execution.get("participantRef") or not execution.get("roleExecutionRef"): raise ConvergenceContractError(f"no canonical role execution for {worker_id}") audience = "report-writer" if role == "report-writer" else "lead" if role in {"lead", "leader"} else "analysis" return {"workerId": worker_id, "audience": audience, "participantRef": str(execution["participantRef"]), "sourceRoleExecutionRef": str(execution["roleExecutionRef"])} def _convergence_config(authority: RunArtifactAuthority) -> Mapping[str, Any]: """convergence 설정의 정본은 task-manifest.json 이다. `render.render_task_manifest` 가 `_build_convergence_block` 결과를 거기에만 쓴다. run manifest 에는 `convergenceStatePath` 뿐이라, 예전처럼 run manifest 를 읽으면 키가 없어 `adversarial=False` / `verificationMode="lightweight"` 기본값으로 조용히 떨어졌다. 그 강등은 무증상이다 — `convergence_engine._parse_groups` 가 적대 모드가 꺼진 다중 출처 그룹을 큐에 넣지 않고 즉시 `full-consensus` 로 확정하므로, 세 워커의 처방이 서로 달라도 합의로 기록된다. 설정이 없으면 기본값으로 떨어지지 않고 실패한다. 여기서 조용히 넘어가면 검증 자체가 사라지고 산출물은 정상으로 읽힌다. """ path = task_manifest_path(authority.project_root, authority.payload) payload = load_owned_json_object(path) config = payload.get("convergence") if not isinstance(config, Mapping): raise ConvergenceContractError( f"task manifest carries no convergence config: {path}" ) return config def _prepare_groups(args: argparse.Namespace) -> tuple[str, Path]: authority = validated_run_authority(args.run_manifest) manifest = authority.payload groups = _parse_grouping_markdown(args.input) worker_ids = list(dict.fromkeys( source["worker"] for group in groups for source in group["sourceItems"] )) output = authority.run_dir / "state" / _canonical_run_artifact_name( "convergence-groups", authority.task_type, authority.state_sequence ) config = _convergence_config(authority) workers = [_group_worker(manifest, worker) for worker in worker_ids] source_refs = [worker["sourceRoleExecutionRef"] for worker in workers] if len(source_refs) != len(set(source_refs)): raise ConvergenceContractError("duplicate sourceRoleExecutionRef") payload = {"schemaVersion": "2.0", "executionIdentityVersion": 2, "taskKey": manifest.get("taskKey"), "runManifestPath": manifest.get("runManifestPath"), "config": {"enabled": bool(config.get("enabled", True)), "adversarial": bool(config.get("adversarial", False)), "maxRounds": int(config.get("maxRounds", 2)), "effectiveMaxRounds": int(config.get("effectiveMaxRounds", 2)), "verificationMode": str(config.get("verificationMode", "lightweight"))}, "workers": workers, "groups": groups} schema_errors = validate_schema( payload, load_named_schema("convergence-groups-v2.0.schema.json") ) if schema_errors: raise ConvergenceContractError( "invalid convergence groups: " + "; ".join(schema_errors) ) seed_working_state( payload, execution_manifest=authority.execution_manifest ) write_json_atomic(output, payload) return "prepared", output def _execute(args: argparse.Namespace) -> tuple[str, Path]: operations: dict[str, Any] = { "prepare-groups": _prepare_groups, "seed": _seed, "plan-round": _plan_round, "collect-results": _collect_results, "apply-round": _apply_round, "apply-critic-gaps": _apply_critic_gaps, "finalize": _finalize, "validate": _validate, } return operations[args.operation](args) def main(argv: list[str] | None = None) -> int: args = _parser().parse_args(argv) if args.operation == "example": print( json.dumps( _EXAMPLES[args.kind], ensure_ascii=False, sort_keys=True, indent=2, ) ) return 0 try: action, path = _execute(args) except (ConvergenceContractError, VerdictBlockError, json.JSONDecodeError, ValueError) as exc: print(f"error: {exc}", file=sys.stderr) return 2 except OSError as exc: print(f"error: {exc}", file=sys.stderr) return 1 if args.operation == "prepare-groups": print("Convergence groups\n" + line("Status", "ready") + line("Action", action), end="") else: _emit(args.operation, action, path) return 0 if __name__ == "__main__": raise SystemExit(main())