diff --git a/docs/engineering/ENGINEERING_REPORTING.md b/docs/engineering/ENGINEERING_REPORTING.md index bd31b220..deb286de 100644 --- a/docs/engineering/ENGINEERING_REPORTING.md +++ b/docs/engineering/ENGINEERING_REPORTING.md @@ -49,9 +49,11 @@ Every terminal report includes these derived sections: - **Statistics Projection** — Mission, execution, Engineering Action and runtime counts are separately scoped. An Engineering execution never implies Mission completion. -- **Deliverable Answer** — an explicit `YES`/`NO`, `PASS`/`FAIL` and `GO`/ - `NO-GO` answer when the prompt requests one, derived from the persisted - terminal checkpoint. +- **Deliverable Answer** — three separately scoped facts: technical delivery + from the terminal checkpoint, EP Run Qualification from the persisted + qualification snapshot, and Forge Mission/autonomy acceptance only when an + owning acceptance source exists. Prompt words such as `PASS` or `GO` never + create an acceptance outcome. `COMPLETE` is technical completion only. - **Commit Strategy** — Genesis Local Commit, Managed Pull Request, Managed Merge, Finalization or managed execution evidence, including the applicable commit and PR evidence. @@ -127,8 +129,11 @@ Engineering Platform consumes only its declared, immutable audit metadata: Producer ID, Type, Version, Correlation ID, optional Mission ID and Engineering Action ID, plus Execution Constraint Version. Reports expose those values in a **Producer** section without exposing Forge implementation details. When a -legacy prompt has no Producer metadata, the report records `HUMAN` and -`legacy`. Producer metadata never changes execution behaviour. +supported legacy run has positively established absence of a modern +submission binding, the report records prompt metadata as limited +`LEGACY_PROMPT_METADATA_ONLY` provenance. Missing, unreadable, corrupt or +identity-conflicting modern evidence fails closed and is never replaced with +prompt metadata. Producer metadata never changes execution behaviour. The report is execution evidence, not Forge Decision Evidence. It does not interpret, recreate or recommend Missions, Runtime Prompts, Runtime Instances @@ -149,6 +154,16 @@ and separately names its canonical root submission. Attempt-scoped immutable constraints are loaded through the persisted run-to-attempt binding. This prevents a canonical root envelope from hiding the action intent, Runtime Prompt or other Forge execution provenance that applied to the actual run. +The attempt, canonical root, retry parent and available qualification lineage +must agree. A historical parent without a local attempt is accepted only when +CENTRAL explicitly binds that parent run to the current retry and its Producer +correlation. A local retry read without its attempt-scoped source is therefore +unavailable, not silently replaced by the root envelope. +An absent expected attempt link, invalid stored constraints, unavailable +CENTRAL read or Producer/root identity conflict is reported with a safe +submission-evidence diagnostic code and prevents publication of a misleading +authoritative report. Raw constraints and prompts are not included in that +diagnostic. ## Forge Mission Recommendation Handoff diff --git a/src/engineering_platform/execution_reporting.py b/src/engineering_platform/execution_reporting.py index 7a78f868..6a554d03 100644 --- a/src/engineering_platform/execution_reporting.py +++ b/src/engineering_platform/execution_reporting.py @@ -25,7 +25,17 @@ from .providers import GitProvider from .qualification import latest_qualification from .recommendation_handoff import ForgeGovernanceHandoff, report_lines as recommendation_handoff_report_lines -from .storage import EngineeringStorageError, load_readiness_evaluation, load_run_qualification_snapshot, load_submission_attempt_for_run, load_submission_for_run, load_run_lineage, load_validation_context +from .storage import ( + EngineeringStorageError, + SUBMISSION_EVIDENCE_LEGACY_ABSENT, + SUBMISSION_EVIDENCE_VALID_MODERN, + load_readiness_evaluation, + load_run_qualification_snapshot, + load_submission_for_run, + load_run_lineage, + load_validation_context, + resolve_submission_attempt_evidence, +) from .telemetry_contract import run_telemetry_snapshot from .provider_recovery import load_recovery_state from .execution_activity import build_terminal_activity_summary, persist_terminal_activity_summary, terminal_activity_summary @@ -134,16 +144,50 @@ def _objective_requirements(objective: str) -> tuple[str, ...]: return (first,) -def _deliverable_answer(objective: str, state: TransactionState) -> str: - """Answer explicit binary delivery requests from the persisted terminal state.""" - requested = re.search(r"\bYES\b|\bPASS\b|\bGO\b|\bNO-GO\b", objective, re.IGNORECASE) - if not requested: - return "Not explicitly requested by the prompt." - if state.phase == "COMPLETE": - return "YES / PASS / GO — the persisted terminal checkpoint is COMPLETE." - if state.phase == "BLOCKED": - return "NO / FAIL / NO-GO — the persisted terminal checkpoint is BLOCKED." - return "NO / FAIL / NO-GO — the persisted terminal checkpoint is FAILED." +def _outcome_projection( + state: TransactionState, + qualification_snapshot: Mapping[str, object] | None, +) -> dict[str, object]: + """Separate technical completion, EP qualification and external acceptance.""" + snapshot = qualification_snapshot if isinstance(qualification_snapshot, Mapping) else {} + conflicts = snapshot.get("projection_conflicts") + conflict_values = ( + tuple(str(value) for value in conflicts if isinstance(value, str) and value) + if isinstance(conflicts, list) else () + ) + persisted_qualification = snapshot.get("run_qualification") + ep_qualification = ( + "EVIDENCE_CONFLICT" + if conflict_values else + str(persisted_qualification) + if persisted_qualification in {"QUALIFIED", "NOT_QUALIFIED"} else + "UNAVAILABLE" + ) + return { + "technical_delivery": state.phase, + "ep_run_qualification": ep_qualification, + "ep_run_qualification_persisted_outcome": ( + str(persisted_qualification) + if persisted_qualification in {"QUALIFIED", "NOT_QUALIFIED"} else None + ), + "ep_run_qualification_snapshot": snapshot.get("qualification_snapshot_id"), + "ep_run_qualification_conflicts": list(conflict_values), + "mission_autonomy_acceptance": "NOT_ESTABLISHED_BY_EP", + "mission_autonomy_acceptance_source": None, + } + + +def _outcome_lines(outcome: Mapping[str, object]) -> tuple[str, ...]: + conflicts = outcome.get("ep_run_qualification_conflicts") + conflict_text = ", ".join(str(value) for value in conflicts) if isinstance(conflicts, list) and conflicts else "NONE" + return ( + f"- Technical Delivery: `{outcome['technical_delivery']}`", + f"- EP Run Qualification: `{outcome['ep_run_qualification']}`", + f"- EP Qualification Snapshot: `{outcome.get('ep_run_qualification_snapshot') or 'UNAVAILABLE'}`", + f"- EP Qualification Conflicts: `{conflict_text}`", + f"- Mission / Autonomy Acceptance: `{outcome['mission_autonomy_acceptance']}`", + "- Interpretation: technical completion does not establish EP qualification or Forge Mission acceptance.", + ) def _next_action_message(action: str) -> str: return { @@ -161,28 +205,39 @@ def _format_terminal_report(state: TransactionState) -> str: def _persisted_producer_submission( root: Path, state: TransactionState, fallback_prompt: str, *, central_database: Path | None = None, -) -> tuple[ProducerMetadata, dict[str, object] | None]: +) -> tuple[ProducerMetadata, dict[str, object] | None, str]: """Use immutable Producer submission evidence before legacy prompt compatibility.""" - try: - submission = load_submission_attempt_for_run( - root, state.run_id, central_database=central_database, + evidence = resolve_submission_attempt_evidence( + root, state.run_id, central_database=central_database, + ) + if evidence.status == SUBMISSION_EVIDENCE_LEGACY_ABSENT: + return parse_producer_metadata(fallback_prompt), None, evidence.provenance + if evidence.status != SUBMISSION_EVIDENCE_VALID_MODERN or evidence.submission is None: + raise EngineeringStorageError( + f"{evidence.diagnostic_code or 'SUBMISSION_EVIDENCE_UNAVAILABLE'}: run_id={state.run_id}" ) - except EngineeringStorageError: - submission = None - if submission is None: - return parse_producer_metadata(fallback_prompt), None + submission = evidence.submission return ProducerMetadata( producer_id=str(submission["producer_id"]), producer_type=str(submission["producer_type"]), producer_version=submission.get("producer_version") if isinstance(submission.get("producer_version"), str) else None, correlation_id=submission.get("correlation_id") if isinstance(submission.get("correlation_id"), str) else None, mission_id=submission.get("mission_id") if isinstance(submission.get("mission_id"), str) else None, engineering_action_id=submission.get("engineering_action_id") if isinstance(submission.get("engineering_action_id"), str) else None, - ), submission + ), submission, evidence.provenance def _producer_submission_contract_lines( submission: dict[str, object] | None, state: TransactionState, root: Path | None = None, + *, evidence_provenance: str = "UNAVAILABLE", ) -> tuple[str, ...]: + legacy_fallback = submission is None and evidence_provenance == "LEGACY_PROMPT_METADATA" + submission_label = submission.get("submission_id") if submission else "legacy" if legacy_fallback else "UNAVAILABLE" + contract_label = submission.get("contract_version") if submission else "legacy prompt" if legacy_fallback else "UNAVAILABLE" + submission_status = ( + "PERSISTED_IMMUTABLY" if submission else + "LEGACY_PROMPT_METADATA_ONLY" if legacy_fallback else + "UNAVAILABLE" + ) context = submission.get("execution_context") if isinstance(submission, dict) else None profile = context.get("validation_profile") if isinstance(context, dict) else None validation_context = None @@ -204,15 +259,16 @@ def _producer_submission_contract_lines( ) return ( "## Producer Submission Contract", - f"- Submission ID: `{submission.get('submission_id') if submission else 'legacy'}`", + f"- Submission ID: `{submission_label}`", *lineage, - f"- Contract Version: `{submission.get('contract_version') if submission else 'legacy prompt'}`", - "- Submission Status: `PERSISTED_IMMUTABLY`", + f"- Contract Version: `{contract_label}`", + f"- Submission Status: `{submission_status}`", + f"- Submission Evidence Provenance: `{evidence_provenance}`", "", "## Execution Context Contract", f"- Execution Context Status: `{'SUPPLIED_BY_PRODUCER' if isinstance(context, dict) else 'NOT_SUPPLIED_BY_PRODUCER'}`", f"- Execution Context Version: `{submission.get('execution_context_version') if isinstance(context, dict) else 'not supplied'}`", - f"- Execution Context Reference: `execution-submission:{submission.get('submission_id') if isinstance(context, dict) else 'legacy'}`", + f"- Execution Context Reference: `execution-submission:{submission.get('submission_id') if isinstance(context, dict) else 'legacy' if legacy_fallback else 'UNAVAILABLE'}`", f"- Action Intent: `{context.get('action_intent', 'not supplied') if isinstance(context, dict) else 'not supplied'}`", f"- Validation Profile: `{profile.get('tier', 'not supplied') if isinstance(profile, dict) else 'not supplied'}`", f"- Validation Profile Source: `{profile_source}`", @@ -735,12 +791,15 @@ def _qualification_projection( state: TransactionState, qualification_status: object, runtime_provider: str, + outcome: Mapping[str, object], ) -> tuple[str, ...]: """Keep execution, qualification, runtime and governance outcomes distinct.""" validation = "recorded" if state.validation_evidence else "not recorded" return ( f"- Execution Status: `{state.phase}`", f"- Platform Qualification Status: `{qualification_status or 'not recorded'}`", + f"- EP Run Qualification: `{outcome['ep_run_qualification']}`", + f"- Mission / Autonomy Acceptance: `{outcome['mission_autonomy_acceptance']}`", f"- Runtime Status: `{'reported' if runtime_provider != 'unavailable' else 'not reported'}`", f"- Validation Status: `{validation}`", "- Governance Status: see the Forge Governance Handoff projection above.", @@ -770,19 +829,28 @@ def _execution_receipt_projection( state: TransactionState, producer: ProducerMetadata, submission: dict[str, object] | None = None, + qualification_snapshot: Mapping[str, object] | None = None, + *, + submission_evidence_resolved: bool = False, ) -> tuple[str, ...]: """Render receipt qualification fields only from the persisted terminal snapshot.""" - try: - snapshot = load_run_qualification_snapshot(root, state.run_id) or {} - except EngineeringStorageError: - snapshot = {} + if qualification_snapshot is not None: + snapshot = dict(qualification_snapshot) + else: + try: + snapshot = load_run_qualification_snapshot(root, state.run_id) or {} + except EngineeringStorageError: + snapshot = {"projection_conflicts": ["QUALIFICATION_SNAPSHOT_READ_FAILED"]} conflicts = snapshot.get("projection_conflicts", []) activity = terminal_activity_summary(root, state.run_id) activity_total = activity.get("activity", {}).get("overall_activity_total", "UNAVAILABLE") if isinstance(activity, dict) else "UNAVAILABLE" delivery_paths = activity.get("terminal_delivery_diff", {}).get("total_unique_changed_paths", "UNAVAILABLE") if isinstance(activity, dict) else "UNAVAILABLE" conflicts_text = ", ".join(conflicts) if isinstance(conflicts, list) and conflicts else "NONE" recovery = load_recovery_state(root, state.run_id) or {} - submission = submission or load_submission_for_run(root, state.run_id) or {} + if submission_evidence_resolved: + submission = submission or {} + else: + submission = submission or load_submission_for_run(root, state.run_id) or {} return ( f"- Receipt ID: `{state.run_id}`", "- Execution Host: `Engineering Platform`", @@ -825,14 +893,18 @@ def _decision_evidence_projection(producer: ProducerMetadata) -> tuple[str, ...] ) -def _evidence_summary(state: TransactionState, bundle: TerminalEvidenceBundle, objective: str) -> str: +def _evidence_summary( + state: TransactionState, + bundle: TerminalEvidenceBundle, + outcome: Mapping[str, object], +) -> str: """Return a compact, machine-readable summary derived only from report evidence.""" return json.dumps( { "repository_commit": bundle.target_commit, "implemented_components": [name for name, _ in _component_inventory(bundle)], "regression_coverage": [path for path in bundle.changed_files if path.startswith("tests/")], - "deliverable_answer": _deliverable_answer(objective, state), + "outcomes": dict(outcome), "commit_strategy": _commit_strategy(state, bundle)[0].removeprefix("- Strategy: `").removesuffix("`"), "execution_strategy": state.execution_mode, "repository_state": bundle.worktree_state, @@ -842,7 +914,13 @@ def _evidence_summary(state: TransactionState, bundle: TerminalEvidenceBundle, o ) -def report_consistency_errors(body: str, state: TransactionState, bundle: TerminalEvidenceBundle, objective: str) -> tuple[str, ...]: +def report_consistency_errors( + body: str, + state: TransactionState, + bundle: TerminalEvidenceBundle, + objective: str, + outcome: Mapping[str, object] | None = None, +) -> tuple[str, ...]: """Validate mandatory Evidence 2.0 sections before a report is published.""" required = ( "## Component Inventory", @@ -862,8 +940,35 @@ def report_consistency_errors(body: str, state: TransactionState, bundle: Termin errors = [f"missing required section: {section}" for section in required if section not in body] if "Implemented Components:\n\nnone recorded" in body: errors.append("component inventory is missing") - if re.search(r"\bYES\b|\bPASS\b|\bGO\b|\bNO-GO\b", objective, re.IGNORECASE) and _deliverable_answer(objective, state) not in body: - errors.append("explicit deliverable answer is missing") + expected_outcome = dict(outcome or _outcome_projection(state, None)) + for line in _outcome_lines(expected_outcome): + if line not in body: + errors.append(f"outcome projection is missing: {line.removeprefix('- ')}") + answer_section = re.search( + r"^## Deliverable Answer\s*$\n(?P.*?)(?=^## |\Z)", + body, + re.MULTILINE | re.DOTALL, + ) + if answer_section is not None and re.search( + r"(?i)\b(?:YES|PASS|GO|NO-GO)\b", + answer_section.group("body"), + ): + errors.append("deliverable answer contains an unscoped acceptance conclusion") + summary = re.search( + r"^## Engineering Evidence Summary\s*$\n```json\n(?P.*?)\n```", + body, + re.MULTILINE | re.DOTALL, + ) + if summary is None: + errors.append("Engineering Evidence Summary machine projection is missing") + else: + try: + summary_payload = json.loads(summary.group("body")) + except json.JSONDecodeError: + errors.append("Engineering Evidence Summary is invalid JSON") + else: + if summary_payload.get("outcomes") != expected_outcome: + errors.append("Engineering Evidence Summary outcome projection conflicts") if bundle.target_commit not in body: errors.append("repository commit is missing") if state.phase == "COMPLETE" and "## Evidence Bundle" not in body: @@ -1043,11 +1148,7 @@ def terminal_report_matches_state(body: str, state: TransactionState) -> bool: def corrected_terminal_report(state: TransactionState) -> str: """Generate a minimal replacement when richer report assembly is inconsistent.""" - try: - producer_prompt = Path(state.prompt_path).read_text(encoding="utf-8") - except OSError: - producer_prompt = "" - producer = parse_producer_metadata(producer_prompt) + producer = ProducerMetadata(producer_id="unavailable", producer_type="UNKNOWN") submission = None return "\n".join( ( @@ -1065,7 +1166,9 @@ def corrected_terminal_report(state: TransactionState) -> str: f"- Engineering Action ID: `{producer.engineering_action_id or 'not supplied'}`", f"- Execution Constraint Version: `{producer.execution_constraint_version or 'not supplied'}`", "", - *_producer_submission_contract_lines(submission, state), + *_producer_submission_contract_lines( + submission, state, evidence_provenance="UNAVAILABLE_CORRECTED_REPORT", + ), "## Execution Target Identity", f"- Execution Host Repository: `{state.repository}`", f"- Execution Mode: `{state.execution_mode}`", @@ -1080,6 +1183,9 @@ def corrected_terminal_report(state: TransactionState) -> str: "## Engineering Outcome", format_terminal_management_summary(state), "", + "## Deliverable Answer", + *_outcome_lines(_outcome_projection(state, None)), + "", *_retry_relationship(state), "## Reviewer Findings", "No reviewer findings were retained. Reviewer observations are advisory initial observations only.", @@ -1201,16 +1307,25 @@ def _format_engineering_outcome(state: TransactionState) -> str: ) -def _managed_autonomy_projection(root: Path, state: TransactionState, bundle: TerminalEvidenceBundle, reviewer_records: tuple[dict[str, object], ...]) -> tuple[str, ...]: +def _managed_autonomy_projection( + root: Path, + state: TransactionState, + bundle: TerminalEvidenceBundle, + reviewer_records: tuple[dict[str, object], ...], + *, + submission: Mapping[str, object] | None = None, + submission_evidence_resolved: bool = False, +) -> tuple[tuple[str, ...], dict[str, object]]: """Project canonical evidence only; legacy runs intentionally fail closed.""" try: lineage = load_run_lineage(root, state.run_id) except EngineeringStorageError: lineage = None - try: - submission = load_submission_for_run(root, state.run_id) - except EngineeringStorageError: - submission = None + if not submission_evidence_resolved: + try: + submission = load_submission_for_run(root, state.run_id) + except EngineeringStorageError: + submission = None snapshot = managed_autonomy_snapshot( root, run_id=state.run_id, execution_outcome=state.phase, implementation_pr=state.implementation_pull_request, finalization_pr=state.finalization_pull_request, @@ -1247,7 +1362,7 @@ def pr_lines(role: str) -> tuple[str, ...]: f" - Required Checks Evidence Reference: `{item.get('evidence_ref', 'UNAVAILABLE')}`", f" - Historical Check Observations: `{item.get('historical_observation_count', 'UNAVAILABLE')}`", ) - return ( + lines = ( "## Run Qualification", f"- Execution: `{snapshot['terminal_execution_state']}`", f"- Action Intent: `{snapshot['action_intent']}`", @@ -1282,6 +1397,7 @@ def pr_lines(role: str) -> tuple[str, ...]: f"- Qualification Reasons: `{', '.join(snapshot['qualification_failure_reasons']) or 'none'}`", "", ) + return lines, snapshot def generate_terminal_report( @@ -1299,7 +1415,6 @@ def generate_terminal_report( central_database.resolve().parent / "artifacts" / "reports" if central_database is not None else root / ".engineering" / "reports" ) - reports.mkdir(mode=0o700, parents=True, exist_ok=True) timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H-%M-%SZ") path = reports / f"{timestamp}_{state.run_id}.md" objective = "Objective unavailable because the prompt file is no longer local." @@ -1315,9 +1430,10 @@ def generate_terminal_report( "Submitted runtime prompt retained at the supplied prompt path; " "non-authoritative input." ) - producer, submission = _persisted_producer_submission( + producer, submission, submission_evidence_provenance = _persisted_producer_submission( root, state, objective, central_database=central_database, ) + reports.mkdir(mode=0o700, parents=True, exist_ok=True) raw_handoff = submission.get("forge_governance_handoff") if isinstance(submission, dict) else None handoff = ForgeGovernanceHandoff.from_snapshot(raw_handoff) if isinstance(raw_handoff, dict) else None manifest = manifest or EngineeringPlatformManifest.load( @@ -1346,6 +1462,12 @@ def generate_terminal_report( if isinstance(value, int) and not isinstance(value, bool) } bundle = collect_terminal_evidence(root, state) + managed_autonomy_lines, qualification_snapshot = _managed_autonomy_projection( + root, state, bundle, reviewer_records, + submission=submission, + submission_evidence_resolved=True, + ) + outcome = _outcome_projection(state, qualification_snapshot) activity_summary = persist_terminal_activity_summary( root, build_terminal_activity_summary(root, state, bundle) ) @@ -1605,7 +1727,10 @@ def duration_value(name: str) -> str: f"- Engineering Action ID: `{producer.engineering_action_id or 'not supplied'}`", f"- Execution Constraint Version: `{producer.execution_constraint_version or 'not supplied'}`", "", - *_producer_submission_contract_lines(submission, state, root), + *_producer_submission_contract_lines( + submission, state, root, + evidence_provenance=submission_evidence_provenance, + ), *_forge_execution_provenance_lines(submission), "## Execution Target Identity", "- Execution Host: `Engineering Platform`", @@ -1732,7 +1857,7 @@ def duration_value(name: str) -> str: "## Engineering Outcome", _format_engineering_outcome(state), "", - *_managed_autonomy_projection(root, state, bundle, reviewer_records), + *managed_autonomy_lines, *_format_assurance_reviews(state), "## Reviewer Findings", "Initial observations only. They are not final repository claims.", @@ -1752,19 +1877,26 @@ def duration_value(name: str) -> str: "", *recommendation_handoff_report_lines(handoff, state.phase), "## Qualification Projection", - *_qualification_projection(state, qualification_status, runtime_provider), + *_qualification_projection(state, qualification_status, runtime_provider, outcome), "", "## Runtime Projection", *_runtime_projection(state, producer, runtime_provider, reported_model), "", "## Execution Receipt Projection", - *_execution_receipt_projection(root, state, producer, submission), + *_execution_receipt_projection( + root, + state, + producer, + submission, + qualification_snapshot, + submission_evidence_resolved=True, + ), "", "## Decision Evidence Projection", *_decision_evidence_projection(producer), "", "## Deliverable Answer", - f"- Final Deliverable Answer: {_deliverable_answer(objective, state)}", + *_outcome_lines(outcome), "", "## Commit Strategy", *_commit_strategy(state, bundle), @@ -1787,7 +1919,7 @@ def duration_value(name: str) -> str: "", "## Engineering Evidence Summary", "```json", - _evidence_summary(state, bundle, objective), + _evidence_summary(state, bundle, outcome), "```", "", evidence_bundle, @@ -1830,6 +1962,8 @@ def duration_value(name: str) -> str: return ReportingCoordinator().deliver( path=path, body=body, - validate=lambda value: report_consistency_errors(value, state, bundle, objective), + validate=lambda value: report_consistency_errors( + value, state, bundle, objective, outcome, + ), terminal_matches=lambda value: terminal_report_matches_state(value, state), ) diff --git a/src/engineering_platform/storage.py b/src/engineering_platform/storage.py index 1fb1179b..9cefe991 100644 --- a/src/engineering_platform/storage.py +++ b/src/engineering_platform/storage.py @@ -9,6 +9,7 @@ from collections.abc import Callable, Iterator from contextlib import contextmanager +from dataclasses import dataclass from datetime import datetime, timezone import argparse import json @@ -39,6 +40,25 @@ class EngineeringStorageError(RuntimeError): """Raised when the local Engineering evidence database is unsafe to use.""" +SUBMISSION_EVIDENCE_VALID_MODERN = "VALID_MODERN" +SUBMISSION_EVIDENCE_LEGACY_ABSENT = "LEGACY_ABSENT" +SUBMISSION_EVIDENCE_MISSING_MODERN_BINDING = "MISSING_MODERN_BINDING" +SUBMISSION_EVIDENCE_STORAGE_UNAVAILABLE = "STORAGE_UNAVAILABLE" +SUBMISSION_EVIDENCE_CORRUPT = "CORRUPT" +SUBMISSION_EVIDENCE_IDENTITY_CONFLICT = "IDENTITY_CONFLICT" + + +@dataclass(frozen=True) +class SubmissionAttemptEvidence: + """Safe typed result for one run-bound submission evidence read.""" + + run_id: str + status: str + submission: dict[str, object] | None = None + diagnostic_code: str | None = None + provenance: str = "UNAVAILABLE" + + class _ClosingSQLiteConnection(sqlite3.Connection): """Close an ``open_storage`` connection when its outer context exits. @@ -2038,69 +2058,257 @@ def load_submission_for_run(root: Path, run_id: str, *, central_database: Path | } -def load_submission_attempt_for_run( +def resolve_submission_attempt_evidence( root: Path, run_id: str, *, central_database: Path | None = None, -) -> dict[str, object] | None: - """Load the concrete admitted submission attempt bound to one run. - - ``load_submission_for_run`` deliberately returns the immutable root - submission for a correlated retry. A terminal report also needs the - attempt identity and its admitted constraints; otherwise it can combine - the root submission ID with retry-scoped qualification evidence. Keep the - root metadata as provenance authority while overlaying only fields that - CENTRAL persisted immutably for the exact run-bound attempt. +) -> SubmissionAttemptEvidence: + """Resolve modern, legacy and invalid submission evidence without fallback. + + A CENTRAL-bound run is modern by construction and therefore requires its + concrete attempt link. A local run without either a submission record or + qualification-lineage record is the only supported legacy absence. The + result contains safe codes and identities only; corrupt payloads are never + copied into diagnostics. """ - root_submission = load_submission_for_run( - root, run_id, central_database=central_database, - ) - if root_submission is None or central_database is None: - return root_submission - connection = _evidence_connection(root, central_database) try: - row = connection.execute( - "SELECT accepted.submission_id,accepted.producer_id,accepted.producer_type," - "accepted.producer_version,accepted.correlation_id,accepted.mission_id," - "accepted.engineering_action_id,accepted.constraints,attempt.canonical_submission_id " - "FROM execution_submission_attempt_links AS link " - "JOIN execution_submission_attempts AS attempt " - "ON attempt.submission_id=link.submission_id " - "JOIN ep_submissions AS accepted " - "ON accepted.submission_id=link.submission_id " - "WHERE link.run_id=?", - (run_id,), - ).fetchone() - finally: - connection.close() + connection = _evidence_connection(root, central_database) + try: + attempt_row = connection.execute( + "SELECT attempt.submission_id,attempt.canonical_submission_id," + "attempt.retry_parent_submission_id,parent.canonical_submission_id," + "parent_link.run_id " + "FROM execution_submission_attempt_links AS link " + "JOIN execution_submission_attempts AS attempt " + "ON attempt.submission_id=link.submission_id " + "LEFT JOIN execution_submission_attempts AS parent " + "ON parent.submission_id=attempt.retry_parent_submission_id " + "LEFT JOIN execution_submission_attempt_links AS parent_link " + "ON parent_link.submission_id=parent.submission_id " + "WHERE link.run_id=?", + (run_id,), + ).fetchone() + lineage_row = connection.execute( + "SELECT submission_id,fresh_submission,retry_parent_run_id,resume_parent_run_id " + "FROM execution_run_qualification_context WHERE run_id=?", + (run_id,), + ).fetchone() + finally: + connection.close() + except (EngineeringStorageError, sqlite3.DatabaseError, OSError): + return SubmissionAttemptEvidence( + run_id=run_id, + status=SUBMISSION_EVIDENCE_STORAGE_UNAVAILABLE, + diagnostic_code="SUBMISSION_EVIDENCE_ATTEMPT_READ_FAILED", + ) + try: + root_submission = load_submission_for_run( + root, run_id, central_database=central_database, + ) + except (EngineeringStorageError, sqlite3.DatabaseError, OSError) as error: + unavailable = "unavailable" in str(error).casefold() or isinstance(error, (sqlite3.DatabaseError, OSError)) + return SubmissionAttemptEvidence( + run_id=run_id, + status=(SUBMISSION_EVIDENCE_STORAGE_UNAVAILABLE if unavailable else SUBMISSION_EVIDENCE_CORRUPT), + diagnostic_code=( + "SUBMISSION_EVIDENCE_STORAGE_READ_FAILED" + if unavailable else "SUBMISSION_EVIDENCE_ROOT_CORRUPT" + ), + ) + if attempt_row is None: + if root_submission is not None or central_database is not None or lineage_row is not None: + return SubmissionAttemptEvidence( + run_id=run_id, + status=SUBMISSION_EVIDENCE_MISSING_MODERN_BINDING, + diagnostic_code="SUBMISSION_EVIDENCE_ATTEMPT_BINDING_MISSING", + ) + return SubmissionAttemptEvidence( + run_id=run_id, + status=SUBMISSION_EVIDENCE_LEGACY_ABSENT, + diagnostic_code="SUBMISSION_EVIDENCE_LEGACY_ABSENT", + provenance="LEGACY_PROMPT_METADATA", + ) + attempt_id = str(attempt_row[0]) + canonical_id = str(attempt_row[1]) + retry_parent_id = attempt_row[2] + parent_canonical_id = attempt_row[3] + parent_run_id = attempt_row[4] + if root_submission is None: + return SubmissionAttemptEvidence( + run_id=run_id, + status=SUBMISSION_EVIDENCE_MISSING_MODERN_BINDING, + diagnostic_code="SUBMISSION_EVIDENCE_ROOT_BINDING_MISSING", + ) + if root_submission.get("submission_id") != canonical_id: + return SubmissionAttemptEvidence( + run_id=run_id, + status=SUBMISSION_EVIDENCE_IDENTITY_CONFLICT, + diagnostic_code="SUBMISSION_EVIDENCE_ROOT_IDENTITY_CONFLICT", + ) + if retry_parent_id is not None and ( + parent_canonical_id is None + or str(parent_canonical_id) != canonical_id + or parent_run_id is None + ): + return SubmissionAttemptEvidence( + run_id=run_id, + status=SUBMISSION_EVIDENCE_IDENTITY_CONFLICT, + diagnostic_code="SUBMISSION_EVIDENCE_RETRY_PARENT_CONFLICT", + ) + if lineage_row is not None: + lineage_submission_id = str(lineage_row[0]) + lineage_fresh = bool(lineage_row[1]) + lineage_retry_run = lineage_row[2] + lineage_resume_run = lineage_row[3] + lineage_conflicts = lineage_submission_id != attempt_id + if retry_parent_id is not None: + historical_parent_verified = False + if lineage_retry_run != parent_run_id and central_database is not None: + try: + connection = _evidence_connection(root, central_database) + try: + historical_parent = connection.execute( + "SELECT submission.producer_id,submission.producer_type," + "submission.correlation_id,dispatch.operator_resolution," + "historical_attempt.submission_id " + "FROM ep_parity_lifecycle_dispatches AS dispatch " + "JOIN ep_submissions AS submission " + "ON submission.submission_id=dispatch.submission_id " + "LEFT JOIN execution_submission_attempts AS historical_attempt " + "ON historical_attempt.submission_id=dispatch.submission_id " + "WHERE dispatch.run_id=? AND dispatch.resolution_submission_id=?", + (lineage_retry_run, attempt_id), + ).fetchone() + finally: + connection.close() + except (EngineeringStorageError, sqlite3.DatabaseError, OSError): + historical_parent = None + historical_parent_verified = bool( + historical_parent is not None + and root_submission.get("correlation_id") is not None + and historical_parent[0] == root_submission.get("producer_id") + and historical_parent[1] == root_submission.get("producer_type") + and historical_parent[2] == root_submission.get("correlation_id") + and historical_parent[3] == "RETRIED" + and historical_parent[4] is None + ) + lineage_conflicts = lineage_conflicts or ( + lineage_fresh + or lineage_resume_run is not None + or ( + lineage_retry_run != parent_run_id + and not historical_parent_verified + ) + ) + elif lineage_retry_run is not None: + lineage_conflicts = True + if lineage_conflicts: + return SubmissionAttemptEvidence( + run_id=run_id, + status=SUBMISSION_EVIDENCE_IDENTITY_CONFLICT, + diagnostic_code="SUBMISSION_EVIDENCE_QUALIFICATION_LINEAGE_CONFLICT", + ) + if central_database is None: + if attempt_id != canonical_id: + return SubmissionAttemptEvidence( + run_id=run_id, + status=SUBMISSION_EVIDENCE_STORAGE_UNAVAILABLE, + diagnostic_code="SUBMISSION_EVIDENCE_ATTEMPT_SOURCE_UNAVAILABLE", + ) + return SubmissionAttemptEvidence( + run_id=run_id, + status=SUBMISSION_EVIDENCE_VALID_MODERN, + submission=root_submission, + provenance="IMMUTABLE_SUBMISSION", + ) + try: + connection = _evidence_connection(root, central_database) + try: + row = connection.execute( + "SELECT accepted.submission_id,accepted.producer_id,accepted.producer_type," + "accepted.producer_version,accepted.correlation_id,accepted.mission_id," + "accepted.engineering_action_id,accepted.constraints,attempt.canonical_submission_id " + "FROM execution_submission_attempt_links AS link " + "JOIN execution_submission_attempts AS attempt " + "ON attempt.submission_id=link.submission_id " + "JOIN ep_submissions AS accepted " + "ON accepted.submission_id=link.submission_id " + "WHERE link.run_id=?", + (run_id,), + ).fetchone() + finally: + connection.close() + except (EngineeringStorageError, sqlite3.DatabaseError, OSError): + return SubmissionAttemptEvidence( + run_id=run_id, + status=SUBMISSION_EVIDENCE_STORAGE_UNAVAILABLE, + diagnostic_code="SUBMISSION_EVIDENCE_ATTEMPT_READ_FAILED", + ) if row is None: - return root_submission + return SubmissionAttemptEvidence( + run_id=run_id, + status=SUBMISSION_EVIDENCE_MISSING_MODERN_BINDING, + diagnostic_code="SUBMISSION_EVIDENCE_ATTEMPT_BINDING_MISSING", + ) if ( - row[8] != root_submission["submission_id"] + row[0] != attempt_id + or row[8] != canonical_id or row[1] != root_submission["producer_id"] or row[2] != root_submission["producer_type"] ): - raise EngineeringStorageError( - "Persisted submission attempt conflicts with canonical Producer provenance." + return SubmissionAttemptEvidence( + run_id=run_id, + status=SUBMISSION_EVIDENCE_IDENTITY_CONFLICT, + diagnostic_code="SUBMISSION_EVIDENCE_IDENTITY_CONFLICT", ) try: constraints = json.loads(row[7]) - except (TypeError, json.JSONDecodeError) as error: - raise EngineeringStorageError( - "Persisted submission attempt constraints are corrupt." - ) from error + except (TypeError, json.JSONDecodeError): + return SubmissionAttemptEvidence( + run_id=run_id, + status=SUBMISSION_EVIDENCE_CORRUPT, + diagnostic_code="SUBMISSION_EVIDENCE_ATTEMPT_CONSTRAINTS_CORRUPT", + ) if not isinstance(constraints, dict): - raise EngineeringStorageError( - "Persisted submission attempt constraints are invalid." + return SubmissionAttemptEvidence( + run_id=run_id, + status=SUBMISSION_EVIDENCE_CORRUPT, + diagnostic_code="SUBMISSION_EVIDENCE_ATTEMPT_CONSTRAINTS_INVALID", ) - return { - **root_submission, - "canonical_submission_id": root_submission["submission_id"], - "submission_id": row[0], - "producer_version": row[3], - "correlation_id": row[4], - "mission_id": row[5], - "engineering_action_id": row[6], - "constraints": constraints, - } + return SubmissionAttemptEvidence( + run_id=run_id, + status=SUBMISSION_EVIDENCE_VALID_MODERN, + submission={ + **root_submission, + "canonical_submission_id": canonical_id, + "submission_id": row[0], + "producer_version": row[3], + "correlation_id": row[4], + "mission_id": row[5], + "engineering_action_id": row[6], + "constraints": constraints, + }, + provenance="IMMUTABLE_RUN_BOUND_ATTEMPT", + ) + + +def load_submission_attempt_for_run( + root: Path, run_id: str, *, central_database: Path | None = None, +) -> dict[str, object] | None: + """Load valid attempt evidence while preserving legacy compatibility. + + Invalid or unavailable modern evidence is never represented as absence. + Callers that need the full state matrix use + :func:`resolve_submission_attempt_evidence`. + """ + result = resolve_submission_attempt_evidence( + root, run_id, central_database=central_database, + ) + if result.status == SUBMISSION_EVIDENCE_VALID_MODERN: + return result.submission + if result.status == SUBMISSION_EVIDENCE_LEGACY_ABSENT: + return None + raise EngineeringStorageError( + f"{result.diagnostic_code or 'SUBMISSION_EVIDENCE_UNAVAILABLE'}: run_id={run_id}" + ) def load_run_lineage(root: Path, run_id: str) -> dict[str, object] | None: diff --git a/tests/engineering/test_execution_host.py b/tests/engineering/test_execution_host.py index 37e6d198..25bd74df 100644 --- a/tests/engineering/test_execution_host.py +++ b/tests/engineering/test_execution_host.py @@ -5646,14 +5646,19 @@ def test_engineering_evidence_2_report_is_self_validating_and_traceable(self) -> "## Engineering Evidence Summary", ): self.assertIn(section, body) - self.assertIn("YES / PASS / GO", body) + self.assertIn("- Technical Delivery: `COMPLETE`", body) + self.assertIn("- EP Run Qualification: `NOT_QUALIFIED`", body) + self.assertIn("- Mission / Autonomy Acceptance: `NOT_ESTABLISHED_BY_EP`", body) + self.assertNotIn("Final Deliverable Answer: YES / PASS / GO", body) self.assertIn("Requirement: Produce a self-validating report.", body) self.assertIn("Runtime evidence: run `evidence-2`; execution mode `MANAGED`.", body) self.assertIn("Execution Status: `COMPLETE`", body) self.assertIn("Receipt ID: `evidence-2`", body) self.assertIn("### Mission Statistics", body) self.assertIn("Executed Validation Command: `Documentation validation`", body) - self.assertIn('"deliverable_answer": "YES / PASS / GO', body) + self.assertIn('"technical_delivery": "COMPLETE"', body) + self.assertIn('"ep_run_qualification": "NOT_QUALIFIED"', body) + self.assertIn('"mission_autonomy_acceptance": "NOT_ESTABLISHED_BY_EP"', body) def test_component_inventory_is_derived_from_implementation_evidence(self) -> None: subprocess.run(("git", "init", "-b", "main", str(self.root)), check=True, capture_output=True) @@ -5705,7 +5710,7 @@ def test_report_consistency_validation_rejects_missing_evidence_2_sections(self) self.assertIn("missing required section: ## Deliverable Projection", errors) self.assertIn("missing required section: ## Qualification Projection", errors) self.assertIn("missing required section: ## Statistics Projection", errors) - self.assertIn("explicit deliverable answer is missing", errors) + self.assertTrue(any(error.startswith("outcome projection is missing:") for error in errors)) def test_report_consistency_rejects_contradictory_fresh_submission(self) -> None: state = TransactionState("lineage-conflict", "pcvantol/djconnect", str(self.prompt), "COMPLETE", terminal=True) diff --git a/tests/engineering/test_reporting_integrity.py b/tests/engineering/test_reporting_integrity.py new file mode 100644 index 00000000..7ac78794 --- /dev/null +++ b/tests/engineering/test_reporting_integrity.py @@ -0,0 +1,600 @@ +"""Regressions for scoped reporting outcomes and submission evidence integrity.""" + +from __future__ import annotations + +import json +from pathlib import Path +import re +import tempfile +import unittest +from unittest.mock import patch + +from engineering_platform.agent_state import TransactionState +from engineering_platform.execution_reporting import ( + _outcome_projection, + _persisted_producer_submission, + collect_terminal_evidence, + generate_terminal_report, + report_consistency_errors, +) +from engineering_platform.managed_autonomy import terminal_snapshot +from engineering_platform.storage import ( + EngineeringStorageError, + SUBMISSION_EVIDENCE_CORRUPT, + SUBMISSION_EVIDENCE_IDENTITY_CONFLICT, + SUBMISSION_EVIDENCE_LEGACY_ABSENT, + SUBMISSION_EVIDENCE_MISSING_MODERN_BINDING, + SUBMISSION_EVIDENCE_STORAGE_UNAVAILABLE, + SUBMISSION_EVIDENCE_VALID_MODERN, + database_path, + record_run_qualification_context, + record_run_qualification_snapshot, + record_submission, + resolve_submission_attempt_evidence, + sqlite_connection, +) + + +class ReportingIntegrityTest(unittest.TestCase): + def setUp(self) -> None: + self.temporary = tempfile.TemporaryDirectory() + self.root = Path(self.temporary.name) + self.prompt = self.root / "prompt.md" + self.prompt.write_text("# Bounded objective\n", encoding="utf-8") + + def tearDown(self) -> None: + self.temporary.cleanup() + + def _state(self, run_id: str, phase: str = "COMPLETE", **values: object) -> TransactionState: + return TransactionState( + run_id, + "pcvantol/engineering-platform", + str(self.prompt), + phase, + terminal=True, + **values, + ) + + def _record_submission( + self, + *, + submission_id: str, + run_id: str, + producer_id: str = "forge", + correlation_id: str = "correlation-1", + retry_parent_submission_id: str | None = None, + constraints: dict[str, object] | None = None, + ) -> None: + record_submission( + self.root, + submission_id=submission_id, + producer_id=producer_id, + producer_type="FORGE", + producer_version="1.2", + contract_version="1.2", + correlation_id=correlation_id, + mission_id="MISSION-0003", + engineering_action_id=f"ACTION-{submission_id}", + prompt_content="bounded modern submission", + prompt_metadata={"constraints": constraints or {"scope": submission_id}}, + target_identity={"repository": "pcvantol/engineering-platform"}, + original_envelope={"kind": "synthetic"}, + received_at="2026-09-17T10:00:00+00:00", + link_run_id=run_id, + retry_parent_submission_id=retry_parent_submission_id, + ) + + def _create_central_submission_rows(self, rows: tuple[tuple[str, str, str], ...]) -> Path: + database = database_path(self.root) + with sqlite_connection(database) as connection: + connection.execute( + "CREATE TABLE ep_submissions(" + "submission_id TEXT PRIMARY KEY,producer_id TEXT,producer_type TEXT," + "producer_version TEXT,correlation_id TEXT,mission_id TEXT," + "engineering_action_id TEXT,constraints TEXT)" + ) + connection.executemany( + "INSERT INTO ep_submissions VALUES(?,?,?,?,?,?,?,?)", + [ + ( + submission_id, + producer_id, + "FORGE", + "1.2", + "correlation-1", + "MISSION-0003", + f"ACTION-{submission_id}", + constraints, + ) + for submission_id, producer_id, constraints in rows + ], + ) + return database + + def _record_qualification_snapshot( + self, + state: TransactionState, + *, + qualification: str, + conflicts: list[str] | None = None, + ) -> dict[str, object]: + snapshot = terminal_snapshot( + self.root, + run_id=state.run_id, + execution_outcome=state.phase, + implementation_pr=None, + finalization_pr=None, + repository_state="MERGED_RECONCILED", + workspace_state="WORKSPACE_READY", + main_origin_sync="YES", + worktree_state="CLEAN", + active_blocker="NONE", + recovery_required="NO", + persist=False, + ) + snapshot.update({ + "run_qualification": qualification, + "required_validation_state": "PASS" if qualification == "QUALIFIED" else "UNRESOLVED", + "qualification_failure_reasons": [] if qualification == "QUALIFIED" else ["synthetic missing evidence"], + "projection_conflicts": conflicts or [], + "qualification_snapshot_id": f"qualification:synthetic:{state.run_id}", + "required_control_snapshot_ref": f"required-controls:synthetic:{state.run_id}", + "terminal_checkpoint_ref": f"terminal-checkpoint:{state.run_id}:{state.phase}", + "persisted_at": "2026-09-17T10:00:00+00:00", + "reconciliation_evidence": { + "repository_state": "MERGED_RECONCILED", + "workspace_state": "WORKSPACE_READY", + }, + }) + return record_run_qualification_snapshot(self.root, snapshot) + + @staticmethod + def _summary(body: str) -> dict[str, object]: + match = re.search( + r"^## Engineering Evidence Summary\s*$\n```json\n(?P.*?)\n```", + body, + re.MULTILINE | re.DOTALL, + ) + if match is None: + raise AssertionError("Engineering Evidence Summary missing") + return json.loads(match.group("body")) + + def test_complete_not_qualified_is_not_general_acceptance(self) -> None: + self.prompt.write_text( + "Never claim PASS for missing evidence. Historical quote: PASS.\n", + encoding="utf-8", + ) + assurance = ({ + "reviewer": "quality", + "status": "PASS", + "candidate_sha": "a" * 40, + "profile_digest": "sha256:" + "b" * 64, + "invocation_id": "quality-review-1", + "findings": [], + },) + body = generate_terminal_report( + self.root, + self._state("complete-not-qualified", assurance_reviews=assurance), + ).read_text(encoding="utf-8") + + self.assertIn("- Technical Delivery: `COMPLETE`", body) + self.assertIn("- EP Run Qualification: `NOT_QUALIFIED`", body) + self.assertIn("- Mission / Autonomy Acceptance: `NOT_ESTABLISHED_BY_EP`", body) + self.assertIn("- Reviewer: `quality`\n - Status: `PASS`", body) + self.assertNotRegex(body, r"Final Deliverable Answer:.*(?:YES|PASS|GO)") + outcomes = self._summary(body)["outcomes"] + self.assertEqual(outcomes["technical_delivery"], "COMPLETE") + self.assertEqual(outcomes["ep_run_qualification"], "NOT_QUALIFIED") + self.assertEqual(outcomes["mission_autonomy_acceptance"], "NOT_ESTABLISHED_BY_EP") + + def test_qualification_and_acceptance_semantics_are_independent(self) -> None: + complete = self._state("outcome-matrix") + missing = _outcome_projection(complete, None) + qualified = _outcome_projection(complete, { + "run_qualification": "QUALIFIED", + "qualification_snapshot_id": "qualification:valid", + "projection_conflicts": [], + }) + conflict = _outcome_projection(complete, { + "run_qualification": "NOT_QUALIFIED", + "qualification_snapshot_id": "qualification:conflict", + "projection_conflicts": ["REQUIRED_CONTROL_CONFLICT"], + }) + + self.assertEqual(missing["ep_run_qualification"], "UNAVAILABLE") + self.assertEqual(qualified["ep_run_qualification"], "QUALIFIED") + self.assertEqual(conflict["ep_run_qualification"], "EVIDENCE_CONFLICT") + self.assertEqual(conflict["ep_run_qualification_persisted_outcome"], "NOT_QUALIFIED") + for projection in (missing, qualified, conflict): + self.assertEqual(projection["technical_delivery"], "COMPLETE") + self.assertEqual(projection["mission_autonomy_acceptance"], "NOT_ESTABLISHED_BY_EP") + + def test_complete_qualified_report_still_does_not_claim_mission_acceptance(self) -> None: + state = self._state("complete-qualified") + self._record_qualification_snapshot(state, qualification="QUALIFIED") + + body = generate_terminal_report(self.root, state).read_text(encoding="utf-8") + + self.assertIn("- Technical Delivery: `COMPLETE`", body) + self.assertIn("- EP Run Qualification: `QUALIFIED`", body) + self.assertIn("- Mission / Autonomy Acceptance: `NOT_ESTABLISHED_BY_EP`", body) + self.assertNotIn("Final Deliverable Answer:", body) + + def test_evidence_conflict_is_visible_in_readable_and_machine_projection(self) -> None: + state = self._state("complete-conflict") + self._record_qualification_snapshot( + state, + qualification="NOT_QUALIFIED", + conflicts=["REQUIRED_CONTROL_CONFLICT"], + ) + + body = generate_terminal_report(self.root, state).read_text(encoding="utf-8") + outcomes = self._summary(body)["outcomes"] + + self.assertIn("- EP Run Qualification: `EVIDENCE_CONFLICT`", body) + self.assertIn("- EP Qualification Conflicts: `REQUIRED_CONTROL_CONFLICT`", body) + self.assertEqual(outcomes["ep_run_qualification"], "EVIDENCE_CONFLICT") + self.assertEqual(outcomes["ep_run_qualification_persisted_outcome"], "NOT_QUALIFIED") + + def test_consistency_validator_rejects_only_unscoped_acceptance_section(self) -> None: + state = self._state("consistency-scope") + body = generate_terminal_report(self.root, state).read_text(encoding="utf-8") + outcome = self._summary(body)["outcomes"] + bundle = collect_terminal_evidence(self.root, state) + self.assertEqual(report_consistency_errors(body, state, bundle, "PASS", outcome), ()) + + tampered = body.replace( + "## Deliverable Answer\n", + "## Deliverable Answer\n- Final Deliverable Answer: YES / PASS / GO\n", + 1, + ) + errors = report_consistency_errors(tampered, state, bundle, "PASS", outcome) + self.assertIn("deliverable answer contains an unscoped acceptance conclusion", errors) + + overall_acceptance = body.replace( + "## Commit Strategy", + "- Overall Acceptance: PASS\n\n## Commit Strategy", + ) + errors = report_consistency_errors( + overall_acceptance, state, bundle, "PASS", outcome, + ) + self.assertIn("deliverable answer contains an unscoped acceptance conclusion", errors) + + missing_machine_summary = re.sub( + r"```json\n.*?\n```", + "machine projection unavailable", + body, + count=1, + flags=re.DOTALL, + ) + errors = report_consistency_errors( + missing_machine_summary, state, bundle, "PASS", outcome, + ) + self.assertIn("Engineering Evidence Summary machine projection is missing", errors) + + def test_blocked_and_failed_keep_factual_execution_status(self) -> None: + for phase in ("BLOCKED", "FAILED"): + with self.subTest(phase=phase): + body = generate_terminal_report( + self.root, + self._state(f"terminal-{phase.casefold()}", phase), + ).read_text(encoding="utf-8") + self.assertIn(f"- Technical Delivery: `{phase}`", body) + self.assertIn("- EP Run Qualification: `NOT_QUALIFIED`", body) + self.assertIn("- Mission / Autonomy Acceptance: `NOT_ESTABLISHED_BY_EP`", body) + self.assertNotRegex(body, r"Final Deliverable Answer:.*(?:YES|PASS|GO)") + + def test_valid_modern_submission_uses_immutable_evidence_without_prompt_parser(self) -> None: + self._record_submission(submission_id="submission-root", run_id="modern-root") + state = self._state("modern-root") + with patch( + "engineering_platform.execution_reporting.parse_producer_metadata", + side_effect=AssertionError("prompt fallback must not run"), + ): + producer, submission, provenance = _persisted_producer_submission( + self.root, state, "Mission ID: WRONG", + ) + self.assertEqual(producer.mission_id, "MISSION-0003") + self.assertEqual(submission["submission_id"], "submission-root") + self.assertEqual(provenance, "IMMUTABLE_SUBMISSION") + + def test_valid_modern_retry_preserves_attempt_and_root_identities(self) -> None: + self._record_submission(submission_id="submission-root", run_id="run-root") + self._record_submission( + submission_id="submission-retry", + run_id="run-retry", + retry_parent_submission_id="submission-root", + constraints={"attempt": "retry"}, + ) + database = self._create_central_submission_rows(( + ("submission-root", "forge", '{"attempt":"root"}'), + ("submission-retry", "forge", '{"attempt":"retry"}'), + )) + + result = resolve_submission_attempt_evidence( + self.root, "run-retry", central_database=database, + ) + + self.assertEqual(result.status, SUBMISSION_EVIDENCE_VALID_MODERN) + self.assertEqual(result.submission["submission_id"], "submission-retry") + self.assertEqual(result.submission["canonical_submission_id"], "submission-root") + self.assertEqual(result.submission["constraints"], {"attempt": "retry"}) + + def test_retry_root_parent_and_qualification_conflicts_fail_closed(self) -> None: + for case in ("local-source", "canonical-root", "qualification-attempt", "qualification-parent"): + with self.subTest(case=case): + root = self.root / case + root.mkdir() + original_root = self.root + self.root = root + try: + self._record_submission(submission_id="submission-root", run_id="run-root") + self._record_submission( + submission_id="submission-retry", + run_id="run-retry", + retry_parent_submission_id="submission-root", + constraints={"attempt": "retry"}, + ) + if case == "local-source": + result = resolve_submission_attempt_evidence(self.root, "run-retry") + self.assertEqual(result.status, SUBMISSION_EVIDENCE_STORAGE_UNAVAILABLE) + self.assertEqual( + result.diagnostic_code, + "SUBMISSION_EVIDENCE_ATTEMPT_SOURCE_UNAVAILABLE", + ) + continue + if case == "canonical-root": + self._record_submission( + submission_id="other-root", + run_id="other-run", + correlation_id="correlation-other", + ) + database = self._create_central_submission_rows(( + ("submission-root", "forge", '{"attempt":"root"}'), + ("submission-retry", "forge", '{"attempt":"retry"}'), + ("other-root", "forge", '{}'), + )) + if case == "canonical-root": + with sqlite_connection(database) as connection: + connection.execute( + "UPDATE execution_submission_attempts " + "SET canonical_submission_id='other-root' " + "WHERE submission_id='submission-retry'" + ) + else: + record_run_qualification_context( + self.root, + run_id="run-retry", + submission_id=( + "wrong-attempt" if case == "qualification-attempt" + else "submission-retry" + ), + fresh_submission=False, + retry_parent_run_id=( + "wrong-parent" if case == "qualification-parent" + else "run-root" + ), + resume_parent_run_id=None, + recorded_at="2026-09-17T10:00:00+00:00", + ) + result = resolve_submission_attempt_evidence( + self.root, "run-retry", central_database=database, + ) + self.assertEqual(result.status, SUBMISSION_EVIDENCE_IDENTITY_CONFLICT) + with patch( + "engineering_platform.execution_reporting.parse_producer_metadata", + side_effect=AssertionError("prompt fallback must not run"), + ), self.assertRaisesRegex(EngineeringStorageError, "SUBMISSION_EVIDENCE_"): + _persisted_producer_submission( + self.root, + self._state("run-retry"), + "Mission ID: WRONG", + central_database=database, + ) + finally: + self.root = original_root + + def test_historical_parent_exception_requires_retried_parent_without_attempt(self) -> None: + for case in ("parent-has-attempt", "dismissed-parent"): + with self.subTest(case=case): + root = self.root / case + root.mkdir() + original_root = self.root + self.root = root + try: + self._record_submission(submission_id="submission-root", run_id="run-root") + historical_submission = "submission-middle" + historical_run = "run-middle" + if case == "parent-has-attempt": + self._record_submission( + submission_id=historical_submission, + run_id=historical_run, + retry_parent_submission_id="submission-root", + ) + self._record_submission( + submission_id="submission-current", + run_id="run-current", + retry_parent_submission_id="submission-root", + ) + database = self._create_central_submission_rows(( + ("submission-root", "forge", '{}'), + (historical_submission, "forge", '{}'), + ("submission-current", "forge", '{}'), + )) + with sqlite_connection(database) as connection: + connection.execute( + "CREATE TABLE ep_parity_lifecycle_dispatches(" + "submission_id TEXT,run_id TEXT,operator_resolution TEXT," + "resolution_submission_id TEXT)" + ) + connection.execute( + "INSERT INTO ep_parity_lifecycle_dispatches VALUES(?,?,?,?)", + ( + historical_submission, + historical_run, + "RETRIED" if case == "parent-has-attempt" else "DISMISSED", + "submission-current", + ), + ) + record_run_qualification_context( + self.root, + run_id="run-current", + submission_id="submission-current", + fresh_submission=False, + retry_parent_run_id=historical_run, + resume_parent_run_id=None, + recorded_at="2026-09-17T10:00:00+00:00", + ) + + result = resolve_submission_attempt_evidence( + self.root, "run-current", central_database=database, + ) + + self.assertEqual(result.status, SUBMISSION_EVIDENCE_IDENTITY_CONFLICT) + self.assertEqual( + result.diagnostic_code, + "SUBMISSION_EVIDENCE_QUALIFICATION_LINEAGE_CONFLICT", + ) + finally: + self.root = original_root + + def test_identity_conflict_and_corrupt_constraints_never_call_prompt_fallback(self) -> None: + for case, producer_id, constraints, expected in ( + ("identity", "other-producer", '{}', SUBMISSION_EVIDENCE_IDENTITY_CONFLICT), + ("corrupt", "forge", '[not-json', SUBMISSION_EVIDENCE_CORRUPT), + ("wrong-type", "forge", '[]', SUBMISSION_EVIDENCE_CORRUPT), + ): + with self.subTest(case=case): + root = self.root / case + root.mkdir() + original_root = self.root + self.root = root + try: + self._record_submission(submission_id="submission-root", run_id="conflict-run") + database = self._create_central_submission_rows(( + ("submission-root", producer_id, constraints), + )) + resolution = resolve_submission_attempt_evidence( + self.root, "conflict-run", central_database=database, + ) + self.assertEqual(resolution.status, expected) + with patch( + "engineering_platform.execution_reporting.parse_producer_metadata", + side_effect=AssertionError("prompt fallback must not run"), + ), self.assertRaisesRegex(EngineeringStorageError, "SUBMISSION_EVIDENCE_"): + _persisted_producer_submission( + self.root, + self._state("conflict-run"), + "Mission ID: WRONG\nProducer ID: attacker", + central_database=database, + ) + finally: + self.root = original_root + + def test_missing_modern_binding_and_storage_failure_are_not_legacy(self) -> None: + self._record_submission(submission_id="submission-root", run_id="missing-attempt") + database = self._create_central_submission_rows(( + ("submission-root", "forge", '{}'), + )) + with sqlite_connection(database) as connection: + connection.execute( + "DELETE FROM execution_submission_attempt_links WHERE run_id=?", + ("missing-attempt",), + ) + missing = resolve_submission_attempt_evidence( + self.root, "missing-attempt", central_database=database, + ) + unavailable = resolve_submission_attempt_evidence( + self.root, + "unavailable-run", + central_database=self.root / "missing-central.sqlite", + ) + self.assertEqual(missing.status, SUBMISSION_EVIDENCE_MISSING_MODERN_BINDING) + self.assertEqual(unavailable.status, SUBMISSION_EVIDENCE_STORAGE_UNAVAILABLE) + with patch( + "engineering_platform.execution_reporting.parse_producer_metadata", + side_effect=AssertionError("prompt fallback must not run"), + ): + for run_id, central_database in ( + ("missing-attempt", database), + ("unavailable-run", self.root / "missing-central.sqlite"), + ): + with self.subTest(run_id=run_id), self.assertRaisesRegex( + EngineeringStorageError, + "SUBMISSION_EVIDENCE_", + ): + _persisted_producer_submission( + self.root, + self._state(run_id), + "Mission ID: WRONG", + central_database=central_database, + ) + + modern_without_submission = self.root / "lineage" + modern_without_submission.mkdir() + record_run_qualification_context( + modern_without_submission, + run_id="modern-lineage", + submission_id="expected-submission", + fresh_submission=True, + retry_parent_run_id=None, + resume_parent_run_id=None, + recorded_at="2026-09-17T10:00:00+00:00", + ) + result = resolve_submission_attempt_evidence(modern_without_submission, "modern-lineage") + self.assertEqual(result.status, SUBMISSION_EVIDENCE_MISSING_MODERN_BINDING) + + def test_supported_legacy_absence_is_the_only_prompt_fallback(self) -> None: + state = self._state("legacy-run") + result = resolve_submission_attempt_evidence(self.root, state.run_id) + self.assertEqual(result.status, SUBMISSION_EVIDENCE_LEGACY_ABSENT) + with patch( + "engineering_platform.execution_reporting.parse_producer_metadata", + wraps=__import__( + "engineering_platform.execution_reporting", + fromlist=["parse_producer_metadata"], + ).parse_producer_metadata, + ) as parser: + _producer, submission, provenance = _persisted_producer_submission( + self.root, + state, + "Producer ID: legacy-human\nMission ID: MISSION-LEGACY", + ) + parser.assert_called_once() + self.assertIsNone(submission) + self.assertEqual(provenance, "LEGACY_PROMPT_METADATA") + + def test_integrated_report_refuses_modern_conflict_without_mutation_or_fallback(self) -> None: + self._record_submission(submission_id="submission-root", run_id="integrated-conflict") + database = self._create_central_submission_rows(( + ("submission-root", "wrong-producer", '{}'), + )) + with sqlite_connection(database) as connection: + before = connection.execute( + "SELECT submission_id,producer_id,prompt_metadata FROM execution_submissions" + ).fetchall() + report_directory = database.parent / "artifacts" / "reports" + + with patch( + "engineering_platform.execution_reporting.parse_producer_metadata", + side_effect=AssertionError("prompt fallback must not run"), + ), self.assertRaisesRegex( + EngineeringStorageError, + "SUBMISSION_EVIDENCE_IDENTITY_CONFLICT", + ): + generate_terminal_report( + self.root, + self._state("integrated-conflict"), + central_database=database, + ) + + with sqlite_connection(database) as connection: + after = connection.execute( + "SELECT submission_id,producer_id,prompt_metadata FROM execution_submissions" + ).fetchall() + self.assertEqual(after, before) + self.assertFalse(report_directory.exists()) + + +if __name__ == "__main__": + unittest.main()