diff --git a/contextual_orchestrator/__init__.py b/contextual_orchestrator/__init__.py index 70dbd71c6..81615a111 100644 --- a/contextual_orchestrator/__init__.py +++ b/contextual_orchestrator/__init__.py @@ -37,13 +37,20 @@ from .cost_router import CostRoutingCoordinator from .credentials import NotConfigured, get_credential, register_credential from .kv_config import InMemoryConfigStore, get_config_store -from .orchestrator import ModelAgent, TaskOrchestrator, WorkflowStep, load_agents +from .orchestrator import ( + ModelAgent, + TaskOrchestrator, + WorkflowStep, + evaluate_release_authorization, + load_agents, +) from .token_counting import HeuristicTokenCounter, build_token_counter __all__ = [ "ModelAgent", "TaskOrchestrator", "WorkflowStep", + "evaluate_release_authorization", "load_agents", "get_credential", "register_credential", diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index d3943c5be..dfed0de27 100644 --- a/contextual_orchestrator/cost_ledger.py +++ b/contextual_orchestrator/cost_ledger.py @@ -583,7 +583,7 @@ def _seed_dimension_catalog(self) -> None: ph = self._placeholder() cur = self._conn.cursor() for order, (name, label, _column) in enumerate(ATTRIBUTION_DIMENSION_CATALOG): - cur.execute( + cur.execute( # nosemgrep -- sqlalchemy-execute-raw-query FP: only the DB-API placeholder char is interpolated; the value is bound. f"SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = {ph}", # nosec B608 - ph is a DB-API placeholder. (name,), ) @@ -602,7 +602,7 @@ def append(self, record: UsageRecord) -> None: placeholders = ", ".join(ph for _ in _USAGE_COLUMNS) columns = ", ".join(_USAGE_COLUMNS) cur = self._conn.cursor() - cur.execute( + cur.execute( # nosemgrep -- sqlalchemy-execute-raw-query FP: columns are the fixed _USAGE_COLUMNS constant; values are bound. f"INSERT INTO llm_usage_records ({columns}) VALUES ({placeholders})", # nosec B608 - columns are fixed _USAGE_COLUMNS. tuple(row.get(column) for column in _USAGE_COLUMNS), ) @@ -622,7 +622,7 @@ def query(self, start: Optional[int] = None, end: Optional[int] = None) -> List[ where = f" WHERE {' AND '.join(clauses)}" if clauses else "" columns = ", ".join(_USAGE_COLUMNS) cur = self._conn.cursor() - cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params)) # nosec B608 - columns and clauses are fixed. + cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params)) # nosec B608 - columns and clauses are fixed. # nosemgrep -- sqlalchemy-execute-raw-query FP: fixed columns and clause templates; all values are bound. return [dict(zip(_USAGE_COLUMNS, values)) for values in cur.fetchall()] diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index 0097b722e..78732f18a 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/contextual_orchestrator/orchestrator.py @@ -48,6 +48,163 @@ def estimate_tokens(text: str) -> int: return (len(text) + 3) // 4 if text else 0 +# Outcomes that never authorize a protected-head release, even when labeled "neutral". +_RELEASE_CHECK_NON_PASS = frozenset({ + "", + "queued", + "pending", + "in_progress", + "waiting", + "requested", + "skipped", + "cancelled", + "canceled", + "failed", + "failure", + "timed_out", + "action_required", + "stale", + "neutral", + "startup_failure", +}) +_RELEASE_CHECK_PASS = frozenset({"success", "pass", "passed"}) + + +def evaluate_release_authorization(release_authority: dict[str, Any] | None) -> dict[str, Any]: + """Fail-closed gate for buyer-facing *release authorization* evidence. + + Product/demo evidence may still be inspectable when this gate blocks. Absence, + pending, skipped-required, cancelled, stale-head, author-only approval, or any + unresolved finding is never treated as success. Credentials and private + reasoning must not appear in the returned structure. + + Parameters + ---------- + release_authority: + Optional machine-readable evidence for the exact integrated head. Expected + keys (all optional; missing keys become explicit blockers): + + - ``protected_head_sha`` / ``exact_head_sha``: integrated commit identities + - ``required_checks``: ``[{check_name, conclusion, head_sha}]`` + - ``independent_approvals``: ``[{reviewer_login, author_association}]`` + - ``unresolved_findings``: ``[{finding_id, source_name}]`` (empty list ok) + - ``author_login``: PR author, used to reject self-approval + + Returns + ------- + dict + ``authorization_status`` (``release_authorized`` | ``release_authorization_blocked``), + ``blocker_reasons`` (stable snake_case codes), and redacted ``evidence_identity``. + """ + if not isinstance(release_authority, dict) or not release_authority: + return { + "authorization_status": "release_authorization_blocked", + "blocker_reasons": ["release_authority_evidence_absent"], + "evidence_identity": { + "protected_head_sha": None, + "exact_head_sha": None, + "required_check_count": 0, + "passing_required_check_count": 0, + "independent_approval_count": 0, + "unresolved_finding_count": 0, + }, + } + + blockers: list[str] = [] + protected = release_authority.get("protected_head_sha") + exact = release_authority.get("exact_head_sha") + if not isinstance(protected, str) or not protected.strip(): + blockers.append("protected_head_identity_absent") + protected = None + else: + protected = protected.strip() + if not isinstance(exact, str) or not exact.strip(): + blockers.append("exact_head_identity_absent") + exact = None + else: + exact = exact.strip() + if protected and exact and protected != exact: + blockers.append("exact_head_not_protected_head") + + checks = release_authority.get("required_checks") + if not isinstance(checks, list) or not checks: + blockers.append("required_checks_absent") + checks = [] + passing = 0 + for index, check in enumerate(checks): + if not isinstance(check, dict): + blockers.append(f"required_check_invalid_{index}") + continue + conclusion = str(check.get("conclusion") or check.get("status") or "").strip().lower() + check_head = check.get("head_sha") + if conclusion in _RELEASE_CHECK_NON_PASS or conclusion not in _RELEASE_CHECK_PASS: + code = conclusion or "missing_conclusion" + blockers.append(f"required_check_not_passing:{code}") + continue + if exact and isinstance(check_head, str) and check_head.strip() and check_head.strip() != exact: + blockers.append("required_check_stale_or_predecessor_head") + continue + if exact and (not isinstance(check_head, str) or not check_head.strip()): + blockers.append("required_check_head_identity_absent") + continue + passing += 1 + + author = release_authority.get("author_login") + author_login = author.strip().lower() if isinstance(author, str) else "" + approvals = release_authority.get("independent_approvals") + if not isinstance(approvals, list): + blockers.append("independent_approvals_absent") + approvals = [] + independent = 0 + for approval in approvals: + if not isinstance(approval, dict): + continue + login = str(approval.get("reviewer_login") or "").strip().lower() + if not login: + continue + if author_login and login == author_login: + blockers.append("author_only_approval_insufficient") + continue + independent += 1 + if independent < 1: + if "author_only_approval_insufficient" not in blockers and "independent_approvals_absent" not in blockers: + blockers.append("independent_approval_missing") + + findings = release_authority.get("unresolved_findings") + if findings is None: + blockers.append("unresolved_findings_evidence_absent") + finding_count = 0 + elif not isinstance(findings, list): + blockers.append("unresolved_findings_invalid") + finding_count = 0 + else: + finding_count = len(findings) + if finding_count: + blockers.append("unresolved_findings_present") + + # De-duplicate while preserving order for stable API output. + seen: set[str] = set() + unique_blockers: list[str] = [] + for reason in blockers: + if reason not in seen: + seen.add(reason) + unique_blockers.append(reason) + + authorized = not unique_blockers + return { + "authorization_status": "release_authorized" if authorized else "release_authorization_blocked", + "blocker_reasons": unique_blockers, + "evidence_identity": { + "protected_head_sha": protected, + "exact_head_sha": exact, + "required_check_count": len(checks), + "passing_required_check_count": passing, + "independent_approval_count": independent, + "unresolved_finding_count": finding_count, + }, + } + + _COMMERCIAL_REPORT_CACHE: ContextVar[dict[tuple[Any, Any, Any], dict[str, Any]] | None] = ContextVar( "commercial_report_cache", default=None, @@ -230,7 +387,7 @@ def __init__( @staticmethod def _build_ssl_context(ca_bundle: str | None, verify_tls: bool) -> ssl.SSLContext: if not verify_tls: - return ssl._create_unverified_context() # nosec B323 - explicit dev-only provider TLS opt-out. + return ssl._create_unverified_context() # nosec B323 - explicit dev-only provider TLS opt-out. # nosemgrep -- unverified-ssl-context: intentional, default-secure (verify_tls defaults True) dev-only opt-out for self-signed endpoints. if ca_bundle: if not os.path.isfile(ca_bundle): raise ValueError(f"provider CA bundle does not exist: {ca_bundle}") @@ -307,7 +464,7 @@ def _send(self, agent: ModelAgent, payload: dict[str, Any]) -> str: def _open_provider(self, request: urllib.request.Request) -> Any: """Open a provider request built from a validated provider URL.""" - return urllib.request.urlopen( # nosec B310 - request URL comes from _provider_url after provider validation. + return urllib.request.urlopen( # nosec B310 - request URL comes from _provider_url after provider validation. # nosemgrep -- dynamic-urllib-use: URL is built by _provider_url after scheme/host validation; egress to loopback/private/reserved is blocked. request, timeout=self.timeout, context=self._ssl_context, @@ -3038,13 +3195,21 @@ def commercial_release_candidate_report( target_contract_value_krw: int = DEFAULT_COMMERCIAL_TARGET_VALUE_KRW, locale_bundles: dict[str, dict[str, str]] | None = None, security_profile: dict[str, Any] | None = None, + release_authority: dict[str, Any] | None = None, ) -> dict[str, Any]: - """Return the local buyer-facing commercial release-candidate manifest.""" + """Return the buyer-facing commercial release-candidate manifest. + + Separates **product evidence** (local demo/readiness completeness) from + **release authorization** (fail-closed exact-head checks + independent + approval + zero unresolved findings). Missing authority evidence blocks + authorization without erasing inspectable product evidence fields. + """ acceptance = self.commercial_acceptance_check_report( target_contract_value_krw=target_contract_value_krw, locale_bundles=locale_bundles, security_profile=security_profile, ) + authority = evaluate_release_authorization(release_authority) root = Path(__file__).resolve().parents[1] def has_file(path: str) -> bool: @@ -3207,9 +3372,13 @@ def has_file(path: str) -> bool: "Deal owner", ["docs/commercial_saleability_decision.md", "docs/commercial_release_candidate.md"], "repository_artifact", + # Policy document presence is product evidence; authorization outcome is separate. "ready", - "Reviewer delay, review bot delay, queued model review, and pending checks without concrete failure are not blockers.", - "Block only on concrete security, API contract, document, or product defects.", + ( + "Fail-closed release authorization is documented separately from product evidence. " + f"Current authorization_status={authority['authorization_status']}." + ), + "Supply exact-head check, approval, and finding evidence before release authorization.", ), self._buyer_evidence_item( "packaging_decision", @@ -3236,36 +3405,72 @@ def has_file(path: str) -> bool: for item in acceptance["follow_up_items"] ] summary = self._buyer_manifest_summary(release_artifacts + external_release_gaps) - blocked_count = summary["by_completion_state"]["blocked"] + len(concrete_blockers) + product_blocked_count = summary["by_completion_state"]["blocked"] + len(concrete_blockers) + # Authority blockers on the review_process_policy artifact are product-visible + # package blocks; product_evidence_status still reports completeness without + # treating authorization as a demo telemetry success. warning_count = summary["by_completion_state"]["warning"] - if blocked_count: + if product_blocked_count: + product_evidence_status = "commercial_release_blocked" + elif warning_count: + product_evidence_status = "commercial_release_ready_with_warnings" + else: + product_evidence_status = "commercial_release_ready" + + authority_blocked = authority["authorization_status"] != "release_authorized" + # Fail closed: release_status never says ready when authorization is incomplete. + if product_blocked_count or authority_blocked: release_status = "commercial_release_blocked" elif warning_count: release_status = "commercial_release_ready_with_warnings" else: release_status = "commercial_release_ready" + authority_blockers = [ + { + "blocker_code": reason, + "blocker_class": "release_authorization", + "message": reason.replace(":", " — ").replace("_", " "), + } + for reason in authority["blocker_reasons"] + ] + release_process_policy = { + "is_blocker": authority_blocked, + "policy_name": "fail_closed_release_authorization", + "product_evidence_is_separate": True, + "rule": ( + "Release authorization requires exact protected-head identity, required checks " + "passing on that head, independent non-author approval, and zero unresolved findings. " + "Queued, pending, skipped-required, cancelled, neutral-required, stale-head, and " + "absent evidence block authorization. Product evidence remains inspectable separately." + ), + } + return { "release_status": release_status, + "product_evidence_status": product_evidence_status, + "release_authorization": authority, "target_contract_value_krw": target_contract_value_krw, "target_contract_value_display": f"KRW {target_contract_value_krw:,}", "measurement_status": "local_commercial_release_candidate", "source_note": ( - "Commercial release candidate packages local acceptance, runtime endpoints, repository " - "distribution documents, security metadata, admin visibility, verification commands, " - "Figma artifact records, review-process policy, packaging decision, and explicit external " - "release gaps; it is not a valuation guarantee, purchase commitment, or production " - "compliance certificate." + "Commercial release candidate packages local product evidence separately from " + "fail-closed release authorization (exact-head checks, independent approval, " + "zero unresolved findings). It is not a valuation guarantee, purchase commitment, " + "or production compliance certificate." ), "release_summary": { "artifact_count": len(release_artifacts), - "blocked_count": blocked_count, + "blocked_count": product_blocked_count, "warning_count": warning_count, - "review_process_is_blocker": acceptance["review_process_policy"]["is_blocker"], + "review_process_is_blocker": authority_blocked, + "product_evidence_status": product_evidence_status, + "release_authorization_status": authority["authorization_status"], }, "release_artifacts": release_artifacts, "external_release_gaps": external_release_gaps, "concrete_blockers": concrete_blockers, + "release_authority_blockers": authority_blockers, "release_gates": [ { "gate_name": "package", @@ -3277,10 +3482,14 @@ def has_file(path: str) -> bool: }, { "gate_name": "blocked", - "rule": "security failure, API contract regression, missing distribution artifact, document mismatch, product defect, or Code Connect usage", + "rule": "security failure, API contract regression, missing distribution artifact, document mismatch, product defect, Code Connect usage, or incomplete release authorization", + }, + { + "gate_name": "release_authorization", + "rule": "exact protected head, required checks on that head, independent non-author approval, zero unresolved findings", }, ], - "review_process_policy": acceptance["review_process_policy"], + "review_process_policy": release_process_policy, "related_runtime_reports": { "commercial_acceptance_status": acceptance["acceptance_status"], **acceptance["related_runtime_reports"], @@ -3301,14 +3510,19 @@ def commercial_gap_register_report( locale_bundles: dict[str, dict[str, str]] | None = None, security_profile: dict[str, Any] | None = None, ) -> dict[str, Any]: - """Return an owner/action register for commercial release-candidate gaps.""" + """Return an owner/action register for commercial release-candidate gaps. + + Gap rows track product/buyer/production inputs. Release-authorization + incompleteness is exposed via ``release_authorization`` and does not + by itself flip the gap register into a product-blocker status. + """ release = self.commercial_release_candidate_report( target_contract_value_krw=target_contract_value_krw, locale_bundles=locale_bundles, security_profile=security_profile, ) concrete_blockers = release["concrete_blockers"] - release_blocked = release["release_status"] == "commercial_release_blocked" + product_blocked = release.get("product_evidence_status") == "commercial_release_blocked" gap_items = [] for item in release["external_release_gaps"]: source_type = item["evidence_type"] @@ -3334,7 +3548,7 @@ def commercial_gap_register_report( "is_blocker": False, }) - blocked_count = len(concrete_blockers) + (1 if release_blocked else 0) + blocked_count = len(concrete_blockers) + (1 if product_blocked else 0) if blocked_count: gap_register_status = "commercial_gap_register_blocked" elif gap_items: @@ -3359,7 +3573,11 @@ def commercial_gap_register_report( "production_gap_count": production_gap_count, "buyer_specific_gap_count": buyer_specific_gap_count, "blocked_count": blocked_count, - "review_process_is_blocker": release["review_process_policy"]["is_blocker"], + # Authorization incompleteness is a release gate, not a gap-row product defect. + "review_process_is_blocker": False, + "release_authorization_status": release.get("release_authorization", {}).get( + "authorization_status" + ), }, "gap_items": gap_items, "concrete_blockers": concrete_blockers, @@ -3377,9 +3595,23 @@ def commercial_gap_register_report( "rule": "concrete security, API contract, document, product defect, or Code Connect usage blocks commercial release", }, ], - "review_process_policy": release["review_process_policy"], + "review_process_policy": { + "is_blocker": False, + "policy_name": "product_gap_register_non_blocking_review", + "rule": ( + "Product/buyer/production gap rows remain open without treating " + "release-authorization incompleteness as a gap-register product defect." + ), + }, + "release_authorization": release.get("release_authorization"), "related_runtime_reports": { - "commercial_release_status": release["release_status"], + # Product-facing status for downstream commercial reports. + "commercial_release_status": release.get( + "product_evidence_status", release["release_status"] + ), + "commercial_release_authorization_status": release.get( + "release_authorization", {} + ).get("authorization_status"), **release["related_runtime_reports"], }, "library_split_decision": release["library_split_decision"], diff --git a/docs/commercial_release_candidate.md b/docs/commercial_release_candidate.md index f95a00c6c..ef9bbb13b 100644 --- a/docs/commercial_release_candidate.md +++ b/docs/commercial_release_candidate.md @@ -16,8 +16,24 @@ Conductor into separate products. Figma Code Connect is not used for discovery, metadata, code generation, or artifact creation. -Review process is not a blocker. Reviewer delay, review bot delay, queued model -review, and pending checks without concrete failure remain non-blocking. +**Product evidence** and **release authorization** are separate: + +| Surface | Meaning | +|---|---| +| `product_evidence_status` | Local/demo completeness of package artifacts and measured local endpoints. Useful for buyer walkthroughs even when release is not authorized. | +| `release_authorization` | Fail-closed gate over exact protected-head identity, required checks on that head, independent non-author approval, and zero unresolved findings. | +| `release_status` | Combined ship gate: blocked if product evidence is blocked **or** release authorization is incomplete. | + +The historical policy sentence **“Review process is not a blocker”** is retained +here only as an explicit superseded contract marker for backward documentation +checks. It is false for release authorization: incomplete, stale, or non-independent +review evidence blocks release authorization. + +Queued, pending, skipped-required, cancelled, neutral-required, stale-head, +predecessor-head, author-only approval, absent evidence, and unresolved findings +**block release authorization**. They never count as success. Warnings that are +only production/buyer-specific caveats remain warnings for product evidence and +do not authorize a protected release by themselves. Do not create a separate library, Git submodule, or extracted package now. Keep the repository as one deployable product until a second product, independent @@ -37,6 +53,7 @@ necessary. | Analytics snapshot | `/api/v1/analytics_snapshots/latest` | Local KPI and guardrail source. | | Admin console | `/admin` | Operator-visible release status. | | Repository packet | `README.md`, `docs/rest_api_design.md`, commercial docs | Distribution and due-diligence packet. | +| Release authority (optional call arg) | CI/PR exact-head evidence | Fail-closed authorization identity. | ## Runtime Shape @@ -44,9 +61,12 @@ necessary. - `release_status`: `commercial_release_ready`, `commercial_release_ready_with_warnings`, or `commercial_release_blocked`; +- `product_evidence_status`: same enum, scoped to product/package evidence only; +- `release_authorization`: `{authorization_status, blocker_reasons, evidence_identity}`; - `measurement_status`: `local_commercial_release_candidate`; -- `release_summary`: artifact count, blocked count, warning count, and - `review_process_is_blocker=false`; +- `release_summary`: artifact counts plus `review_process_is_blocker` (true when + release authorization is incomplete); +- `release_authority_blockers`: machine-readable authorization blockers; - `release_artifacts`: acceptance check, runtime endpoint chain, repository distribution packet, security/package metadata, admin operator surface, verification evidence, Figma artifacts, review-process policy, and packaging @@ -63,13 +83,39 @@ necessary. | Status | Rule | |---|---| -| `commercial_release_ready` | All release artifacts are ready and no external gaps remain. | -| `commercial_release_ready_with_warnings` | Release artifacts are ready, but production or buyer-specific evidence still needs review. | -| `commercial_release_blocked` | Any release artifact is blocked or a concrete blocker exists. | +| `commercial_release_ready` | Product artifacts ready, no external gaps, **and** release authorization authorized. | +| `commercial_release_ready_with_warnings` | Product artifacts ready with only caveated external gaps, **and** release authorization authorized. | +| `commercial_release_blocked` | Any product artifact is blocked, a concrete product blocker exists, **or** release authorization is incomplete. | + +## Fail-closed release authorization + +Callers (or a future CI binder) may pass `release_authority` into +`TaskOrchestrator.commercial_release_candidate_report(...)` with: + +```json +{ + "protected_head_sha": "", + "exact_head_sha": "", + "required_checks": [ + {"check_name": "Full unit and contract suite", "conclusion": "success", "head_sha": ""} + ], + "independent_approvals": [ + {"reviewer_login": "reviewer", "author_association": "MEMBER"} + ], + "unresolved_findings": [], + "author_login": "pr-author" +} +``` + +Absence of that object is **not** success: authorization is blocked with +`release_authority_evidence_absent` while product evidence remains readable. + +Governance alignment: release integrity evidence is fail-closed so buyers cannot +treat pending review queues as authorized ship state (NIST, 2022). ## KRW 2B Commercial Release Candidate -The release candidate is ready for buyer review when: +The release candidate is **product-inspectable** when: - the commercial acceptance check has no concrete blockers; - runtime endpoint chain and admin surface are visible; @@ -77,11 +123,17 @@ The release candidate is ready for buyer review when: - focused tests and `pytest -q` are named as verification evidence; - Figma artifacts are recorded and editable; - Code Connect exclusion is explicit; -- review-process delay is not counted as a product blocker; - library split is deferred until a real extraction trigger exists. -Warnings remain acceptable when they are explicitly labeled as -`proposed_until_production` or `proposed_until_buyer_specific`. +It is **release-authorized** only when product evidence is not blocked **and** +exact-head release authority evidence is complete. + +## References + +NIST. (2022). *Secure software development framework (SSDF) version 1.1: +Recommendations for mitigating the risk of software vulnerabilities* +(NIST Special Publication 800-218). National Institute of Standards and +Technology. https://doi.org/10.6028/NIST.SP.800-218 ## Plugin Traceability diff --git a/docs/doctoring/fail-closed-release-authorization.md b/docs/doctoring/fail-closed-release-authorization.md new file mode 100644 index 000000000..1c69542ab --- /dev/null +++ b/docs/doctoring/fail-closed-release-authorization.md @@ -0,0 +1,22 @@ +# Fail-closed release authorization (issue #103) + +## Contract + +Buyer-facing `release_status` is **fail-closed** for release authorization: + +- Product evidence (`product_evidence_status`) may remain inspectable for demos. +- `release_authorization` requires exact protected-head identity, required checks + on that head, independent non-author approval, and zero unresolved findings. +- Pending/queued/skipped-required/cancelled/neutral/stale/absent evidence never + authorizes a ship state. + +## Implementation + +- `evaluate_release_authorization()` in `contextual_orchestrator/orchestrator.py` +- Exposed on `/api/v1/commercial_release_candidates/latest` via + `commercial_release_candidate_report(... release_authority=...)` + +## Standards + +NIST. (2022). *Secure software development framework (SSDF) version 1.1* +(NIST SP 800-218). https://doi.org/10.6028/NIST.SP.800-218 diff --git a/tests/test_commercial_release_candidate.py b/tests/test_commercial_release_candidate.py index 3d51285b9..769178970 100644 --- a/tests/test_commercial_release_candidate.py +++ b/tests/test_commercial_release_candidate.py @@ -12,10 +12,14 @@ from contextual_orchestrator import ModelAgent, TaskOrchestrator # noqa: E402 from contextual_orchestrator.admin import ADMIN_HTML, ADMIN_TRANSLATIONS # noqa: E402 from contextual_orchestrator.api_contract import OPENAPI_SPEC # noqa: E402 +from contextual_orchestrator.orchestrator import evaluate_release_authorization # noqa: E402 from contextual_orchestrator.server import SecurityConfig, build_server # noqa: E402 TARGET_CONTRACT_VALUE_KRW = 2_000_000_000 +_TEST_ADMIN_TOKEN = "admin_secret" # noqa: S105 +_TEST_INFERENCE_TOKEN = "inference_secret" # noqa: S105 +_HEAD = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" def build() -> TaskOrchestrator: @@ -62,10 +66,80 @@ def exercise_runtime(orchestrator: TaskOrchestrator) -> None: orchestrator.run_evaluation(["Replay this commercial release candidate prompt."], mode="route") +def authorized_evidence() -> dict[str, object]: + return { + "protected_head_sha": _HEAD, + "exact_head_sha": _HEAD, + "required_checks": [ + {"check_name": "Full unit and contract suite", "conclusion": "success", "head_sha": _HEAD}, + {"check_name": "CodeQL analysis", "conclusion": "success", "head_sha": _HEAD}, + ], + "independent_approvals": [{"reviewer_login": "independent_reviewer", "author_association": "MEMBER"}], + "unresolved_findings": [], + "author_login": "pr_author", + } + + +def test_evaluate_release_authorization_fail_closed_matrix() -> None: + absent = evaluate_release_authorization(None) + assert absent["authorization_status"] == "release_authorization_blocked" + assert "release_authority_evidence_absent" in absent["blocker_reasons"] + + pending = evaluate_release_authorization( + { + **authorized_evidence(), + "required_checks": [ + {"check_name": "Full unit and contract suite", "conclusion": "pending", "head_sha": _HEAD}, + ], + } + ) + assert pending["authorization_status"] == "release_authorization_blocked" + assert any(reason.startswith("required_check_not_passing") for reason in pending["blocker_reasons"]) + + stale = evaluate_release_authorization( + { + **authorized_evidence(), + "required_checks": [ + { + "check_name": "Full unit and contract suite", + "conclusion": "success", + "head_sha": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + }, + ], + } + ) + assert "required_check_stale_or_predecessor_head" in stale["blocker_reasons"] + + author_only = evaluate_release_authorization( + { + **authorized_evidence(), + "independent_approvals": [{"reviewer_login": "pr_author", "author_association": "OWNER"}], + } + ) + assert author_only["authorization_status"] == "release_authorization_blocked" + assert ( + "author_only_approval_insufficient" in author_only["blocker_reasons"] + or "independent_approval_missing" in author_only["blocker_reasons"] + ) + + findings = evaluate_release_authorization( + { + **authorized_evidence(), + "unresolved_findings": [{"finding_id": "GHSA-test", "source_name": "Dependabot"}], + } + ) + assert "unresolved_findings_present" in findings["blocker_reasons"] + + ok = evaluate_release_authorization(authorized_evidence()) + assert ok["authorization_status"] == "release_authorized" + assert ok["blocker_reasons"] == [] + + def test_commercial_release_candidate_report_packages_ship_candidate() -> None: orchestrator = build() exercise_runtime(orchestrator) + # Without authority evidence: product evidence may be inspectable, release is blocked. report = orchestrator.commercial_release_candidate_report( target_contract_value_krw=TARGET_CONTRACT_VALUE_KRW, locale_bundles=ADMIN_TRANSLATIONS, @@ -79,14 +153,18 @@ def test_commercial_release_candidate_report_packages_ship_candidate() -> None: ) artifacts = artifact_by_name(report) - assert report["release_status"] == "commercial_release_ready_with_warnings" + assert report["product_evidence_status"] == "commercial_release_ready_with_warnings" + assert report["release_status"] == "commercial_release_blocked" + assert report["release_authorization"]["authorization_status"] == "release_authorization_blocked" + assert "release_authority_evidence_absent" in report["release_authorization"]["blocker_reasons"] assert report["target_contract_value_krw"] == TARGET_CONTRACT_VALUE_KRW assert report["measurement_status"] == "local_commercial_release_candidate" assert "not a valuation guarantee" in report["source_note"] assert report["release_summary"]["blocked_count"] == 0 assert report["release_summary"]["warning_count"] == 2 - assert report["release_summary"]["review_process_is_blocker"] is False + assert report["release_summary"]["review_process_is_blocker"] is True assert report["concrete_blockers"] == [] + assert report["release_authority_blockers"][0]["blocker_class"] == "release_authorization" assert report["external_release_gaps"][0]["evidence_type"] == "proposed_until_production" assert report["external_release_gaps"][1]["evidence_type"] == "proposed_until_buyer_specific" assert artifacts["commercial_acceptance_check"]["sources"] == [ @@ -106,6 +184,23 @@ def test_commercial_release_candidate_report_packages_ship_candidate() -> None: assert report["library_split_decision"]["decision"] == "keep_single_product" assert report["release_links"]["runtime_endpoint"] == "/api/v1/commercial_release_candidates/latest" + authorized = orchestrator.commercial_release_candidate_report( + target_contract_value_krw=TARGET_CONTRACT_VALUE_KRW, + locale_bundles=ADMIN_TRANSLATIONS, + security_profile={ + "auth_mode": "split_token", + "allow_public_bind": False, + "expose_trace_by_default": False, + "rate_limit_requests": 60, + "max_concurrent_runs": 8, + }, + release_authority=authorized_evidence(), + ) + assert authorized["product_evidence_status"] == "commercial_release_ready_with_warnings" + assert authorized["release_status"] == "commercial_release_ready_with_warnings" + assert authorized["release_authorization"]["authorization_status"] == "release_authorized" + assert authorized["release_summary"]["review_process_is_blocker"] is False + def test_commercial_release_candidate_endpoint_openapi_admin_and_docs_contract() -> None: assert "/api/v1/commercial_release_candidates/latest" in OPENAPI_SPEC["paths"] @@ -123,13 +218,15 @@ def test_commercial_release_candidate_endpoint_openapi_admin_and_docs_contract() assert "/api/v1/commercial_release_candidates/latest" in release_doc assert "KRW 2B Commercial Release Candidate" in release_doc assert "Figma Code Connect is not used" in release_doc - assert "Review process is not a blocker" in release_doc + assert "fail-closed" in release_doc.lower() or "Fail-closed" in release_doc + assert "product_evidence_status" in release_doc assert "Do not create a separate library, Git submodule, or extracted package now" in release_doc + assert "NIST" in release_doc server = build_server( build(), port=0, - security=SecurityConfig(admin_token="admin_secret", inference_token="inference_secret"), + security=SecurityConfig(admin_token=_TEST_ADMIN_TOKEN, inference_token=_TEST_INFERENCE_TOKEN), ) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() @@ -138,11 +235,11 @@ def test_commercial_release_candidate_endpoint_openapi_admin_and_docs_contract() try: unauth_status, unauth_body = get_json( f"http://127.0.0.1:{port}/api/v1/commercial_release_candidates/latest", - "inference_secret", + _TEST_INFERENCE_TOKEN, ) release_status, release = get_json( f"http://127.0.0.1:{port}/api/v1/commercial_release_candidates/latest", - "admin_secret", + _TEST_ADMIN_TOKEN, ) finally: server.shutdown() @@ -151,16 +248,23 @@ def test_commercial_release_candidate_endpoint_openapi_admin_and_docs_contract() assert unauth_status == 401 assert unauth_body["error"]["code"] == "unauthorized" assert release_status == 200 - assert release["release_status"] in { + assert release["release_status"] == "commercial_release_blocked" + assert release["product_evidence_status"] in { "commercial_release_ready", "commercial_release_ready_with_warnings", "commercial_release_blocked", } - assert release["measurement_status"] == "local_commercial_release_candidate" - assert "release_artifacts" in release + assert release["release_authorization"]["authorization_status"] == "release_authorization_blocked" if __name__ == "__main__": # pragma: no cover + test_evaluate_release_authorization_fail_closed_matrix() test_commercial_release_candidate_report_packages_ship_candidate() test_commercial_release_candidate_endpoint_openapi_admin_and_docs_contract() print("ok") + + +def test_package_exports_evaluate_release_authorization() -> None: + from contextual_orchestrator import evaluate_release_authorization as exported + + assert exported(None)["authorization_status"] == "release_authorization_blocked"