"""Deterministically render one validated capability-fit assessment as Markdown."""
from __future__ import annotations
import hashlib
import json
from pathlib import Path
from typing import Any, Iterable, Mapping, Sequence
from jsonschema import Draft202012Validator, FormatChecker
from jsonschema.exceptions import SchemaError
MAX_ASSESSMENT_BYTES = 16 * 1024 * 1024
STATUS_DEFINITIONS = (
(
"verified",
"Implemented and directly exercised by evidence appropriate to the stated scope.",
),
(
"available_unconfigured",
"Implemented with supporting evidence, but not configured and exercised in the target.",
),
(
"partial",
"A useful subset exists, but a material part of the requirement is missing or unproved.",
),
(
"scaffold",
"Contracts or structure exist, but the end-to-end capability is not usable.",
),
(
"external_system",
"The deployment or another system must supply the capability.",
),
(
"planned",
"Only a concept, backlog item, or design direction exists.",
),
(
"not_fit",
"Evidence shows that the assessed composition cannot meet the requirement.",
),
(
"not_assessed",
"The requirement or target environment is not sufficiently known.",
),
)
class AssessmentGenerationError(ValueError):
"""The assessment cannot be safely validated or rendered."""
def load_bounded_json(path: Path, *, label: str) -> dict[str, Any]:
try:
size = path.stat().st_size
except OSError as exc:
raise AssessmentGenerationError(f"Could not inspect {label}: {exc}") from exc
if size > MAX_ASSESSMENT_BYTES:
raise AssessmentGenerationError(
f"{label} exceeds the {MAX_ASSESSMENT_BYTES}-byte input limit"
)
try:
payload = json.loads(
path.read_text(encoding="utf-8"),
object_pairs_hook=_unique_object,
)
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
raise AssessmentGenerationError(f"Could not read {label}: {exc}") from exc
if not isinstance(payload, dict):
raise AssessmentGenerationError(f"{label} must contain one JSON object")
return payload
def validate_report_input(
*,
assessment: Mapping[str, Any],
schema: Mapping[str, Any],
) -> None:
try:
Draft202012Validator.check_schema(schema)
except SchemaError as exc:
raise AssessmentGenerationError(
f"Assessment schema is invalid: {exc.message}"
) from exc
errors = sorted(
Draft202012Validator(
schema,
format_checker=FormatChecker(),
).iter_errors(assessment),
key=lambda item: tuple(str(part) for part in item.absolute_path),
)
if errors:
details = "; ".join(
f"{_json_path(error.absolute_path)}: {error.message}"
for error in errors[:20]
)
raise AssessmentGenerationError(f"Assessment does not match schema: {details}")
_validate_references(assessment)
_reject_sensitive_keys(assessment)
def render_assessment_markdown(assessment: Mapping[str, Any]) -> str:
"""Return stable Markdown derived only from a validated assessment object."""
assessment_hash = hashlib.sha256(
json.dumps(
assessment,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=True,
).encode("utf-8")
).hexdigest()
scope = _mapping(assessment["scope"])
release = _mapping(assessment["release"])
profile = _mapping(assessment["deployment_profile"])
lines = [
"# GovOPlaN Capability and IT-Infrastructure Fit Assessment",
"",
"> Generated from [`capability-fit-current.json`](../../capability-fit-current.json).",
"> Edit and validate the machine-readable assessment, then regenerate this file;",
"> do not maintain conclusions independently in Markdown.",
"",
"This is an evidence-based fit assessment, not a production approval or",
"security certification. Repository or manifest existence alone never counts",
"as an implemented capability. Unknown target requirements remain explicitly",
"`not_assessed`.",
"",
"## Assessment record",
"",
"| Field | Value |",
"| --- | --- |",
f"| Assessment ID | `{_cell(assessment['assessment_id'])}` |",
f"| Schema version | `govoplan.fit-assessment/{_cell(assessment['schema_version'])}` |",
f"| Assessed on | {_cell(assessment['assessed_at'])} |",
f"| Scope | {_cell(scope['title'])} |",
f"| Release | `{_cell(release['ref'])}` ({_cell(release['kind'])}) |",
f"| Meta commit | `{_cell(release['meta_commit'])}` |",
f"| Deployment profile | `{_cell(profile['id'])}` · `{_cell(profile['status'])}` |",
f"| Configuration packages | {_inline_list(release['configuration_packages'], code=True)} |",
f"| Canonical input SHA-256 | `{assessment_hash}` |",
"",
"## Controlled status vocabulary",
"",
"| Status | Meaning |",
"| --- | --- |",
]
lines.extend(
f"| `{status}` | {_cell(description)} |"
for status, description in STATUS_DEFINITIONS
)
lines.extend(
[
"",
"## Scope and reference journeys",
"",
"Reference journeys:",
"",
*_bullets(scope["reference_journeys"]),
"",
"Explicitly postponed:",
"",
*_bullets(scope["postponed"]),
"",
"## Facts",
"",
*_bullets(assessment["facts"]),
"",
"## Decisions",
"",
*_bullets(assessment["decisions"]),
"",
"## Assumptions",
"",
*_bullets(assessment["assumptions"]),
"",
"## Unresolved decisions",
"",
*_bullets(assessment["open_questions"]),
"",
"## Pinned release and composition",
"",
f"Release reproducible: **{'yes' if release['reproducible'] else 'no'}**.",
"",
]
)
lines.extend(_notes(release.get("notes", [])))
lines.extend(
[
"",
"| Module | Repository and commit | Manifest version | Enabled | Role |",
"| --- | --- | --- | --- | --- |",
]
)
for module_value in assessment["composition"]:
module = _mapping(module_value)
lines.append(
"| `{}` | `{}` @ `{}` | `{}` | {} | {} |".format(
_cell(module["module_id"]),
_cell(module["repository"]),
_cell(module["commit"]),
_cell(module["manifest_version"]),
"yes" if module["enabled"] else "no",
_cell(module["role"]),
)
)
lines.extend(
[
"",
"## Deployment profile",
"",
f"Status: `{_cell(profile['status'])}`",
"",
_text(profile["description"]),
"",
"Evidence:",
"",
*_bullets(_evidence_labels(profile["evidence"])),
"",
"## Recommended scenarios",
"",
]
)
for scenario_value in assessment["scenarios"]:
scenario = _mapping(scenario_value)
lines.extend(
[
f"### {_text(scenario['label'])}",
"",
f"Status: `{_cell(scenario['status'])}`",
"",
_text(scenario["recommendation"]),
"",
f"Composition: {_inline_list(scenario['composition'], code=True)}.",
"",
"Topology:",
"",
*_bullets(scenario["topology"]),
"",
"Conditions:",
"",
*_bullets(scenario["conditions"]),
"",
]
)
functional_context = _mapping(assessment["functional_context"])
lines.extend(
[
"## Functional matrix context",
"",
"### Required modules",
"",
*_bullets(functional_context["required_modules"]),
"",
"### Optional modules",
"",
*_bullets(functional_context["optional_modules"]),
"",
"### External systems and connectors",
"",
*_bullets(functional_context["external_systems"]),
"",
"### Missing contracts",
"",
*_bullets(functional_context["missing_contracts"]),
"",
"### Policy decisions",
"",
*_bullets(functional_context["policy_decisions"]),
"",
"### Manual workarounds",
"",
*_bullets(functional_context["manual_workarounds"]),
"",
"### Blockers",
"",
*_bullets(functional_context["blockers"]),
"",
]
)
lines.extend(
[
"## Assessment questionnaire",
"",
"Every required area remains visible even when its target answer is unknown.",
"",
"| Area | Question | State | Answer | Evidence |",
"| --- | --- | --- | --- | --- |",
]
)
questionnaire = _mapping(assessment["questionnaire"])
for area, answers in questionnaire.items():
for answer_value in _sequence(answers):
answer = _mapping(answer_value)
raw_answer = answer["answer"]
answer_text = (
_inline_list(raw_answer)
if isinstance(raw_answer, list)
else _text(raw_answer) if raw_answer is not None else "—"
)
lines.append(
"| {} | {} | `{}` | {} | {} |".format(
_cell(area.replace("_", " ").title()),
_cell(answer["question"]),
_cell(answer["state"]),
_cell(answer_text),
_cell("; ".join(_evidence_labels(answer["evidence"])) or "—"),
)
)
lines.extend(_assessed_matrix("Functional capability matrix", assessment["capabilities"]))
lines.extend(_assessed_matrix("Infrastructure matrix", assessment["infrastructure"]))
lines.extend(
[
"## Data flows and trust boundaries",
"",
"| Flow | From → to | Data | Trust boundary | Controls |",
"| --- | --- | --- | --- | --- |",
]
)
for flow_value in assessment["data_flows"]:
flow = _mapping(flow_value)
lines.append(
"| `{}` | {} → {} | {} | {} | {} |".format(
_cell(flow["id"]),
_cell(flow["from"]),
_cell(flow["to"]),
_cell(_inline_list(flow["data"])),
_cell(flow["trust_boundary"]),
_cell(_inline_list(flow["controls"])),
)
)
lines.extend(
[
"",
"## Risks and residual risks",
"",
"| Risk | Impact | Treatment | Owner | Residual risk |",
"| --- | --- | --- | --- | --- |",
]
)
for risk_value in assessment["risks"]:
risk = _mapping(risk_value)
lines.append(
"| **{}**
{} | {} | {} | {} | {} |".format(
_cell(risk["id"]),
_cell(risk["statement"]),
_cell(risk["impact"]),
_cell(risk["treatment"]),
_cell(risk["owner"] or "unassigned"),
_cell(risk["residual_risk"]),
)
)
lines.extend(
[
"",
"## Recommendations",
"",
*_bullets(assessment["recommendations"]),
"",
"## Proof-of-concept and promotion checks",
"",
*_numbered(assessment["proof_checks"]),
"",
"## Generation contract",
"",
"This report is deterministic output from the schema-validated JSON companion.",
"The generator rejects duplicate JSON keys, schema drift, secret-bearing field",
"names, stale checked-in output, and oversized inputs. A new assessment or",
"release changes the canonical input hash and requires review of the affected",
"evidence and conclusions through the release-aware reassessment tool.",
"",
]
)
return "\n".join(lines)
def _assessed_matrix(title: str, values: object) -> list[str]:
lines = [
"",
f"## {title}",
"",
"| Requirement | Status | Evidence | Conditions and gaps | Recommendation and proof |",
"| --- | --- | --- | --- | --- |",
]
for item_value in _sequence(values):
item = _mapping(item_value)
conditions = [f"Condition: {value}" for value in item["conditions"]]
gaps = [f"Gap: {value}" for value in item["gaps"]]
risks = [f"Risk: {value}" for value in item["risks"]]
lines.append(
"| **{}**
{} | `{}` | {} | {} | {}
**Proof:** {} |".format(
_cell(item["id"]),
_cell(item["requirement"]),
_cell(item["status"]),
_cell("; ".join(_evidence_labels(item["evidence"])) or "Explicit absence of evidence"),
_cell("; ".join([*conditions, *gaps, *risks]) or "—"),
_cell(item["recommendation"] or "—"),
_cell(item["proof_check"] or "—"),
)
)
return lines
def _evidence_labels(values: object) -> list[str]:
labels: list[str] = []
for value in _sequence(values):
item = _mapping(value)
label = f"{item['kind']}/{item['scope']}: {item['locator']}"
if item.get("note"):
label += f" ({item['note']})"
labels.append(label)
return labels
def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
result: dict[str, Any] = {}
for key, value in pairs:
if key in result:
raise AssessmentGenerationError(f"Duplicate JSON key: {key!r}")
result[key] = value
return result
def _reject_sensitive_keys(value: object, path: tuple[str, ...] = ()) -> None:
forbidden = {
"access_token",
"api_key",
"credential_value",
"password",
"private_key",
"refresh_token",
"secret",
}
if isinstance(value, Mapping):
for key, nested in value.items():
normalized = str(key).strip().casefold()
if normalized in forbidden:
raise AssessmentGenerationError(
f"Assessment contains forbidden sensitive field {_json_path((*path, str(key)))}"
)
_reject_sensitive_keys(nested, (*path, str(key)))
elif isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
for index, nested in enumerate(value):
_reject_sensitive_keys(nested, (*path, str(index)))
def _validate_references(assessment: Mapping[str, Any]) -> None:
composition = [_mapping(item) for item in _sequence(assessment["composition"])]
module_ids = [str(item["module_id"]) for item in composition]
if len(module_ids) != len(set(module_ids)):
raise AssessmentGenerationError("Composition contains duplicate module IDs")
modules = {str(item["module_id"]): item for item in composition}
context = _mapping(assessment["functional_context"])
required = {str(item) for item in _sequence(context["required_modules"])}
optional = {str(item) for item in _sequence(context["optional_modules"])}
unknown_context = (required | optional) - set(modules)
if unknown_context:
raise AssessmentGenerationError(
"Functional context references unknown modules: "
+ ", ".join(sorted(unknown_context))
)
if required & optional:
raise AssessmentGenerationError(
"Functional context cannot mark a module both required and optional"
)
for scenario_value in _sequence(assessment["scenarios"]):
scenario = _mapping(scenario_value)
referenced = {str(item) for item in _sequence(scenario["composition"])}
unknown = referenced - set(modules)
if unknown:
raise AssessmentGenerationError(
f"Scenario {scenario['id']!r} references unknown modules: "
+ ", ".join(sorted(unknown))
)
disabled = sorted(
module_id
for module_id in referenced
if not bool(modules[module_id]["enabled"])
)
if disabled:
raise AssessmentGenerationError(
f"Scenario {scenario['id']!r} references disabled modules: "
+ ", ".join(disabled)
)
for collection in ("capabilities", "infrastructure", "data_flows", "risks"):
identifiers = [
str(_mapping(item)["id"])
for item in _sequence(assessment[collection])
]
if len(identifiers) != len(set(identifiers)):
raise AssessmentGenerationError(
f"Assessment contains duplicate {collection} IDs"
)
def _mapping(value: object) -> Mapping[str, Any]:
if not isinstance(value, Mapping):
raise AssessmentGenerationError("Validated assessment contains a non-object value")
return value
def _sequence(value: object) -> Sequence[Any]:
if not isinstance(value, Sequence) or isinstance(value, (str, bytes)):
raise AssessmentGenerationError("Validated assessment contains a non-list value")
return value
def _text(value: object) -> str:
return str(value).strip()
def _cell(value: object) -> str:
return _text(value).replace("|", "\\|").replace("\r", " ").replace("\n", " ")
def _inline_list(values: object, *, code: bool = False) -> str:
items = [_text(item) for item in _sequence(values)]
if not items:
return "none"
if code:
return ", ".join(f"`{_cell(item)}`" for item in items)
return "; ".join(items)
def _bullets(values: object) -> list[str]:
items = [_text(item) for item in _sequence(values)]
return [f"- {item}" for item in items] or ["- None recorded."]
def _numbered(values: object) -> list[str]:
return [f"{index}. {_text(item)}" for index, item in enumerate(_sequence(values), 1)]
def _notes(values: object) -> list[str]:
items = _bullets(values)
return ["Release notes:", "", *items]
def _json_path(parts: Iterable[object]) -> str:
suffix = "".join(f"[{part}]" if str(part).isdigit() else f".{part}" for part in parts)
return f"${suffix}"
__all__ = (
"AssessmentGenerationError",
"MAX_ASSESSMENT_BYTES",
"load_bounded_json",
"render_assessment_markdown",
"validate_report_input",
)