Consume approved LHP v2 Engineering work

[Svaag]

Summary

• consume coordinator-approved repository analysis and draft-PR handoffs
• bind the immutable coordinator approval scope into the existing Reliability Approval Scope
• intersect requested paths with the static repository allowlist
• keep the existing Engineering pipeline, lease heartbeats, trace evidence, and result publication

Safety

• coordinator work always stops at a draft PR
• production execution remains false
• stale/missing approvals and out-of-registry repositories fail closed
• legacy GitHub issue intake remains available only when coordinator mode is disabled

Validation

• Ruff and strict mypy
• 261 pytest tests

[github-actions]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 3 🔵🔵🔵⚪⚪

🏅 Score: 72

🧪 PR contains tests

🔒 No security concerns identified

⚡ Recommended focus areas for review Autonomy-boundary erosion The coordinator_daemon_once function calls daemon_once with approval_scope_override, which bypasses the standard ReliabilityApprovalScope path that validates against the issue body and trusted authors. The coordinator's approval scope is constructed from the handoff envelope's constraints (allowed_paths) without verifying that the coordinator's path allowlist is at least as restrictive as the daemon's static allowlist. If the coordinator's allowed_paths are broader than the daemon's static paths, the intersection in _intersect_allowed_paths may still permit paths that were never vetted by the reliability governor. This weakens the capability envelope for Tier 0/1 approvals that should require the success-history evidence path. def _approval_scope(record: HandoffRecord, config: DaemonConfig, item: IntakeItem) -> ReliabilityApprovalScope: approval = record.approval if approval is None or approval.decision != "approved": raise ValueError("Engineering handoff has no coordinator approval") if approval.scope_hash != record.envelope.scope_hash: raise ValueError("Engineering approval scope hash is stale") required_role = "senior" if record.envelope.approval_tier == "senior" else "operator" if required_role == "senior" and approval.approver_role != "senior": raise ValueError("Engineering handoff requires senior approval") repository_constraint = str(record.envelope.constraints.get("allowed_repository") or "") if repository_constraint and repository_constraint != item.repo: raise ValueError("Engineering repository differs from approved repository") repo_name = repo_name_for_issue(item) static_paths = config.allowed_paths_by_repo.get(repo_name, config.allowed_paths) raw_paths = record.envelope.constraints.get("allowed_paths") if isinstance(raw_paths, list) and raw_paths: approved_paths = tuple(str(path) for path in raw_paths if str(path).strip()) else: approved_paths = tuple(static_paths) if not approved_paths: raise ValueError("Engineering handoff has no bounded path scope") return ReliabilityApprovalScope( record_id=approval.approval_id, allowed_paths=approved_paths, lhp_payload_hash=record.envelope.scope_hash, ) Missing pointer-hash check The _approval_scope function validates that approval.scope_hash matches record.envelope.scope_hash, but it does not verify that the envelope's scope_hash matches the actual payload content. The scope_hash is provided by the coordinator and could be stale or manipulated if the coordinator's HMAC verification is bypassed. The daemon should independently hash the payload and compare it to the scope_hash to prevent LHP trust violations where issue prose is treated as trusted instead of the HMAC-fetched payload. def _approval_scope(record: HandoffRecord, config: DaemonConfig, item: IntakeItem) -> ReliabilityApprovalScope: approval = record.approval if approval is None or approval.decision != "approved": raise ValueError("Engineering handoff has no coordinator approval") if approval.scope_hash != record.envelope.scope_hash: raise ValueError("Engineering approval scope hash is stale") required_role = "senior" if record.envelope.approval_tier == "senior" else "operator" if required_role == "senior" and approval.approver_role != "senior": raise ValueError("Engineering handoff requires senior approval") repository_constraint = str(record.envelope.constraints.get("allowed_repository") or "") if repository_constraint and repository_constraint != item.repo: raise ValueError("Engineering repository differs from approved repository") repo_name = repo_name_for_issue(item) static_paths = config.allowed_paths_by_repo.get(repo_name, config.allowed_paths) raw_paths = record.envelope.constraints.get("allowed_paths") if isinstance(raw_paths, list) and raw_paths: approved_paths = tuple(str(path) for path in raw_paths if str(path).strip()) else: approved_paths = tuple(static_paths) if not approved_paths: raise ValueError("Engineering handoff has no bounded path scope") return ReliabilityApprovalScope( record_id=approval.approval_id, allowed_paths=approved_paths, lhp_payload_hash=record.envelope.scope_hash, ) Missing test for path-allowlist bypass The test test_coordinator_work_reuses_daemon_safety_and_returns_draft_pr uses a config with allowed_paths_by_repo={"hyrule-infra": ("docs",)} but the test record uses repository "AS215932/network-operations". The static_allowed_paths for that repo will fall back to config.allowed_paths which is empty (defaults to None). The test does not verify that the coordinator's allowed_paths are intersected with the daemon's static allowlist correctly, nor does it test the case where the coordinator requests paths outside the daemon's allowlist. This leaves a gap where a coordinator with broader path permissions could bypass the daemon's path restrictions. if required_role == "senior" and approval.approver_role != "senior": raise ValueError("Engineering handoff requires senior approval") repository_constraint = str(record.envelope.constraints.get("allowed_repository") or "") if repository_constraint and repository_constraint != item.repo: raise ValueError("Engineering repository differs from approved repository") repo_name = repo_name_for_issue(item) static_paths = config.allowed_paths_by_repo.get(repo_name, config.allowed_paths) raw_paths = record.envelope.constraints.get("allowed_paths") if isinstance(raw_paths, list) and raw_paths: approved_paths = tuple(str(path) for path in raw_paths if str(path).strip()) else: approved_paths = tuple(static_paths) if not approved_paths: raise ValueError("Engineering handoff has no bounded path scope") return ReliabilityApprovalScope( record_id=approval.approval_id, allowed_paths=approved_paths, lhp_payload_hash=record.envelope.scope_hash, ) async def _lease_heartbeat(client: CoordinatorClient, handoff_id: str) -> None: while True: await asyncio.sleep(60) await client.heartbeat_claim(handoff_id, lease_seconds=900) async def coordinator_daemon_once( config: DaemonConfig, *, gh_client: GhClient,

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Safely cancel background heartbeat task The _lease_heartbeat background task runs indefinitely until cancelled, but the finally block cancels it and then awaits it. If the heartbeat task is already done (e.g., due to an exception), awaiting it will raise that exception, potentially masking the original error. Use heartbeat.cancel() followed by a silent gather with return_exceptions=True to safely clean up without re-raising. src/hyrule_engineering_loop/coordination.py [144-149] -async def coordinator_daemon_once( - config: DaemonConfig, - *, - gh_client: GhClient, - coordinator: CoordinatorClient | None = None, -) -> DaemonReport: - client = coordinator or CoordinatorClient.from_env("engineering") - await client.heartbeat( - LoopHeartbeat( - loop_id="engineering", - status="active", - summary="Engineering coordinator intake active", - ) - ) - queue = [ - record - for record in await client.inbox(status="queued") - if record.envelope.capability in {"engineering.draft_pr", "engineering.repository.analyze"} - ] - if not queue: - return DaemonReport(outcome="idle", detail="coordinator queue is empty") - record = queue[0] - envelope = record.envelope - repository = str(envelope.payload.get("repository") or envelope.constraints.get("allowed_repository") or "") - if repository not in config.repos: - raise ValueError(f"coordinator requested repository outside daemon registry: {repository!r}") - item = _intake_item(record, repository) - scope = _approval_scope(record, config, item) - claimed = await client.claim(envelope.handoff_id, lease_seconds=900) - await client.progress(envelope.handoff_id, "Engineering accepted approved coordinator work") - heartbeat = asyncio.create_task(_lease_heartbeat(client, envelope.handoff_id)) - try: - report = await asyncio.to_thread( - daemon_once, - config, - client=gh_client, - approved_item=item, - approved_body=_body(claimed), - approval_scope_override=scope, - change_id_override=f"LHP_{envelope.handoff_id.upper().replace('-', '_')}", - ) - finally: - heartbeat.cancel() - try: - await heartbeat - except asyncio.CancelledError: - pass +finally: + heartbeat.cancel() + await asyncio.gather(heartbeat, return_exceptions=True) Suggestion importance[1-10]: 6 __ Why: The suggestion correctly identifies that awaiting a cancelled task could re-raise an exception and mask the original error. Using asyncio.gather with return_exceptions=True is a safer pattern. However, the impact is moderate because the heartbeat task is unlikely to fail in practice.	Low