← Files VeraARCHIVED FILE
modules/bandi-agevolazioni/scripts/validate_application.py
43.9 KB · Oct 4, 2026 · 12:28 UTC
"""Validate mechanical integrity of a Bandi e agevolazioni workbench."""
from __future__ import annotations
import argparse
import hashlib
import logging
import re
from pathlib import Path
from typing import Any, Iterable
from case_core import (
PLUGIN_NAME,
canonical_json_sha256,
case_lock,
iso_now,
load_running_context,
prohibited_secret_paths,
require_run_artifact,
safe_identifier,
write_private_json,
)
from deterministic_rules import RuleContractError, validate_deterministic_rule
from record_review import SCOPES, current_scope_hash
from schema_validation import validate_artifact_schema
__all__ = ["validate_application", "main"]
LOGGER = logging.getLogger(__name__)
SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
READINESS = {"ready", "missing", "verify", "not_applicable"}
REVIEW_STATUS = {"proposed", "confirmed", "rejected", "blocked"}
def _issue(issues: list[dict[str, str]], code: str, path: str, message: str) -> None:
issues.append({"code": code, "path": path, "message": message})
def _items(
payload: dict[str, Any], key: str, *, issues: list[dict[str, str]]
) -> list[dict[str, Any]]:
value = payload.get(key)
if not isinstance(value, list):
_issue(issues, "invalid_collection", key, f"{key} must be a list")
return []
items: list[dict[str, Any]] = []
for index, item in enumerate(value):
if not isinstance(item, dict):
_issue(
issues,
"invalid_item",
f"{key}[{index}]",
"item must be an object",
)
else:
items.append(item)
return items
def _object(payload: dict[str, Any], key: str) -> dict[str, Any]:
"""Return a nested object or an empty object after schema issues are recorded."""
value = payload.get(key)
return value if isinstance(value, dict) else {}
def _has_material_value(value: object) -> bool:
"""Return whether a mechanically required draft value is present."""
return value is not None and not (isinstance(value, str) and not value.strip())
def _id_set(
items: Iterable[dict[str, Any]],
field: str,
*,
path: str,
issues: list[dict[str, str]],
) -> set[str]:
result: set[str] = set()
for index, item in enumerate(items):
try:
identifier = safe_identifier(item.get(field), field=f"{path}.{field}")
except ValueError as exc:
_issue(issues, "invalid_id", f"{path}[{index}].{field}", str(exc))
continue
if identifier in result:
_issue(
issues,
"duplicate_id",
f"{path}[{index}].{field}",
f"duplicate {identifier}",
)
result.add(identifier)
return result
def _check_refs(
values: object,
known: set[str],
*,
path: str,
issues: list[dict[str, str]],
require_one: bool = False,
) -> list[str]:
if not isinstance(values, list):
_issue(issues, "invalid_references", path, "references must be a list")
return []
refs = [value for value in values if isinstance(value, str)]
if len(refs) != len(values):
_issue(issues, "invalid_reference", path, "references must be strings")
if require_one and not refs:
_issue(issues, "missing_reference", path, "at least one reference is required")
if len(refs) != len(set(refs)):
_issue(issues, "duplicate_reference", path, "references must be unique")
unknown = sorted(set(refs) - known)
if unknown:
_issue(
issues,
"unknown_reference",
path,
"unknown references: " + ", ".join(unknown),
)
return refs
def _review_status(
item: dict[str, Any], *, path: str, issues: list[dict[str, str]]
) -> str:
status = str(item.get("review_status") or "")
if status not in REVIEW_STATUS:
_issue(issues, "invalid_review_status", path, "invalid review_status")
return status
def _readiness(item: dict[str, Any], *, path: str, issues: list[dict[str, str]]) -> str:
readiness = str(item.get("readiness") or "")
if readiness not in READINESS:
_issue(issues, "invalid_readiness", path, "invalid readiness")
if readiness == "ready" and item.get("review_status") != "confirmed":
_issue(
issues,
"ready_requires_confirmed_review",
path,
"ready items must be professionally confirmed",
)
if readiness == "not_applicable" and not str(item.get("rationale") or "").strip():
_issue(
issues,
"not_applicable_requires_rationale",
path,
"not_applicable requires a reviewed rationale",
)
if readiness == "not_applicable" and item.get("review_status") != "confirmed":
_issue(
issues,
"not_applicable_requires_confirmed_review",
path,
"not_applicable items must be professionally confirmed",
)
return readiness
def _check_not_applicable_outcome(
item: dict[str, Any], *, path: str, issues: list[dict[str, str]]
) -> None:
# A proposed check outcome is not documentary readiness. It must be possible
# to package it for review without falsely confirming the professional's
# decision. Ready-disposition checks below still require confirmed outcomes.
if not str(item.get("rationale") or "").strip():
_issue(
issues,
"not_applicable_requires_rationale",
path,
"not_applicable check outcomes require a rationale",
)
def _latest_review_state(
output_dir: Path,
*,
run_id: str,
review_log: dict[str, Any],
issues: list[dict[str, str]],
) -> dict[str, str]:
events = review_log.get("events")
if not isinstance(events, list):
_issue(
issues, "invalid_review_log", "review_log.events", "events must be a list"
)
events = []
states: dict[str, str] = {}
for scope in SCOPES:
expected_hash = current_scope_hash(output_dir, run_id=run_id, scope=scope)
matching = [
event
for event in events
if isinstance(event, dict)
and event.get("scope") == scope
and event.get("scope_sha256") == expected_hash
and event.get("confirmation_basis") == "explicit_user_confirmation"
and event.get("identity_assurance") == "asserted_not_authenticated"
]
states[scope] = (
str(matching[-1].get("decision")) if matching else "stale_or_missing"
)
return states
def validate_application(
*, output_dir: Path, client_engagement: Path
) -> dict[str, Any]:
"""Validate contracts and traceability without deciding semantic correctness."""
context = load_running_context(client_engagement, output_dir=output_dir)
output_dir = output_dir.resolve()
with case_lock(output_dir):
return _validate_application_locked(output_dir=output_dir, context=context)
def _validate_application_locked(
*, output_dir: Path, context: dict[str, Any]
) -> dict[str, Any]:
"""Validate one immutable case snapshot while cooperative writers are locked."""
run_id = safe_identifier(context["run_id"], field="run_id")
intake = require_run_artifact(output_dir / "case_intake.json", run_id=run_id)
sources = require_run_artifact(output_dir / "source_register.json", run_id=run_id)
workbench = require_run_artifact(
output_dir / "application_workbench.json", run_id=run_id
)
intelligence = require_run_artifact(
output_dir / "intelligence_register.json", run_id=run_id
)
reviews = require_run_artifact(output_dir / "review_log.json", run_id=run_id)
run_state = require_run_artifact(output_dir / "run_state.json", run_id=run_id)
issues: list[dict[str, str]] = []
for label, payload in (
("case_intake", intake),
("source_register", sources),
("application_workbench", workbench),
("intelligence_register", intelligence),
("review_log", reviews),
("run_state", run_state),
):
for secret_path in prohibited_secret_paths(payload, path=label):
_issue(
issues,
"secret_or_session_material_forbidden",
secret_path,
"credentials, signatures, tokens, cookies, and session material are forbidden",
)
issues.extend(validate_artifact_schema(label, payload))
if payload.get("plugin") != PLUGIN_NAME or payload.get("run_id") != run_id:
_issue(issues, "artifact_identity_mismatch", label, "plugin/run mismatch")
# Exact scope binding is mechanically reproducible; portal meaning is reviewed.
preparation = _object(run_state, "portal_preparation")
if preparation and preparation.get("approved_scope_sha256") != current_scope_hash(
output_dir, run_id=run_id, scope="dossier"
):
_issue(
issues,
"portal_preparation_approval_stale",
"run_state.portal_preparation.approved_scope_sha256",
"portal preparation approval must match the current dossier scope",
)
submission = _object(run_state, "submission_approval")
if submission and submission.get("approved_scope_sha256") != current_scope_hash(
output_dir, run_id=run_id, scope="dossier"
):
_issue(
issues,
"submission_approval_stale",
"run_state.submission_approval.approved_scope_sha256",
"submission approval must match the current dossier scope",
)
if (
preparation
and submission
and preparation.get("destination") != submission.get("destination")
):
_issue(
issues,
"submission_destination_mismatch",
"run_state.submission_approval.destination",
"submission destination must match the prepared application",
)
applying_intelligence = [
str(item.get("intelligence_run_id"))
for item in intelligence.get("runs", [])
if isinstance(item, dict) and item.get("status") == "APPLYING"
]
if applying_intelligence:
_issue(
issues,
"intelligence_application_incomplete",
"intelligence_register.runs",
"interrupted intelligence application requires recovery: "
+ ", ".join(applying_intelligence),
)
if run_state.get("source_set_revision") != sources.get("source_set_revision"):
_issue(
issues,
"source_set_revision_mismatch",
"run_state.source_set_revision",
"run state and source register revisions must match",
)
source_items = _items(sources, "sources", issues=issues)
source_ids = _id_set(source_items, "source_id", path="sources", issues=issues)
source_by_id = {str(item.get("source_id")): item for item in source_items}
for index, source in enumerate(source_items):
if not SHA256_RE.fullmatch(str(source.get("sha256") or "")):
_issue(
issues, "invalid_sha256", f"sources[{index}].sha256", "invalid SHA-256"
)
relationships = source.get("relationships")
if not isinstance(relationships, list):
_issue(
issues,
"invalid_relationships",
f"sources[{index}].relationships",
"relationships must be a list",
)
else:
normalized_relationships = [
(
item.get("kind"),
item.get("target_source_id"),
)
for item in relationships
if isinstance(item, dict)
and isinstance(item.get("kind"), str)
and isinstance(item.get("target_source_id"), str)
]
if len(normalized_relationships) != len(set(normalized_relationships)):
_issue(
issues,
"duplicate_source_relationship",
f"sources[{index}].relationships",
"source relationships must be unique",
)
if any(
target_source_id == source.get("source_id")
for _, target_source_id in normalized_relationships
):
_issue(
issues,
"self_source_relationship",
f"sources[{index}].relationships",
"a source cannot relate to itself",
)
_check_refs(
[
item.get("target_source_id")
for item in relationships
if isinstance(item, dict)
],
source_ids,
path=f"sources[{index}].relationships",
issues=issues,
)
requirements = _items(workbench, "requirements", issues=issues)
facts = _items(workbench, "facts", issues=issues)
assessments = _items(workbench, "assessments", issues=issues)
documents = _items(workbench, "document_checklist", issues=issues)
expenses = _items(workbench, "expenses", issues=issues)
form_fields = _items(workbench, "form_fields", issues=issues)
narratives = _items(workbench, "narratives", issues=issues)
consistency_checks = _items(workbench, "consistency_checks", issues=issues)
issue_items = _items(workbench, "issues", issues=issues)
requirement_ids = _id_set(
requirements, "requirement_id", path="requirements", issues=issues
)
fact_ids = _id_set(facts, "fact_id", path="facts", issues=issues)
assessment_ids = _id_set(
assessments, "assessment_id", path="assessments", issues=issues
)
document_ids = _id_set(
documents, "document_id", path="document_checklist", issues=issues
)
expense_ids = _id_set(expenses, "expense_id", path="expenses", issues=issues)
form_field_ids = _id_set(form_fields, "field_id", path="form_fields", issues=issues)
narrative_ids = _id_set(
narratives, "narrative_id", path="narratives", issues=issues
)
consistency_check_ids = _id_set(
consistency_checks,
"check_id",
path="consistency_checks",
issues=issues,
)
issue_ids = _id_set(issue_items, "issue_id", path="issues", issues=issues)
fact_by_id = {str(item.get("fact_id")): item for item in facts}
for index, requirement in enumerate(requirements):
path = f"requirements[{index}]"
_review_status(requirement, path=path, issues=issues)
refs = requirement.get("source_refs")
if not isinstance(refs, list) or not refs:
_issue(
issues,
"requirement_missing_source",
f"{path}.source_refs",
"requirements need at least one exact source fragment",
)
continue
for ref_index, ref in enumerate(refs):
ref_path = f"{path}.source_refs[{ref_index}]"
if not isinstance(ref, dict):
_issue(
issues,
"invalid_source_ref",
ref_path,
"source ref must be an object",
)
continue
source_id = str(ref.get("source_id") or "")
if source_id not in source_ids:
_issue(
issues,
"unknown_reference",
f"{ref_path}.source_id",
f"unknown source {source_id}",
)
if not str(ref.get("locator") or "").strip():
_issue(
issues,
"missing_locator",
f"{ref_path}.locator",
"source locator is required",
)
if not SHA256_RE.fullmatch(str(ref.get("excerpt_sha256") or "")):
_issue(
issues,
"invalid_excerpt_sha256",
f"{ref_path}.excerpt_sha256",
"invalid excerpt SHA-256",
)
excerpt = ref.get("excerpt")
if isinstance(excerpt, str) and excerpt:
expected_excerpt_hash = hashlib.sha256(
excerpt.encode("utf-8")
).hexdigest()
if ref.get("excerpt_sha256") != expected_excerpt_hash:
_issue(
issues,
"excerpt_sha256_mismatch",
f"{ref_path}.excerpt_sha256",
"excerpt_sha256 must hash the exact stored UTF-8 excerpt",
)
if (
requirement.get("review_status") == "confirmed"
and source_by_id.get(source_id, {}).get("review_status") != "reviewed"
):
_issue(
issues,
"confirmed_requirement_uses_unreviewed_source",
ref_path,
"confirmed requirements require reviewed sources",
)
for index, fact in enumerate(facts):
_review_status(fact, path=f"facts[{index}]", issues=issues)
_check_refs(
fact.get("source_ids"),
source_ids,
path=f"facts[{index}].source_ids",
issues=issues,
)
negative_assessment = False
for index, assessment in enumerate(assessments):
path = f"assessments[{index}]"
_review_status(assessment, path=path, issues=issues)
_readiness(assessment, path=path, issues=issues)
_check_refs(
[assessment.get("requirement_id")],
requirement_ids,
path=f"{path}.requirement_id",
issues=issues,
require_one=True,
)
fact_refs = _check_refs(
assessment.get("fact_ids"), fact_ids, path=f"{path}.fact_ids", issues=issues
)
outcome = assessment.get("outcome")
negative_assessment = negative_assessment or outcome == "not_satisfied"
if (
assessment.get("readiness") == "not_applicable"
and outcome != "not_applicable"
):
_issue(
issues,
"not_applicable_outcome_mismatch",
path,
"not_applicable readiness requires not_applicable outcome",
)
if assessment.get("readiness") == "ready" and outcome != "not_applicable":
if not fact_refs:
_issue(
issues,
"ready_assessment_requires_fact",
f"{path}.fact_ids",
"a ready applicable assessment requires at least one reviewed fact",
)
unconfirmed = [
fact_id
for fact_id in fact_refs
if fact_by_id.get(fact_id, {}).get("review_status") != "confirmed"
]
if unconfirmed:
_issue(
issues,
"ready_assessment_uses_unconfirmed_fact",
f"{path}.fact_ids",
"ready assessments require confirmed facts: "
+ ", ".join(sorted(unconfirmed)),
)
evaluation_method = assessment.get("evaluation_method")
rule = assessment.get("deterministic_rule")
if evaluation_method == "model_led" and rule is not None:
_issue(
issues,
"model_led_assessment_has_deterministic_rule",
f"{path}.deterministic_rule",
"model-led assessments cannot claim a deterministic rule",
)
if evaluation_method in {"deterministic", "hybrid"}:
try:
validate_deterministic_rule(rule, assessment_outcome=outcome)
except RuleContractError as exc:
_issue(
issues,
"deterministic_rule_not_reproducible",
f"{path}.deterministic_rule",
str(exc),
)
for key, items in (
("document_checklist", documents),
("expenses", expenses),
("form_fields", form_fields),
("narratives", narratives),
):
for index, item in enumerate(items):
path = f"{key}[{index}]"
_review_status(item, path=path, issues=issues)
_readiness(item, path=path, issues=issues)
requirement_refs = _check_refs(
item.get("requirement_ids"),
requirement_ids,
path=f"{path}.requirement_ids",
issues=issues,
)
if key in {"document_checklist", "expenses"}:
source_field = (
"material_source_ids"
if key == "document_checklist"
else "source_ids"
)
source_refs = _check_refs(
item.get(source_field),
source_ids,
path=f"{path}.{source_field}",
issues=issues,
)
if item.get("readiness") == "ready" and (
not requirement_refs or not source_refs
):
_issue(
issues,
"ready_item_missing_traceability",
path,
"ready documents and expenses require requirement and source links",
)
if key in {"form_fields", "narratives"}:
fact_refs = _check_refs(
item.get("fact_ids"),
fact_ids,
path=f"{path}.fact_ids",
issues=issues,
)
if key == "form_fields":
protected = any(
bool(item.get(field))
for field in (
"declaration_control",
"signature_control",
"submission_control",
)
)
if protected and not item.get("manual_only"):
_issue(
issues,
"protected_field_must_be_manual",
path,
"declaration, signature, and submission controls must be manual_only",
)
if protected and item.get("proposed_value") not in (None, ""):
_issue(
issues,
"protected_field_must_be_empty",
path,
"protected controls cannot have a proposed value",
)
if (
item.get("readiness") == "ready"
and not protected
and (not requirement_refs or not fact_refs)
):
_issue(
issues,
"ready_item_missing_traceability",
path,
"ready ordinary form fields require requirement and fact links",
)
if (
item.get("readiness") == "ready"
and not protected
and not _has_material_value(item.get("proposed_value"))
):
_issue(
issues,
"ready_form_field_missing_value",
f"{path}.proposed_value",
"ready ordinary form fields require a proposed value",
)
if (
key == "narratives"
and item.get("readiness") == "ready"
and (not requirement_refs or not fact_refs)
):
_issue(
issues,
"ready_item_missing_traceability",
path,
"ready narratives require requirement and fact links",
)
if (
key == "narratives"
and item.get("readiness") == "ready"
and not str(item.get("draft") or "").strip()
):
_issue(
issues,
"ready_narrative_missing_draft",
f"{path}.draft",
"ready narratives require a non-empty draft",
)
if (
key == "expenses"
and item.get("readiness") == "not_applicable"
and item.get("outcome") != "not_assessed"
):
_issue(
issues,
"expense_not_applicable_outcome_mismatch",
path,
"not_applicable expenses require outcome not_assessed",
)
consistency_conflict = False
for index, item in enumerate(consistency_checks):
path = f"consistency_checks[{index}]"
_review_status(item, path=path, issues=issues)
consistency_fact_refs = _check_refs(
item.get("fact_ids"),
fact_ids,
path=f"{path}.fact_ids",
issues=issues,
)
consistency_source_refs = _check_refs(
item.get("source_ids"),
source_ids,
path=f"{path}.source_ids",
issues=issues,
)
if (
item.get("outcome") == "consistent"
and item.get("review_status") == "confirmed"
and len(set(consistency_fact_refs + consistency_source_refs)) < 2
):
_issue(
issues,
"consistency_check_missing_evidence",
path,
"a confirmed consistency check requires at least two evidence links",
)
outcome = item.get("outcome")
consistency_conflict = consistency_conflict or outcome == "conflict"
if outcome == "not_applicable":
_check_not_applicable_outcome(item, path=path, issues=issues)
related_ids = (
source_ids
| requirement_ids
| fact_ids
| assessment_ids
| document_ids
| expense_ids
| form_field_ids
| narrative_ids
| consistency_check_ids
| issue_ids
)
authority = workbench.get("authority_simulation")
authority_checks: list[dict[str, Any]] = []
if not isinstance(authority, dict):
_issue(
issues,
"invalid_authority_simulation",
"authority_simulation",
"authority_simulation must be an object",
)
authority = {}
else:
authority_checks = _items(authority, "checks", issues=issues)
authority_check_ids = _id_set(
authority_checks,
"check_id",
path="authority_simulation.checks",
issues=issues,
)
# IDs form one global namespace because issue and authority links carry no
# type discriminator. Global uniqueness makes every reference auditable.
id_collections = (
("source", source_ids),
("requirement", requirement_ids),
("fact", fact_ids),
("assessment", assessment_ids),
("document", document_ids),
("expense", expense_ids),
("form_field", form_field_ids),
("narrative", narrative_ids),
("consistency_check", consistency_check_ids),
("issue", issue_ids),
("authority_check", authority_check_ids),
)
owners: dict[str, str] = {}
for kind, identifiers in id_collections:
for identifier in identifiers:
if identifier in owners:
_issue(
issues,
"cross_type_duplicate_id",
kind,
f"{identifier} is already used as {owners[identifier]}",
)
else:
owners[identifier] = kind
for index, item in enumerate(authority_checks):
path = f"authority_simulation.checks[{index}]"
_review_status(item, path=path, issues=issues)
_check_refs(
item.get("related_ids"),
related_ids,
path=f"{path}.related_ids",
issues=issues,
)
if item.get("outcome") == "not_applicable":
_check_not_applicable_outcome(item, path=path, issues=issues)
open_blockers = [
item
for item in issue_items
if item.get("severity") in {"blocking", "review_required"}
and item.get("status") == "open"
]
for index, item in enumerate(issue_items):
_review_status(item, path=f"issues[{index}]", issues=issues)
_check_refs(
item.get("related_ids"),
related_ids,
path=f"issues[{index}].related_ids",
issues=issues,
)
if (
item.get("severity") in {"blocking", "review_required"}
and item.get("status") != "open"
and item.get("review_status") != "confirmed"
):
_issue(
issues,
"material_issue_closure_requires_confirmed_review",
f"issues[{index}]",
"closing a material issue requires confirmed professional review",
)
dossier = workbench.get("dossier")
if not isinstance(dossier, dict):
_issue(issues, "invalid_dossier", "dossier", "dossier must be an object")
dossier = {}
if dossier.get("ready_to_file") is not False:
_issue(
issues,
"ready_to_file_forbidden",
"dossier.ready_to_file",
"ready_to_file must remain false",
)
review_states = _latest_review_state(
output_dir,
run_id=run_id,
review_log=reviews,
issues=issues,
)
disposition = dossier.get("disposition")
all_readiness = [
item.get("readiness")
for collection in (assessments, documents, expenses, form_fields, narratives)
for item in collection
]
if disposition == "ready_for_authorized_review":
application = _object(intake, "application")
applicant = _object(intake, "applicant")
project = _object(intake, "project")
required_nonempty = {
"sources": source_items,
"requirements": requirements,
"assessments": assessments,
"document_checklist": documents,
"expenses": expenses,
"form_fields": form_fields,
"narratives": narratives,
"consistency_checks": consistency_checks,
"authority_simulation.checks": authority_checks,
}
empty = sorted(name for name, items in required_nonempty.items() if not items)
if empty:
_issue(
issues,
"ready_disposition_incomplete_dossier",
"dossier.disposition",
"ready disposition requires non-empty reviewed sections: "
+ ", ".join(empty),
)
intake_checks = {
"application.status": application.get("status"),
"applicant.confirmation_status": applicant.get("confirmation_status"),
"project.confirmation_status": project.get("confirmation_status"),
}
unconfirmed_intake = sorted(
field for field, value in intake_checks.items() if value != "confirmed"
)
required_text = {
"application.title": application.get("title"),
"application.issuing_authority": application.get("issuing_authority"),
"application.procedure_id": application.get("procedure_id"),
"applicant.legal_name": applicant.get("legal_name"),
"project.title": project.get("title"),
"project.summary": project.get("summary"),
"professional_question": intake.get("professional_question"),
"workbench.case_summary": workbench.get("case_summary"),
}
unconfirmed_intake.extend(
field
for field, value in required_text.items()
if not str(value or "").strip()
)
if unconfirmed_intake:
_issue(
issues,
"ready_disposition_has_unconfirmed_intake",
"case_intake",
"ready disposition requires confirmed material intake: "
+ ", ".join(sorted(set(unconfirmed_intake))),
)
if any(source.get("review_status") != "reviewed" for source in source_items):
_issue(
issues,
"ready_disposition_has_unreviewed_sources",
"source_register.sources",
"every source in a ready dossier must be reviewed",
)
governing_calls = [
source
for source in source_items
if source.get("source_type") == "call"
and source.get("review_status") == "reviewed"
]
if not governing_calls:
_issue(
issues,
"ready_disposition_missing_governing_call",
"source_register.sources",
"ready disposition requires at least one reviewed governing call",
)
formal_source_types = {"call", "formal_amendment", "official_faq"}
undated_official_sources = sorted(
str(source.get("source_id"))
for source in source_items
if source.get("source_type") in formal_source_types
and (
source.get("publication_date") is None
or source.get("effective_from") is None
)
)
if undated_official_sources:
_issue(
issues,
"ready_disposition_has_undated_official_sources",
"source_register.sources",
"publication_date and effective_from are required for: "
+ ", ".join(undated_official_sources),
)
relationship_gaps: list[str] = []
for source in source_items:
source_type = source.get("source_type")
expected_kinds = (
{"clarifies"}
if source_type == "official_faq"
else (
{"amends", "supersedes"}
if source_type == "formal_amendment"
else set()
)
)
if not expected_kinds:
continue
valid_relationship = any(
isinstance(relationship, dict)
and relationship.get("kind") in expected_kinds
and source_by_id.get(str(relationship.get("target_source_id")), {}).get(
"source_type"
)
in {"call", "formal_amendment"}
for relationship in source.get("relationships", [])
)
if not valid_relationship:
relationship_gaps.append(str(source.get("source_id")))
if relationship_gaps:
_issue(
issues,
"ready_disposition_has_unbound_dependent_sources",
"source_register.sources",
"FAQ and amendment sources require explicit formal-source relationships: "
+ ", ".join(sorted(relationship_gaps)),
)
if any(
requirement.get("review_status") != "confirmed"
for requirement in requirements
):
_issue(
issues,
"ready_disposition_has_unconfirmed_requirements",
"requirements",
"every requirement in a ready dossier must be confirmed",
)
if any(fact.get("review_status") != "confirmed" for fact in facts):
_issue(
issues,
"ready_disposition_has_unconfirmed_facts",
"facts",
"every fact in a ready dossier must be confirmed",
)
empty_fact_ids = sorted(
str(fact.get("fact_id"))
for fact in facts
if not _has_material_value(fact.get("value"))
)
if empty_fact_ids:
_issue(
issues,
"ready_disposition_has_empty_facts",
"facts",
"ready disposition requires material fact values: "
+ ", ".join(empty_fact_ids),
)
assessed_requirement_ids = [
str(assessment.get("requirement_id")) for assessment in assessments
]
missing_assessments = sorted(requirement_ids - set(assessed_requirement_ids))
duplicate_assessments = sorted(
requirement_id
for requirement_id in set(assessed_requirement_ids)
if assessed_requirement_ids.count(requirement_id) > 1
)
if missing_assessments or duplicate_assessments:
_issue(
issues,
"ready_disposition_has_assessment_coverage_gap",
"assessments",
"one assessment is required per requirement; missing="
+ ",".join(missing_assessments)
+ "; duplicate="
+ ",".join(duplicate_assessments),
)
if any(value in {"missing", "verify"} for value in all_readiness):
_issue(
issues,
"ready_disposition_has_unresolved_items",
"dossier.disposition",
"ready disposition cannot contain missing or verify items",
)
unresolved_outcomes = [
str(item.get("assessment_id"))
for item in assessments
if item.get("outcome") not in {"satisfied", "not_applicable"}
]
unresolved_consistency = [
str(item.get("check_id"))
for item in consistency_checks
if item.get("outcome") not in {"consistent", "not_applicable"}
or item.get("review_status") != "confirmed"
]
unresolved_authority = [
str(item.get("check_id"))
for item in authority_checks
if item.get("outcome") not in {"pass", "not_applicable"}
or item.get("review_status") != "confirmed"
]
authority_covered_ids = {
related_id
for item in authority_checks
for related_id in item.get("related_ids", [])
if isinstance(related_id, str)
}
material_ids = (
requirement_ids
| fact_ids
| assessment_ids
| document_ids
| expense_ids
| form_field_ids
| narrative_ids
| consistency_check_ids
| {
str(item.get("issue_id"))
for item in issue_items
if item.get("severity") in {"blocking", "review_required"}
}
)
missing_authority_coverage = sorted(material_ids - authority_covered_ids)
if missing_authority_coverage:
_issue(
issues,
"authority_simulation_coverage_gap",
"authority_simulation.checks",
"authority simulation does not cover: "
+ ", ".join(missing_authority_coverage),
)
if (
authority.get("status") != "reviewed"
or authority.get("overall_outcome") != "pass"
or unresolved_authority
):
_issue(
issues,
"ready_disposition_has_unreviewed_authority_simulation",
"authority_simulation",
"ready disposition requires a reviewed passing authority simulation",
)
if unresolved_consistency or consistency_conflict:
_issue(
issues,
"ready_disposition_has_unresolved_consistency",
"consistency_checks",
"ready disposition requires confirmed consistent checks",
)
if negative_assessment or unresolved_outcomes or open_blockers:
_issue(
issues,
"ready_disposition_has_adverse_result",
"dossier.disposition",
"ready disposition cannot contain unresolved assessments or open review issues",
)
adverse_expenses = [
str(item.get("expense_id"))
for item in expenses
if item.get("readiness") == "ready" and item.get("outcome") != "eligible"
]
if adverse_expenses:
_issue(
issues,
"ready_disposition_has_adverse_expenses",
"expenses",
"ready disposition requires eligible expense outcomes: "
+ ", ".join(adverse_expenses),
)
stale = [
scope for scope, decision in review_states.items() if decision != "accepted"
]
if stale:
_issue(
issues,
"ready_disposition_has_stale_reviews",
"dossier.disposition",
"accepted current reviews required: " + ", ".join(stale),
)
if disposition == "not_eligible_or_excluded":
if not negative_assessment:
_issue(
issues,
"negative_disposition_without_negative_assessment",
"dossier.disposition",
"negative disposition requires a not_satisfied assessment",
)
if review_states.get("assessments") != "accepted":
_issue(
issues,
"negative_disposition_requires_review",
"dossier.disposition",
"negative disposition requires accepted current assessment review",
)
artifact_hashes = {
"case_intake": canonical_json_sha256(intake),
"source_register": canonical_json_sha256(sources),
"application_workbench": canonical_json_sha256(workbench),
"intelligence_register": canonical_json_sha256(intelligence),
"review_log": canonical_json_sha256(reviews),
"run_state": canonical_json_sha256(run_state),
}
audit = {
"schema_version": "1.1",
"plugin": PLUGIN_NAME,
"run_id": run_id,
"validated_at": iso_now(),
"status": "passed" if not issues else "failed",
"ready_to_file": False,
"portal_actions_performed": run_state.get("portal_actions_performed"),
"signature_actions_performed": run_state.get("signature_actions_performed"),
"submission_actions_performed": run_state.get("submission_actions_performed"),
"artifact_hashes": artifact_hashes,
"review_states": review_states,
"counts": {
"sources": len(source_items),
"requirements": len(requirements),
"facts": len(facts),
"assessments": len(assessments),
"documents": len(documents),
"expenses": len(expenses),
"form_fields": len(form_fields),
"narratives": len(narratives),
"consistency_checks": len(consistency_checks),
"authority_checks": len(authority_checks),
"issues": len(issue_items),
"intelligence_runs": len(intelligence.get("runs", [])),
"model_suggestions_pending": sum(
isinstance(item, dict) and item.get("status") == "MODEL_SUGGESTED"
for item in intelligence.get("runs", [])
),
},
"issues": issues,
"limitations": [
"Mechanical validation does not establish source authority, legal interpretation, eligibility, cost admissibility, or filing readiness."
],
}
write_private_json(output_dir / "validation_audit.json", audit)
return audit
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--output-dir", required=True, type=Path)
parser.add_argument("--client-engagement", required=True, type=Path)
args = parser.parse_args(argv)
audit = validate_application(
output_dir=args.output_dir,
client_engagement=args.client_engagement,
)
LOGGER.info("Validation: %s (%s issues)", audit["status"], len(audit["issues"]))
return 0 if audit["status"] == "passed" else 1
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO, format="%(message)s")
raise SystemExit(main())
SHA-256: 2c8b5489e1d6c01cf06483782eb0e8f24bffe0caff332bb072348242d603e97b