diff --git a/architecture/README.md b/architecture/README.md index e2c9b244e..8c1179902 100644 --- a/architecture/README.md +++ b/architecture/README.md @@ -36,6 +36,8 @@ architecture/ ├── layer-contract.schema.json ├── layer-baseline.json ├── layer-baseline.schema.json +├── lifecycle-contract.json +├── lifecycle-contract.schema.json ├── deltas/ │ └── .json └── README.md @@ -51,6 +53,8 @@ architecture/ - `layer-contract.schema.json`:layer contract 的結構 schema。 - `layer-baseline.json`:**已核准的 layer baseline**(grandfathered 跨層違規+每 service 零寬鬆的 violation budget);在 GitHub PR 上同樣只能相對 PR base 單調縮減,不能用 candidate 自己擴張的 baseline 替新違規開脫。 - `layer-baseline.schema.json`:layer baseline 的結構 schema。 +- `lifecycle-contract.json`:**review-session / endpoint-lease / stage-binding 三個 coordinator 端狀態機的機器定義**(Phase 4):states、observed transitions、forbidden shortcuts、evidence gates、reentry 規則、cross-machine 規則與 readiness evidence binding。 +- `lifecycle-contract.schema.json`:lifecycle contract 的結構 schema。 ## 3. 第一版硬規則 @@ -62,6 +66,7 @@ architecture/ | `ARCH-CALL-001` | 新 service edge 必須同時存在於 desired contract 與 change delta | semantic validator | | `ARCH-GRAPH-001` | 不得新增 dependency cycle | observed-graph ratchet(已 active) | | `ARCH-LAYER-001` | service 內部 module 只能依賴 layer set 允許的層;新跨層違規一律擋下 | layer-boundary ratchet(已 active) | +| `ARCH-LIFECYCLE-001` | 三個 lifecycle 機器必須 well-formed、state 集與 TS union 同步、forbidden shortcut 無直達邊、readiness binding 與 policy 一致 | lifecycle-contract gate(已 active) | | `ARCH-READY-001` | `ready` 必須同時有 Kit-side 與 browser-side evidence,包括 first frame 與 stage match | semantic validator + existing runtime evidence | | `ARCH-UI-001` | user-facing capability 必須前端可操作並有 browser E2E evidence | delegated to existing frontend operability gates | | `ARCH-DELTA-001` | Lane G / S 架構變更必須提交 architecture delta | delta schema + semantic validator | @@ -117,6 +122,10 @@ python -m pytest tests/test_observed_architecture.py -q -p no:cacheprovider python scripts/dev/check_layered_architecture.py --repo-root . --strict python scripts/dev/check_layered_architecture.py --repo-root . --report-only --output artifacts/architecture/layer-report.json python -m pytest tests/test_layered_architecture.py -q -p no:cacheprovider + +# lifecycle contracts(Phase 4) +python scripts/dev/check_lifecycle_contracts.py --repo-root . --strict +python -m pytest tests/test_lifecycle_contracts.py -q -p no:cacheprovider ``` `--report-only` 產出的 report 是 **可重生的本機產物**(`artifacts/architecture/` 已 gitignore),同一份 source tree 在 Windows 與 Linux 會得到 byte-identical 輸出。入庫的權威是 `observed-baseline.json`。 @@ -189,10 +198,10 @@ finding 以下仍是後續 phase,不應被目前文件或 PR 誤報為已完成: -- `review-session`、`endpoint-lease`、`stage-binding` executable state machines(Phase 4)。 - architecture quality grade 與定期 architecture garbage collection(Phase 5)。 - 跨 service 的 module-level layer 比對;目前只在 service 內部判定。 - 動態 import 與執行期才決定的 module 名稱;靜態掃描看不到,因此不宣稱涵蓋。 +- lifecycle contract 對 transition **行為**的執行期驗證;Phase 4 的 gate 只驗 contract 一致性與 state 集同步,行為由各 service 自己的 runtime 與測試持有(見 Phase 4 界線)。 ### Phase 2 的已知偏離與界線(誠實揭露) @@ -215,3 +224,13 @@ finding - **warning 不會讓 `status` 變 failed。** 與 Phase 2 相同:`layer.baseline_stale`、`layer.rule.unused`、`layer.budget_unknown_service` 都是 warning,CLI 不加 `--strict` 時 exit 0。真正把 warning 當紅燈的是 `tests/test_layered_architecture.py::test_canonical_repository_layer_ratchet_passes` 的 `warning_count == 0` 斷言,那才是 CI 的 oracle。 - **`bim-streaming-server` 與 `kit-manager-api` 只有 `exact` 規則。** 因為 suffix 規則已被禁止、這兩個 service 也沒有可用的 anchored prefix,所以在它們裡面**新增任何 Python module 都必須同時改 contract**,否則 `layer.module.unassigned` 會紅。這是刻意的成本。 - **`apps/kit-manager-web` 的 undeclared-node debt 未被解決。** 本 gate 只約束它內部的分層;contract node 宣告仍是 `observed-baseline.json` 持有的既有債務。 + +### Phase 4 的已知偏離與界線(誠實揭露) + +- **gate 驗的是 contract 一致性與 state 集同步,不是 runtime 行為。** `scripts/lib/lifecycle_contracts.py` 證明三件事:(1) contract 是 well-formed machine(initial 可達性、terminal 封閉、forbidden pair 無直達邊、evidence 引用可解析、同 `(from, trigger)` 唯一目標);(2) 每個 machine 宣告的 state 集與擁有它的 TypeScript union(`SessionStatus`/`ViewerLeaseStatus`/`StageBindingStatus`)**字面完全一致**——單獨改任一側都會紅;(3) readiness binding 與 architecture contract 的 `review-session-ready` policy 的 evidence 集雙向相等。transition 在執行期是否真的照 contract 走,由 `bim-review-coordinator` 自己的測試持有(`stageBindingState` 等),本 gate 不執行 runtime、也不宣稱行為已被證明。 +- **state 集同步只支援純字面 TS union。** source 掃描認的是 `export type X = "a" | "b";` 這種單純形狀;union 一旦引用別的 type alias、換成單引號、或摻入註解,掃描**整個 fail closed**(`lifecycle.source_sync.union_unparsed`),不會部分解析。這是刻意的:寧可紅燈逼人來改 contract 或 checker,也不默默漏抓。 +- **Kit 側狀態面不在 gate 範圍。** `bim-streaming-server` 的 `loading_state`(idle|busy)與 `runtime_state`(unchanged|changed_failed|changed_unconfirmed)是 streaming 端的 observed 面,記在 contract 的 notes;kit-manager-api 的 KitInstance 生命週期同樣未被機器化。Phase 4 的三個 machine 是任務指名的 coordinator 端權威。 +- **`review-session.failed` 是 declared-only。** `SessionStatus` union 宣告了 `failed`,但目前 runtime **沒有任何寫入路徑**(`sessionStore.setStatus` 存在但零呼叫者)。contract 以 `runtime_write_path: "declared_only"` 誠實記錄,該 state 不參與任何 transition;哪天真的接上失敗路徑,屬 behavioral state-machine change,必須在 delta 申報並更新 contract 與 pin。 +- **`created → active` 不存在 runtime 轉移。** activation 是建立時決定的(有 kit binding 即 active),不是事後轉移;contract 因此把兩者都標成 initial,而不是虛構一條沒有 runtime 對應的 transition。 +- **放寬 lifecycle contract 本身,gate 不會抓。** 與 Phase 3 同型的界線:刪一條 forbidden shortcut、放寬某個 transition 的 `evidence_required`、或改 `source_binding` 指向別的檔案,對 checker 而言都是合法輸入。防線是 `tests/test_lifecycle_contracts.py` 裡 `PINNED_STATES`/`PINNED_FORBIDDEN`/`PINNED_EVIDENCE_GATED_TRANSITIONS`/`PINNED_SOURCE_BINDINGS`/`PINNED_READINESS_EVIDENCE` 這幾組**寫死的 pin**:放寬 contract 必須連同改測試,才會在 review diff 裡看得見。這一層是 **review-enforced,不是 gate-enforced**。 +- **cross-machine 規則只驗引用完整性。** 「stage-binding 需要 open session+active primary lease」與「session close 級聯釋放 lease」在 contract 中宣告並由 checker 驗證引用的 machine/state/trigger/transition 都存在,但**跨機器的執行期時序**(例如 close 是否真的先釋放 lease 再改 status)由 coordinator 的實作與測試持有,本 gate 不模擬多機器組合狀態空間。 diff --git a/architecture/architecture-contract.json b/architecture/architecture-contract.json index 53fa9e97e..21c28b51c 100644 --- a/architecture/architecture-contract.json +++ b/architecture/architecture-contract.json @@ -362,6 +362,16 @@ "rule": "no-new-layer-violations" } }, + { + "id": "ARCH-LIFECYCLE-001", + "statement": "The review-session, endpoint-lease, and stage-binding lifecycles are declared in architecture/lifecycle-contract.json as well-formed machines whose state sets stay synchronized with the owning TypeScript unions, whose forbidden shortcuts have no direct edge, and whose readiness binding matches the review-session-ready policy.", + "severity": "error", + "enforcement": { + "status": "active", + "mode": "lifecycle-contract-gate", + "rule": "lifecycle-machines-consistent-and-source-synced" + } + }, { "id": "ARCH-READY-001", "statement": "Review-session readiness requires the conjunction of Kit-side and browser-side evidence, including first frame and stage match.", diff --git a/architecture/deltas/introduce-executable-architecture-contracts.json b/architecture/deltas/introduce-executable-architecture-contracts.json index 3b273f599..d64b462b0 100644 --- a/architecture/deltas/introduce-executable-architecture-contracts.json +++ b/architecture/deltas/introduce-executable-architecture-contracts.json @@ -29,10 +29,31 @@ "contract": "layer boundary ratchet", "change_type": "additive", "description": "Adds a per-service module-to-layer contract and an approved layer baseline. New cross-layer violations are rejected, grandfathered ones carry attributed debt, and per-service budgets carry no slack. The named third-party tools (dependency-cruiser, import-linter) were evaluated and not adopted; the deviation is recorded in architecture/layer-contract.json under tooling_deviation. No product API or event contract changes." + }, + { + "contract": "lifecycle contract gate", + "change_type": "additive", + "description": "Adds architecture/lifecycle-contract.json plus a standard-library checker that validates machine well-formedness, keeps declared state sets synchronized with the owning TypeScript unions, and cross-checks the readiness binding against the review-session-ready policy. Declaration of existing runtime behavior only; no product API, event contract, or runtime behavior changes." } ], "data_ownership_changes": [], - "state_machine_changes": [], + "state_machine_changes": [ + { + "machine": "review-session", + "change_type": "additive", + "description": "First machine-readable declaration of the existing SessionStatus lifecycle (created/active/closing/closed, failed recorded as declared-only) with close-path forbidden shortcuts and idempotent close reentry. No runtime behavior change." + }, + { + "machine": "endpoint-lease", + "change_type": "additive", + "description": "First machine-readable declaration of the existing ViewerLeaseStatus lifecycle (active/released/expired) with resurrection forbidden shortcuts, nonce-replay reentry, and the lease-held readiness evidence fields. No runtime behavior change." + }, + { + "machine": "stage-binding", + "change_type": "additive", + "description": "First machine-readable declaration of the existing StageBindingStatus transaction lifecycle (pending/executing/active/failed/superseded) with attributed failure codes, evidence-gated consume/confirm transitions, and the pending-to-active forbidden shortcut. No runtime behavior change." + } + ], "exceptions": [], "approval": { "required": false, diff --git a/architecture/lifecycle-contract.json b/architecture/lifecycle-contract.json new file mode 100644 index 000000000..4f666d546 --- /dev/null +++ b/architecture/lifecycle-contract.json @@ -0,0 +1,494 @@ +{ + "$schema": "./lifecycle-contract.schema.json", + "schema_version": "ai-bim-lifecycle-contract/v1", + "purpose": "Executable lifecycle contracts for the three coordinator-owned runtime state machines: review-session, endpoint-lease, and stage-binding. Each machine records the states, the observed runtime transitions, the shortcuts the runtime forbids, and the evidence a transition demands. scripts/lib/lifecycle_contracts.py validates the contract's internal consistency and keeps the declared state sets synchronized with the TypeScript source unions that own them.", + "enforcement_note": "This contract describes the current coordinator-side runtime truth, not target intent. The gate proves three things only: the contract is a well-formed machine (reachable states, closed terminals, forbidden pairs have no direct edge, evidence references resolve), the declared state sets are byte-equal to the owning TypeScript union literals, and the readiness binding matches the architecture contract's review-session-ready policy. Transition behavior itself is enforced by each service's own runtime and tests; this gate does not execute the runtime. Kit-side stage loading states (loading_state idle|busy, runtime_state unchanged|changed_failed|changed_unconfirmed in bim-streaming-server stage_loading.py) are an observed surface and are out of this gate's scope.", + "machines": [ + { + "id": "review-session", + "title": "Review session lifecycle", + "owner_service": "bim-review-coordinator", + "source_binding": { + "file": "bim-review-coordinator/src/types.ts", + "type_name": "SessionStatus" + }, + "states": [ + { + "id": "created", + "kind": "initial", + "description": "Session record exists without kit instance bindings. Entered at creation time when no kit capacity was bound." + }, + { + "id": "active", + "kind": "initial", + "description": "Session with kit instance bindings. Entered at creation time when bindings were allocated; there is no observed runtime transition from created to active, activation is decided when the record is created (sessionStore.create)." + }, + { + "id": "closing", + "kind": "intermediate", + "description": "Close has been accepted: viewer leases released and kit bindings marked draining. The close handler moves the session synchronously through closing to closed within one request; closing is not a resting state that outlives the handler." + }, + { + "id": "closed", + "kind": "terminal", + "description": "Kit bindings released, participants cleared, sessionClosed and kitInstanceReleased appended to the audit log." + }, + { + "id": "failed", + "kind": "terminal", + "runtime_write_path": "declared_only", + "description": "Declared by the SessionStatus union but no runtime code path writes it: sessionStore.setStatus exists with zero callers. Recorded honestly as declared-only; wiring a real failure path is a future behavioral change that must be declared in a delta." + } + ], + "transitions": [ + { + "id": "close-from-created", + "from": "created", + "to": "closing", + "trigger": "close-session", + "evidence_required": [], + "effects": [ + "release-viewer-leases", + "drain-kit-bindings" + ], + "description": "POST close on a created session releases every active viewer lease and marks kit bindings draining before the status moves." + }, + { + "id": "close-from-active", + "from": "active", + "to": "closing", + "trigger": "close-session", + "evidence_required": [], + "effects": [ + "release-viewer-leases", + "drain-kit-bindings" + ], + "description": "POST close on an active session releases every active viewer lease and marks kit bindings draining before the status moves." + }, + { + "id": "finalize-close", + "from": "closing", + "to": "closed", + "trigger": "close-session-finalize", + "evidence_required": [], + "effects": [ + "release-kit-bindings", + "clear-participants" + ], + "description": "The same close handler synchronously releases kit bindings, clears participants, and appends sessionClosed plus kitInstanceReleased." + } + ], + "forbidden_shortcuts": [ + { + "id": "no-direct-close-from-created", + "from": "created", + "to": "closed", + "reason": "A session must pass through closing so viewer leases are released and kit bindings are drained before release; skipping closing would leak leases and bindings.", + "enforced_by": "The close handler is the only writer of closed and always writes closing first within the same request." + }, + { + "id": "no-direct-close-from-active", + "from": "active", + "to": "closed", + "reason": "Same lease-release and binding-drain ordering as from created; closed may only follow closing.", + "enforced_by": "The close handler is the only writer of closed and always writes closing first within the same request." + } + ], + "evidence": [], + "reentry_rules": [ + { + "id": "close-idempotent", + "states": [ + "closing", + "closed" + ], + "trigger": "close-session", + "behavior": "no-op-return-current-state", + "description": "Repeated close on a closing or closed session returns the current record without appending duplicate audit events (append-only ledger cannot be cleaned afterwards)." + } + ], + "notes": [ + "Session mutability gate: stage-binding and other runtime mutations require status in {created, active} (isSessionMutable); the runtime rejects others with session_lifecycle_blocked.", + "There is no observed created-to-active transition: kit binding allocation happens at creation (store.create decides created vs active). A later activation path would be a behavioral state-machine change requiring a delta.", + "failed is declared-only: the SessionStatus union declares it, but no caller writes it today." + ] + }, + { + "id": "endpoint-lease", + "title": "Viewer endpoint lease lifecycle", + "owner_service": "bim-review-coordinator", + "source_binding": { + "file": "bim-review-coordinator/src/services/viewerLeaseStore.ts", + "type_name": "ViewerLeaseStatus" + }, + "states": [ + { + "id": "active", + "kind": "initial", + "description": "Claimed lease bound to a kit instance endpoint. Claim enforces at most one active primary lease per session and replays idempotently on the same client nonce." + }, + { + "id": "released", + "kind": "terminal", + "description": "Explicitly released by the caller or cascaded from session close (releaseSession). Runtime authorization rejects it with lease_released." + }, + { + "id": "expired", + "kind": "terminal", + "description": "TTL elapsed without heartbeat; the sweep marks the lease expired lazily on the next store access. Runtime authorization rejects it with lease_expired." + } + ], + "transitions": [ + { + "id": "release-lease", + "from": "active", + "to": "released", + "trigger": "release-lease", + "evidence_required": [], + "effects": [], + "description": "Caller-initiated release; expires_at is clamped to the release instant." + }, + { + "id": "session-close-release", + "from": "active", + "to": "released", + "trigger": "session-close-cascade", + "evidence_required": [], + "effects": [], + "description": "Session close releases every active lease of that session before the session status moves to closing." + }, + { + "id": "ttl-expire", + "from": "active", + "to": "expired", + "trigger": "ttl-sweep", + "evidence_required": [], + "effects": [], + "description": "Lazy sweep on store access marks leases whose expires_at has passed; heartbeats extend expires_at and are not a state transition." + } + ], + "forbidden_shortcuts": [ + { + "id": "no-released-resurrection", + "from": "released", + "to": "active", + "reason": "A released lease must never regain streaming authority; a new claim creates a new lease with a new token instead.", + "enforced_by": "No store API writes active onto an existing lease; authorization rejects released leases with lease_released." + }, + { + "id": "no-expired-resurrection", + "from": "expired", + "to": "active", + "reason": "An expired lease must never regain streaming authority; a new claim creates a new lease with a new token instead.", + "enforced_by": "No store API writes active onto an existing lease; authorization rejects expired leases with lease_expired." + } + ], + "evidence": [ + { + "id": "datachannel-ready", + "description": "Browser reported the Kit DataChannel is open, recorded on the lease by heartbeat.", + "source": "viewerLeaseStore heartbeat field datachannel_ready" + }, + { + "id": "first-frame-at", + "description": "First decoded video frame timestamp, set once by heartbeat and never overwritten.", + "source": "viewerLeaseStore heartbeat field first_frame_at (monotonic set-once)" + }, + { + "id": "stage-matched", + "description": "Loaded stage URL equals the expected stage URL for the session's composition.", + "source": "viewerLeaseStore heartbeat field stage_match (loaded vs expected URL equivalence)" + } + ], + "reentry_rules": [ + { + "id": "claim-nonce-replay", + "states": [ + "active" + ], + "trigger": "claim-lease", + "behavior": "idempotent-replay", + "description": "A claim carrying the same session, viewer, user, and client nonce as an active lease returns that lease as an idempotent replay instead of allocating a second one. When the claim names an explicit role, only an active lease holding that same role replays; a changed explicit role is not a replay and goes through normal claim arbitration (auto matches any active role)." + } + ], + "notes": [ + "The lease carries readiness evidence fields (datachannel_ready, first_frame_at, stage_match); they gate review-session readiness, not the lease's own transitions.", + "Runtime mutation authority requires an active primary lease and rejects otherwise with reasons lease_invalid (lease_not_found, lease_expired, lease_released, lease_inactive), spectator_readonly (spectator_lease), or unauthorized_source_client (source_client_mismatch, cross_session_lease).", + "Heartbeat extends expires_at and updates evidence fields; it is deliberately not modeled as a transition because the status does not change." + ] + }, + { + "id": "stage-binding", + "title": "Stage binding preauthorization transaction lifecycle", + "owner_service": "bim-review-coordinator", + "source_binding": { + "file": "bim-review-coordinator/src/services/runtimeMutationAuthority/stageBindingState.ts", + "type_name": "StageBindingStatus" + }, + "states": [ + { + "id": "pending", + "kind": "initial", + "description": "Browser preauthorized a stage composition. Creation is refused while another transaction is executing (transaction_executing), when the client intent was already cancelled (request_cancelled), or when capacity is exceeded." + }, + { + "id": "executing", + "kind": "intermediate", + "description": "Kit consumed the authorization: the attempt matched the pending transaction on every identity field, so the runtime stage mutation is in flight." + }, + { + "id": "active", + "kind": "terminal", + "description": "Kit confirmed the runtime load succeeded. The session's active binding snapshot is replaced and the previous one is retained as last-good. Terminal per transaction: a later revision is a new transaction, not a transition of this one." + }, + { + "id": "failed", + "kind": "terminal", + "description": "Terminal failure with an attributed failure_code: pending_expired, preauthorization_cancelled, authorization_unavailable, runtime_stage_load_failed, or executing_expired." + }, + { + "id": "superseded", + "kind": "terminal", + "description": "A newer pending preauthorization for the same session replaced this pending transaction (failure code superseded_by_new_pending)." + } + ], + "transitions": [ + { + "id": "kit-consume", + "from": "pending", + "to": "executing", + "trigger": "kit-consume-authorization", + "evidence_required": [ + "attempt-binding-match" + ], + "effects": [], + "description": "consume() moves pending to executing only when the attempt matches the transaction's authorization id, session, revision, lease, source client, and full composition." + }, + { + "id": "confirm-load-success", + "from": "executing", + "to": "active", + "trigger": "runtime-load-success-confirmation", + "evidence_required": [ + "attempt-binding-match", + "runtime-load-outcome" + ], + "effects": [ + "replace-active-binding-snapshot", + "retain-last-good-binding" + ], + "description": "complete(outcome=success) requires the executing attempt to match on every field including requestId and eventType; repeated identical confirmations replay idempotently, mismatches are rejected with completion_mismatch." + }, + { + "id": "confirm-load-failure", + "from": "executing", + "to": "failed", + "trigger": "runtime-load-failure-confirmation", + "evidence_required": [ + "attempt-binding-match", + "runtime-load-outcome" + ], + "failure_code": "runtime_stage_load_failed", + "effects": [], + "description": "complete(outcome=failed) records the runtime load failure against the matching executing attempt." + }, + { + "id": "pending-expire", + "from": "pending", + "to": "failed", + "trigger": "pending-ttl-sweep", + "evidence_required": [], + "failure_code": "pending_expired", + "effects": [], + "description": "Sweep fails a pending transaction whose pending TTL elapsed before Kit consumed it." + }, + { + "id": "executing-expire", + "from": "executing", + "to": "failed", + "trigger": "executing-ttl-sweep", + "evidence_required": [], + "failure_code": "executing_expired", + "effects": [], + "description": "Sweep fails an executing transaction whose executing TTL elapsed without a completion confirmation." + }, + { + "id": "cancel-pending", + "from": "pending", + "to": "failed", + "trigger": "browser-cancel-intent", + "evidence_required": [ + "cancellation-intent-match" + ], + "failure_code": "preauthorization_cancelled", + "effects": [], + "description": "The browser's deadline cancellation fences the pending transaction by client intent; only a cancellation matching the transaction's preauthorization intent identity cancels it, and the tombstone also fences a late duplicate POST that has not created a transaction yet." + }, + { + "id": "authorization-unavailable-pending", + "from": "pending", + "to": "failed", + "trigger": "authorization-unavailable", + "evidence_required": [ + "attempt-binding-match" + ], + "failure_code": "authorization_unavailable", + "effects": [], + "description": "failBeforeMutation() records that the coordinator could not uphold the authorization before any runtime mutation happened; the attempt must match the pending transaction's base identity fields or the call is rejected with transaction_mismatch." + }, + { + "id": "authorization-unavailable-executing", + "from": "executing", + "to": "failed", + "trigger": "authorization-unavailable", + "evidence_required": [ + "attempt-binding-match" + ], + "failure_code": "authorization_unavailable", + "effects": [], + "description": "failBeforeMutation() on a matching executing attempt records the same terminal failure; non-matching attempts are rejected with transaction_mismatch instead of clobbering the transaction." + }, + { + "id": "supersede-pending", + "from": "pending", + "to": "superseded", + "trigger": "new-pending-preauthorization", + "evidence_required": [], + "failure_code": "superseded_by_new_pending", + "effects": [], + "description": "A newer preauthorization for the same session replaces the existing pending transaction; executing transactions are never superseded." + } + ], + "forbidden_shortcuts": [ + { + "id": "no-pending-direct-active", + "from": "pending", + "to": "active", + "reason": "A binding may only become active after Kit actually consumed the authorization (executing) and confirmed the runtime load outcome; activating a never-executed binding would assert a stage state Kit never reported.", + "enforced_by": "complete() rejects non-executing transactions with transaction_not_executing; consume() is the only writer of executing." + }, + { + "id": "no-executing-supersede", + "from": "executing", + "to": "superseded", + "reason": "Once Kit consumed the authorization the browser cannot roll it back by starting a new preauthorization; createPending refuses with transaction_executing instead.", + "enforced_by": "createPending() returns transaction_executing while a non-terminal executing transaction exists; only pending transactions are superseded." + } + ], + "evidence": [ + { + "id": "attempt-binding-match", + "description": "The attempt matches the transaction on authorization id, session, binding revision, lease, source client, and the full stage composition (and on requestId plus eventType at completion).", + "source": "stageBindingState matchesBase/matchesAttempt field-by-field comparison" + }, + { + "id": "cancellation-intent-match", + "description": "The cancellation names the same session, principal, lease, source client, and client request id the pending transaction was created with; non-matching intents leave the transaction untouched (transaction_not_abortable for non-pending matches).", + "source": "stageBindingState cancelPendingByIntent preauthorization intent key comparison" + }, + { + "id": "runtime-load-outcome", + "description": "Kit's reported outcome for the runtime stage load; success is required for active, failure records runtime_stage_load_failed.", + "source": "complete() outcome parameter reported by the Kit-side stage load confirmation" + } + ], + "reentry_rules": [ + { + "id": "complete-idempotent-replay", + "states": [ + "active", + "failed" + ], + "trigger": "runtime-load-confirmation", + "behavior": "idempotent-replay", + "description": "A completion identical to the recorded attempt and outcome replays the stored result, but only on terminal records that carry a completion outcome: active, runtime_stage_load_failed, and authorization_unavailable. Failed records without a completion outcome (pending_expired, preauthorization_cancelled, executing_expired) and superseded records are not replayable; the runtime rejects them with transaction_not_executing or transaction_not_abortable. Any attempt or outcome mismatch is rejected with completion_mismatch instead of rewriting history." + } + ], + "notes": [ + "Preauthorization requires an open review session and the caller's active primary viewer lease; the runtime rejects otherwise with session_lifecycle_blocked and primary_lease_required (see cross-machine rules).", + "The per-session active/last-good binding summary is a projection maintained at confirmation time, not a state of the transaction machine.", + "Kit-side loading_state (idle|busy) and runtime_state (unchanged|changed_failed|changed_unconfirmed) are the streaming server's observed surface and are out of this gate's scope.", + "Completed transactions are retained for a bounded window and then evicted; eviction is storage hygiene, not a transition." + ] + } + ], + "cross_machine_rules": [ + { + "id": "stage-binding-requires-open-session-and-primary-lease", + "statement": "Creating a stage-binding preauthorization requires the review session to be mutable (created or active) and the caller to hold the session's active primary endpoint lease; the runtime rejects violations with session_lifecycle_blocked and primary_lease_required.", + "machines": [ + "review-session", + "endpoint-lease", + "stage-binding" + ], + "required_states": [ + { + "machine": "review-session", + "any_of": [ + "created", + "active" + ] + }, + { + "machine": "endpoint-lease", + "any_of": [ + "active" + ] + } + ], + "runtime_rejections": [ + "session_lifecycle_blocked", + "primary_lease_required" + ] + }, + { + "id": "session-close-cascades-lease-release", + "statement": "Entering the review session close path releases every active viewer lease of that session before the session status leaves the open set; the lease machine records this as the session-close-cascade transition.", + "machines": [ + "review-session", + "endpoint-lease" + ], + "cascade": { + "from_machine": "review-session", + "on_trigger": "close-session", + "to_machine": "endpoint-lease", + "applies_transition": "session-close-release" + } + } + ], + "readiness_binding": { + "policy_id": "review-session-ready", + "evidence_bindings": [ + { + "evidence_id": "kit-process-alive", + "provider": "kit-manager-api", + "policy_source": "kit-manager-api", + "surface": "Kit process supervision reports the managed Kit process alive." + }, + { + "evidence_id": "opened-stage-result", + "provider": "bim-streaming-server", + "policy_source": "bim-streaming-server", + "surface": "Kit-side stage load confirmation for the opened stage." + }, + { + "evidence_id": "datachannel-ready", + "provider": "endpoint-lease", + "policy_source": "web-viewer-sample", + "surface": "Lease heartbeat field datachannel_ready, produced by the browser viewer." + }, + { + "evidence_id": "first-frame-at", + "provider": "endpoint-lease", + "policy_source": "web-viewer-sample", + "surface": "Lease heartbeat field first_frame_at (set-once), produced by the browser viewer." + }, + { + "evidence_id": "stage-matched", + "provider": "endpoint-lease", + "policy_source": "web-viewer-sample", + "surface": "Lease heartbeat field stage_match, produced by the browser viewer." + } + ] + } +} diff --git a/architecture/lifecycle-contract.schema.json b/architecture/lifecycle-contract.schema.json new file mode 100644 index 000000000..b5aaa2554 --- /dev/null +++ b/architecture/lifecycle-contract.schema.json @@ -0,0 +1,440 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "$id": "urn:ai-bim-governance:lifecycle-contract:v1", + "title": "AI-BIM lifecycle contract", + "type": "object", + "additionalProperties": false, + "required": [ + "$schema", + "schema_version", + "purpose", + "enforcement_note", + "machines", + "cross_machine_rules", + "readiness_binding" + ], + "properties": { + "$schema": { + "type": "string", + "minLength": 1 + }, + "schema_version": { + "const": "ai-bim-lifecycle-contract/v1" + }, + "purpose": { + "$ref": "#/definitions/nonEmptyString" + }, + "enforcement_note": { + "$ref": "#/definitions/nonEmptyString" + }, + "machines": { + "type": "array", + "minItems": 1, + "items": { + "$ref": "#/definitions/machine" + } + }, + "cross_machine_rules": { + "type": "array", + "minItems": 1, + "items": { + "$ref": "#/definitions/crossMachineRule" + } + }, + "readiness_binding": { + "$ref": "#/definitions/readinessBinding" + } + }, + "definitions": { + "identifier": { + "type": "string", + "pattern": "^[a-z0-9]+(?:-[a-z0-9]+)*$" + }, + "snakeCode": { + "type": "string", + "pattern": "^[a-z0-9]+(?:_[a-z0-9]+)*$" + }, + "nonEmptyString": { + "type": "string", + "minLength": 1 + }, + "identifierArray": { + "type": "array", + "items": { + "$ref": "#/definitions/identifier" + }, + "uniqueItems": true + }, + "repoRelativePath": { + "type": "string", + "pattern": "^[A-Za-z0-9_.-]+(?:/[A-Za-z0-9_.-]+)*$" + }, + "machine": { + "type": "object", + "additionalProperties": false, + "required": [ + "id", + "title", + "owner_service", + "source_binding", + "states", + "transitions", + "forbidden_shortcuts", + "evidence", + "reentry_rules", + "notes" + ], + "properties": { + "id": { + "$ref": "#/definitions/identifier" + }, + "title": { + "$ref": "#/definitions/nonEmptyString" + }, + "owner_service": { + "$ref": "#/definitions/identifier" + }, + "source_binding": { + "$ref": "#/definitions/sourceBinding" + }, + "states": { + "type": "array", + "minItems": 1, + "items": { + "$ref": "#/definitions/state" + } + }, + "transitions": { + "type": "array", + "minItems": 1, + "items": { + "$ref": "#/definitions/transition" + } + }, + "forbidden_shortcuts": { + "type": "array", + "items": { + "$ref": "#/definitions/forbiddenShortcut" + } + }, + "evidence": { + "type": "array", + "items": { + "$ref": "#/definitions/evidenceDeclaration" + } + }, + "reentry_rules": { + "type": "array", + "items": { + "$ref": "#/definitions/reentryRule" + } + }, + "notes": { + "type": "array", + "items": { + "$ref": "#/definitions/nonEmptyString" + } + } + } + }, + "sourceBinding": { + "type": "object", + "additionalProperties": false, + "required": [ + "file", + "type_name" + ], + "properties": { + "file": { + "$ref": "#/definitions/repoRelativePath" + }, + "type_name": { + "type": "string", + "pattern": "^[A-Za-z][A-Za-z0-9]*$" + } + } + }, + "state": { + "type": "object", + "additionalProperties": false, + "required": [ + "id", + "kind", + "description" + ], + "properties": { + "id": { + "$ref": "#/definitions/identifier" + }, + "kind": { + "enum": [ + "initial", + "intermediate", + "terminal" + ] + }, + "runtime_write_path": { + "enum": [ + "observed", + "declared_only" + ] + }, + "description": { + "$ref": "#/definitions/nonEmptyString" + } + } + }, + "transition": { + "type": "object", + "additionalProperties": false, + "required": [ + "id", + "from", + "to", + "trigger", + "evidence_required", + "description" + ], + "properties": { + "id": { + "$ref": "#/definitions/identifier" + }, + "from": { + "$ref": "#/definitions/identifier" + }, + "to": { + "$ref": "#/definitions/identifier" + }, + "trigger": { + "$ref": "#/definitions/identifier" + }, + "evidence_required": { + "$ref": "#/definitions/identifierArray" + }, + "effects": { + "$ref": "#/definitions/identifierArray" + }, + "failure_code": { + "$ref": "#/definitions/snakeCode" + }, + "description": { + "$ref": "#/definitions/nonEmptyString" + } + } + }, + "forbiddenShortcut": { + "type": "object", + "additionalProperties": false, + "required": [ + "id", + "from", + "to", + "reason", + "enforced_by" + ], + "properties": { + "id": { + "$ref": "#/definitions/identifier" + }, + "from": { + "$ref": "#/definitions/identifier" + }, + "to": { + "$ref": "#/definitions/identifier" + }, + "reason": { + "$ref": "#/definitions/nonEmptyString" + }, + "enforced_by": { + "$ref": "#/definitions/nonEmptyString" + } + } + }, + "evidenceDeclaration": { + "type": "object", + "additionalProperties": false, + "required": [ + "id", + "description", + "source" + ], + "properties": { + "id": { + "$ref": "#/definitions/identifier" + }, + "description": { + "$ref": "#/definitions/nonEmptyString" + }, + "source": { + "$ref": "#/definitions/nonEmptyString" + } + } + }, + "reentryRule": { + "type": "object", + "additionalProperties": false, + "required": [ + "id", + "states", + "trigger", + "behavior", + "description" + ], + "properties": { + "id": { + "$ref": "#/definitions/identifier" + }, + "states": { + "type": "array", + "minItems": 1, + "items": { + "$ref": "#/definitions/identifier" + }, + "uniqueItems": true + }, + "trigger": { + "$ref": "#/definitions/identifier" + }, + "behavior": { + "enum": [ + "no-op-return-current-state", + "idempotent-replay" + ] + }, + "description": { + "$ref": "#/definitions/nonEmptyString" + } + } + }, + "crossMachineRule": { + "type": "object", + "additionalProperties": false, + "required": [ + "id", + "statement", + "machines" + ], + "properties": { + "id": { + "$ref": "#/definitions/identifier" + }, + "statement": { + "$ref": "#/definitions/nonEmptyString" + }, + "machines": { + "type": "array", + "minItems": 2, + "items": { + "$ref": "#/definitions/identifier" + }, + "uniqueItems": true + }, + "required_states": { + "type": "array", + "items": { + "$ref": "#/definitions/requiredStates" + } + }, + "cascade": { + "$ref": "#/definitions/cascade" + }, + "runtime_rejections": { + "type": "array", + "items": { + "$ref": "#/definitions/snakeCode" + }, + "uniqueItems": true + } + } + }, + "requiredStates": { + "type": "object", + "additionalProperties": false, + "required": [ + "machine", + "any_of" + ], + "properties": { + "machine": { + "$ref": "#/definitions/identifier" + }, + "any_of": { + "type": "array", + "minItems": 1, + "items": { + "$ref": "#/definitions/identifier" + }, + "uniqueItems": true + } + } + }, + "cascade": { + "type": "object", + "additionalProperties": false, + "required": [ + "from_machine", + "on_trigger", + "to_machine", + "applies_transition" + ], + "properties": { + "from_machine": { + "$ref": "#/definitions/identifier" + }, + "on_trigger": { + "$ref": "#/definitions/identifier" + }, + "to_machine": { + "$ref": "#/definitions/identifier" + }, + "applies_transition": { + "$ref": "#/definitions/identifier" + } + } + }, + "readinessBinding": { + "type": "object", + "additionalProperties": false, + "required": [ + "policy_id", + "evidence_bindings" + ], + "properties": { + "policy_id": { + "$ref": "#/definitions/identifier" + }, + "evidence_bindings": { + "type": "array", + "minItems": 1, + "items": { + "$ref": "#/definitions/evidenceBinding" + } + } + } + }, + "evidenceBinding": { + "type": "object", + "additionalProperties": false, + "required": [ + "evidence_id", + "provider", + "policy_source", + "surface" + ], + "properties": { + "evidence_id": { + "$ref": "#/definitions/identifier" + }, + "provider": { + "$ref": "#/definitions/identifier" + }, + "policy_source": { + "$ref": "#/definitions/identifier" + }, + "surface": { + "$ref": "#/definitions/nonEmptyString" + } + } + } + } +} diff --git a/openspec/changes/introduce-executable-architecture-contracts/design.md b/openspec/changes/introduce-executable-architecture-contracts/design.md index 5f830414e..403fb0b64 100644 --- a/openspec/changes/introduce-executable-architecture-contracts/design.md +++ b/openspec/changes/introduce-executable-architecture-contracts/design.md @@ -188,5 +188,5 @@ Mitigation:只接入 `verification-manifest.json` 與 root pytest,不新增 1. Phase 1:desired contract、delta、semantic validator、pytest、manifest dispatch。 2. Phase 2(2026-07-30 完成):deterministic static desired-vs-observed graph report + no-new-edge / no-new-cycle ratchet。GitNexus 改列 advisory,理由見上。 3. Phase 3(2026-08-03 完成):~~TypeScript dependency-cruiser + Python Import Linter~~ → 純標準函式庫的 layer boundary ratchet(`scripts/lib/layered_architecture.py`)。**Phase 3 更正:** 原文指名的兩個第三方工具未採用——canonical root-contract CI job 只裝 `pytest`/`jsonschema`,`apps/kit-manager-web` 沒有 lockfile 可釘版本,且兩者都不保證本 repo 要求的 Windows/Linux byte-identical 輸出。改為重用 Phase 2 已對抗硬化的 module graph extractor,沿用同一套 baseline ratchet。任務產出(可執行的分層邊界契約)不變,工具不同;偏離記於 `architecture/layer-contract.json` 的 `tooling_deviation` 並由測試斷言,只能 supersede 不能刪除。 -4. Phase 4:review-session / endpoint-lease / stage-binding executable state machines。 +4. Phase 4(2026-08-05 完成):review-session / endpoint-lease / stage-binding executable state machines(`architecture/lifecycle-contract.json` + `scripts/lib/lifecycle_contracts.py`)。**Phase 4 定調:** contract 描述 current runtime truth 而非 target intent——`failed` 記為 declared-only(零寫入路徑)、`created→active` 不虛構 runtime 轉移;gate 驗 machine well-formedness、TS union state 集同步(fail-closed 純字面 union 掃描)與 readiness binding 一致性,transition 行為仍由各 service 測試持有。 5. Phase 5:將 recurring `$improve-codebase-architecture` findings 編譯成 permanent rules 與 quality grade。 diff --git a/openspec/changes/introduce-executable-architecture-contracts/tasks.md b/openspec/changes/introduce-executable-architecture-contracts/tasks.md index fd22df4b2..59d7a603f 100644 --- a/openspec/changes/introduce-executable-architecture-contracts/tasks.md +++ b/openspec/changes/introduce-executable-architecture-contracts/tasks.md @@ -303,10 +303,75 @@ module;掃不到 module 的 service;observed layer 集合與宣告不符的 ## Phase 4 — Executable lifecycle contracts -- [ ] 4.1 Define `review-session` state machine. -- [ ] 4.2 Define `endpoint-lease` state machine. -- [ ] 4.3 Define `stage-binding` state machine. -- [ ] 4.4 Add model-based tests for forbidden shortcuts and evidence-gated transitions. +- [x] 4.1 Define `review-session` state machine. +- [x] 4.2 Define `endpoint-lease` state machine. +- [x] 4.3 Define `stage-binding` state machine. +- [x] 4.4 Add model-based tests for forbidden shortcuts and evidence-gated transitions. + +### Phase 4 交付紀錄 — 2026-08-05 + +**落地內容。** `architecture/lifecycle-contract.json`(+ Draft-07 schema)以機器可讀形式宣告三個 +coordinator 端狀態機的 current runtime truth:states(含 kind 與 declared-only 標記)、observed +transitions(含 trigger、evidence_required、failure_code、effects)、forbidden shortcuts(附 +enforced_by)、evidence 宣告、reentry 規則、兩條 cross-machine 規則(stage-binding 需 open +session+active primary lease;session close 級聯釋放 lease)與 readiness binding(五個 +evidence 綁到 provider)。`scripts/lib/lifecycle_contracts.py`(純標準函式庫,重用 Phase 1/3 的 +`_load_document`/`validate_schema_instance`)驗證:machine well-formedness(initial 可達性 +BFS、terminal 封閉、forbidden pair 無直達邊、同 `(from, trigger)` 唯一目標、evidence/ +cross-machine/readiness 引用完整性、duplicate 偵測、declared-only state 不得接線)、**TS union +state 集同步**(`SessionStatus`/`ViewerLeaseStatus`/`StageBindingStatus` 的字面 union 與 +contract 雙向相等;非純字面 union 整個 fail closed)、readiness binding 與 +`review-session-ready` policy 的 evidence 集雙向相等。`ARCH-LIFECYCLE-001` 因 gate 實際跑在 +canonical root-contract dispatch 而標 `active`;delta 以 additive `state_machine_changes` +申報三個 machine(首次機器化宣告,無 runtime 行為變更)。 + +**Runtime 真相的兩個誠實記錄。** `review-session.failed` 由 union 宣告但零 runtime 寫入路徑 +(`sessionStore.setStatus` 存在、零呼叫者),contract 記為 `runtime_write_path: "declared_only"` +且不參與 transition;`created → active` 不存在 runtime 轉移(activation 於建立時由 kit binding +有無決定),contract 把兩者都標 initial 而非虛構轉移。接上任一路徑都屬 behavioral +state-machine change,須申報 delta。 + +**4.4 model-based tests。** `tests/test_lifecycle_contracts.py`(52 項)從 canonical contract +載入 transition system 後**枚舉全部 simple paths** 斷言性質,而非手寫個案:每條 forbidden pair +無單步邊;`pending → active` 的每條路徑必經 `executing` 且 evidence 聯集恰為 +`{attempt-binding-match, runtime-load-outcome}`;每條 forbidden-pair 繞行路徑必經 intermediate +state;terminal 全封閉;全 state 可達或 declared-only;`(from, trigger)` 決定性;lease terminal +不可達 `active`(復活禁止)。Pin 防線與 Phase 3 同型:`PINNED_STATES`/`PINNED_FORBIDDEN`/ +`PINNED_EVIDENCE_GATED_TRANSITIONS`/`PINNED_SOURCE_BINDINGS`/`PINNED_READINESS_EVIDENCE`/ +schema load-bearing keys 全部寫死在測試檔,放寬 contract 必須連同改測試。負例覆蓋:缺檔/ +非物件/vacuous schema(`{}`、`{"type":"object"}` 等四型)/schema_version 錯/terminal 出邊/ +forbidden 直達邊矛盾/未宣告 evidence/不可達 state/declared-only 接線或標 initial/ +duplicate(machine、state、transition、self-loop)/nondeterministic/unknown owner service/ +cross-machine 與 cascade 的 unknown machine・state・trigger・transition/readiness 缺綁・多綁・ +unknown policy・unknown provider・machine evidence 不符/source 刪 state・加 state・type 改名・ +union 引用他型・檔案缺失/`..` path escape/unused evidence 降 warning 不 fail。另有獨立於 +checker 的 `test_runtime_source_unions_match_pinned_states` 直接讀三個 TS 檔比對 pin。 + +**驗證(Windows governed worktree,Python 3.12.7/pytest 8.2.2/jsonschema 4.25.1):** + +- `python -m pytest tests -q -p no:cacheprovider` — **445 passed, 9 skipped**(本 change 之前 + 乾淨樹為 393 passed,新增 52)。 +- `python scripts/dev/check_lifecycle_contracts.py --repo-root . --strict` — PASSED;3 machines、 + 13 states、15 transitions、0 error、0 warning,exit 0。 +- `python scripts/dev/check_layered_architecture.py --repo-root . --strict` — PASSED(Phase 3 + ratchet 未受影響)。 +- `python scripts/dev/export_observed_architecture.py --repo-root . --strict` — PASSED(Phase 2 + ratchet 未受影響)。 +- `python scripts/dev/validate_architecture_contract.py --repo-root . --strict` — PASSED + (`ARCH-LIFECYCLE-001` 與 delta 的 `state_machine_changes` 均通過 Phase 1 semantic validator)。 +- `node scripts/tests/test-verification-plan.mjs` — 23/23;`root-contracts` path class 與 target + 已含 `scripts/lib/lifecycle_contracts.py` 與 `scripts/dev/check_lifecycle_contracts.py`。 +- `node scripts/tests/test-openspec-machine-truth.mjs` — 24/24。 +- `npx openspec validate introduce-executable-architecture-contracts --strict` — passed。 +- `npx openspec validate --all --strict` — **71 passed, 0 failed**。 +- `git diff --check` — clean。 + +**未宣稱解決的已知界線**(完整清單見 `architecture/README.md` §「Phase 4 的已知偏離與界線」): +gate 驗 contract 一致性與 state 集同步,**不驗 transition 的執行期行為**(由各 service 測試持 +有);source 掃描只支援純字面 TS union(其他形狀 fail closed);Kit 側 `loading_state`/ +`runtime_state` 與 kit-manager-api 的 KitInstance 生命週期不在本 gate;放寬 contract 屬 +review-enforced(pin 逼 diff 現形),非 gate-enforced;cross-machine 規則只驗引用完整性, +不模擬多機器組合時序。 ## Phase 5 — Continuous architecture learning diff --git a/openspec/lifecycle-ledger.json b/openspec/lifecycle-ledger.json index 9099d583e..800be9025 100644 --- a/openspec/lifecycle-ledger.json +++ b/openspec/lifecycle-ledger.json @@ -1381,18 +1381,18 @@ "id": "introduce-executable-architecture-contracts", "status": "active", "owner": "repository-maintainer", - "current_slice": "Phase 4 可執行生命週期契約", + "current_slice": "Phase 5 持續架構學習", "blocked_by": [], - "last_verified": "2026-08-03T02:00:00Z", + "last_verified": "2026-08-05T14:05:13Z", "task_ledger": { - "completed": 19, + "completed": 23, "total": 26 }, "evidence_refs": [ "openspec/changes/introduce-executable-architecture-contracts/proposal.md", "openspec/changes/introduce-executable-architecture-contracts/tasks.md" ], - "subject_commit": "c5b9089ebf11eabe9979b8eb717270c7c7a7308e", + "subject_commit": "7c5dd985597711b3afa625df97ef5079a4a18ccb", "archive_debt": null }, { diff --git a/scripts/dev/check_lifecycle_contracts.py b/scripts/dev/check_lifecycle_contracts.py new file mode 100644 index 000000000..8a6cf9259 --- /dev/null +++ b/scripts/dev/check_lifecycle_contracts.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 +"""Check the lifecycle contract's machine consistency and source synchronization.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +import sys + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Validate architecture/lifecycle-contract.json: well-formed state machines, " + "forbidden shortcuts without direct edges, evidence reference integrity, state " + "sets synchronized with the owning TypeScript unions, and a readiness binding " + "matching the architecture contract." + ) + ) + parser.add_argument( + "--repo-root", + default=".", + help="Repository root containing architecture/ and scripts/ (default: current directory).", + ) + parser.add_argument( + "--format", + choices=("human", "json"), + default="human", + help="Output format (default: human).", + ) + parser.add_argument( + "--output", + help="Optional output path. Stdout is still used when omitted.", + ) + parser.add_argument( + "--strict", + action="store_true", + help="Treat warnings as a failing result in addition to errors.", + ) + return parser.parse_args() + + +def main() -> int: + args = _parse_args() + repo_root = Path(args.repo_root).resolve() + sys.path.insert(0, str(repo_root)) + + try: + from scripts.lib.lifecycle_contracts import check_lifecycle_contracts + except ModuleNotFoundError as exc: + print( + f"ERROR: could not import scripts.lib.lifecycle_contracts from {repo_root}: {exc}", + file=sys.stderr, + ) + return 2 + + result = check_lifecycle_contracts(repo_root) + # The rendered verdict must agree with the process outcome: under --strict a + # warning-only run exits 1, so it must not be labelled PASSED. The library's + # error-only `status` is preserved in the JSON payload; `cli_status` carries + # the strict-aware verdict this process actually returns. + failed = result.error_count > 0 or (args.strict and result.warning_count > 0) + cli_status = "failed" if failed else result.status + if args.format == "json": + payload = result.to_dict() + payload["strict"] = args.strict + payload["cli_status"] = cli_status + rendered = json.dumps(payload, indent=2, ensure_ascii=False, sort_keys=True) + "\n" + else: + lines = [ + f"Lifecycle contracts: {cli_status.upper()}", + f"Repository: {result.repo_root}", + f"Machines: {result.machine_count}; states: {result.state_count}; " + f"transitions: {result.transition_count}", + f"Errors: {result.error_count}; Warnings: {result.warning_count}", + ] + if result.issues: + lines.append("") + for issue in result.issues: + lines.append(f"[{issue.severity.upper()}] {issue.code} {issue.path}: {issue.message}") + rendered = "\n".join(lines) + "\n" + + _emit(rendered, args.output, repo_root) + return 1 if failed else 0 + + +def _emit(rendered: str, output: str | None, repo_root: Path) -> None: + if output: + path = Path(output) + if not path.is_absolute(): + path = repo_root / path + path.parent.mkdir(parents=True, exist_ok=True) + # newline="" keeps the LF endings the renderer produced, so the same tree + # yields a byte-identical file on Windows and Linux. + with open(path, "w", encoding="utf-8", newline="") as handle: + handle.write(rendered) + else: + sys.stdout.write(rendered) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/lib/architecture_contract.py b/scripts/lib/architecture_contract.py index b23a3ec88..42728c2e9 100644 --- a/scripts/lib/architecture_contract.py +++ b/scripts/lib/architecture_contract.py @@ -39,6 +39,8 @@ "ARCH-SVC-001", "ARCH-CALL-001", "ARCH-GRAPH-001", + "ARCH-LAYER-001", + "ARCH-LIFECYCLE-001", "ARCH-READY-001", "ARCH-UI-001", "ARCH-DELTA-001", diff --git a/scripts/lib/lifecycle_contracts.py b/scripts/lib/lifecycle_contracts.py new file mode 100644 index 000000000..3f0d35fb4 --- /dev/null +++ b/scripts/lib/lifecycle_contracts.py @@ -0,0 +1,1224 @@ +"""Executable lifecycle contracts (Phase 4 of the architecture contract). + +Phases 1-3 made the desired service topology, the observed dependency graph, +and the intra-service layer boundaries machine-checkable. This module does the +same for the three coordinator-owned runtime state machines: review-session, +endpoint-lease, and stage-binding. + +The gate proves exactly three things and claims nothing beyond them: + +1. ``architecture/lifecycle-contract.json`` describes well-formed machines: + every state is reachable from an initial state, terminal states have no + outgoing edges, forbidden shortcuts have no direct edge, transitions are + deterministic per (from, trigger), and every evidence reference resolves. +2. The declared state set of each machine is exactly the literal union of the + TypeScript type that owns it (``source_binding``). Editing either side alone + fails the gate, so the contract cannot drift from the code silently. +3. The readiness binding names the same evidence set as the architecture + contract's ``review-session-ready`` policy, so the two documents cannot + diverge about what "ready" means. + +Transition *behavior* is enforced by each service's own runtime and tests; +this gate never executes the runtime. Kit-side stage loading states are an +observed surface recorded in the contract's notes, not a gated machine. + +Fail-closed posture, mirroring Phases 2-3: a run that could not load and +compare the contract must never report ``passed``. Missing or corrupt files, +vacuous schemas, an unresolvable source binding, a union the parser does not +understand -- all of these are findings, not silent successes. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +import re +from typing import Any, Iterable, Mapping, TypeGuard + +from scripts.lib.architecture_contract import ( + ValidationIssue, + _is_mapping, + _is_sequence, + _issue, +) +from scripts.lib.layered_architecture import _load_document + +CONTRACT_SCHEMA_VERSION = "ai-bim-lifecycle-contract/v1" +ARCHITECTURE_CONTRACT_SCHEMA_VERSION = "ai-bim-architecture-contract/v1" + +CONTRACT_RELATIVE_PATH = "architecture/lifecycle-contract.json" +CONTRACT_SCHEMA_RELATIVE_PATH = "architecture/lifecycle-contract.schema.json" +ARCHITECTURE_CONTRACT_RELATIVE_PATH = "architecture/architecture-contract.json" +ARCHITECTURE_CONTRACT_SCHEMA_RELATIVE_PATH = "architecture/architecture-contract.schema.json" + +STATE_KINDS = ("initial", "intermediate", "terminal") + + +# --------------------------------------------------------------------------- # +# Data model +# --------------------------------------------------------------------------- # + + +@dataclass(frozen=True, slots=True) +class TransitionModel: + """One declared transition, reduced to the fields the model checks need.""" + + id: str + from_state: str + to_state: str + trigger: str + evidence_required: tuple[str, ...] + failure_code: str | None = None + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "from": self.from_state, + "to": self.to_state, + "trigger": self.trigger, + "evidence_required": list(self.evidence_required), + "failure_code": self.failure_code, + } + + +@dataclass(frozen=True, slots=True) +class MachineModel: + """One declared machine, reduced to the fields the model checks need.""" + + id: str + owner_service: str + source_file: str + source_type_name: str + states: tuple[str, ...] + initial_states: tuple[str, ...] + terminal_states: tuple[str, ...] + declared_only_states: tuple[str, ...] + transitions: tuple[TransitionModel, ...] + forbidden_pairs: tuple[tuple[str, str], ...] + evidence_ids: tuple[str, ...] + + def edges(self) -> set[tuple[str, str]]: + return {(item.from_state, item.to_state) for item in self.transitions} + + def reachable_states(self) -> set[str]: + """States reachable from the initial set by declared transitions.""" + + adjacency: dict[str, set[str]] = {} + for item in self.transitions: + adjacency.setdefault(item.from_state, set()).add(item.to_state) + seen = set(self.initial_states) + frontier = list(self.initial_states) + while frontier: + current = frontier.pop() + for target in adjacency.get(current, ()): + if target not in seen: + seen.add(target) + frontier.append(target) + return seen + + def simple_paths(self, source: str, target: str) -> list[tuple[TransitionModel, ...]]: + """Every cycle-free transition path from ``source`` to ``target``. + + The state spaces here are tiny (five states at most), so full + enumeration is deterministic and cheap; the model-based tests assert + properties over every path instead of sampling. + """ + + adjacency: dict[str, list[TransitionModel]] = {} + for item in self.transitions: + adjacency.setdefault(item.from_state, []).append(item) + paths: list[tuple[TransitionModel, ...]] = [] + + def walk(state: str, visited: frozenset[str], trail: tuple[TransitionModel, ...]) -> None: + if state == target: + paths.append(trail) + return + for item in adjacency.get(state, ()): + if item.to_state in visited: + continue + walk(item.to_state, visited | {item.to_state}, trail + (item,)) + + walk(source, frozenset({source}), ()) + return paths + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "owner_service": self.owner_service, + "source_binding": {"file": self.source_file, "type_name": self.source_type_name}, + "states": list(self.states), + "initial_states": list(self.initial_states), + "terminal_states": list(self.terminal_states), + "declared_only_states": list(self.declared_only_states), + "transitions": [item.to_dict() for item in self.transitions], + "forbidden_pairs": [list(pair) for pair in self.forbidden_pairs], + "evidence_ids": list(self.evidence_ids), + } + + +@dataclass(frozen=True, slots=True) +class LifecycleCheckResult: + """Result returned by :func:`check_lifecycle_contracts`.""" + + repo_root: str + compared: bool + machines: tuple[MachineModel, ...] + issues: tuple[ValidationIssue, ...] + + @property + def machine_count(self) -> int: + return len(self.machines) + + @property + def state_count(self) -> int: + return sum(len(machine.states) for machine in self.machines) + + @property + def transition_count(self) -> int: + return sum(len(machine.transitions) for machine in self.machines) + + @property + def error_count(self) -> int: + return sum(issue.severity == "error" for issue in self.issues) + + @property + def warning_count(self) -> int: + return sum(issue.severity == "warning" for issue in self.issues) + + @property + def status(self) -> str: + # A run that never reached the comparison must not report success even + # with zero recorded issues (Phase 2 lesson). + return "passed" if self.compared and self.error_count == 0 else "failed" + + def to_dict(self) -> dict[str, Any]: + return { + "schema_version": "ai-bim-lifecycle-check-result/v1", + "repo_root": self.repo_root, + "status": self.status, + "compared": self.compared, + "summary": { + "machines": self.machine_count, + "states": self.state_count, + "transitions": self.transition_count, + "errors": self.error_count, + "warnings": self.warning_count, + }, + "machines": [machine.to_dict() for machine in self.machines], + "issues": [issue.to_dict() for issue in self.issues], + } + + +# --------------------------------------------------------------------------- # +# Loading +# --------------------------------------------------------------------------- # + + +def _load_lifecycle_contract( + repo_root: Path, +) -> tuple[Mapping[str, Any] | None, list[ValidationIssue]]: + return _load_document( + repo_root, + CONTRACT_RELATIVE_PATH, + CONTRACT_SCHEMA_RELATIVE_PATH, + CONTRACT_SCHEMA_VERSION, + "lifecycle_contract", + ) + + +def _load_architecture_contract( + repo_root: Path, +) -> tuple[Mapping[str, Any] | None, list[ValidationIssue]]: + return _load_document( + repo_root, + ARCHITECTURE_CONTRACT_RELATIVE_PATH, + ARCHITECTURE_CONTRACT_SCHEMA_RELATIVE_PATH, + ARCHITECTURE_CONTRACT_SCHEMA_VERSION, + "lifecycle.architecture_contract", + ) + + +def _non_empty_string(value: Any) -> TypeGuard[str]: + return isinstance(value, str) and bool(value.strip()) + + +def _list_of_mappings(value: Any) -> list[Mapping[str, Any]]: + if not _is_sequence(value): + return [] + return [item for item in value if _is_mapping(item)] + + +def _as_mapping(value: Any) -> Mapping[str, Any] | None: + return value if isinstance(value, Mapping) else None + + +def _string_items(value: Any) -> list[str]: + if not _is_sequence(value): + return [] + return [item for item in value if _non_empty_string(item)] + + +def _duplicates(values: Iterable[str]) -> list[str]: + seen: set[str] = set() + repeated: list[str] = [] + for value in values: + if value in seen and value not in repeated: + repeated.append(value) + seen.add(value) + return sorted(repeated) + + +# --------------------------------------------------------------------------- # +# Machine semantics +# --------------------------------------------------------------------------- # + + +def _build_machine( + machine: Mapping[str, Any], + path: str, +) -> tuple[MachineModel | None, list[ValidationIssue]]: + """Validate one machine's internal consistency and build its model. + + The JSON Schema has already reported shape violations; this function is + defensive about shapes anyway so a schema-invalid document still produces + stable semantic findings instead of a crash. + """ + + issues: list[ValidationIssue] = [] + machine_id = machine.get("id") if _non_empty_string(machine.get("id")) else None + if machine_id is None: + issues.append(_issue("lifecycle.machine.id", path, "Machine is missing a usable id.")) + return None, issues + + states = _list_of_mappings(machine.get("states")) + state_ids = [sid for state in states if _non_empty_string(sid := state.get("id"))] + for duplicate in _duplicates(state_ids): + issues.append( + _issue( + "lifecycle.state.duplicate", + f"{path}/states", + f"State {duplicate!r} is declared more than once in machine {machine_id!r}.", + ) + ) + state_set = set(state_ids) + + kinds: dict[str, str] = {} + declared_only: list[str] = [] + for state in states: + raw_state_id = state.get("id") + if not _non_empty_string(raw_state_id): + continue + state_id = str(raw_state_id) + kind = state.get("kind") + kinds[state_id] = kind if kind in STATE_KINDS else "" + if kind not in STATE_KINDS: + issues.append( + _issue( + "lifecycle.state.kind", + f"{path}/states/{state_id}", + f"State {state_id!r} has kind {kind!r}; expected one of {list(STATE_KINDS)!r}.", + ) + ) + if state.get("runtime_write_path") == "declared_only": + declared_only.append(state_id) + if kind == "initial": + issues.append( + _issue( + "lifecycle.state.declared_only_initial", + f"{path}/states/{state_id}", + f"State {state_id!r} is declared_only but marked initial; an initial state is " + "by definition entered by the runtime.", + ) + ) + + initial_states = sorted(state for state, kind in kinds.items() if kind == "initial") + terminal_states = sorted(state for state, kind in kinds.items() if kind == "terminal") + if not initial_states: + issues.append( + _issue( + "lifecycle.machine.no_initial", + path, + f"Machine {machine_id!r} declares no initial state.", + ) + ) + if not terminal_states: + issues.append( + _issue( + "lifecycle.machine.no_terminal", + path, + f"Machine {machine_id!r} declares no terminal state.", + ) + ) + + transitions_raw = _list_of_mappings(machine.get("transitions")) + transition_models: list[TransitionModel] = [] + transition_ids: list[str] = [] + seen_from_trigger: dict[tuple[str, str], str] = {} + for index, transition in enumerate(transitions_raw): + transition_id = transition.get("id") + transition_path = f"{path}/transitions[{index}]" + if not _non_empty_string(transition_id): + issues.append( + _issue("lifecycle.transition.id", transition_path, "Transition is missing a usable id.") + ) + continue + transition_ids.append(transition_id) + from_state = transition.get("from") + to_state = transition.get("to") + trigger = transition.get("trigger") + usable = True + for field_name, value in (("from", from_state), ("to", to_state)): + if not _non_empty_string(value) or value not in state_set: + issues.append( + _issue( + "lifecycle.transition.unknown_state", + f"{transition_path}/{field_name}", + f"Transition {transition_id!r} references undeclared state {value!r}.", + ) + ) + usable = False + if not _non_empty_string(trigger): + issues.append( + _issue( + "lifecycle.transition.trigger", + f"{transition_path}/trigger", + f"Transition {transition_id!r} is missing a usable trigger.", + ) + ) + usable = False + if not usable: + continue + from_state = str(from_state) + to_state = str(to_state) + trigger = str(trigger) + key = (from_state, trigger) + if key in seen_from_trigger: + issues.append( + _issue( + "lifecycle.transition.nondeterministic", + transition_path, + f"Transitions {seen_from_trigger[key]!r} and {transition_id!r} share (from={from_state!r}, " + f"trigger={trigger!r}); a trigger must lead to exactly one target state.", + ) + ) + else: + seen_from_trigger[key] = transition_id + if kinds.get(from_state) == "terminal": + issues.append( + _issue( + "lifecycle.terminal.outgoing", + transition_path, + f"Transition {transition_id!r} leaves terminal state {from_state!r}; terminal states " + "must be closed.", + ) + ) + for endpoint in {from_state, to_state}: + if endpoint in declared_only: + issues.append( + _issue( + "lifecycle.state.declared_only_wired", + transition_path, + f"Transition {transition_id!r} touches declared_only state {endpoint!r}; a state " + "with no runtime write path cannot participate in transitions.", + ) + ) + raw_failure_code = transition.get("failure_code") + transition_models.append( + TransitionModel( + id=transition_id, + from_state=from_state, + to_state=to_state, + trigger=trigger, + evidence_required=tuple(_string_items(transition.get("evidence_required"))), + failure_code=raw_failure_code if _non_empty_string(raw_failure_code) else None, + ) + ) + for duplicate in _duplicates(transition_ids): + issues.append( + _issue( + "lifecycle.transition.duplicate", + f"{path}/transitions", + f"Transition id {duplicate!r} is declared more than once in machine {machine_id!r}.", + ) + ) + + edge_set = {(item.from_state, item.to_state) for item in transition_models} + forbidden_raw = _list_of_mappings(machine.get("forbidden_shortcuts")) + forbidden_pairs: list[tuple[str, str]] = [] + forbidden_ids: list[str] = [] + for index, shortcut in enumerate(forbidden_raw): + shortcut_path = f"{path}/forbidden_shortcuts[{index}]" + shortcut_id = shortcut.get("id") + if _non_empty_string(shortcut_id): + forbidden_ids.append(shortcut_id) + from_state = shortcut.get("from") + to_state = shortcut.get("to") + usable = True + for field_name, value in (("from", from_state), ("to", to_state)): + if not _non_empty_string(value) or value not in state_set: + issues.append( + _issue( + "lifecycle.forbidden.unknown_state", + f"{shortcut_path}/{field_name}", + f"Forbidden shortcut {shortcut_id!r} references undeclared state {value!r}.", + ) + ) + usable = False + if not usable: + continue + from_state = str(from_state) + to_state = str(to_state) + if from_state == to_state: + issues.append( + _issue( + "lifecycle.forbidden.self_loop", + shortcut_path, + f"Forbidden shortcut {shortcut_id!r} declares from == to ({from_state!r}), which " + "forbids nothing.", + ) + ) + continue + pair = (from_state, to_state) + if pair in forbidden_pairs: + issues.append( + _issue( + "lifecycle.forbidden.duplicate", + shortcut_path, + f"Forbidden pair {from_state!r} -> {to_state!r} is declared more than once.", + ) + ) + continue + forbidden_pairs.append(pair) + if pair in edge_set: + issues.append( + _issue( + "lifecycle.forbidden.direct_edge_exists", + shortcut_path, + f"Machine {machine_id!r} declares {from_state!r} -> {to_state!r} both as a transition " + "and as a forbidden shortcut; the contract contradicts itself.", + ) + ) + for duplicate in _duplicates(forbidden_ids): + issues.append( + _issue( + "lifecycle.forbidden.duplicate_id", + f"{path}/forbidden_shortcuts", + f"Forbidden shortcut id {duplicate!r} is declared more than once.", + ) + ) + + evidence_raw = _list_of_mappings(machine.get("evidence")) + evidence_ids = [eid for item in evidence_raw if _non_empty_string(eid := item.get("id"))] + for duplicate in _duplicates(evidence_ids): + issues.append( + _issue( + "lifecycle.evidence.duplicate", + f"{path}/evidence", + f"Evidence id {duplicate!r} is declared more than once in machine {machine_id!r}.", + ) + ) + evidence_set = set(evidence_ids) + for item in transition_models: + for evidence_id in item.evidence_required: + if evidence_id not in evidence_set: + issues.append( + _issue( + "lifecycle.evidence.unknown", + f"{path}/transitions/{item.id}", + f"Transition {item.id!r} requires undeclared evidence {evidence_id!r}.", + ) + ) + + reentry_raw = _list_of_mappings(machine.get("reentry_rules")) + reentry_ids: list[str] = [] + for index, rule in enumerate(reentry_raw): + rule_path = f"{path}/reentry_rules[{index}]" + rule_id = rule.get("id") + if _non_empty_string(rule_id): + reentry_ids.append(rule_id) + for state in _string_items(rule.get("states")): + if state not in state_set: + issues.append( + _issue( + "lifecycle.reentry.unknown_state", + rule_path, + f"Reentry rule {rule_id!r} references undeclared state {state!r}.", + ) + ) + for duplicate in _duplicates(reentry_ids): + issues.append( + _issue( + "lifecycle.reentry.duplicate", + f"{path}/reentry_rules", + f"Reentry rule id {duplicate!r} is declared more than once.", + ) + ) + + source_binding = machine.get("source_binding") + source_file = "" + source_type_name = "" + source_binding = _as_mapping(source_binding) + if source_binding is not None: + raw_file = source_binding.get("file") + raw_type = source_binding.get("type_name") + source_file = raw_file if _non_empty_string(raw_file) else "" + source_type_name = raw_type if _non_empty_string(raw_type) else "" + + model = MachineModel( + id=machine_id, + owner_service=str(machine.get("owner_service") or ""), + source_file=source_file, + source_type_name=source_type_name, + states=tuple(sorted(state_set)), + initial_states=tuple(initial_states), + terminal_states=tuple(terminal_states), + declared_only_states=tuple(sorted(declared_only)), + transitions=tuple(transition_models), + forbidden_pairs=tuple(forbidden_pairs), + evidence_ids=tuple(sorted(evidence_set)), + ) + + if initial_states: + reachable = model.reachable_states() + for state in sorted(state_set - reachable - set(declared_only)): + issues.append( + _issue( + "lifecycle.state.unreachable", + f"{path}/states/{state}", + f"State {state!r} in machine {machine_id!r} is not reachable from any initial state " + "and is not marked declared_only.", + ) + ) + + # Failure attribution: a failure_code names why a machine reached a failure + # terminal, so it may only appear on transitions into a terminal state, and + # every entry into the same target must agree on whether it is attributed. + # Without this, a new unattributed edge into `failed` (or an attributed edge + # into a live state) would silently blur the failure taxonomy. + coded_targets: dict[str, bool] = {} + for item in transition_models: + has_code = item.failure_code is not None + if has_code and kinds.get(item.to_state) != "terminal": + issues.append( + _issue( + "lifecycle.transition.failure_code_nonterminal", + f"{path}/transitions/{item.id}", + f"Transition {item.id!r} carries failure_code {item.failure_code!r} but its target " + f"{item.to_state!r} is not terminal; failure attribution belongs to terminal entries only.", + ) + ) + if item.to_state in coded_targets and coded_targets[item.to_state] != has_code: + issues.append( + _issue( + "lifecycle.transition.failure_code_inconsistent", + f"{path}/transitions/{item.id}", + f"Transitions into {item.to_state!r} in machine {machine_id!r} disagree on failure " + "attribution; every entry into the same target must either carry a failure_code or none.", + ) + ) + coded_targets.setdefault(item.to_state, has_code) + + # A non-terminal state with no outgoing transition is a dead end: it claims + # the machine can continue but declares no way out. Without this check, + # demoting a terminal state to intermediate would silently weaken the + # closed-state semantics while every other assertion stays green. + outgoing_sources = {item.from_state for item in transition_models} + for state_id in sorted(state_set): + if kinds.get(state_id) == "terminal" or state_id in declared_only: + continue + if state_id not in outgoing_sources: + issues.append( + _issue( + "lifecycle.state.dead_end", + f"{path}/states/{state_id}", + f"State {state_id!r} in machine {machine_id!r} is neither terminal nor declared_only " + "but has no outgoing transition; declare it terminal or wire its exit.", + ) + ) + + return model, issues + + +# --------------------------------------------------------------------------- # +# Source synchronization +# --------------------------------------------------------------------------- # + + +def _strip_ts_comments(text: str) -> tuple[str | None, str | None]: + """Blank out ``//`` and ``/* */`` comments, string-aware. + + Quoted content (including template literals) is left untouched so a + ``"http://..."`` literal never opens a comment. An unterminated block + comment fails closed: silently keeping the tail would let commented-out + code be read as source. + """ + + out: list[str] = [] + index = 0 + length = len(text) + quote: str | None = None + while index < length: + char = text[index] + if quote is not None: + out.append(char) + if char == "\\" and index + 1 < length: + out.append(text[index + 1]) + index += 2 + continue + if char == quote: + quote = None + index += 1 + continue + if char in ('"', "'", "`"): + quote = char + out.append(char) + index += 1 + continue + if char == "/" and index + 1 < length and text[index + 1] == "/": + newline = text.find("\n", index) + if newline == -1: + break + index = newline + continue + if char == "/" and index + 1 < length and text[index + 1] == "*": + closing = text.find("*/", index + 2) + if closing == -1: + return None, "unterminated_block_comment" + # Preserve the newlines so line anchoring stays meaningful. + out.append(text.count("\n", index, closing) * "\n" or " ") + index = closing + 2 + continue + out.append(char) + index += 1 + return "".join(out), None + + +def _extract_union_literals(text: str, type_name: str) -> tuple[list[str] | None, str | None]: + """Extract the string literals of ``export type = "a" | "b";``. + + Returns ``(literals, None)`` on success or ``(None, reason)`` when the + union cannot be interpreted. Anything but a pure string-literal union is + rejected (fail closed) rather than partially parsed: a referenced type + alias, a comment, or single-quoted literals all leave residue. + """ + + # Comments are stripped first (string-aware), so neither a `// export type` + # line nor a block-commented copy is ever read as source; the anchored match + # then requires the declaration to be real code, and more than one anchored + # declaration is ambiguous and fails closed. + stripped, comment_error = _strip_ts_comments(text) + if stripped is None: + return None, comment_error + pattern = re.compile( + r"(?m)^[ \t]*export\s+type\s+" + re.escape(type_name) + r"\s*=\s*(?P[^;]*);" + ) + matches = list(pattern.finditer(stripped)) + if not matches: + return None, "type_not_found" + if len(matches) > 1: + return None, "ambiguous_declaration" + body = matches[0].group("body") + literals = re.findall(r'"([^"\\]*)"', body) + residue = re.sub(r'"[^"\\]*"', "", body).replace("|", " ").strip() + if residue: + return None, "unsupported_union" + if not literals: + return None, "no_literals" + if len(set(literals)) != len(literals): + return None, "duplicate_literals" + if any(not literal for literal in literals): + return None, "empty_literal" + return literals, None + + +def _check_source_sync( + repo_root: Path, + machine: MachineModel, + path: str, +) -> list[ValidationIssue]: + issues: list[ValidationIssue] = [] + if not machine.source_file or not machine.source_type_name: + issues.append( + _issue( + "lifecycle.source_sync.binding_missing", + path, + f"Machine {machine.id!r} has no usable source_binding; the state set cannot be " + "synchronized with the owning source union.", + ) + ) + return issues + parts = machine.source_file.replace("\\", "/").split("/") + if any(part in ("", ".", "..") for part in parts): + issues.append( + _issue( + "lifecycle.source_binding.path_escape", + path, + f"Machine {machine.id!r} source_binding file {machine.source_file!r} contains empty, " + "'.' or '..' segments; only plain repo-relative paths are allowed.", + ) + ) + return issues + source_path = repo_root / machine.source_file + try: + text = source_path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError) as exc: + issues.append( + _issue( + "lifecycle.source_sync.file_unreadable", + path, + f"Machine {machine.id!r} source file {machine.source_file!r} could not be read: {exc}.", + ) + ) + return issues + literals, reason = _extract_union_literals(text, machine.source_type_name) + if literals is None: + issues.append( + _issue( + "lifecycle.source_sync.union_unparsed", + path, + f"Machine {machine.id!r} could not synchronize with type {machine.source_type_name!r} in " + f"{machine.source_file!r}: {reason}. Only a pure string-literal union is supported; " + "anything else must fail closed rather than partially parse.", + ) + ) + return issues + source_states = set(literals) + contract_states = set(machine.states) + for state in sorted(source_states - contract_states): + issues.append( + _issue( + "lifecycle.source_sync.state_missing_in_contract", + path, + f"Source union {machine.source_type_name!r} declares state {state!r} that machine " + f"{machine.id!r} does not; update the lifecycle contract in the same change.", + ) + ) + for state in sorted(contract_states - source_states): + issues.append( + _issue( + "lifecycle.source_sync.state_missing_in_source", + path, + f"Machine {machine.id!r} declares state {state!r} that source union " + f"{machine.source_type_name!r} does not; the contract may only describe the current " + "runtime truth.", + ) + ) + return issues + + +# --------------------------------------------------------------------------- # +# Contract-level semantics +# --------------------------------------------------------------------------- # + + +def _check_cross_machine_rules( + contract: Mapping[str, Any], + machines: Mapping[str, MachineModel], +) -> list[ValidationIssue]: + issues: list[ValidationIssue] = [] + rules = _list_of_mappings(contract.get("cross_machine_rules")) + rule_ids: list[str] = [] + for index, rule in enumerate(rules): + rule_path = f"$.cross_machine_rules[{index}]" + rule_id = rule.get("id") + if _non_empty_string(rule_id): + rule_ids.append(rule_id) + listed = _string_items(rule.get("machines")) + for machine_id in listed: + if machine_id not in machines: + issues.append( + _issue( + "lifecycle.cross.unknown_machine", + rule_path, + f"Cross-machine rule {rule_id!r} references undeclared machine {machine_id!r}.", + ) + ) + for requirement in _list_of_mappings(rule.get("required_states")): + machine_id = requirement.get("machine") + if machine_id not in machines: + issues.append( + _issue( + "lifecycle.cross.unknown_machine", + rule_path, + f"Cross-machine rule {rule_id!r} requires states of undeclared machine " + f"{machine_id!r}.", + ) + ) + continue + if machine_id not in listed: + issues.append( + _issue( + "lifecycle.cross.unlisted_machine", + rule_path, + f"Cross-machine rule {rule_id!r} requires states of machine {machine_id!r} " + "without listing it under machines.", + ) + ) + declared = set(machines[str(machine_id)].states) + for state in _string_items(requirement.get("any_of")): + if state not in declared: + issues.append( + _issue( + "lifecycle.cross.unknown_state", + rule_path, + f"Cross-machine rule {rule_id!r} requires undeclared state {state!r} of " + f"machine {machine_id!r}.", + ) + ) + cascade = _as_mapping(rule.get("cascade")) + if cascade is not None: + from_machine = cascade.get("from_machine") + to_machine = cascade.get("to_machine") + for field_name, machine_id in (("from_machine", from_machine), ("to_machine", to_machine)): + if machine_id not in machines: + issues.append( + _issue( + "lifecycle.cross.unknown_machine", + f"{rule_path}/cascade/{field_name}", + f"Cross-machine rule {rule_id!r} cascade references undeclared machine " + f"{machine_id!r}.", + ) + ) + elif machine_id not in listed: + issues.append( + _issue( + "lifecycle.cross.unlisted_machine", + f"{rule_path}/cascade/{field_name}", + f"Cross-machine rule {rule_id!r} cascade uses machine {machine_id!r} " + "without listing it under machines.", + ) + ) + if from_machine in machines: + triggers = {item.trigger for item in machines[str(from_machine)].transitions} + on_trigger = cascade.get("on_trigger") + if on_trigger not in triggers: + issues.append( + _issue( + "lifecycle.cross.unknown_trigger", + f"{rule_path}/cascade/on_trigger", + f"Cross-machine rule {rule_id!r} cascade fires on trigger {on_trigger!r}, " + f"which no transition of machine {from_machine!r} declares.", + ) + ) + if to_machine in machines: + transition_ids = {item.id for item in machines[str(to_machine)].transitions} + applied = cascade.get("applies_transition") + if applied not in transition_ids: + issues.append( + _issue( + "lifecycle.cross.unknown_transition", + f"{rule_path}/cascade/applies_transition", + f"Cross-machine rule {rule_id!r} cascade applies transition {applied!r}, " + f"which machine {to_machine!r} does not declare.", + ) + ) + for duplicate in _duplicates(rule_ids): + issues.append( + _issue( + "lifecycle.cross.duplicate", + "$.cross_machine_rules", + f"Cross-machine rule id {duplicate!r} is declared more than once.", + ) + ) + return issues + + +def _check_readiness_binding( + contract: Mapping[str, Any], + machines: Mapping[str, MachineModel], + architecture: Mapping[str, Any] | None, + known_services: set[str], +) -> list[ValidationIssue]: + issues: list[ValidationIssue] = [] + binding = _as_mapping(contract.get("readiness_binding")) + if binding is None: + issues.append( + _issue( + "lifecycle.readiness.missing", + "$.readiness_binding", + "readiness_binding must be an object binding the readiness policy evidence to providers.", + ) + ) + return issues + + policy_id = binding.get("policy_id") + bindings = _list_of_mappings(binding.get("evidence_bindings")) + bound_ids = [bid for item in bindings if _non_empty_string(bid := item.get("evidence_id"))] + for duplicate in _duplicates(bound_ids): + issues.append( + _issue( + "lifecycle.readiness.duplicate_evidence", + "$.readiness_binding", + f"Evidence id {duplicate!r} is bound more than once.", + ) + ) + for index, item in enumerate(bindings): + provider = item.get("provider") + item_path = f"$.readiness_binding.evidence_bindings[{index}]" + if not _non_empty_string(provider): + issues.append( + _issue( + "lifecycle.readiness.provider", + item_path, + "Evidence binding is missing a usable provider.", + ) + ) + continue + if provider in machines: + evidence_id = item.get("evidence_id") + if not _non_empty_string(evidence_id): + continue + if evidence_id not in set(machines[provider].evidence_ids): + issues.append( + _issue( + "lifecycle.readiness.unknown_machine_evidence", + item_path, + f"Evidence {evidence_id!r} is attributed to machine {provider!r}, which does not " + "declare it.", + ) + ) + elif provider not in known_services: + issues.append( + _issue( + "lifecycle.readiness.unknown_provider", + item_path, + f"Provider {provider!r} is neither a declared machine nor an architecture-contract " + "service.", + ) + ) + + if architecture is None: + issues.append( + _issue( + "lifecycle.readiness.policy_unverifiable", + "$.readiness_binding", + "The architecture contract could not be loaded, so the readiness binding cannot be " + "verified against the readiness policy; failing closed.", + ) + ) + return issues + + policy_rows = _list_of_mappings(architecture.get("readiness_policies")) + policy_ids = [pid for row in policy_rows if _non_empty_string(pid := row.get("id"))] + duplicate_policies = set(_duplicates(policy_ids)) + for duplicate in sorted(duplicate_policies): + issues.append( + _issue( + "lifecycle.readiness.duplicate_policy", + "$.readiness_binding.policy_id", + f"Readiness policy {duplicate!r} is declared more than once in the architecture " + "contract; the binding cannot be verified against an ambiguous policy.", + ) + ) + if isinstance(policy_id, str) and policy_id in duplicate_policies: + # Selecting either copy would silently endorse one of two conflicting + # readiness semantics; fail closed instead. + return issues + policies = { + pid: policy + for policy in policy_rows + if _non_empty_string(pid := policy.get("id")) + } + policy = policies.get(policy_id) if isinstance(policy_id, str) else None + if policy is None: + issues.append( + _issue( + "lifecycle.readiness.unknown_policy", + "$.readiness_binding.policy_id", + f"Readiness policy {policy_id!r} is not declared by the architecture contract.", + ) + ) + return issues + policy_sources = { + rid: item.get("source") + for item in _list_of_mappings(policy.get("required_evidence")) + if _non_empty_string(rid := item.get("id")) + } + required = set(policy_sources) + bound = set(bound_ids) + for missing in sorted(required - bound): + issues.append( + _issue( + "lifecycle.readiness.evidence_unbound", + "$.readiness_binding", + f"Required readiness evidence {missing!r} has no provider binding.", + ) + ) + for extra in sorted(bound - required): + issues.append( + _issue( + "lifecycle.readiness.evidence_undeclared", + "$.readiness_binding", + f"Evidence {extra!r} is bound but the readiness policy does not require it.", + ) + ) + + # ID-set equality alone would let a binding keep the evidence id but move it + # to another provider, hiding readiness-source drift. Each binding must + # restate the policy's declared source, and a service provider must BE that + # source; machine providers hold evidence produced elsewhere (for example + # the lease carries browser-produced fields), so only their declared + # policy_source is pinned against the policy. + for index, item in enumerate(bindings): + evidence_id = item.get("evidence_id") + # A non-string id (e.g. a list) is a schema violation already reported; + # guard here so the dictionary lookup cannot raise on an unhashable + # value and abort the structured result. + if not _non_empty_string(evidence_id) or evidence_id not in policy_sources: + continue + item_path = f"$.readiness_binding.evidence_bindings[{index}]" + declared_source = item.get("policy_source") + expected_source = policy_sources[evidence_id] + if not _non_empty_string(declared_source) or declared_source != expected_source: + issues.append( + _issue( + "lifecycle.readiness.source_mismatch", + item_path, + f"Evidence {evidence_id!r} declares policy_source {declared_source!r} but the " + f"readiness policy declares source {expected_source!r}.", + ) + ) + provider = item.get("provider") + if _non_empty_string(provider) and provider in known_services and provider != expected_source: + issues.append( + _issue( + "lifecycle.readiness.provider_source_mismatch", + item_path, + f"Evidence {evidence_id!r} is bound to service provider {provider!r} but the " + f"readiness policy declares source {expected_source!r}; a service provider must " + "be the policy's declared source.", + ) + ) + return issues + + +def _check_unused_evidence( + contract: Mapping[str, Any], + machines: Mapping[str, MachineModel], +) -> list[ValidationIssue]: + """Evidence nobody consumes is drift, reported as a warning. + + An evidence declaration is consumed either by a transition's + ``evidence_required`` or by a readiness evidence binding whose provider is + the declaring machine. + """ + + issues: list[ValidationIssue] = [] + readiness = _as_mapping(contract.get("readiness_binding")) + readiness_used: set[tuple[str, str]] = set() + if readiness is not None: + for item in _list_of_mappings(readiness.get("evidence_bindings")): + provider = item.get("provider") + evidence_id = item.get("evidence_id") + if _non_empty_string(provider) and _non_empty_string(evidence_id): + readiness_used.add((provider, evidence_id)) + for machine in machines.values(): + used = { + evidence_id + for transition in machine.transitions + for evidence_id in transition.evidence_required + } + for evidence_id in machine.evidence_ids: + if evidence_id in used or (machine.id, evidence_id) in readiness_used: + continue + issues.append( + _issue( + "lifecycle.evidence.unused", + f"$.machines/{machine.id}/evidence/{evidence_id}", + f"Evidence {evidence_id!r} of machine {machine.id!r} is consumed by no transition " + "and no readiness binding.", + severity="warning", + ) + ) + return issues + + +# --------------------------------------------------------------------------- # +# Entry point +# --------------------------------------------------------------------------- # + + +def check_lifecycle_contracts(repo_root: Path) -> LifecycleCheckResult: + """Validate the lifecycle contract and its source synchronization.""" + + repo_root = Path(repo_root).resolve() + issues: list[ValidationIssue] = [] + + contract, contract_issues = _load_lifecycle_contract(repo_root) + issues.extend(contract_issues) + architecture, architecture_issues = _load_architecture_contract(repo_root) + issues.extend(architecture_issues) + + if contract is None: + return LifecycleCheckResult( + repo_root=str(repo_root), + compared=False, + machines=(), + issues=_finalize(issues), + ) + + machines: dict[str, MachineModel] = {} + machine_ids: list[str] = [] + for index, machine in enumerate(_list_of_mappings(contract.get("machines"))): + model, machine_issues = _build_machine(machine, f"$.machines[{index}]") + issues.extend(machine_issues) + if model is None: + continue + machine_ids.append(model.id) + if model.id not in machines: + machines[model.id] = model + for duplicate in _duplicates(machine_ids): + issues.append( + _issue( + "lifecycle.machine.duplicate", + "$.machines", + f"Machine id {duplicate!r} is declared more than once.", + ) + ) + + known_services: set[str] = set() + if architecture is not None: + known_services = { + service_id + for service in _list_of_mappings(architecture.get("services")) + if _non_empty_string(service_id := service.get("id")) + } + for machine in machines.values(): + if machine.owner_service not in known_services: + issues.append( + _issue( + "lifecycle.machine.unknown_service", + f"$.machines/{machine.id}", + f"Machine {machine.id!r} names owner service {machine.owner_service!r}, which the " + "architecture contract does not declare.", + ) + ) + else: + issues.append( + _issue( + "lifecycle.machine.owner_unverifiable", + "$.machines", + "The architecture contract could not be loaded, so machine owner services cannot be " + "verified; failing closed.", + ) + ) + + for machine in machines.values(): + issues.extend(_check_source_sync(repo_root, machine, f"$.machines/{machine.id}/source_binding")) + + issues.extend(_check_cross_machine_rules(contract, machines)) + issues.extend(_check_readiness_binding(contract, machines, architecture, known_services)) + issues.extend(_check_unused_evidence(contract, machines)) + + return LifecycleCheckResult( + repo_root=str(repo_root), + compared=True, + machines=tuple(machines[machine_id] for machine_id in sorted(machines)), + issues=_finalize(issues), + ) + + +def _finalize(issues: Iterable[ValidationIssue]) -> tuple[ValidationIssue, ...]: + return tuple( + sorted(set(issues), key=lambda item: (item.severity, item.path, item.code, item.message)) + ) diff --git a/scripts/verification-manifest.json b/scripts/verification-manifest.json index be05be930..b677376ad 100644 --- a/scripts/verification-manifest.json +++ b/scripts/verification-manifest.json @@ -115,6 +115,8 @@ "scripts/dev/export_observed_architecture.py", "scripts/lib/layered_architecture.py", "scripts/dev/check_layered_architecture.py", + "scripts/lib/lifecycle_contracts.py", + "scripts/dev/check_lifecycle_contracts.py", "scripts/dev/check_governance_trust_root.py" ] }, @@ -309,7 +311,7 @@ ], "targets": [ { - "id": "root-contracts", "display_name": "tests (contracts+fakes)", "path_globs": ["tests/**", "scripts/tests/**", "scripts/gen_routing.py", ".claude/workflows/**", "docs/contracts/**", "bim-review-coordinator/**", "bim-streaming-server/**", "governance-service/**", "rules/**", "web-viewer-sample/**", "services/kit-manager-api/**", "apps/kit-manager-web/**", "compose.*.yml", "architecture/**", "scripts/lib/architecture_contract.py", "scripts/dev/validate_architecture_contract.py", "scripts/lib/observed_architecture.py", "scripts/dev/export_observed_architecture.py", "scripts/lib/layered_architecture.py", "scripts/dev/check_layered_architecture.py", "scripts/dev/check_governance_trust_root.py"], + "id": "root-contracts", "display_name": "tests (contracts+fakes)", "path_globs": ["tests/**", "scripts/tests/**", "scripts/gen_routing.py", ".claude/workflows/**", "docs/contracts/**", "bim-review-coordinator/**", "bim-streaming-server/**", "governance-service/**", "rules/**", "web-viewer-sample/**", "services/kit-manager-api/**", "apps/kit-manager-web/**", "compose.*.yml", "architecture/**", "scripts/lib/architecture_contract.py", "scripts/dev/validate_architecture_contract.py", "scripts/lib/observed_architecture.py", "scripts/dev/export_observed_architecture.py", "scripts/lib/layered_architecture.py", "scripts/dev/check_layered_architecture.py", "scripts/lib/lifecycle_contracts.py", "scripts/dev/check_lifecycle_contracts.py", "scripts/dev/check_governance_trust_root.py"], "owner": "contract-owner", "fast_gates": [], "contract_gates": ["root-contracts"], "slow_evidence_gates": [], "required_when": { "predicate": "changed_path_class", "any_of": ["root-contracts"] }, "skip_reason": "path_not_affected", "default_profiles": ["developer", "developer-py"], "ci_output": "root_contracts", diff --git a/tests/test_lifecycle_contracts.py b/tests/test_lifecycle_contracts.py new file mode 100644 index 000000000..04ec8e558 --- /dev/null +++ b/tests/test_lifecycle_contracts.py @@ -0,0 +1,1247 @@ +"""Fail-closed and model-based tests for the lifecycle contracts (Phase 4). + +Two protection layers, mirroring Phase 3: + +* **Pinned literals.** The machine ids, state sets, forbidden shortcuts, + evidence-gated transitions, source bindings, and the schema files' + load-bearing constraints are pinned here as literals. A ratchet over a + contract cannot detect the contract itself being loosened, so loosening it + must require an edit to this file that is visible in the same diff. +* **Model-based properties.** The tests load the canonical contract as a + transition system and assert properties over *every* enumerated path rather + than hand-picked examples: forbidden pairs have no single-step edge, every + pending-to-active path crosses the evidence gates, terminal states are + closed, and every state is reachable or explicitly declared_only. +""" + +from __future__ import annotations + +from copy import deepcopy +import json +import os +from pathlib import Path +import subprocess +import sys + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from scripts.lib.architecture_contract import validate_schema_instance # noqa: E402 +from scripts.lib.lifecycle_contracts import ( # noqa: E402 + CONTRACT_SCHEMA_VERSION, + MachineModel, + _extract_union_literals, + check_lifecycle_contracts, +) + + +# --------------------------------------------------------------------------- # +# Pinned literals (loosening the contract must edit this file) +# --------------------------------------------------------------------------- # + +PINNED_MACHINE_IDS = frozenset({"review-session", "endpoint-lease", "stage-binding"}) + +PINNED_STATES: dict[str, frozenset[str]] = { + "review-session": frozenset({"created", "active", "closing", "closed", "failed"}), + "endpoint-lease": frozenset({"active", "released", "expired"}), + "stage-binding": frozenset({"pending", "executing", "active", "failed", "superseded"}), +} + +PINNED_SOURCE_BINDINGS: dict[str, tuple[str, str]] = { + "review-session": ("bim-review-coordinator/src/types.ts", "SessionStatus"), + "endpoint-lease": ( + "bim-review-coordinator/src/services/viewerLeaseStore.ts", + "ViewerLeaseStatus", + ), + "stage-binding": ( + "bim-review-coordinator/src/services/runtimeMutationAuthority/stageBindingState.ts", + "StageBindingStatus", + ), +} + +PINNED_FORBIDDEN: frozenset[tuple[str, str, str]] = frozenset( + { + ("review-session", "created", "closed"), + ("review-session", "active", "closed"), + ("endpoint-lease", "released", "active"), + ("endpoint-lease", "expired", "active"), + ("stage-binding", "pending", "active"), + ("stage-binding", "executing", "superseded"), + } +) + +PINNED_EVIDENCE_GATED_TRANSITIONS: dict[tuple[str, str], frozenset[str]] = { + ("stage-binding", "kit-consume"): frozenset({"attempt-binding-match"}), + ("stage-binding", "confirm-load-success"): frozenset( + {"attempt-binding-match", "runtime-load-outcome"} + ), + ("stage-binding", "confirm-load-failure"): frozenset( + {"attempt-binding-match", "runtime-load-outcome"} + ), + ("stage-binding", "authorization-unavailable-pending"): frozenset({"attempt-binding-match"}), + ("stage-binding", "authorization-unavailable-executing"): frozenset({"attempt-binding-match"}), + ("stage-binding", "cancel-pending"): frozenset({"cancellation-intent-match"}), +} + +# The two canonical cross-machine rules, pinned so emptying or renaming them +# requires an edit visible in the same diff. +PINNED_CROSS_MACHINE_RULE_IDS = frozenset( + { + "stage-binding-requires-open-session-and-primary-lease", + "session-close-cascades-lease-release", + } +) + +PINNED_READINESS_EVIDENCE = frozenset( + {"kit-process-alive", "opened-stage-result", "datachannel-ready", "first-frame-at", "stage-matched"} +) + +# (evidence_id, provider, policy_source): the full readiness wiring is pinned so +# neither the provider nor the declared policy source can drift silently. +PINNED_READINESS_BINDINGS = frozenset( + { + ("kit-process-alive", "kit-manager-api", "kit-manager-api"), + ("opened-stage-result", "bim-streaming-server", "bim-streaming-server"), + ("datachannel-ready", "endpoint-lease", "web-viewer-sample"), + ("first-frame-at", "endpoint-lease", "web-viewer-sample"), + ("stage-matched", "endpoint-lease", "web-viewer-sample"), + } +) + +PINNED_DECLARED_ONLY: frozenset[tuple[str, str]] = frozenset({("review-session", "failed")}) + +# Load-bearing constraints of the schema file. Replacing the schema with a stub +# must show up either here or in the checker's vacuous-schema guard. +PINNED_SCHEMA_TOP_REQUIRED = frozenset( + { + "$schema", + "schema_version", + "purpose", + "enforcement_note", + "machines", + "cross_machine_rules", + "readiness_binding", + } +) +PINNED_SCHEMA_MACHINE_REQUIRED = frozenset( + { + "id", + "title", + "owner_service", + "source_binding", + "states", + "transitions", + "forbidden_shortcuts", + "evidence", + "reentry_rules", + "notes", + } +) +PINNED_SCHEMA_TRANSITION_REQUIRED = frozenset( + {"id", "from", "to", "trigger", "evidence_required", "description"} +) +PINNED_SCHEMA_STATE_KINDS = ("initial", "intermediate", "terminal") + + +# --------------------------------------------------------------------------- # +# Helpers +# --------------------------------------------------------------------------- # + + +CONTRACT_PATH = ROOT / "architecture" / "lifecycle-contract.json" +SCHEMA_PATH = ROOT / "architecture" / "lifecycle-contract.schema.json" +ARCHITECTURE_PATH = ROOT / "architecture" / "architecture-contract.json" +ARCHITECTURE_SCHEMA_PATH = ROOT / "architecture" / "architecture-contract.schema.json" + + +def load_contract() -> dict: + return json.loads(CONTRACT_PATH.read_text(encoding="utf-8")) + + +def load_schema() -> dict: + return json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) + + +def load_architecture() -> dict: + return json.loads(ARCHITECTURE_PATH.read_text(encoding="utf-8")) + + +def issue_codes(result) -> set[str]: + return {issue.code for issue in result.issues} + + +def machine_by_id(contract: dict, machine_id: str) -> dict: + return next(machine for machine in contract["machines"] if machine["id"] == machine_id) + + +def model_by_id(result, machine_id: str) -> MachineModel: + return next(machine for machine in result.machines if machine.id == machine_id) + + +def build_tmp_repo( + tmp_path: Path, + *, + contract: dict | None = None, + schema: object = None, + architecture: dict | None = None, + drop_contract: bool = False, + drop_architecture: bool = False, + source_overrides: dict[str, str] | None = None, + drop_sources: tuple[str, ...] = (), +) -> Path: + """Copy the canonical contract set into a scratch repository and mutate it.""" + + repo = tmp_path / "repo" + arch_dir = repo / "architecture" + arch_dir.mkdir(parents=True) + + if not drop_contract: + document = contract if contract is not None else load_contract() + (arch_dir / "lifecycle-contract.json").write_text( + json.dumps(document, ensure_ascii=False, indent=2), encoding="utf-8" + ) + schema_document = schema if schema is not None else load_schema() + (arch_dir / "lifecycle-contract.schema.json").write_text( + json.dumps(schema_document, ensure_ascii=False, indent=2), encoding="utf-8" + ) + if not drop_architecture: + architecture_document = architecture if architecture is not None else load_architecture() + (arch_dir / "architecture-contract.json").write_text( + json.dumps(architecture_document, ensure_ascii=False, indent=2), encoding="utf-8" + ) + (arch_dir / "architecture-contract.schema.json").write_text( + ARCHITECTURE_SCHEMA_PATH.read_text(encoding="utf-8"), encoding="utf-8" + ) + + overrides = source_overrides or {} + for source_file, _type_name in PINNED_SOURCE_BINDINGS.values(): + if source_file in drop_sources: + continue + destination = repo / source_file + destination.parent.mkdir(parents=True, exist_ok=True) + text = overrides.get(source_file) + if text is None: + text = (ROOT / source_file).read_text(encoding="utf-8") + destination.write_text(text, encoding="utf-8") + return repo + + +# --------------------------------------------------------------------------- # +# Canonical repository +# --------------------------------------------------------------------------- # + + +def test_canonical_repository_lifecycle_contracts_pass() -> None: + result = check_lifecycle_contracts(ROOT) + + assert result.compared is True + assert result.error_count == 0, [issue.to_dict() for issue in result.issues] + # Unused evidence or any other warning on the canonical tree means the gate + # is rotting quietly; the canonical repository must carry none. + assert result.warning_count == 0, [issue.to_dict() for issue in result.issues] + assert result.status == "passed" + assert result.machine_count == len(PINNED_MACHINE_IDS) + + +def test_canonical_contract_matches_its_schema() -> None: + assert validate_schema_instance(load_contract(), load_schema()) == [] + + +def test_canonical_machines_states_and_declared_only_are_pinned() -> None: + contract = load_contract() + machines = {machine["id"]: machine for machine in contract["machines"]} + assert set(machines) == set(PINNED_MACHINE_IDS) + for machine_id, expected_states in PINNED_STATES.items(): + declared = {state["id"] for state in machines[machine_id]["states"]} + assert declared == set(expected_states), machine_id + declared_only = { + (machine["id"], state["id"]) + for machine in contract["machines"] + for state in machine["states"] + if state.get("runtime_write_path") == "declared_only" + } + assert declared_only == set(PINNED_DECLARED_ONLY) + + +def test_canonical_forbidden_shortcuts_are_pinned() -> None: + contract = load_contract() + declared = { + (machine["id"], shortcut["from"], shortcut["to"]) + for machine in contract["machines"] + for shortcut in machine["forbidden_shortcuts"] + } + assert declared == set(PINNED_FORBIDDEN) + + +def test_canonical_source_bindings_are_pinned() -> None: + contract = load_contract() + declared = { + machine["id"]: (machine["source_binding"]["file"], machine["source_binding"]["type_name"]) + for machine in contract["machines"] + } + assert declared == PINNED_SOURCE_BINDINGS + + +def test_canonical_evidence_gated_transitions_are_pinned() -> None: + contract = load_contract() + gated = { + (machine["id"], transition["id"]): frozenset(transition["evidence_required"]) + for machine in contract["machines"] + for transition in machine["transitions"] + if transition["evidence_required"] + } + assert gated == PINNED_EVIDENCE_GATED_TRANSITIONS + + +def test_canonical_readiness_binding_matches_architecture_policy_and_pins() -> None: + contract = load_contract() + architecture = load_architecture() + binding = contract["readiness_binding"] + assert binding["policy_id"] == "review-session-ready" + bound = {item["evidence_id"] for item in binding["evidence_bindings"]} + assert bound == set(PINNED_READINESS_EVIDENCE) + wiring = { + (item["evidence_id"], item["provider"], item["policy_source"]) + for item in binding["evidence_bindings"] + } + assert wiring == set(PINNED_READINESS_BINDINGS) + policy = next( + policy + for policy in architecture["readiness_policies"] + if policy["id"] == "review-session-ready" + ) + required = {item["id"] for item in policy["required_evidence"]} + assert bound == required + policy_sources = {item["id"]: item["source"] for item in policy["required_evidence"]} + for item in binding["evidence_bindings"]: + assert item["policy_source"] == policy_sources[item["evidence_id"]], item["evidence_id"] + + +def test_lifecycle_invariant_is_pinned_in_the_architecture_validator() -> None: + from scripts.lib.architecture_contract import REQUIRED_INVARIANT_IDS + + assert "ARCH-LIFECYCLE-001" in REQUIRED_INVARIANT_IDS + assert "ARCH-LAYER-001" in REQUIRED_INVARIANT_IDS + + +def test_canonical_cross_machine_rules_are_pinned() -> None: + contract = load_contract() + rules = {rule["id"]: rule for rule in contract["cross_machine_rules"]} + assert set(rules) == set(PINNED_CROSS_MACHINE_RULE_IDS) + gate_rule = rules["stage-binding-requires-open-session-and-primary-lease"] + required = { + requirement["machine"]: frozenset(requirement["any_of"]) + for requirement in gate_rule["required_states"] + } + assert required == { + "review-session": frozenset({"created", "active"}), + "endpoint-lease": frozenset({"active"}), + } + cascade_rule = rules["session-close-cascades-lease-release"] + assert cascade_rule["cascade"] == { + "from_machine": "review-session", + "on_trigger": "close-session", + "to_machine": "endpoint-lease", + "applies_transition": "session-close-release", + } + # Emptying the rule list is now also a schema violation (minItems 1). + schema = load_schema() + assert schema["properties"]["cross_machine_rules"]["minItems"] == 1 + + +def test_failure_code_on_nonterminal_target_rejected(tmp_path) -> None: + contract = load_contract() + machine_by_id(contract, "stage-binding")["transitions"].append( + { + "id": "coded-into-live-state", + "from": "pending", + "to": "executing", + "trigger": "coded-live-trigger", + "evidence_required": [], + "failure_code": "not_a_terminal_entry", + "description": "failure_code on a non-terminal target as a counterexample.", + } + ) + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.transition.failure_code_nonterminal" in issue_codes(result) + + +def test_unattributed_entry_into_a_coded_failure_target_rejected(tmp_path) -> None: + contract = load_contract() + machine_by_id(contract, "stage-binding")["transitions"].append( + { + "id": "uncoded-failure-entry", + "from": "pending", + "to": "failed", + "trigger": "uncoded-failure-trigger", + "evidence_required": [], + "description": "Entry into failed without a failure_code as a counterexample.", + } + ) + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.transition.failure_code_inconsistent" in issue_codes(result) + + +def test_schema_files_keep_their_load_bearing_constraints() -> None: + schema = load_schema() + assert schema["additionalProperties"] is False + assert set(schema["required"]) == set(PINNED_SCHEMA_TOP_REQUIRED) + machine = schema["definitions"]["machine"] + assert machine["additionalProperties"] is False + assert set(machine["required"]) == set(PINNED_SCHEMA_MACHINE_REQUIRED) + transition = schema["definitions"]["transition"] + assert transition["additionalProperties"] is False + assert set(transition["required"]) == set(PINNED_SCHEMA_TRANSITION_REQUIRED) + assert tuple(schema["definitions"]["state"]["properties"]["kind"]["enum"]) == ( + PINNED_SCHEMA_STATE_KINDS + ) + assert schema["properties"]["schema_version"]["const"] == CONTRACT_SCHEMA_VERSION + + +# --------------------------------------------------------------------------- # +# Model-based properties over the canonical contract +# --------------------------------------------------------------------------- # + + +@pytest.fixture(scope="module") +def canonical_result(): + return check_lifecycle_contracts(ROOT) + + +def test_model_forbidden_pairs_have_no_single_step_edge(canonical_result) -> None: + for machine in canonical_result.machines: + edges = machine.edges() + for pair in machine.forbidden_pairs: + assert pair not in edges, (machine.id, pair) + + +def test_model_every_pending_to_active_path_crosses_the_evidence_gates(canonical_result) -> None: + machine = model_by_id(canonical_result, "stage-binding") + paths = machine.simple_paths("pending", "active") + assert paths, "pending must be able to reach active through the declared transitions" + for path in paths: + visited = [transition.to_state for transition in path] + assert "executing" in visited, [transition.id for transition in path] + evidence = { + evidence_id + for transition in path + for evidence_id in transition.evidence_required + } + assert evidence == {"attempt-binding-match", "runtime-load-outcome"}, ( + [transition.id for transition in path] + ) + + +def test_model_every_forbidden_pair_path_crosses_an_intermediate_state(canonical_result) -> None: + """No multi-step bypass reaches a forbidden target without an intermediate. + + A forbidden shortcut bans the *single-step* edge; a longer path is legal + exactly because it crosses the machine's intermediate evidence-carrying + states. Enumerating every simple path proves there is no second direct + route hiding in the model. + """ + + for machine in canonical_result.machines: + kinds_intermediate = { + state + for state in machine.states + if state not in machine.terminal_states and state not in machine.initial_states + } + for source, target in machine.forbidden_pairs: + for path in machine.simple_paths(source, target): + assert len(path) >= 2, (machine.id, source, target) + crossed = {transition.to_state for transition in path[:-1]} + assert crossed & kinds_intermediate, (machine.id, source, target, crossed) + + +def test_model_terminal_states_have_no_outgoing_transitions(canonical_result) -> None: + for machine in canonical_result.machines: + terminal = set(machine.terminal_states) + for transition in machine.transitions: + assert transition.from_state not in terminal, (machine.id, transition.id) + + +def test_model_all_states_reachable_or_declared_only(canonical_result) -> None: + for machine in canonical_result.machines: + reachable = machine.reachable_states() + unreachable = set(machine.states) - reachable - set(machine.declared_only_states) + assert not unreachable, (machine.id, unreachable) + + +def test_model_transitions_deterministic_per_from_and_trigger(canonical_result) -> None: + for machine in canonical_result.machines: + seen: dict[tuple[str, str], str] = {} + for transition in machine.transitions: + key = (transition.from_state, transition.trigger) + assert key not in seen, (machine.id, transition.id, seen[key]) + seen[key] = transition.id + + +def test_model_lease_terminal_states_cannot_reach_active(canonical_result) -> None: + machine = model_by_id(canonical_result, "endpoint-lease") + for source in ("released", "expired"): + assert machine.simple_paths(source, "active") == [] + + +# --------------------------------------------------------------------------- # +# Fail-closed loading +# --------------------------------------------------------------------------- # + + +def test_missing_contract_file_fails_closed(tmp_path) -> None: + repo = build_tmp_repo(tmp_path, drop_contract=True) + result = check_lifecycle_contracts(repo) + assert result.compared is False + assert result.status == "failed" + assert "file.read" in issue_codes(result) + + +def test_non_object_contract_fails_closed(tmp_path) -> None: + repo = build_tmp_repo(tmp_path) + (repo / "architecture" / "lifecycle-contract.json").write_text("null", encoding="utf-8") + result = check_lifecycle_contracts(repo) + assert result.compared is False + assert result.status == "failed" + assert "lifecycle_contract.not_object" in issue_codes(result) + + +def test_vacuous_schema_rejected(tmp_path) -> None: + for index, stub in enumerate(({}, {"type": "object"}, {"properties": {}}, {"required": []})): + repo = build_tmp_repo(tmp_path / f"case-{index}", schema=stub) + result = check_lifecycle_contracts(repo) + assert result.status == "failed", stub + codes = issue_codes(result) + assert "lifecycle_contract.schema_vacuous" in codes or ( + "lifecycle_contract.schema_not_object" in codes + ), stub + + +def test_wrong_schema_version_rejected(tmp_path) -> None: + contract = load_contract() + contract["schema_version"] = "ai-bim-lifecycle-contract/v0" + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle_contract.schema_version" in issue_codes(result) + + +def test_missing_architecture_contract_fails_closed(tmp_path) -> None: + repo = build_tmp_repo(tmp_path, drop_architecture=True) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + codes = issue_codes(result) + assert "lifecycle.machine.owner_unverifiable" in codes + assert "lifecycle.readiness.policy_unverifiable" in codes + + +# --------------------------------------------------------------------------- # +# Fail-closed machine semantics +# --------------------------------------------------------------------------- # + + +def test_unknown_transition_state_rejected(tmp_path) -> None: + contract = load_contract() + machine_by_id(contract, "endpoint-lease")["transitions"][0]["to"] = "ghost" + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.transition.unknown_state" in issue_codes(result) + + +def test_terminal_outgoing_transition_rejected(tmp_path) -> None: + contract = load_contract() + machine_by_id(contract, "review-session")["transitions"].append( + { + "id": "reopen", + "from": "closed", + "to": "active", + "trigger": "reopen-session", + "evidence_required": [], + "description": "Terminal escape used as a counterexample.", + } + ) + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.terminal.outgoing" in issue_codes(result) + + +def test_forbidden_pair_with_direct_edge_contradiction_rejected(tmp_path) -> None: + contract = load_contract() + machine_by_id(contract, "stage-binding")["transitions"].append( + { + "id": "shortcut", + "from": "pending", + "to": "active", + "trigger": "shortcut-trigger", + "evidence_required": [], + "description": "Forbidden shortcut materialized as a counterexample.", + } + ) + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.forbidden.direct_edge_exists" in issue_codes(result) + + +def test_unknown_evidence_reference_rejected(tmp_path) -> None: + contract = load_contract() + machine_by_id(contract, "stage-binding")["transitions"][0]["evidence_required"] = [ + "no-such-evidence" + ] + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.evidence.unknown" in issue_codes(result) + + +def test_unreachable_state_rejected(tmp_path) -> None: + contract = load_contract() + machine = machine_by_id(contract, "endpoint-lease") + machine["states"].append( + { + "id": "orphaned", + "kind": "intermediate", + "description": "No transition reaches this state.", + } + ) + # Keep the source union untouched: the orphan also desynchronizes the + # source binding, and both findings must appear. + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + codes = issue_codes(result) + assert "lifecycle.state.unreachable" in codes + assert "lifecycle.source_sync.state_missing_in_source" in codes + + +def test_declared_only_state_with_transition_rejected(tmp_path) -> None: + contract = load_contract() + machine_by_id(contract, "review-session")["transitions"].append( + { + "id": "fail-session", + "from": "active", + "to": "failed", + "trigger": "fail-session", + "evidence_required": [], + "description": "Wiring a declared_only state as a counterexample.", + } + ) + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.state.declared_only_wired" in issue_codes(result) + + +def test_declared_only_initial_state_rejected(tmp_path) -> None: + contract = load_contract() + machine = machine_by_id(contract, "review-session") + failed_state = next(state for state in machine["states"] if state["id"] == "failed") + failed_state["kind"] = "initial" + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.state.declared_only_initial" in issue_codes(result) + + +def test_duplicate_machine_rejected(tmp_path) -> None: + contract = load_contract() + contract["machines"].append(deepcopy(contract["machines"][0])) + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.machine.duplicate" in issue_codes(result) + + +def test_duplicate_state_rejected(tmp_path) -> None: + contract = load_contract() + machine = machine_by_id(contract, "endpoint-lease") + machine["states"].append(deepcopy(machine["states"][0])) + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.state.duplicate" in issue_codes(result) + + +def test_duplicate_transition_id_rejected(tmp_path) -> None: + contract = load_contract() + machine = machine_by_id(contract, "endpoint-lease") + clone = deepcopy(machine["transitions"][0]) + clone["trigger"] = "another-trigger" + machine["transitions"].append(clone) + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.transition.duplicate" in issue_codes(result) + + +def test_nondeterministic_transitions_rejected(tmp_path) -> None: + contract = load_contract() + machine = machine_by_id(contract, "endpoint-lease") + machine["transitions"].append( + { + "id": "ttl-expire-elsewhere", + "from": "active", + "to": "released", + "trigger": "ttl-sweep", + "evidence_required": [], + "description": "Same (from, trigger) with a different target as a counterexample.", + } + ) + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.transition.nondeterministic" in issue_codes(result) + + +def test_forbidden_self_loop_rejected(tmp_path) -> None: + contract = load_contract() + machine_by_id(contract, "endpoint-lease")["forbidden_shortcuts"].append( + { + "id": "self-loop", + "from": "active", + "to": "active", + "reason": "Nonsense pair as a counterexample.", + "enforced_by": "nothing", + } + ) + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.forbidden.self_loop" in issue_codes(result) + + +def test_unknown_owner_service_rejected(tmp_path) -> None: + contract = load_contract() + machine_by_id(contract, "endpoint-lease")["owner_service"] = "ghost-service" + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.machine.unknown_service" in issue_codes(result) + + +# --------------------------------------------------------------------------- # +# Fail-closed cross-machine rules +# --------------------------------------------------------------------------- # + + +def test_cross_rule_unknown_machine_rejected(tmp_path) -> None: + contract = load_contract() + contract["cross_machine_rules"][0]["machines"] = ["review-session", "ghost-machine"] + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.cross.unknown_machine" in issue_codes(result) + + +def test_cross_rule_unknown_required_state_rejected(tmp_path) -> None: + contract = load_contract() + contract["cross_machine_rules"][0]["required_states"][0]["any_of"] = ["ghost-state"] + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.cross.unknown_state" in issue_codes(result) + + +def test_cascade_unknown_trigger_rejected(tmp_path) -> None: + contract = load_contract() + contract["cross_machine_rules"][1]["cascade"]["on_trigger"] = "ghost-trigger" + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.cross.unknown_trigger" in issue_codes(result) + + +def test_cascade_unknown_transition_rejected(tmp_path) -> None: + contract = load_contract() + contract["cross_machine_rules"][1]["cascade"]["applies_transition"] = "ghost-transition" + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.cross.unknown_transition" in issue_codes(result) + + +# --------------------------------------------------------------------------- # +# Fail-closed readiness binding +# --------------------------------------------------------------------------- # + + +def test_readiness_missing_required_evidence_rejected(tmp_path) -> None: + contract = load_contract() + bindings = contract["readiness_binding"]["evidence_bindings"] + contract["readiness_binding"]["evidence_bindings"] = [ + item for item in bindings if item["evidence_id"] != "first-frame-at" + ] + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.readiness.evidence_unbound" in issue_codes(result) + + +def test_readiness_extra_evidence_rejected(tmp_path) -> None: + contract = load_contract() + contract["readiness_binding"]["evidence_bindings"].append( + { + "evidence_id": "extra-evidence", + "provider": "kit-manager-api", + "policy_source": "kit-manager-api", + "surface": "Invented evidence as a counterexample.", + } + ) + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.readiness.evidence_undeclared" in issue_codes(result) + + +def test_readiness_service_provider_drift_rejected(tmp_path) -> None: + """Moving an evidence binding to another known service must fail. + + ID-set equality alone accepted this (the codex ship-gate counterexample): + kit-process-alive re-bound to web-viewer-sample kept the same ids. + """ + + contract = load_contract() + binding = next( + item + for item in contract["readiness_binding"]["evidence_bindings"] + if item["evidence_id"] == "kit-process-alive" + ) + binding["provider"] = "web-viewer-sample" + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.readiness.provider_source_mismatch" in issue_codes(result) + + +def test_readiness_policy_source_mismatch_rejected(tmp_path) -> None: + contract = load_contract() + binding = next( + item + for item in contract["readiness_binding"]["evidence_bindings"] + if item["evidence_id"] == "first-frame-at" + ) + binding["policy_source"] = "bim-review-coordinator" + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.readiness.source_mismatch" in issue_codes(result) + + +def test_demoting_a_terminal_state_to_intermediate_is_a_dead_end(tmp_path) -> None: + """The codex ship-gate counterexample: released -> intermediate passed.""" + + contract = load_contract() + machine = machine_by_id(contract, "endpoint-lease") + released = next(state for state in machine["states"] if state["id"] == "released") + released["kind"] = "intermediate" + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.state.dead_end" in issue_codes(result) + + +def test_readiness_unknown_policy_rejected(tmp_path) -> None: + contract = load_contract() + contract["readiness_binding"]["policy_id"] = "ghost-policy" + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.readiness.unknown_policy" in issue_codes(result) + + +def test_readiness_unknown_machine_evidence_rejected(tmp_path) -> None: + contract = load_contract() + binding = next( + item + for item in contract["readiness_binding"]["evidence_bindings"] + if item["evidence_id"] == "first-frame-at" + ) + assert binding["provider"] == "endpoint-lease" + binding["evidence_id"] = "stage-matched-typo" + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + codes = issue_codes(result) + assert "lifecycle.readiness.unknown_machine_evidence" in codes + + +def test_readiness_unknown_provider_rejected(tmp_path) -> None: + contract = load_contract() + contract["readiness_binding"]["evidence_bindings"][0]["provider"] = "ghost-provider" + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.readiness.unknown_provider" in issue_codes(result) + + +def test_unused_evidence_warns_without_failing(tmp_path) -> None: + contract = load_contract() + machine_by_id(contract, "stage-binding")["evidence"].append( + { + "id": "dangling-evidence", + "description": "Declared but consumed by nothing.", + "source": "nowhere", + } + ) + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert "lifecycle.evidence.unused" in issue_codes(result) + assert result.warning_count >= 1 + assert result.error_count == 0 + assert result.status == "passed" + + +# --------------------------------------------------------------------------- # +# Source synchronization +# --------------------------------------------------------------------------- # + + +def test_source_sync_state_missing_in_source_rejected(tmp_path) -> None: + source_file, _ = PINNED_SOURCE_BINDINGS["review-session"] + text = (ROOT / source_file).read_text(encoding="utf-8") + assert '"closing" | "closed" | "failed";' in text + mutated = text.replace('"closing" | "closed" | "failed";', '"closing" | "closed";', 1) + repo = build_tmp_repo(tmp_path, source_overrides={source_file: mutated}) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.source_sync.state_missing_in_source" in issue_codes(result) + + +def test_source_sync_state_missing_in_contract_rejected(tmp_path) -> None: + source_file, _ = PINNED_SOURCE_BINDINGS["review-session"] + text = (ROOT / source_file).read_text(encoding="utf-8") + mutated = text.replace('"failed";', '"failed" | "paused";', 1) + assert mutated != text + repo = build_tmp_repo(tmp_path, source_overrides={source_file: mutated}) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.source_sync.state_missing_in_contract" in issue_codes(result) + + +def test_source_sync_type_not_found_rejected(tmp_path) -> None: + source_file, type_name = PINNED_SOURCE_BINDINGS["endpoint-lease"] + text = (ROOT / source_file).read_text(encoding="utf-8") + mutated = text.replace(f"export type {type_name} =", f"export type {type_name}Renamed =", 1) + assert mutated != text + repo = build_tmp_repo(tmp_path, source_overrides={source_file: mutated}) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.source_sync.union_unparsed" in issue_codes(result) + + +def test_source_sync_union_referencing_another_type_rejected(tmp_path) -> None: + source_file, type_name = PINNED_SOURCE_BINDINGS["endpoint-lease"] + text = (ROOT / source_file).read_text(encoding="utf-8") + mutated = text.replace( + f'export type {type_name} = "active"', + f'export type {type_name} = SomeAlias | "active"', + 1, + ) + assert mutated != text + repo = build_tmp_repo(tmp_path, source_overrides={source_file: mutated}) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.source_sync.union_unparsed" in issue_codes(result) + + +def test_source_sync_missing_file_rejected(tmp_path) -> None: + source_file, _ = PINNED_SOURCE_BINDINGS["stage-binding"] + repo = build_tmp_repo(tmp_path, drop_sources=(source_file,)) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.source_sync.file_unreadable" in issue_codes(result) + + +def test_source_sync_ignores_commented_out_stale_union(tmp_path) -> None: + """A dead comment carrying the old literals must not mask real drift. + + The codex ship-gate counterexample: a preceding `// export type ...` with + the stale union let a real `"paused"` addition go unnoticed. + """ + + source_file, type_name = PINNED_SOURCE_BINDINGS["review-session"] + text = (ROOT / source_file).read_text(encoding="utf-8") + stale = f'// export type {type_name} = "created" | "active" | "closing" | "closed" | "failed";\n' + mutated = stale + text.replace('"failed";', '"failed" | "paused";', 1) + assert mutated != stale + text + repo = build_tmp_repo(tmp_path, source_overrides={source_file: mutated}) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.source_sync.state_missing_in_contract" in issue_codes(result) + + +def test_source_sync_duplicate_declarations_fail_closed(tmp_path) -> None: + source_file, type_name = PINNED_SOURCE_BINDINGS["endpoint-lease"] + text = (ROOT / source_file).read_text(encoding="utf-8") + mutated = text + f'\nexport type {type_name} = "active" | "released" | "expired";\n' + repo = build_tmp_repo(tmp_path, source_overrides={source_file: mutated}) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.source_sync.union_unparsed" in issue_codes(result) + + +def test_source_sync_block_commented_union_is_not_source(tmp_path) -> None: + """The codex counterexample: live declaration replaced by a re-export while + the old union survives inside a block comment must not report passed.""" + + source_file, type_name = PINNED_SOURCE_BINDINGS["review-session"] + text = (ROOT / source_file).read_text(encoding="utf-8") + live = 'export type SessionStatus = "created" | "active" | "closing" | "closed" | "failed";' + assert live in text + mutated = text.replace( + live, + f'/*\n{live}\n*/\nexport type {type_name}X = "unrelated";', + 1, + ) + repo = build_tmp_repo(tmp_path, source_overrides={source_file: mutated}) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.source_sync.union_unparsed" in issue_codes(result) + + +def test_strip_ts_comments_behaviors() -> None: + from scripts.lib.lifecycle_contracts import _strip_ts_comments + + stripped, error = _strip_ts_comments('const url = "http://x/*y*/z"; // tail\ncode();') + assert error is None + assert stripped is not None + assert '"http://x/*y*/z"' in stripped + assert "tail" not in stripped + assert "code();" in stripped + + blocked, error = _strip_ts_comments("before /* comment */ after") + assert error is None and blocked is not None + assert "comment" not in blocked and "before" in blocked and "after" in blocked + + unterminated, error = _strip_ts_comments("code /* never closed") + assert unterminated is None and error == "unterminated_block_comment" + + +def test_duplicate_readiness_policy_rejected(tmp_path) -> None: + architecture = load_architecture() + architecture["readiness_policies"].append( + deepcopy(architecture["readiness_policies"][0]) + ) + architecture["readiness_policies"][-1]["operator"] = "any" + repo = build_tmp_repo(tmp_path, architecture=architecture) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.readiness.duplicate_policy" in issue_codes(result) + + +def test_unhashable_evidence_id_fails_closed_without_crashing(tmp_path) -> None: + contract = load_contract() + contract["readiness_binding"]["evidence_bindings"][0]["evidence_id"] = [] + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert result.compared is True + + +def test_source_binding_path_escape_rejected(tmp_path) -> None: + contract = load_contract() + machine_by_id(contract, "review-session")["source_binding"]["file"] = ( + "bim-review-coordinator/../secrets.ts" + ) + repo = build_tmp_repo(tmp_path, contract=contract) + result = check_lifecycle_contracts(repo) + assert result.status == "failed" + assert "lifecycle.source_binding.path_escape" in issue_codes(result) + + +def test_extract_union_literals_behaviors() -> None: + ok, reason = _extract_union_literals( + 'export type S = "a" | "b" | "c";', "S" + ) + assert reason is None and ok == ["a", "b", "c"] + + multiline, reason = _extract_union_literals( + 'export type S =\n | "a"\n | "b";', "S" + ) + assert reason is None and multiline == ["a", "b"] + + missing, reason = _extract_union_literals('export type T = "a";', "S") + assert missing is None and reason == "type_not_found" + + aliased, reason = _extract_union_literals('export type S = Other | "a";', "S") + assert aliased is None and reason == "unsupported_union" + + empty, reason = _extract_union_literals("export type S = ;", "S") + assert empty is None and reason == "no_literals" + + duplicated, reason = _extract_union_literals('export type S = "a" | "a";', "S") + assert duplicated is None and reason == "duplicate_literals" + + hollow, reason = _extract_union_literals('export type S = "" | "a";', "S") + assert hollow is None and reason == "empty_literal" + + # A commented-out stale declaration is not source; the anchored match reads + # only the real one. + commented, reason = _extract_union_literals( + '// export type S = "a";\nexport type S = "a" | "b";', "S" + ) + assert reason is None and commented == ["a", "b"] + + only_comment, reason = _extract_union_literals('// export type S = "a";', "S") + assert only_comment is None and reason == "type_not_found" + + twice, reason = _extract_union_literals( + 'export type S = "a";\nexport type S = "a" | "b";', "S" + ) + assert twice is None and reason == "ambiguous_declaration" + + +# --------------------------------------------------------------------------- # +# Canonical runtime unions still match the pins directly +# --------------------------------------------------------------------------- # + + +def test_runtime_source_unions_match_pinned_states() -> None: + """Reads the real TypeScript files, independently of the checker.""" + + for machine_id, (source_file, type_name) in PINNED_SOURCE_BINDINGS.items(): + text = (ROOT / source_file).read_text(encoding="utf-8") + literals, reason = _extract_union_literals(text, type_name) + assert reason is None, (machine_id, reason) + assert literals is not None + assert set(literals) == set(PINNED_STATES[machine_id]), machine_id + + +# --------------------------------------------------------------------------- # +# Developer entry point +# --------------------------------------------------------------------------- # + + +def test_check_script_passes_on_canonical_repository() -> None: + completed = subprocess.run( + [ + sys.executable, + str(ROOT / "scripts" / "dev" / "check_lifecycle_contracts.py"), + "--repo-root", + str(ROOT), + "--format", + "json", + "--strict", + ], + capture_output=True, + text=True, + check=False, + ) + assert completed.returncode == 0, completed.stdout + completed.stderr + payload = json.loads(completed.stdout) + assert payload["status"] == "passed" + assert payload["cli_status"] == "passed" + assert payload["strict"] is True + assert payload["compared"] is True + assert payload["summary"]["machines"] == len(PINNED_MACHINE_IDS) + + +def test_check_script_strict_fails_on_warning_only_repository(tmp_path) -> None: + """--strict must exit 1 AND render a failed verdict on a warning-only run. + + The library keeps its error-only `status`; the CLI's rendered verdict and + `cli_status` must agree with the process outcome, otherwise a strict CI log + reads PASSED while the step fails. + """ + + contract = load_contract() + machine_by_id(contract, "stage-binding")["evidence"].append( + { + "id": "dangling-evidence", + "description": "Declared but consumed by nothing.", + "source": "nowhere", + } + ) + repo = build_tmp_repo(tmp_path, contract=contract) + # The scratch repo carries data only; the checker code under test comes from + # this checkout via PYTHONPATH. + env = {**os.environ, "PYTHONPATH": str(ROOT)} + completed = subprocess.run( + [ + sys.executable, + str(ROOT / "scripts" / "dev" / "check_lifecycle_contracts.py"), + "--repo-root", + str(repo), + "--format", + "json", + "--strict", + ], + capture_output=True, + text=True, + check=False, + env=env, + ) + assert completed.returncode == 1, completed.stdout + completed.stderr + payload = json.loads(completed.stdout) + assert payload["status"] == "passed" + assert payload["cli_status"] == "failed" + assert payload["strict"] is True + assert any(issue["code"] == "lifecycle.evidence.unused" for issue in payload["issues"]) + + human = subprocess.run( + [ + sys.executable, + str(ROOT / "scripts" / "dev" / "check_lifecycle_contracts.py"), + "--repo-root", + str(repo), + "--strict", + ], + capture_output=True, + text=True, + check=False, + env=env, + ) + assert human.returncode == 1 + assert human.stdout.startswith("Lifecycle contracts: FAILED"), human.stdout + # Without --strict the same run passes, and the rendered verdict says so. + lenient = subprocess.run( + [ + sys.executable, + str(ROOT / "scripts" / "dev" / "check_lifecycle_contracts.py"), + "--repo-root", + str(repo), + ], + capture_output=True, + text=True, + check=False, + env=env, + ) + assert lenient.returncode == 0 + assert lenient.stdout.startswith("Lifecycle contracts: PASSED"), lenient.stdout + + +def test_check_script_output_writes_relative_path_under_repo_root(tmp_path) -> None: + repo = build_tmp_repo(tmp_path) + env = {**os.environ, "PYTHONPATH": str(ROOT)} + completed = subprocess.run( + [ + sys.executable, + str(ROOT / "scripts" / "dev" / "check_lifecycle_contracts.py"), + "--repo-root", + str(repo), + "--format", + "json", + "--strict", + "--output", + "artifacts/lifecycle/report.json", + ], + capture_output=True, + text=True, + check=False, + env=env, + ) + assert completed.returncode == 0, completed.stdout + completed.stderr + assert completed.stdout == "" + report_path = repo / "artifacts" / "lifecycle" / "report.json" + assert report_path.is_file() + payload = json.loads(report_path.read_text(encoding="utf-8")) + assert payload["cli_status"] == "passed" + assert payload["summary"]["machines"] == len(PINNED_MACHINE_IDS)