diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index bd7caac5f..758d23e4d 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -7,6 +7,10 @@ See the CHANGELOG rule in [AGENTS.md](../AGENTS.md). - QA Telegram probes, button presses and the `/start` preflight return once the bot's response stays unchanged for 3 s; the wait stays the upper bound and silence still runs to it. +- A QA temporary-access grant or revoke calls the target's users-core capability with the stored + `USERS_GRANT_CAPABILITY` instead of redeploying the product, so QA access no longer costs a deploy. +- A temporary-access operation takes its Run's dispatch claim before any external write, and every revoke + re-derives the earlier-workflow stop from all the grant's Runs, so withdrawal or retry cannot skip either. ## 2026-10-17 diff --git a/docs/contracts/generated-service-grants.md b/docs/contracts/generated-service-grants.md index 90b3138c8..f49280ca5 100644 --- a/docs/contracts/generated-service-grants.md +++ b/docs/contracts/generated-service-grants.md @@ -152,7 +152,8 @@ intents, even when it carries a deployment SHA. Temporary QA access uses the same generated-service capability but a distinct durable record. It binds the central QA Telegram identity, application, base URL, and SHA before dispatch. Grant and revoke each require the matching active or -inactive `/users/access` readback; the capability remains in deploy-local memory. +inactive `/users/access` readback without a redeploy; the stored capability is +decrypted only in the deploy consumer's memory. The reconciler retries only the record's immutable target. Missing or stale grant and revoke operation Runs consume their separately recorded, bounded attempt budgets. A revoke that exceeds its attempt or unrevoked-time bound receives one diff --git a/docs/contracts/lifecycle-invariants.md b/docs/contracts/lifecycle-invariants.md index 4757a9178..004d6ff79 100644 --- a/docs/contracts/lifecycle-invariants.md +++ b/docs/contracts/lifecycle-invariants.md @@ -201,26 +201,56 @@ older active deploys instead of allowing an earlier run to restore the value. Canonical contracts: `dto/temporary_access.py` and `dto/qa_ssh_grant.py`. Persist the immutable QA identity and exact deployed-service target before the -capability operation is dispatched. The post-health deploy worker resolves the -generated capability only in `secret_values`, then proves grant or revoke with -the matching access readback. A grant or revoke only ever acts on an existing -deployment of the grant's recorded `target_application_id`: the deploy consumer -reads that application's allocations and never creates any -(`existing_application_allocations`), and never asks the repository for "its" -application, since a repository can hold one per server. When the target exists -and holds no allocations it is not deployed: a revoke completes `SUCCESS` with -no precheck, SSH or DevOps run, because the access went with the deployment, and -the reconciler closes the grant as for any proved revoke; a grant fails -`owner_access_proof_failed` through the ordinary grant retry. A grant, target or -allocation set that cannot be read fails either operation closed with -`owner_access_proof_failed`, never as that proof. Both target columns are NOT NULL: every record has +capability operation is dispatched. The operation is a DEPLOY Run whose metadata +names the grant and `grant` or `revoke`; the deploy consumer validates it after +the deploy-lock claim and handles it before any product deploy preparation: +malformed or conflicting access metadata (a grant intent, a lifecycle action) +fails `owner_access_proof_failed` and never falls through to a product deploy. +It deploys nothing: no checkout, environment contract, secret generation or +persistence, GitHub secret or workflow dispatch, SSH, smoke, settings seeding or +`service_deployments` row. It reads only the grant's recorded +`target_application_id` and that application's existing allocations +(`existing_application_allocations`), never the repository's "its" application, +since a repository can hold one per server. When the target exists and holds no +allocations it is not deployed: a revoke completes `SUCCESS`, because the access +went with the deployment, and a grant fails `owner_access_proof_failed` through +the ordinary grant retry. A grant, target or allocation set that cannot be read +fails either operation closed with `owner_access_proof_failed`, never as that +proof. Either marker key in the Run metadata, even with a null value, makes the +Run such an operation. Before its first external write the operation takes its +Run's dispatch claim (`claim_dispatch`), the same locked admission a deploy takes +before GitHub, and renews it before each later external write; a withdrawal or a +lapsed lease that wins any renewal ends the Run `cancelled` with no further +external write, so a withdrawn operation never changes access. A revoke owes a +stop of earlier grant workflows when its message carries `fence_active_deploys`, +when any grant operation Run of the record crossed the dispatch boundary without +completing `SUCCESS`, or when the record's current grant Run is missing from that +history (`shared.temporary_access_cleanup.grant_cleanup_owes_workflow_stop`). The +history is the project's complete deploy Run listing filtered by the Run +metadata's grant id and operation, so replacing the record's grant Run does not +hide a predecessor; an unreadable listing fails the revoke. The scheduler derives +the flag this way on every revoke publication and the consumer re-derives it, so +no retry, replacement or redispatch loses it. The stop proves every +unfinished `deploy.yml` run of the repository stopped (`fence_active_deploys`), +renewing admission and checking the deploy lock immediately before each +cancellation request; it fails `owner_access_proof_failed` when a run cannot be +proved stopped, and a lost lock ends the Run `deploy_lock_lost`. Only then may a +revoke of an undeployed target record `SUCCESS`. For a deployed target the +consumer decrypts the project's stored `USERS_GRANT_CAPABILITY` in memory, +renews admission, re-reads the record and requires its project, target, base +URL, identity, head SHA, in-flight status and own Run id, checks the lease and +then the deploy lock with nothing awaited before the product call, and calls only +the target base URL's `/users/grant` or `/users/revoke` and `/users/access` +readback. A grant needs the identity active, a revoke inactive with no proof +failure; an absent or undecryptable capability, a refusal or an unverified +readback fails `owner_access_proof_failed` with a bounded reason and never mints +a capability. The terminal Run outcome is written last. +Both target columns are NOT NULL: every record has a target. The target lock and partial unique index scope contention to `(project_id, target_application_id)`. Cancelled deploy-lock or fence operations are redispatched against their stored target without consuming a grant or revoke proof budget; failed, missing, and -stale operations remain independently bounded. Immediately before either remote -operation, the executor re-reads the record and requires its own Run id and -matching in-flight state. Recovery changes that durable operation authority +stale operations remain independently bounded. Recovery changes that durable operation authority before withdrawing the predecessor and dispatching fenced cleanup, so a delayed grant cannot restore access after revoke proof. Cancelled revoke redispatches retain their attempt budget only before the absolute unrevoked deadline. diff --git a/services/langgraph/src/consumers/deploy.py b/services/langgraph/src/consumers/deploy.py index 4a4c4bcbf..15a699d6b 100644 --- a/services/langgraph/src/consumers/deploy.py +++ b/services/langgraph/src/consumers/deploy.py @@ -9,19 +9,22 @@ from __future__ import annotations from dataclasses import dataclass +from datetime import UTC, datetime from typing import Any import structlog from shared.allocation_disposition import attempt_disposition, may_terminate_story -from shared.clients.github import WorkflowCancellationUnprovenError +from shared.clients.github import GitHubAppClient, WorkflowCancellationUnprovenError from shared.contracts.dto.application import ( DEFAULT_APPLICATION_RESERVED_RAM_MB, + ApplicationDTO, ApplicationStatus, ) from shared.contracts.dto.project import ProjectDTO from shared.contracts.dto.run import RunStatus from shared.contracts.dto.run_result import DeployRunResult, DeploySkipReason, MissingUserSecret +from shared.contracts.dto.temporary_access import TemporaryAccessGrantDTO from shared.contracts.dto.users_grant import USERS_GRANT_INTENT_KEY from shared.contracts.env_overrides import ( EMPTY_OVERRIDES_DIGEST, @@ -36,6 +39,7 @@ from shared.contracts.service_ports import DEPLOY_INFRA_PORT_SERVICES from shared.queues import DEPLOY_QUEUE from shared.redis import RedisStreamClient +from shared.temporary_access_cleanup import grant_cleanup_owes_workflow_stop from ..allocations import AllocationError from ..clients.api import api_client @@ -43,6 +47,12 @@ from ..deploy_fence import DeployFence, DeployFenceLost, DeployWrite from ..runtime_identity import project_runtime_slug from ..subgraphs.devops import create_devops_subgraph +from ..subgraphs.devops.deployer import ( + DeployDispatchWithdrawnError, + DeployFenceUnprovenError, + fence_active_deploys, + require_live_lease, +) from ._base import start_worker, validate_queued_message from ._events import publish_callback_event from ._live_work import live_work_cancel_key, live_work_settled, live_work_unsettled @@ -54,6 +64,11 @@ _handle_smoke_failure, user_secret_classification, ) +from .temporary_access_operation import ( + admit_temporary_access, + apply_temporary_access_operation, + confirm_current_operation, +) logger = structlog.get_logger(__name__) @@ -71,30 +86,14 @@ class AccessTargetUnresolvedError(Exception): """A temporary-access operation whose recorded target could not be read. Only a target application that exists and holds no allocations proves there - is no deployment. Anything short of reading it — no grant, no target, a row - that is gone, an API error — fails the operation closed, never as that proof. + is no deployment. A target row that is gone or an API error fails the + operation closed, never as that proof. """ -async def _allocate_resources( - project_id: str, - project: ProjectDTO, - *, - access: DeployAccessContext | None = None, -) -> dict | str: +async def _allocate_resources(project_id: str, project: ProjectDTO) -> dict | str: """Get or create allocations. Returns dict of resources or error string. - This is the deploy path's only allocation call, and the one place that - decides whether it may create. A temporary-access grant or revoke redeploys - what its grant's recorded target application already runs, so for one only - that application's allocations are read: an empty result means the target - has no deployment, and nothing is allocated for it. The repository's first - application is never asked instead — a repository can have one per server, - and another row's empty allocations say nothing about the target's access. - On 2026-09-25 a revoke that arrived after its target was undeployed - re-allocated all four ports, failed its precheck, and left them owned by an - application nobody runs. - An `AllocationError` is deliberately *not* caught here. It is the one failure on this path that is about the platform rather than the project, and it carries the classification the scheduler needs; flattening it into this @@ -108,8 +107,6 @@ async def _allocate_resources( dict.fromkeys([*config.get("modules", ["backend"]), *DEPLOY_INFRA_PORT_SERVICES]) ) min_ram_mb = config.get("estimated_ram_mb", DEFAULT_APPLICATION_RESERVED_RAM_MB) - if access is not None and access.temporary_access_operation is not None: - return await _access_target_allocations(access, min_ram_mb) # Get repo_id from primary repository primary_repo = await api_client.get_primary_repository(project_id) @@ -127,21 +124,22 @@ async def _allocate_resources( ) -async def _access_target_allocations(access: DeployAccessContext, min_ram_mb: int) -> dict: - """Read the grant's recorded target application's allocations, or fail closed. +async def _access_target( + grant: TemporaryAccessGrantDTO, min_ram_mb: int +) -> tuple[ApplicationDTO, dict]: + """Read the grant's recorded target application and its allocations, or fail closed. - An admission refusal of the target's host stays the typed `AllocationError` - every placement raises; every other way of not reading the target raises + Only that application is read and nothing is allocated: a repository can hold + one application per server, and another row's allocations say nothing about + the target's access. An admission refusal of the target's host stays the + typed `AllocationError`; every other way of not reading the target raises `AccessTargetUnresolvedError`. """ from ..allocations import existing_application_allocations - grant = access.temporary_access_grant - if grant is None: - raise AccessTargetUnresolvedError("temporary access operation has no grant") try: application = await api_client.get_application(grant.target_application_id) - return await existing_application_allocations( + allocations = await existing_application_allocations( application_id=application.id, server_handle=application.server_handle, min_ram_mb=min_ram_mb, @@ -153,6 +151,7 @@ async def _access_target_allocations(access: DeployAccessContext, min_ram_mb: in f"target application {grant.target_application_id} of temporary access grant " f"{grant.id} could not be read: {type(error).__name__}" ) from error + return application, allocations async def _record_infrastructure_wait( @@ -525,9 +524,9 @@ async def _resolve_deploy_access( metadata = getattr(run, "run_metadata", None) or {} stored_temporary_access_grant = metadata.get("temporary_access_grant_id") stored_temporary_access_operation = metadata.get("temporary_access_operation") - # Either key marks a capability operation, which must never fall through to - # a product deploy that places the application: it needs both. - if stored_temporary_access_grant is not None or stored_temporary_access_operation is not None: + # Either key marks a capability operation, even with a null value, and one + # must never fall through to a product deploy that places the application. + if "temporary_access_grant_id" in metadata or "temporary_access_operation" in metadata: temporary_access_operation = stored_temporary_access_operation if not isinstance(stored_temporary_access_grant, str) or not isinstance( temporary_access_operation, str @@ -561,6 +560,8 @@ async def _resolve_deploy_access( temporary_access_grant.project_id != msg.project_id or temporary_access_grant.head_sha != msg.head_sha or temporary_access_operation not in {"grant", "revoke"} + or grant_intent is not None + or msg.action in LIFECYCLE_ACTIONS ): return await _deploy_failure_terminal( msg, @@ -640,22 +641,12 @@ async def _allocate_deploy_resources( fence: DeployFence, ) -> tuple[dict, dict[str, str]] | DeployTerminal: """Resolve placement and effective environment for a normal deploy.""" - operation = base.access.temporary_access_operation try: - alloc_result = await _allocate_resources(msg.project_id, base.project, access=base.access) + alloc_result = await _allocate_resources(msg.project_id, base.project) except AllocationError as error: return DeployTerminal( await _record_infrastructure_wait(msg.task_id, msg.project_id, error, fence) ) - except AccessTargetUnresolvedError as error: - logger.warning("temporary_access_target_unresolved", task_id=msg.task_id, error=str(error)) - return await _deploy_failure_terminal( - msg, - redis, - fence, - str(error), - deploy_outcome=DeployOutcome.OWNER_ACCESS_PROOF_FAILED, - ) if isinstance(alloc_result, str): await fence.ensure_held(DeployWrite.RUN_STATE) @@ -671,9 +662,6 @@ async def _allocate_deploy_resources( ) return DeployTerminal(live_work_unsettled({"status": "failed", "error": alloc_result})) - if operation is not None and not alloc_result: - return await _access_operation_without_deployment(base, msg, redis, fence) - try: env_overrides = _effective_env_overrides(base.project, msg.env_overrides) except ValueError as error: @@ -687,19 +675,173 @@ async def _allocate_deploy_resources( return alloc_result, env_overrides +async def _run_temporary_access_operation( + base: DeployBaseContext, + msg: DeployMessage, + redis: RedisStreamClient, + fence: DeployFence, +) -> dict: + """Grant or revoke the QA identity through its target's users-core capability. + + Nothing is deployed. In order: read the recorded target's existing allocations + and the cleanup obligation, take the Run's atomic dispatch admission, prove any + earlier grant workflow stopped, confirm the current operation, prove the exact + identity's readback, and record the terminal outcome last. A withdrawal that + wins the admission at any of its renewals ends the Run cancelled with no + further external write. + """ + grant = base.access.temporary_access_grant + operation = base.access.temporary_access_operation + min_ram_mb = (base.project.config or {}).get( + "estimated_ram_mb", DEFAULT_APPLICATION_RESERVED_RAM_MB + ) + try: + application, allocations = await _access_target(grant, min_ram_mb) + except AllocationError as error: + return await _record_infrastructure_wait(msg.task_id, msg.project_id, error, fence) + except AccessTargetUnresolvedError as error: + logger.warning("temporary_access_target_unresolved", task_id=msg.task_id, error=str(error)) + return await _access_failure_response(msg, redis, fence, str(error)) + must_fence = msg.fence_active_deploys or ( + operation == "revoke" and await grant_cleanup_owes_workflow_stop(api_client, grant) + ) + + try: + await admit_temporary_access(msg.task_id, fence) + if must_fence: + refusal = await _fence_earlier_deploy_workflows(msg, fence) + if refusal is not None: + return await _access_failure_response(msg, redis, fence, refusal) + if not allocations: + return await _access_operation_without_deployment( + base, msg, redis, fence, application.id + ) + failure = await apply_temporary_access_operation( + task_id=msg.task_id, + project_id=msg.project_id, + project=base.project, + application_id=application.id, + grant=grant, + operation=operation, + fence=fence, + ) + except DeployDispatchWithdrawnError as withdrawn: + return await _record_access_withdrawn(msg, fence, withdrawn) + if failure is not None: + logger.warning( + "temporary_access_proof_failed", + task_id=msg.task_id, + grant_id=grant.id, + operation=operation, + reason=failure, + ) + return await _access_failure_response( + msg, redis, fence, f"temporary access {operation} was not proved: {failure}" + ) + + logger.info( + "temporary_access_operation_proved", + task_id=msg.task_id, + grant_id=grant.id, + target_application_id=application.id, + operation=operation, + ) + await fence.ensure_held(DeployWrite.RUN_STATE) + await api_client.patch( + f"runs/{msg.task_id}", + json={ + "status": RunStatus.COMPLETED.value, + "result": DeployRunResult( + deploy_outcome=DeployOutcome.SUCCESS, + application_id=application.id, + action=msg.action, + ).model_dump(mode="json"), + }, + ) + await publish_callback_event( + redis, + msg.callback_stream, + "completed", + msg.task_id, + f"Temporary access {operation} proved", + telegram_chat_id=msg.telegram_chat_id, + project_id=msg.project_id, + ) + return live_work_settled({"status": "success", "operation": operation}) + + +async def _record_access_withdrawn( + msg: DeployMessage, fence: DeployFence, withdrawn: DeployDispatchWithdrawnError +) -> dict: + """End an operation whose Run was withdrawn before its effect, as a cancelled deploy.""" + logger.info("temporary_access_operation_withdrawn", task_id=msg.task_id, error=str(withdrawn)) + await fence.ensure_held(DeployWrite.RUN_STATE) + await api_client.patch( + f"runs/{msg.task_id}", + json={ + "status": RunStatus.CANCELLED.value, + "error_message": "Temporary access operation was withdrawn before its effect", + "result": DeployRunResult( + deploy_outcome=DeployOutcome.CANCELLED, + action=msg.action, + ).model_dump(mode="json"), + }, + ) + return live_work_unsettled({"status": "cancelled"}) + + +async def _access_failure_response( + msg: DeployMessage, redis: RedisStreamClient, fence: DeployFence, error_msg: str +) -> dict: + """Record a retryable temporary-access failure the reconciler retries.""" + return ( + await _deploy_failure_terminal( + msg, + redis, + fence, + error_msg, + deploy_outcome=DeployOutcome.OWNER_ACCESS_PROOF_FAILED, + ) + ).response + + +async def _fence_earlier_deploy_workflows(msg: DeployMessage, fence: DeployFence) -> str | None: + """Stop every earlier deploy workflow of the project's repository, or say why not. + + Each cancellation renews this Run's admission first, so a withdrawal or a + lapsed lease stops the next cancellation as well as the product effect. + """ + primary_repo = await api_client.get_primary_repository(msg.project_id) + if primary_repo is None: + return f"project {msg.project_id} has no repository whose deploy runs can be fenced" + owner, repo = primary_repo.git_url.rstrip("/").removesuffix(".git").split("/")[-2:] + + async def admit() -> None: + await admit_temporary_access(msg.task_id, fence) + + try: + await fence_active_deploys( + GitHubAppClient(), owner, repo, msg.project_id, fence, admit=admit + ) + except DeployFenceUnprovenError as error: + logger.error("temporary_access_workflow_fence_unproven", task_id=msg.task_id) + return str(error) + return None + + async def _access_operation_without_deployment( base: DeployBaseContext, msg: DeployMessage, redis: RedisStreamClient, fence: DeployFence, -) -> DeployTerminal: + application_id: int, +) -> dict: """Settle a temporary-access operation whose target is no longer deployed. - No precheck, SSH or DevOps run follows: there is nothing to redeploy. Access - to a deployment that no longer exists went away with it, so a revoke records - exactly what a proved revoke records and the supervisor settles the grant as - revoked. A grant has nothing to grant access to and fails through the grant - failure path, which the supervisor already retries and then cleans up. + Access to a deployment that no longer exists went away with it, so a revoke + that still holds its admission and current operation records exactly what a + proved revoke records. A grant has nothing to grant access to and fails + through the grant failure path the reconciler retries. """ operation = base.access.temporary_access_operation grant = base.access.temporary_access_grant @@ -712,13 +854,25 @@ async def _access_operation_without_deployment( operation=operation, ) if operation != "revoke": - return await _deploy_failure_terminal( + return await _access_failure_response( msg, redis, fence, f"application {grant.target_application_id} has no deployment to grant access to", - deploy_outcome=DeployOutcome.OWNER_ACCESS_PROOF_FAILED, ) + claim, failure = await confirm_current_operation( + task_id=msg.task_id, + project_id=msg.project_id, + application_id=application_id, + grant=grant, + operation=operation, + fence=fence, + ) + if failure is not None: + return await _access_failure_response( + msg, redis, fence, f"temporary access {operation} was not proved: {failure}" + ) + require_live_lease(claim, datetime.now(UTC)) await fence.ensure_held(DeployWrite.RUN_STATE) await api_client.patch( f"runs/{msg.task_id}", @@ -739,7 +893,7 @@ async def _access_operation_without_deployment( telegram_chat_id=msg.telegram_chat_id, project_id=msg.project_id, ) - return DeployTerminal(live_work_settled({"status": "success", "reason": "not_deployed"})) + return live_work_settled({"status": "success", "reason": "not_deployed"}) async def _maybe_skip_redundant_deploy( @@ -752,11 +906,7 @@ async def _maybe_skip_redundant_deploy( ) -> DeployTerminal | None: """Complete a deploy immediately when its exact commit/environment is already live.""" access = base.access - if ( - access.grant_intent is not None - or access.temporary_access_grant is not None - or msg.fence_active_deploys - ): + if access.grant_intent is not None or msg.fence_active_deploys: return None application_id = await _already_deployed_application( @@ -932,8 +1082,6 @@ async def _route_deploy_result( msg=msg, application_id=result.get("application_id"), grant_intent=access.grant_intent, - temporary_access_grant=access.temporary_access_grant, - temporary_access_operation=access.temporary_access_operation, fence=fence, ) @@ -1136,6 +1284,9 @@ async def process_deploy_job(job_data: dict, redis: RedisStreamClient) -> dict: if isinstance(base, DeployTerminal): return base.response + if base.access.temporary_access_operation is not None: + return await _run_temporary_access_operation(base, msg, redis, fence) + prepared = await _prepare_deploy(base, msg, job_data, redis, fence) if isinstance(prepared, DeployTerminal): return prepared.response diff --git a/services/langgraph/src/consumers/deploy_result_handler.py b/services/langgraph/src/consumers/deploy_result_handler.py index 77d9daafa..f47f60459 100644 --- a/services/langgraph/src/consumers/deploy_result_handler.py +++ b/services/langgraph/src/consumers/deploy_result_handler.py @@ -20,7 +20,6 @@ settings_seed_failure_detail, settings_seed_failure_kinds, ) -from shared.contracts.dto.temporary_access import TemporaryAccessGrantDTO, TemporaryAccessStatus from shared.contracts.dto.users_grant import ( GrantIntent, GrantIntentKind, @@ -136,8 +135,6 @@ async def _handle_deploy_success( # noqa: PLR0913 fence: DeployFence, application_id: int | None = None, grant_intent: GrantIntent | None = None, - temporary_access_grant: TemporaryAccessGrantDTO | None = None, - temporary_access_operation: str | None = None, ) -> dict: """Handle successful deploy — update run, no story transitions. @@ -168,70 +165,29 @@ async def _handle_deploy_success( # noqa: PLR0913 fence=fence, ) - if temporary_access_grant is not None and temporary_access_operation is not None: - failure = await _apply_temporary_access_operation( + settings_seed = await _seed_initial_settings( + task_id=task_id, + project_id=project_id, + story_id=story_id, + deployed_url=result["deployed_url"], + secret_values=result.get("secret_values", {}), + fence=fence, + ) + failures = settings_seed_failure_kinds(settings_seed) + if failures: + return await _handle_settings_seed_failure( + result=result, task_id=task_id, project_id=project_id, + callback_stream=callback_stream, + telegram_chat_id=telegram_chat_id, + redis=redis, + failures=failures, application_id=application_id, - secret_values=result.get("secret_values", {}), - grant=temporary_access_grant, - operation=temporary_access_operation, + settings_seed=settings_seed, + deploy_fix_attempt=msg.deploy_fix_attempt, fence=fence, ) - if failure is not None: - return await _handle_owner_access_failure( - result=result, - task_id=task_id, - project_id=project_id, - callback_stream=callback_stream, - telegram_chat_id=telegram_chat_id, - redis=redis, - reason=failure, - application_id=application_id, - deploy_fix_attempt=msg.deploy_fix_attempt, - fence=fence, - ) - - if temporary_access_operation is not None: - # A capability operation redeploys the artifact the target is already - # running for one reason: to write or take away the QA identity's access. - # Its proof is the readback above. Seeding the project's confirmed - # settings here would be the work of a story deploy done against somebody - # else's artifact — and on 2026-09-19 it was: a revoke that had already - # proved the access inactive was recorded `settings_seed_failed` because - # the *next* story's confirmed brief declares a key the still-deployed - # commit does not, so the grant never reached REVOKED and held the target - # against that story's own QA handoff. - settings_seed: list[SettingSeedOutcome] = [] - logger.info( - "deploy_settings_seed_not_a_product_deploy", - task_id=task_id, - temporary_access_operation=temporary_access_operation, - ) - else: - settings_seed = await _seed_initial_settings( - task_id=task_id, - project_id=project_id, - story_id=story_id, - deployed_url=result["deployed_url"], - secret_values=result.get("secret_values", {}), - fence=fence, - ) - failures = settings_seed_failure_kinds(settings_seed) - if failures: - return await _handle_settings_seed_failure( - result=result, - task_id=task_id, - project_id=project_id, - callback_stream=callback_stream, - telegram_chat_id=telegram_chat_id, - redis=redis, - failures=failures, - application_id=application_id, - settings_seed=settings_seed, - deploy_fix_attempt=msg.deploy_fix_attempt, - fence=fence, - ) logger.info( "deploy_job_success", @@ -343,71 +299,6 @@ async def _apply_grant_intent( # noqa: PLR0913 return None -async def _apply_temporary_access_operation( - *, - task_id: str, - project_id: str, - application_id: int | None, - secret_values: dict, - grant: TemporaryAccessGrantDTO, - operation: str, - fence: DeployFence, -) -> str | None: - """Use this deploy's capability only while its durable operation is current. - - The remote grant or revoke checks the deploy lock after the grant is re-read, - immediately before it is made. - """ - if ( - grant.project_id != project_id - or application_id != grant.target_application_id - or operation not in {"grant", "revoke"} - ): - return "temporary_access_target_mismatch" - # The grant was read before the deploy began. Recovery can replace it with - # cleanup while that deploy is still queued or running, so the durable - # operation transition is re-read immediately before the remote effect. - # A delayed delivery must never re-grant after a proved revoke. - current = await api_client.get_temporary_access_grant(grant.id) - expected_run_id = current.grant_run_id if operation == "grant" else current.revoke_run_id - expected_status = ( - TemporaryAccessStatus.GRANTING if operation == "grant" else TemporaryAccessStatus.REVOKING - ) - if current.status is not expected_status or expected_run_id != task_id: - logger.info( - "temporary_access_operation_superseded", - grant_id=grant.id, - task_id=task_id, - operation=operation, - status=current.status.value, - ) - return "temporary_access_operation_superseded" - if ( - current.project_id != project_id - or application_id != current.target_application_id - or current.head_sha != grant.head_sha - ): - return "temporary_access_target_mismatch" - capability = secret_values.get(_USERS_GRANT_CAPABILITY) - if not isinstance(capability, str) or not capability: - return "capability_unavailable" - client = GeneratedServiceGrantClient(current.target_base_url) - await fence.ensure_held(DeployWrite.PRODUCT_ACCESS) - if operation == "grant": - proof = await client.grant_and_resolve( - channel=current.channel, external_id=current.external_id, capability=capability - ) - if proof.active: - return None - else: - proof = await client.revoke_and_resolve( - channel=current.channel, external_id=current.external_id, capability=capability - ) - if not proof.active and proof.failure is None: - return None - return proof.failure.value if proof.failure is not None else "unverified" - - async def _seed_initial_settings( *, task_id: str, @@ -433,9 +324,7 @@ async def _seed_initial_settings( none — the owner-grant deploy of a fresh project is the only *product* deploy such a product gets — reads the project's latest confirmed brief carrying settings, so the product never reaches QA unseeded. Either way one - event says which brief was found, or that there was nothing to seed. A - temporary-access capability operation also names no story and is not a - product deploy at all; its caller keeps it out of here. + event says which brief was found, or that there was nothing to seed. """ if story_id: route = "story" diff --git a/services/langgraph/src/consumers/temporary_access_operation.py b/services/langgraph/src/consumers/temporary_access_operation.py new file mode 100644 index 000000000..07a0f3f3b --- /dev/null +++ b/services/langgraph/src/consumers/temporary_access_operation.py @@ -0,0 +1,134 @@ +"""The one proof of a temporary QA access grant or revoke on its recorded target.""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from cryptography.fernet import InvalidToken +import structlog + +from shared.contracts.dto.deploy_dispatch import DeployDispatchClaim +from shared.contracts.dto.project import ProjectDTO +from shared.contracts.dto.temporary_access import TemporaryAccessGrantDTO, TemporaryAccessStatus +from shared.crypto import decrypt_dict + +from ..clients.api import api_client +from ..clients.users_grant import GeneratedServiceGrantClient +from ..deploy_fence import DeployFence, DeployWrite +from ..subgraphs.devops.deployer import claim_dispatch, require_live_lease +from ._qa_redaction import USERS_GRANT_CAPABILITY + +logger = structlog.get_logger(__name__) + + +def _stored_capability(project: ProjectDTO) -> tuple[str | None, str | None]: + """The project's stored grant capability decrypted in memory, or a bounded failure.""" + encrypted = ((project.config or {}).get("secrets") or {}).get(USERS_GRANT_CAPABILITY) + if not isinstance(encrypted, str) or not encrypted: + return None, "capability_unavailable" + try: + capability = decrypt_dict({USERS_GRANT_CAPABILITY: encrypted})[USERS_GRANT_CAPABILITY] + except InvalidToken: + return None, "capability_unreadable" + if not capability: + return None, "capability_unavailable" + return capability, None + + +async def admit_temporary_access(task_id: str, fence: DeployFence) -> DeployDispatchClaim: + """Take or renew this operation Run's admission across the external boundary. + + The dispatch claim is decided against the locked Run row, so a withdrawal that + landed first refuses it (`DeployDispatchWithdrawnError`), as does a lapsed lease. + """ + claim = await claim_dispatch(task_id, fence) + require_live_lease(claim, datetime.now(UTC)) + return claim + + +async def confirm_current_operation( # noqa: PLR0913 + *, + task_id: str, + project_id: str, + application_id: int, + grant: TemporaryAccessGrantDTO, + operation: str, + fence: DeployFence, +) -> tuple[DeployDispatchClaim, str | None]: + """Renew admission, then require the re-read record to still name this operation. + + Returns the renewed claim and, when the record moved on, a bounded reason, so + a delayed delivery never re-grants after cleanup. + """ + claim = await admit_temporary_access(task_id, fence) + current = await api_client.get_temporary_access_grant(grant.id) + expected_run_id = current.grant_run_id if operation == "grant" else current.revoke_run_id + expected_status = ( + TemporaryAccessStatus.GRANTING if operation == "grant" else TemporaryAccessStatus.REVOKING + ) + if current.status is not expected_status or expected_run_id != task_id: + logger.info( + "temporary_access_operation_superseded", + grant_id=grant.id, + task_id=task_id, + operation=operation, + status=current.status.value, + ) + return claim, "temporary_access_operation_superseded" + if ( + current.project_id != project_id + or current.target_application_id != application_id + or current.target_application_id != grant.target_application_id + or current.target_base_url != grant.target_base_url + or current.head_sha != grant.head_sha + or (current.channel, current.external_id) != (grant.channel, grant.external_id) + ): + return claim, "temporary_access_target_mismatch" + return claim, None + + +async def apply_temporary_access_operation( # noqa: PLR0913 + *, + task_id: str, + project_id: str, + project: ProjectDTO, + application_id: int, + grant: TemporaryAccessGrantDTO, + operation: str, + fence: DeployFence, +) -> str | None: + """Grant or revoke the record's identity on its exact target, or name why not. + + Admission, the current record and the lease are confirmed before the deploy + lock is checked with nothing awaited between it and the product call. + Returned reasons are bounded and never carry the capability. + """ + capability, failure = _stored_capability(project) + if failure is not None: + return failure + claim, failure = await confirm_current_operation( + task_id=task_id, + project_id=project_id, + application_id=application_id, + grant=grant, + operation=operation, + fence=fence, + ) + if failure is not None: + return failure + client = GeneratedServiceGrantClient(grant.target_base_url) + require_live_lease(claim, datetime.now(UTC)) + await fence.ensure_held(DeployWrite.PRODUCT_ACCESS) + if operation == "grant": + proof = await client.grant_and_resolve( + channel=grant.channel, external_id=grant.external_id, capability=capability + ) + if proof.active: + return None + else: + proof = await client.revoke_and_resolve( + channel=grant.channel, external_id=grant.external_id, capability=capability + ) + if not proof.active and proof.failure is None: + return None + return proof.failure.value if proof.failure is not None else "unverified" diff --git a/services/langgraph/src/subgraphs/devops/deployer.py b/services/langgraph/src/subgraphs/devops/deployer.py index 66a86b076..53a23dc7d 100644 --- a/services/langgraph/src/subgraphs/devops/deployer.py +++ b/services/langgraph/src/subgraphs/devops/deployer.py @@ -11,6 +11,7 @@ from shared.clients.github import ( GitHubAppClient, RegistrySecretsNotRefreshedError, + RequestGuard, deploy_pin_tag, registry_repository_secrets, ) @@ -75,7 +76,7 @@ class DeployDispatchWithdrawnError(RuntimeError): """ -def _require_live_lease(claim: DeployDispatchClaim | None, moment: datetime) -> None: +def require_live_lease(claim: DeployDispatchClaim | None, moment: datetime) -> None: """Refuse to dispatch once the claim's deadline has passed. Holding the boundary is a lease, not a possession, and this is the promise @@ -105,6 +106,89 @@ def _require_live_lease(claim: DeployDispatchClaim | None, moment: datetime) -> ) +async def claim_dispatch( + run_id: str | None, deploy_fence: DeployFence +) -> DeployDispatchClaim | None: + """Take the dispatch boundary, or refuse to cross it. + + Called immediately before every call that starts work outside the system, + on GitHub Actions or a temporary-access effect. The claim is itself a + write, so it checks the project deploy lock first; the effect that follows + checks again after the claim returns. + A plain read of the run status cannot do this job: between reading it and + dispatching, a revoke can cancel the run, see no Actions run to fence, + clear the value and finish, and only then does this deploy write the + value back. The claim and that cancellation are decided against the same + locked row, so one of them loses and knows it. + + Raises: + DeployDispatchWithdrawnError: the run was stopped first. Nothing was + dispatched and nothing needs stopping outside. + DeployFenceLost: another deploy holds the project's deploy lock now. + """ + await deploy_fence.ensure_held(DeployWrite.WORKFLOW_DISPATCH) + if not run_id: + return None + claim = await api_client.claim_deploy_dispatch(run_id) + if not claim.granted: + logger.info( + "deploy_dispatch_withdrawn", + run_id=run_id, + run_status=claim.run_status.value, + ) + raise DeployDispatchWithdrawnError( + f"deploy run {run_id} was {claim.run_status.value} before it was dispatched" + ) + return claim + + +async def fence_active_deploys( + github: GitHubAppClient, + owner: str, + repo: str, + project_id: str, + deploy_fence: DeployFence, + *, + admit: RequestGuard | None = None, +) -> list[int]: + """Stop every deploy run that could still write what the caller replaces. + + The project deploy lock only serialises consumers, and it expires; a GitHub + Actions run it started does not stop when it does. GitHub is asynchronous, so + a dispatch already in flight can still be accepted afterwards: whoever needs + an old value gone confirms that by reading the deployed service. An unproven + stop raises `DeployFenceUnprovenError` rather than allowing a claim of removal. + + Each cancellation request is preceded by `admit`, when given, and then by the + deploy-lock check, so the lock check is the last await before every write. + Their own refusals (`DeployFenceLost`, a withdrawn dispatch) propagate as is. + """ + + async def guard() -> None: + if admit is not None: + await admit() + await deploy_fence.ensure_held(DeployWrite.WORKFLOW_FENCE) + + await deploy_fence.ensure_held(DeployWrite.WORKFLOW_FENCE) + try: + fenced = await github.fence_workflow(owner, repo, DEPLOY_WORKFLOW, guard=guard) + except Exception as e: + if type(e).__name__ not in _CANCELLATION_ERRORS: + raise + raise DeployFenceUnprovenError( + f"an earlier {DEPLOY_WORKFLOW} run in {owner}/{repo} could not be proven stopped: {e}" + ) from e + if fenced: + logger.info( + "deploy_fenced_active_runs", + project_id=project_id, + owner=owner, + repo=repo, + run_ids=fenced, + ) + return fenced + + async def _create_deployment_record( project_id: str, service_name: str, @@ -306,8 +390,8 @@ async def _try_deploy_rerun( # A rerun restarts the same external effect, so it crosses the same # boundary and asks the same question first. - claim = await self._claim_dispatch(deploy_run_id, deploy_fence) - _require_live_lease(claim, datetime.now(UTC)) + claim = await claim_dispatch(deploy_run_id, deploy_fence) + require_live_lease(claim, datetime.now(UTC)) # The claim is an API round trip the lock can expire during. await deploy_fence.ensure_held(DeployWrite.WORKFLOW_DISPATCH) await github.rerun_failed_jobs(owner, repo, run_id) @@ -376,49 +460,6 @@ async def _remove_pin_tag( return e return None - async def _fence_active_deploys( - self, - github: GitHubAppClient, - owner: str, - repo: str, - project_id: str, - deploy_fence: DeployFence, - ) -> list[int]: - """Stop every deploy run that could still write the payload this one replaced. - - The project deploy lock only serialises consumers, and it expires; the - GitHub Actions run it started does not stop when it does. So a deploy that - exists to remove a value stops the runs that can still write the old one: - called after the new payload is in the repository secrets, it covers the - runs that already read the old payload, while the write itself covers - everything created afterwards. - - Together they make the window small, not empty. GitHub is asynchronous - and not ours, so a dispatch already in flight can still be accepted after - all of this. Whoever needs the old value gone confirms that by reading - the deployed service; this only shortens the wait. An unproven stop - refuses the deploy rather than reporting a removal it cannot claim. - """ - await deploy_fence.ensure_held(DeployWrite.WORKFLOW_FENCE) - try: - fenced = await github.fence_workflow(owner, repo, DEPLOY_WORKFLOW) - except Exception as e: - if type(e).__name__ not in _CANCELLATION_ERRORS: - raise - raise DeployFenceUnprovenError( - f"an earlier {DEPLOY_WORKFLOW} run in {owner}/{repo} could not be proven " - f"stopped: {e}" - ) from e - if fenced: - logger.info( - "deploy_fenced_active_runs", - project_id=project_id, - owner=owner, - repo=repo, - run_ids=fenced, - ) - return fenced - async def _dispatch_and_wait( self, github: GitHubAppClient, @@ -452,11 +493,11 @@ async def _dispatch_and_wait( # Last thing before the deploy leaves the system. After this the run # exists on GitHub Actions and can only be stopped there. - claim = await self._claim_dispatch(run_id, deploy_fence) + claim = await claim_dispatch(run_id, deploy_fence) # Record dispatch time BEFORE triggering (for race condition safety) dispatch_time = datetime.now(UTC) - _require_live_lease(claim, dispatch_time) + require_live_lease(claim, dispatch_time) # The claim is an API round trip the lock can expire during, so the # dispatch itself checks again with nothing awaited in between. await deploy_fence.ensure_held(DeployWrite.WORKFLOW_DISPATCH) @@ -553,40 +594,6 @@ async def _run_cancelled(self, run_id: str | None) -> bool: run = await api_client.get(f"runs/{run_id}") return run.get("status") == "cancelled" - async def _claim_dispatch( - self, run_id: str | None, deploy_fence: DeployFence - ) -> DeployDispatchClaim | None: - """Take the dispatch boundary, or refuse to cross it. - - Called immediately before every call that starts work on GitHub Actions. - The claim is itself a write, so it checks the project deploy lock first; - the dispatch or rerun that follows checks again after the claim returns. - A plain read of the run status cannot do this job: between reading it and - dispatching, a revoke can cancel the run, see no Actions run to fence, - clear the value and finish, and only then does this deploy write the - value back. The claim and that cancellation are decided against the same - locked row, so one of them loses and knows it. - - Raises: - DeployDispatchWithdrawnError: the run was stopped first. Nothing was - dispatched and nothing needs stopping outside. - DeployFenceLost: another deploy holds the project's deploy lock now. - """ - await deploy_fence.ensure_held(DeployWrite.WORKFLOW_DISPATCH) - if not run_id: - return None - claim = await api_client.claim_deploy_dispatch(run_id) - if not claim.granted: - logger.info( - "deploy_dispatch_withdrawn", - run_id=run_id, - run_status=claim.run_status.value, - ) - raise DeployDispatchWithdrawnError( - f"deploy run {run_id} was {claim.run_status.value} before it was dispatched" - ) - return claim - @staticmethod async def _published_images( references: dict[str, str], @@ -797,7 +804,7 @@ async def _deploy(self, state: DevOpsState) -> dict: # noqa: PLR0911 # removal by reading the deployed service. if state.get("fence_active_deploys"): try: - await self._fence_active_deploys(github, owner, repo, project_id, deploy_fence) + await fence_active_deploys(github, owner, repo, project_id, deploy_fence) except DeployRefusedError as e: logger.error( "deploy_fence_unproven", diff --git a/services/langgraph/tests/unit/consumers/test_deploy_access_allocations.py b/services/langgraph/tests/unit/consumers/test_deploy_access_allocations.py index 8ad6f7697..f44807dca 100644 --- a/services/langgraph/tests/unit/consumers/test_deploy_access_allocations.py +++ b/services/langgraph/tests/unit/consumers/test_deploy_access_allocations.py @@ -2,9 +2,7 @@ These drive `process_deploy_job` end to end against the real allocation code in `src.allocations`, backed by an in-memory allocation table instead of the API. -What the table ends up holding is the assertion: on 2026-09-25 a QA revoke that -arrived after the suite undeployed its target re-created all four ports, failed -its precheck, and left the application owning allocations nothing runs on. +What the table ends up holding is the assertion. The allocations read are always those of the grant's recorded target application. A repository can hold one application per server, and the product @@ -14,7 +12,7 @@ from __future__ import annotations -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from types import SimpleNamespace from unittest.mock import AsyncMock, patch @@ -22,13 +20,16 @@ import pytest from shared.contracts.dto.application import ApplicationDTO +from shared.contracts.dto.deploy_dispatch import DeployDispatchClaim from shared.contracts.dto.run import RunStatus +from shared.contracts.dto.run_result import DeployRunResult from shared.contracts.dto.temporary_access import ( TemporaryAccessGrantDTO, TemporaryAccessRevokeReason, TemporaryAccessStatus, ) from shared.contracts.queues.deploy import DeployOutcome, DeployTrigger +from shared.crypto import encrypt_dict from tests.unit.factories import ( make_project, make_repository, @@ -226,13 +227,28 @@ def deploy_api(): patch("src.consumers.deploy_result_handler.api_client", api), patch("src.consumers.deploy_failure_handler.api_client", api), patch("src.consumers.deploy_precheck.api_client", api), + patch("src.subgraphs.devops.deployer.api_client", api), ): api.patch = AsyncMock() + api.claim_deploy_dispatch = AsyncMock( + return_value=DeployDispatchClaim( + run_id=TASK_ID, + granted=True, + run_status=RunStatus.RUNNING, + lease_expires_at=datetime.now(UTC) + timedelta(minutes=5), + ) + ) api.get = AsyncMock(return_value=[]) api.get_run = AsyncMock(return_value=make_run(id=TASK_ID)) api.start_run = AsyncMock(return_value=make_run_start(run_id=TASK_ID)) api.get_project = AsyncMock( - return_value=make_project(config={"modules": ["backend"], "estimated_ram_mb": 512}) + return_value=make_project( + config={ + "modules": ["backend"], + "estimated_ram_mb": 512, + "secrets": encrypt_dict({"USERS_GRANT_CAPABILITY": "capability-value"}), + } + ) ) api.get_primary_repository = AsyncMock( return_value=make_repository(id=REPO_ID, git_url="https://github.com/org/p") @@ -273,6 +289,20 @@ def _use_access_operation(deploy_api, operation: str) -> None: ) ) deploy_api.get_temporary_access_grant = AsyncMock(return_value=grant) + # The record's grant Run finished without crossing dispatch: no stop is owed. + deploy_api.list_project_runs = AsyncMock( + return_value=[ + make_run( + id=grant.grant_run_id, + status=RunStatus.COMPLETED, + result=DeployRunResult(deploy_outcome=DeployOutcome.SUCCESS), + run_metadata={ + "temporary_access_grant_id": grant.id, + "temporary_access_operation": "grant", + }, + ) + ] + ) def _run_patches(deploy_api) -> list[dict]: @@ -293,7 +323,8 @@ async def _process(table: AllocationTable, redis, deploy_api, action: str = "fea with ( patch("src.allocations.api_client", table), patch("src.allocations.get_settings", return_value=settings), - patch("src.consumers.deploy_result_handler.GeneratedServiceGrantClient") as client, + patch("src.consumers.temporary_access_operation.api_client", deploy_api), + patch("src.consumers.temporary_access_operation.GeneratedServiceGrantClient") as client, ): client.return_value.grant_and_resolve = AsyncMock( return_value=SimpleNamespace(active=True, failure=None) @@ -354,7 +385,7 @@ async def test_grant_without_deployment_fails_through_grant_failure_without_allo @pytest.mark.asyncio @pytest.mark.parametrize("operation", ["grant", "revoke"]) -async def test_access_operation_on_a_deployment_reads_its_allocations_and_proceeds( +async def test_access_operation_on_a_deployment_reads_its_allocations_and_calls_the_product( redis, deploy_api, precheck, devops, operation ): _use_access_operation(deploy_api, operation) @@ -367,13 +398,13 @@ async def test_access_operation_on_a_deployment_reads_its_allocations_and_procee assert table.allocations == _deployed_allocations() table.allocate_next_port.assert_not_awaited() - precheck.assert_awaited_once() - allocated_resources = precheck.await_args.args[0] - assert sorted(entry["port"] for entry in allocated_resources.values()) == [8000, 8001, 8002] - devops.return_value.ainvoke.assert_awaited_once() + table.get_application_allocations.assert_awaited_once_with(TARGET_APPLICATION_ID) + precheck.assert_not_awaited() + devops.assert_not_called() assert result["status"] == "success" [recorded] = _run_patches(deploy_api) assert recorded["status"] == RunStatus.COMPLETED.value + assert recorded["result"]["application_id"] == TARGET_APPLICATION_ID @pytest.mark.asyncio @@ -415,13 +446,12 @@ async def test_revoke_of_a_deployed_target_ignores_an_undeployed_sibling_listed_ result = await _process(table, redis, deploy_api) table.get_application_allocations.assert_awaited_once_with(TARGET_APPLICATION_ID) - precheck.assert_awaited_once() - allocated_resources = precheck.await_args.args[0] - assert {entry["application_id"] for entry in allocated_resources.values()} == { - TARGET_APPLICATION_ID - } - devops.return_value.ainvoke.assert_awaited_once() + precheck.assert_not_awaited() + devops.assert_not_called() + assert result["status"] == "success" assert result.get("reason") != "not_deployed" + [recorded] = _run_patches(deploy_api) + assert recorded["result"]["application_id"] == TARGET_APPLICATION_ID table.allocate_next_port.assert_not_awaited() @@ -541,3 +571,28 @@ async def test_first_product_deploy_still_places_the_application( table.get_or_create_application.assert_awaited_once() assert len(table.allocations) == 3 assert result["status"] == "success" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "run_metadata", + [ + {"temporary_access_grant_id": None, "temporary_access_operation": None}, + {"temporary_access_grant_id": None}, + {"temporary_access_operation": None}, + {"temporary_access_grant_id": 7, "temporary_access_operation": "grant"}, + ], + ids=["both-null", "null-grant-only", "null-operation-only", "wrong-type"], +) +async def test_a_present_access_marker_is_never_an_ordinary_deploy( + redis, deploy_api, precheck, devops, run_metadata +): + """A marker key with no usable value fails closed; only an unmarked Run deploys.""" + deploy_api.get_run = AsyncMock(return_value=make_run(id=TASK_ID, run_metadata=run_metadata)) + table = AllocationTable(applications=[_undeployed_application()]) + + result = await _process(table, redis, deploy_api) + + assert table.allocations == [] + table.get_application_allocations.assert_not_awaited() + _assert_failed_closed(result, deploy_api, table, precheck, devops) diff --git a/services/langgraph/tests/unit/consumers/test_deploy_fence.py b/services/langgraph/tests/unit/consumers/test_deploy_fence.py index 53c517755..fe18d431c 100644 --- a/services/langgraph/tests/unit/consumers/test_deploy_fence.py +++ b/services/langgraph/tests/unit/consumers/test_deploy_fence.py @@ -26,13 +26,11 @@ from shared.contracts.dto.deploy_dispatch import DeployDispatchClaim from shared.contracts.dto.product_brief import InitialSetting, ProductBriefContent from shared.contracts.dto.run import RunStatus -from shared.contracts.dto.temporary_access import TemporaryAccessStatus from shared.contracts.dto.users_grant import USERS_GRANT_INTENT_KEY, GrantIntent, GrantIntentKind from shared.contracts.queues.deploy import DeployOutcome, DeployTrigger from src.clients.product_settings import SettingSeedProof from src.clients.users_grant import GrantProof from src.deploy_fence import DeployFence, DeployFenceLost, DeployWrite -from tests.unit.deploy_fences import held_deploy_fence from tests.unit.factories import ( make_product_brief, make_project, @@ -673,57 +671,3 @@ async def test_two_claims_of_one_job_never_share_a_token(): await first.ensure_held(DeployWrite.RUN_STATE) await first.release() assert await redis.get(second.lock_key) == second.token - - -@pytest.mark.asyncio -@pytest.mark.parametrize("operation", ["grant", "revoke"]) -async def test_no_temporary_access_operation_is_made_after_the_lock_is_lost(monkeypatch, operation): - """A QA capability redeploy writes the identity's access only under its lock.""" - from src.consumers.deploy_result_handler import _handle_deploy_success - from tests.unit.consumers.test_deploy_routing import _make_deploy_msg, _temporary_grant - - grant = ( - _temporary_grant() - if operation == "grant" - else _temporary_grant( - status=TemporaryAccessStatus.REVOKING, revoke_run_id="temporary-access-grant-1" - ) - ) - fence = held_deploy_fence(project_id="proj-1", task_id=grant.grant_run_id) - loss = LockLoss(DeployWrite.PRODUCT_ACCESS) - loss.install(monkeypatch) - - with ( - patch("src.consumers.deploy_result_handler.api_client") as api, - patch("src.consumers.deploy_result_handler.GeneratedServiceGrantClient") as client, - pytest.raises(DeployFenceLost), - ): - api.get_temporary_access_grant = AsyncMock(return_value=grant) - api.patch = AsyncMock() - await _handle_deploy_success( - result={ - "deployed_url": grant.target_base_url, - "secret_values": {"USERS_GRANT_CAPABILITY": "capability"}, - }, - smoke_result=None, - task_id=grant.grant_run_id, - project_id="proj-1", - project=make_project(), - callback_stream="", - telegram_chat_id="", - story_id="", - redis=AsyncMock(), - msg=_make_deploy_msg(task_id=grant.grant_run_id), - fence=fence, - application_id=grant.target_application_id, - temporary_access_grant=grant, - temporary_access_operation=operation, - ) - - assert loss.fired - # The grant was re-read first; the remote grant or revoke was never made. - api.get_temporary_access_grant.assert_awaited_once_with(grant.id) - client.return_value.grant_and_resolve.assert_not_called() - client.return_value.revoke_and_resolve.assert_not_called() - api.patch.assert_not_awaited() - assert await fence.redis.get(fence.lock_key) == OTHER_DEPLOY diff --git a/services/langgraph/tests/unit/consumers/test_deploy_routing.py b/services/langgraph/tests/unit/consumers/test_deploy_routing.py index 61def7d91..bb00f09d9 100644 --- a/services/langgraph/tests/unit/consumers/test_deploy_routing.py +++ b/services/langgraph/tests/unit/consumers/test_deploy_routing.py @@ -4,14 +4,11 @@ and do NOT perform story transitions (dispatcher's job). """ -from datetime import UTC, datetime -from types import SimpleNamespace from unittest.mock import AsyncMock, patch import pytest from shared.contracts.dto.project import ProjectDTO, ProjectStatus -from shared.contracts.dto.temporary_access import TemporaryAccessGrantDTO, TemporaryAccessStatus from shared.contracts.queues.deploy import DeployMessage, DeployOutcome, DeployTrigger from tests.unit.deploy_fences import held_deploy_fence @@ -35,36 +32,6 @@ def _make_deploy_msg(**overrides) -> DeployMessage: return DeployMessage.model_validate(defaults) -def _temporary_grant(**overrides) -> TemporaryAccessGrantDTO: - now = datetime.now(UTC) - values = { - "id": "tempaccess-qa-1", - "project_id": "proj-1", - "channel": "telegram", - "external_id": "8202532144", - "target_application_id": 42, - "target_base_url": "https://exact.example.com", - "head_sha": "a" * 40, - "qa_run_id": "qa-1", - "grant_run_id": "temporary-access-grant-1", - "qa_message": { - "story_id": "story-1", - "project_id": "proj-1", - "initiating_run_id": "deploy-1", - "telegram_chat_id": "", - "deployed_url": "https://exact.example.com", - "application_id": 42, - "acceptance_criteria": "bot admission", - "run_id": "qa-1", - }, - "status": TemporaryAccessStatus.GRANTING, - "granted_at": now, - "created_at": now, - } - values.update(overrides) - return TemporaryAccessGrantDTO(**values) - - def _project() -> ProjectDTO: return ProjectDTO( id="00000000-0000-0000-0000-000000000001", @@ -167,261 +134,6 @@ async def test_no_story_transitions(self): # No QA message published mock_redis.publish_message.assert_not_called() - @pytest.mark.asyncio - async def test_temporary_revoke_requires_inactive_proof_without_disclosing_capability(self): - from src.consumers.deploy_result_handler import _handle_deploy_success - - mock_redis = AsyncMock() - grant = _temporary_grant( - status=TemporaryAccessStatus.REVOKING, - revoke_run_id="temporary-access-revoke-1", - ) - with ( - patch(f"{_HANDLER_PATCH}.api_client") as mock_api, - patch(f"{_HANDLER_PATCH}.GeneratedServiceGrantClient") as grant_client, - ): - mock_api.patch = AsyncMock() - # This story has no Product Brief, so nothing is seeded. - mock_api.get_product_brief_by_story = AsyncMock(return_value=None) - mock_api.get_temporary_access_grant = AsyncMock(return_value=grant) - grant_client.return_value.revoke_and_resolve = AsyncMock( - return_value=SimpleNamespace(active=False, failure=None) - ) - result = await _handle_deploy_success( - result={ - "deployed_url": "https://other.example.com", - "secret_values": {"USERS_GRANT_CAPABILITY": "capability-value"}, - }, - smoke_result=None, - task_id="temporary-access-revoke-1", - project_id="proj-1", - project=_project(), - callback_stream="cb:1", - telegram_chat_id="123", - story_id="story-1", - redis=mock_redis, - fence=held_deploy_fence(), - msg=_make_deploy_msg(), - application_id=42, - temporary_access_grant=grant, - temporary_access_operation="revoke", - ) - - assert result["status"] == "success" - grant_client.assert_called_once_with("https://exact.example.com") - grant_client.return_value.revoke_and_resolve.assert_awaited_once_with( - channel="telegram", external_id="8202532144", capability="capability-value" - ) - assert "capability-value" not in str(mock_api.patch.await_args) - - @pytest.mark.asyncio - async def test_temporary_revoke_still_active_readback_fails_the_run(self): - """Without an inactive readback the revoke run never records SUCCESS.""" - from src.consumers.deploy_result_handler import _handle_deploy_success - - mock_redis = AsyncMock() - grant = _temporary_grant( - status=TemporaryAccessStatus.REVOKING, - revoke_run_id="temporary-access-revoke-1", - ) - with ( - patch(f"{_HANDLER_PATCH}.api_client") as mock_api, - patch(f"{_HANDLER_PATCH}.GeneratedServiceGrantClient") as grant_client, - ): - mock_api.patch = AsyncMock() - mock_api.get_product_brief_by_story = AsyncMock(return_value=None) - mock_api.get_temporary_access_grant = AsyncMock(return_value=grant) - grant_client.return_value.revoke_and_resolve = AsyncMock( - return_value=SimpleNamespace(active=True, failure=None) - ) - result = await _handle_deploy_success( - result={ - "deployed_url": "https://exact.example.com", - "secret_values": {"USERS_GRANT_CAPABILITY": "capability-value"}, - }, - smoke_result=None, - task_id="temporary-access-revoke-1", - project_id="proj-1", - project=_project(), - callback_stream="cb:1", - telegram_chat_id="123", - story_id="story-1", - redis=mock_redis, - fence=held_deploy_fence(), - msg=_make_deploy_msg(), - application_id=42, - temporary_access_grant=grant, - temporary_access_operation="revoke", - ) - - assert result["status"] == "failed" - patched = mock_api.patch.await_args.kwargs["json"] - assert patched["status"] == "failed" - assert patched["result"]["deploy_outcome"] == DeployOutcome.OWNER_ACCESS_PROOF_FAILED.value - assert patched["result"]["error_details"] == "unverified" - - @pytest.mark.asyncio - async def test_temporary_grant_failure_never_records_success_or_releases_handoff(self): - from src.consumers.deploy_result_handler import _handle_deploy_success - - mock_redis = AsyncMock() - with ( - patch(f"{_HANDLER_PATCH}.api_client") as mock_api, - patch(f"{_HANDLER_PATCH}.GeneratedServiceGrantClient") as grant_client, - ): - mock_api.patch = AsyncMock() - # This story has no Product Brief, so nothing is seeded. - mock_api.get_product_brief_by_story = AsyncMock(return_value=None) - mock_api.get_temporary_access_grant = AsyncMock(return_value=_temporary_grant()) - grant_client.return_value.grant_and_resolve = AsyncMock( - return_value=SimpleNamespace( - active=False, failure=SimpleNamespace(value="inactive") - ) - ) - result = await _handle_deploy_success( - result={ - "deployed_url": "https://exact.example.com", - "secret_values": {"USERS_GRANT_CAPABILITY": "capability-value"}, - }, - smoke_result=None, - task_id="deploy-1", - project_id="proj-1", - project=_project(), - callback_stream="cb:1", - telegram_chat_id="123", - story_id="story-1", - redis=mock_redis, - fence=held_deploy_fence(), - msg=_make_deploy_msg(), - application_id=42, - temporary_access_grant=_temporary_grant(), - temporary_access_operation="grant", - ) - - assert result["status"] == "failed" - patched = mock_api.patch.await_args.kwargs["json"] - assert patched["result"]["deploy_outcome"] == DeployOutcome.OWNER_ACCESS_PROOF_FAILED.value - assert "capability-value" not in str(patched) - mock_redis.publish_message.assert_not_called() - - @pytest.mark.asyncio - async def test_temporary_grant_requires_exact_target_and_active_readback(self): - from src.consumers.deploy_result_handler import _apply_temporary_access_operation - - with ( - patch(f"{_HANDLER_PATCH}.api_client") as mock_api, - patch(f"{_HANDLER_PATCH}.GeneratedServiceGrantClient") as grant_client, - ): - mock_api.get_temporary_access_grant = AsyncMock(return_value=_temporary_grant()) - grant_client.return_value.grant_and_resolve = AsyncMock( - return_value=SimpleNamespace(active=True, failure=None) - ) - mismatch = await _apply_temporary_access_operation( - task_id="temporary-access-grant-1", - project_id="proj-1", - application_id=41, - secret_values={"USERS_GRANT_CAPABILITY": "capability-value"}, - grant=_temporary_grant(), - operation="grant", - fence=held_deploy_fence(), - ) - proof = await _apply_temporary_access_operation( - task_id="temporary-access-grant-1", - project_id="proj-1", - application_id=42, - secret_values={"USERS_GRANT_CAPABILITY": "capability-value"}, - grant=_temporary_grant(), - operation="grant", - fence=held_deploy_fence(), - ) - - assert mismatch == "temporary_access_target_mismatch" - grant_client.return_value.grant_and_resolve.assert_awaited_once_with( - channel="telegram", external_id="8202532144", capability="capability-value" - ) - assert proof is None - - @pytest.mark.asyncio - @pytest.mark.parametrize( - ("operation", "current_grant"), - [ - ( - "grant", - _temporary_grant( - status=TemporaryAccessStatus.REVOKED, - revoked_at=datetime.now(UTC), - ), - ), - ( - "revoke", - _temporary_grant( - status=TemporaryAccessStatus.REVOKED, - revoke_run_id="temporary-access-revoke-1", - revoked_at=datetime.now(UTC), - ), - ), - ], - ) - async def test_temporary_operation_refuses_terminal_or_superseded_record_before_effect( - self, operation, current_grant - ): - from src.consumers.deploy_result_handler import _apply_temporary_access_operation - - attempted = _temporary_grant( - revoke_run_id="temporary-access-revoke-1" if operation == "revoke" else None, - status=TemporaryAccessStatus.REVOKING - if operation == "revoke" - else TemporaryAccessStatus.GRANTING, - ) - task_id = attempted.revoke_run_id if operation == "revoke" else attempted.grant_run_id - with ( - patch(f"{_HANDLER_PATCH}.api_client") as mock_api, - patch(f"{_HANDLER_PATCH}.GeneratedServiceGrantClient") as grant_client, - ): - mock_api.get_temporary_access_grant = AsyncMock(return_value=current_grant) - - refusal = await _apply_temporary_access_operation( - task_id=task_id, - project_id="proj-1", - application_id=42, - secret_values={"USERS_GRANT_CAPABILITY": "capability-value"}, - grant=attempted, - operation=operation, - fence=held_deploy_fence(), - ) - - assert refusal == "temporary_access_operation_superseded" - grant_client.assert_not_called() - - @pytest.mark.asyncio - async def test_delayed_grant_cannot_reapply_after_revoke_proved_inactive(self): - from src.consumers.deploy_result_handler import _apply_temporary_access_operation - - stale_grant = _temporary_grant() - revoked_grant = _temporary_grant( - status=TemporaryAccessStatus.REVOKED, - revoke_run_id="temporary-access-revoke-1", - revoked_at=datetime.now(UTC), - ) - with ( - patch(f"{_HANDLER_PATCH}.api_client") as mock_api, - patch(f"{_HANDLER_PATCH}.GeneratedServiceGrantClient") as grant_client, - ): - mock_api.get_temporary_access_grant = AsyncMock(return_value=revoked_grant) - - refusal = await _apply_temporary_access_operation( - task_id=stale_grant.grant_run_id, - project_id="proj-1", - application_id=42, - secret_values={"USERS_GRANT_CAPABILITY": "capability-value"}, - grant=stale_grant, - operation="grant", - fence=held_deploy_fence(), - ) - - assert refusal == "temporary_access_operation_superseded" - grant_client.assert_not_called() - class TestHandleSmokeFailure: """_handle_smoke_failure stores a retry outcome, no story transitions.""" @@ -461,114 +173,8 @@ async def test_stores_retry_outcome(self): assert result["status"] == "failed" -class TestCapabilityOperationIsNotAProductDeploy: - """A grant/revoke redeploy carries the QA identity, not the brief's settings. - - Stand-e2e run 35470184817: a revoke that had already proved the QA access - inactive was stored `settings_seed_failed`, because the same handler then - seeded the *project's* confirmed brief — by then the second story's - revision 3 — into the first story's still-deployed commit, which does not - declare its key. The scheduler reads a capability operation as proved only - when the run is `COMPLETED`/`SUCCESS`, so that refusal meant the grant never - reached `REVOKED` and went on holding the target application against the - second story's own QA handoff. - """ - - @pytest.mark.asyncio - async def test_a_revoke_operation_seeds_nothing_and_stays_proved(self): - from src.consumers.deploy_result_handler import _handle_deploy_success - - grant = _temporary_grant( - status=TemporaryAccessStatus.REVOKING, - revoke_run_id="temporary-access-revoke-1", - ) - with ( - patch(f"{_HANDLER_PATCH}.api_client") as mock_api, - patch(f"{_HANDLER_PATCH}.GeneratedServiceGrantClient") as grant_client, - patch(f"{_HANDLER_PATCH}.GeneratedServiceSettingsClient") as settings_client, - ): - mock_api.patch = AsyncMock() - mock_api.get_temporary_access_grant = AsyncMock(return_value=grant) - mock_api.get_product_brief_by_story = AsyncMock() - mock_api.get_project_initial_settings_brief = AsyncMock() - grant_client.return_value.revoke_and_resolve = AsyncMock( - return_value=SimpleNamespace(active=False, failure=None) - ) - result = await _handle_deploy_success( - result={ - "deployed_url": "https://exact.example.com", - "secret_values": { - "USERS_GRANT_CAPABILITY": "capability-value", - "SETTINGS_WRITE_CAPABILITY": "settings-capability-value", - }, - }, - smoke_result=None, - task_id="temporary-access-revoke-1", - project_id="proj-1", - project=_project(), - callback_stream="cb:1", - telegram_chat_id="123", - # A capability operation names no story: it redeploys whatever - # the target is running. - story_id="", - redis=AsyncMock(), - fence=held_deploy_fence(), - msg=_make_deploy_msg(story_id=""), - application_id=42, - temporary_access_grant=grant, - temporary_access_operation="revoke", - ) - - assert result["status"] == "success" - patched = mock_api.patch.await_args.kwargs["json"] - assert patched["status"] == "completed" - assert patched["result"]["deploy_outcome"] == DeployOutcome.SUCCESS.value - assert patched["result"]["settings_seed"] == [] - # No brief is read and no setting is written against somebody else's commit. - mock_api.get_project_initial_settings_brief.assert_not_called() - mock_api.get_product_brief_by_story.assert_not_called() - settings_client.assert_not_called() - - @pytest.mark.asyncio - async def test_a_grant_operation_seeds_nothing_either(self): - from src.consumers.deploy_result_handler import _handle_deploy_success - - grant = _temporary_grant() - with ( - patch(f"{_HANDLER_PATCH}.api_client") as mock_api, - patch(f"{_HANDLER_PATCH}.GeneratedServiceGrantClient") as grant_client, - patch(f"{_HANDLER_PATCH}.GeneratedServiceSettingsClient") as settings_client, - ): - mock_api.patch = AsyncMock() - mock_api.get_temporary_access_grant = AsyncMock(return_value=grant) - mock_api.get_project_initial_settings_brief = AsyncMock() - grant_client.return_value.grant_and_resolve = AsyncMock( - return_value=SimpleNamespace(active=True, failure=None) - ) - result = await _handle_deploy_success( - result={ - "deployed_url": "https://exact.example.com", - "secret_values": {"USERS_GRANT_CAPABILITY": "capability-value"}, - }, - smoke_result=None, - task_id="temporary-access-grant-1", - project_id="proj-1", - project=_project(), - callback_stream="cb:1", - telegram_chat_id="123", - story_id="", - redis=AsyncMock(), - fence=held_deploy_fence(), - msg=_make_deploy_msg(story_id=""), - application_id=42, - temporary_access_grant=grant, - temporary_access_operation="grant", - ) - - assert result["status"] == "success" - assert mock_api.patch.await_args.kwargs["json"]["result"]["settings_seed"] == [] - mock_api.get_project_initial_settings_brief.assert_not_called() - settings_client.assert_not_called() +class TestStorylessDeploySeeding: + """A deploy that names no story seeds the project's latest confirmed brief.""" @pytest.mark.asyncio async def test_an_ordinary_story_deploy_still_seeds_the_confirmed_brief(self): diff --git a/services/langgraph/tests/unit/consumers/test_deploy_temporary_access.py b/services/langgraph/tests/unit/consumers/test_deploy_temporary_access.py new file mode 100644 index 000000000..bf17fa1d2 --- /dev/null +++ b/services/langgraph/tests/unit/consumers/test_deploy_temporary_access.py @@ -0,0 +1,951 @@ +"""A temporary QA grant or revoke calls the target's users-core capability and deploys nothing. + +These drive `process_deploy_job` with the scheduler's operation message. The API, +including the Run's dispatch claim and withdrawal, the allocation reads, GitHub +Actions and the deployed product are fakes at their boundaries. The product +speaks HTTP through `httpx.MockTransport` and GitHub through `respx`, so the real +capability client and the real workflow fence are what is exercised. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +import json +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest +import respx +from structlog.testing import capture_logs + +from shared.clients.github import GitHubAppClient +from shared.contracts.dto.application import ApplicationDTO +from shared.contracts.dto.deploy_dispatch import DISPATCH_CLAIMED_AT_KEY, DeployDispatchClaim +from shared.contracts.dto.run import RunStatus, RunType +from shared.contracts.dto.run_result import DeployRunResult +from shared.contracts.dto.temporary_access import ( + TemporaryAccessGrantDTO, + TemporaryAccessRevokeReason, + TemporaryAccessStatus, +) +from shared.contracts.dto.users_grant import USERS_GRANT_INTENT_KEY +from shared.contracts.queues.deploy import DeployAction, DeployOutcome, DeployTrigger +from shared.crypto import encrypt_dict +from src.clients.users_grant import GeneratedServiceGrantClient +from src.deploy_fence import DeployFence, DeployWrite +from tests.unit.factories import ( + make_project, + make_repository, + make_run, + make_run_start, + make_server, +) + +TASK_ID = "temporary-access-op-1" +PROJECT_ID = "proj-1" +TARGET_APPLICATION_ID = 17 +OTHER_APPLICATION_ID = 11 +TARGET_URL = "https://target.example.com" +HEAD_SHA = "a" * 40 +CAPABILITY = "stored-capability-canary" +QA_IDENTITY = {"channel": "telegram", "external_id": "8202532144"} +_NOW = datetime.now(UTC) +SERVER = make_server(handle="srv-1", public_ip="1.2.3.4", last_health_check=_NOW) + + +def _grant(operation: str, **overrides) -> TemporaryAccessGrantDTO: + values = { + "id": "tempaccess-qa-1", + "project_id": PROJECT_ID, + **QA_IDENTITY, + "target_application_id": TARGET_APPLICATION_ID, + "target_base_url": TARGET_URL, + "head_sha": HEAD_SHA, + "qa_run_id": "qa-1", + "grant_run_id": TASK_ID if operation == "grant" else "temporary-access-grant-0", + "revoke_run_id": TASK_ID if operation == "revoke" else None, + "revoke_reason": ( + TemporaryAccessRevokeReason.RUN_TERMINAL if operation == "revoke" else None + ), + "qa_message": { + "story_id": "story-1", + "project_id": PROJECT_ID, + "initiating_run_id": "deploy-1", + "telegram_chat_id": "", + "deployed_url": TARGET_URL, + "application_id": TARGET_APPLICATION_ID, + "acceptance_criteria": "bot admission", + "run_id": "qa-1", + }, + "status": ( + TemporaryAccessStatus.GRANTING + if operation == "grant" + else TemporaryAccessStatus.REVOKING + ), + "granted_at": _NOW, + "created_at": _NOW, + } + values.update(overrides) + return TemporaryAccessGrantDTO(**values) + + +def _application(application_id: int, *, server_handle: str = SERVER.handle) -> ApplicationDTO: + return ApplicationDTO( + id=application_id, + repo_id="repo-1", + server_handle=server_handle, + service_name=f"app-{application_id}", + status="running", + reserved_ram_mb=512, + created_at=_NOW, + updated_at=_NOW, + ) + + +def _allocations(application_id: int) -> list[dict]: + return [ + { + "server_handle": SERVER.handle, + "port": 8000 + index, + "server_ip": SERVER.public_ip, + "service_name": module, + "application_id": application_id, + } + for index, module in enumerate(["backend", "postgres", "redis"]) + ] + + +def _grant_run( + grant: TemporaryAccessGrantDTO, + run_id: str, + status: RunStatus, + *, + claimed: bool = False, + succeeded: bool = False, +): + """A grant operation Run of `grant`, as the scheduler created it and its worker left it.""" + metadata = {"temporary_access_grant_id": grant.id, "temporary_access_operation": "grant"} + if claimed: + metadata[DISPATCH_CLAIMED_AT_KEY] = _NOW.isoformat() + return make_run( + id=run_id, + status=status, + run_metadata=metadata, + result=DeployRunResult(deploy_outcome=DeployOutcome.SUCCESS) if succeeded else None, + ) + + +class FakeApi: + """The API slice an access operation reads and writes, with the Run kept like the API. + + It has no allocation-creating, deployment-record or secret-writing method, so + a call to one fails the operation instead of passing silently. + """ + + def __init__(self, operation: str, *, secrets: dict | None) -> None: + self.grant = _grant(operation) + self.current_grant = self.grant + self.run_metadata = { + "temporary_access_grant_id": self.grant.id, + "temporary_access_operation": operation, + } + config = {"modules": ["backend"], "estimated_ram_mb": 512} + if secrets is not None: + config["secrets"] = secrets + self.project = make_project(config=config) + self.applications = { + TARGET_APPLICATION_ID: _application(TARGET_APPLICATION_ID), + OTHER_APPLICATION_ID: _application(OTHER_APPLICATION_ID), + } + self.allocations = { + TARGET_APPLICATION_ID: _allocations(TARGET_APPLICATION_ID), + OTHER_APPLICATION_ID: _allocations(OTHER_APPLICATION_ID), + } + self.grant_reads = 0 + self.run_status = RunStatus.QUEUED + self.run_writes: list[dict] = [] + self.other_writes: list[tuple[str, dict]] = [] + self.allocation_reads: list[int] = [] + self.get_users_grant_intent = AsyncMock() + self.events: list[str] = [] + self.on_allocation_read = None + # The record's grant operation Runs: its current one finished, unclaimed, + # so by default no earlier workflow is owed a stop. + self.history = [ + _grant_run(self.grant, self.grant.grant_run_id, RunStatus.COMPLETED, succeeded=True) + ] + self.history_unreadable = False + + async def get_run(self, run_id: str): + assert run_id == TASK_ID + return make_run(id=TASK_ID, status=self.run_status, run_metadata=self.run_metadata) + + async def list_project_runs(self, project_id: str, run_type: RunType): + """`GET runs/?project_id&run_type`: the whole set, this operation Run included.""" + assert (project_id, run_type) == (PROJECT_ID, RunType.DEPLOY) + if self.history_unreadable: + request = httpx.Request("GET", "http://api/runs/") + raise httpx.HTTPStatusError( + "503", request=request, response=httpx.Response(503, request=request) + ) + operation_run = make_run(id=TASK_ID, status=self.run_status, run_metadata=self.run_metadata) + return [operation_run, *self.history] + + async def claim_deploy_dispatch(self, run_id: str) -> DeployDispatchClaim: + """The locked dispatch claim: refused once the Run is terminal.""" + assert run_id == TASK_ID + granted = self.run_status is RunStatus.RUNNING + self.events.append("claim" if granted else "claim-refused") + if granted: + self.run_metadata[DISPATCH_CLAIMED_AT_KEY] = _NOW.isoformat() + return DeployDispatchClaim( + run_id=run_id, + granted=granted, + run_status=self.run_status, + lease_expires_at=datetime.now(UTC) + timedelta(minutes=5), + ) + + def withdraw(self) -> None: + """What `POST /runs/{id}/dispatch-withdraw` does to a Run not yet terminal.""" + self.run_status = RunStatus.CANCELLED + + async def start_run(self, task_id: str): + self.run_status = RunStatus.RUNNING + return make_run_start(run_id=task_id) + + async def get_project(self, project_id: str, **_kwargs): + return self.project + + async def get_temporary_access_grant(self, grant_id: str): + assert grant_id == self.grant.id + self.grant_reads += 1 + return self.grant if self.grant_reads == 1 else self.current_grant + + async def get_application(self, application_id: int): + return self.applications[application_id] + + async def get_application_allocations(self, application_id: int): + self.allocation_reads.append(application_id) + if self.on_allocation_read is not None: + self.on_allocation_read() + return self.allocations[application_id] + + async def get_server(self, handle: str): + return SERVER + + async def list_active_incidents(self): + return [] + + async def get_primary_repository(self, project_id: str): + return make_repository(git_url="https://github.com/org/product") + + async def patch(self, path: str, json: dict): + if path != f"runs/{TASK_ID}": + self.other_writes.append((path, json)) + return {} + self.run_writes.append(json) + self.run_status = RunStatus(json["status"]) + return {} + + +class FakeProduct: + """The deployed product's users-core grant, revoke and access endpoints.""" + + def __init__(self, *, active: bool) -> None: + self.active = active + self.requests: list[httpx.Request] = [] + self.base_urls: list[str] = [] + self.refuse: str | None = None + self.access_payload: object | None = None + self.events: list[str] = [] + + def handle(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + path = request.url.path + self.events.append(path) + if path == self.refuse: + return httpx.Response(500) + if path == "/users/grant": + self.active = True + return httpx.Response(200, json={}) + if path == "/users/revoke": + self.active = False + return httpx.Response(200, json={}) + payload = self.access_payload or { + **QA_IDENTITY, + "status": "active" if self.active else "inactive", + } + return httpx.Response(200, json=payload) + + def client(self, base_url: str) -> GeneratedServiceGrantClient: + self.base_urls.append(base_url) + transport = httpx.AsyncClient(transport=httpx.MockTransport(self.handle)) + return GeneratedServiceGrantClient(base_url, transport=transport) + + def calls(self) -> list[tuple[str, str]]: + return [(request.method, request.url.path) for request in self.requests] + + +class FakeGitHub: + """GitHub Actions of the product repository: run listing, cancellation, run status. + + A cancelled run reports completed on its next read, so the fence never waits. + """ + + def __init__(self, events: list[str]) -> None: + self.runs: dict[int, str] = {} + self.events = events + self.requests: list[tuple[str, str]] = [] + self.refuse_cancel = False + self.on_list = None + self.on_cancel = None + + def client(self) -> GitHubAppClient: + client = GitHubAppClient() + client.get_token = AsyncMock(return_value="github-token") + return client + + def handle(self, request: httpx.Request) -> httpx.Response: + path = request.url.path + self.requests.append((request.method, path)) + if path.endswith("/actions/workflows/deploy.yml/runs"): + status = request.url.params["status"] + if self.on_list is not None: + self.on_list() + listed = [{"id": i, "status": s} for i, s in self.runs.items() if s == status] + return httpx.Response(200, json={"workflow_runs": listed}) + if path.endswith("/cancel"): + run_id = int(path.split("/")[-2]) + self.events.append(f"cancel:{run_id}") + if self.refuse_cancel: + return httpx.Response(422) + self.runs[run_id] = "completed" + if self.on_cancel is not None: + self.on_cancel(run_id) + return httpx.Response(202) + run_id = int(path.rsplit("/", 1)[-1]) + return httpx.Response(200, json={"id": run_id, "status": self.runs[run_id]}) + + def cancels(self) -> list[int]: + return [int(path.split("/")[-2]) for _, path in self.requests if path.endswith("/cancel")] + + def writes(self) -> list[tuple[str, str]]: + return [(method, path) for method, path in self.requests if method != "GET"] + + +class LockRedis: + """The project deploy lock: `SET NX`, the hold check and compare-and-delete.""" + + def __init__(self) -> None: + self.value: str | None = None + + async def set(self, key: str, value: str, *, nx: bool, ex: int) -> bool: + if self.value is not None: + return False + self.value = value + return True + + async def eval(self, script: str, numkeys: int, key: str, token: str) -> int: + held = self.value == token + if held and "DEL" in script: + self.value = None + return int(held) + + async def exists(self, key: str) -> int: + return 0 + + +@pytest.fixture +def operation_env(): + """Everything an access operation could touch, wired to fakes.""" + + def build(operation: str, *, secrets: dict | None = None, stored: bool = True): + if stored and secrets is None: + secrets = encrypt_dict({"USERS_GRANT_CAPABILITY": CAPABILITY}) + api = FakeApi(operation, secrets=secrets) + product = FakeProduct(active=operation == "revoke") + product.events = api.events + return SimpleNamespace( + operation=operation, + api=api, + product=product, + github=FakeGitHub(api.events), + devops=MagicMock(), + precheck=AsyncMock(return_value=None), + stream=AsyncMock(redis=LockRedis()), + ) + + return build + + +async def _process(env, **message) -> dict: + from src.consumers.deploy import process_deploy_job + + settings = SimpleNamespace( + allocation_ram_reserve_mb=256, allocation_metrics_freshness_seconds=300 + ) + job = { + "task_id": TASK_ID, + "project_id": PROJECT_ID, + "unaddressed_reason": "temporary QA capability operation", + "callback_stream": "", + "story_id": "", + "triggered_by": DeployTrigger.ADMIN.value, + "action": DeployAction.FEATURE.value, + "head_sha": HEAD_SHA, + "deployed_commit_sha": "b" * 40, + **message, + } + with ( + patch("src.consumers.deploy.api_client", env.api), + patch("src.consumers.deploy_failure_handler.api_client", env.api), + patch("src.consumers.deploy_result_handler.api_client", env.api), + patch("src.consumers.temporary_access_operation.api_client", env.api), + patch("src.subgraphs.devops.deployer.api_client", env.api), + patch("src.allocations.api_client", env.api), + patch("src.allocations.get_settings", return_value=settings), + patch("src.consumers.deploy.create_devops_subgraph", env.devops), + patch("src.consumers.deploy._run_deploy_precheck", env.precheck), + patch("src.consumers.deploy.GitHubAppClient", side_effect=env.github.client), + patch( + "src.consumers.temporary_access_operation.GeneratedServiceGrantClient", + side_effect=env.product.client, + ), + respx.mock(assert_all_called=False) as github, + capture_logs() as logs, + ): + github.route(host="api.github.com").mock(side_effect=env.github.handle) + result = await process_deploy_job(job, env.stream) + env.logs = logs + return result + + +def _proves_operation(write: dict) -> bool: + """What the reconciler and the API accept as a grant or revoke proof.""" + result = DeployRunResult.model_validate(write["result"]) + return ( + write["status"] == RunStatus.COMPLETED.value + and result.deploy_outcome is DeployOutcome.SUCCESS + and result.skipped_reason is None + ) + + +def _assert_deployed_nothing(env) -> None: + env.devops.assert_not_called() + env.precheck.assert_not_awaited() + assert all(path.endswith("/cancel") for _, path in env.github.writes()) + assert env.api.other_writes == [] + + +def _assert_credential_safe(env) -> None: + recorded = json.dumps(env.api.run_writes) + json.dumps(env.logs, default=str) + assert CAPABILITY not in recorded + + +def _assert_retryable_failure(env, reason: str) -> None: + [write] = env.api.run_writes + assert write["status"] == RunStatus.FAILED.value + result = DeployRunResult.model_validate(write["result"]) + assert result.deploy_outcome is DeployOutcome.OWNER_ACCESS_PROOF_FAILED + assert reason in result.error_details + assert not _proves_operation(write) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("operation", "effect_path"), [("grant", "/users/grant"), ("revoke", "/users/revoke")] +) +async def test_operation_calls_only_the_targets_capability_and_readback( + operation_env, operation, effect_path +): + env = operation_env(operation) + + result = await _process(env) + + assert result["status"] == "success" + assert env.product.base_urls == [TARGET_URL] + assert env.product.calls() == [("POST", effect_path), ("GET", "/users/access")] + effect, readback = env.product.requests + assert effect.headers["X-Grant-Capability"] == CAPABILITY + assert json.loads(effect.content) == QA_IDENTITY + assert dict(readback.url.params) == QA_IDENTITY + assert "X-Grant-Capability" not in readback.headers + assert env.product.active is (operation == "grant") + [write] = env.api.run_writes + assert _proves_operation(write) + assert write["result"]["application_id"] == TARGET_APPLICATION_ID + assert env.api.allocation_reads == [TARGET_APPLICATION_ID] + _assert_deployed_nothing(env) + _assert_credential_safe(env) + assert env.stream.redis.value is None + + +@pytest.mark.asyncio +async def test_grant_acts_on_the_recorded_target_not_the_repositorys_other_application( + operation_env, +): + env = operation_env("grant") + env.api.grant = env.api.current_grant = _grant( + "grant", target_application_id=OTHER_APPLICATION_ID, target_base_url="https://other.test" + ) + + result = await _process(env) + + assert result["status"] == "success" + assert env.api.allocation_reads == [OTHER_APPLICATION_ID] + assert env.product.base_urls == ["https://other.test"] + assert env.api.run_writes[0]["result"]["application_id"] == OTHER_APPLICATION_ID + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("operation", "current"), + [ + pytest.param( + "grant", + {"status": TemporaryAccessStatus.REVOKING, "revoke_run_id": "revoke-2"}, + id="grant-after-cleanup-started", + ), + pytest.param("grant", {"grant_run_id": "temporary-access-grant-9"}, id="grant-replaced"), + pytest.param( + "revoke", + {"status": TemporaryAccessStatus.REVOKED, "revoked_at": _NOW}, + id="revoke-already-settled", + ), + ], +) +async def test_a_superseded_operation_never_reaches_the_product(operation_env, operation, current): + env = operation_env(operation) + env.api.current_grant = _grant(operation, **current) + + await _process(env) + + assert env.api.grant_reads == 2 + assert env.product.requests == [] + _assert_retryable_failure(env, "temporary_access_operation_superseded") + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "current", + [ + {"target_application_id": OTHER_APPLICATION_ID}, + {"head_sha": "c" * 40}, + {"external_id": "1"}, + ], + ids=["target", "head-sha", "identity"], +) +async def test_a_record_that_no_longer_matches_never_reaches_the_product(operation_env, current): + env = operation_env("grant") + env.api.current_grant = _grant("grant", **current) + + await _process(env) + + assert env.product.requests == [] + _assert_retryable_failure(env, "temporary_access_target_mismatch") + + +def _lose_lock_at(monkeypatch, env, write: DeployWrite, occurrence: int) -> None: + check = DeployFence.ensure_held + seen = [] + + async def ensure_held(fence: DeployFence, checked: DeployWrite) -> None: + if checked is write: + seen.append(checked) + if len(seen) == occurrence: + env.stream.redis.value = "another-deploy" + await check(fence, checked) + + monkeypatch.setattr(DeployFence, "ensure_held", ensure_held) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("operation", ["grant", "revoke"]) +async def test_a_lock_lost_before_the_effect_makes_no_product_call( + operation_env, monkeypatch, operation +): + env = operation_env(operation) + _lose_lock_at(monkeypatch, env, DeployWrite.PRODUCT_ACCESS, 1) + + result = await _process(env) + + assert env.product.requests == [] + assert result["reason"] == DeployOutcome.DEPLOY_LOCK_LOST.value + [write] = env.api.run_writes + assert write["result"]["deploy_outcome"] == DeployOutcome.DEPLOY_LOCK_LOST.value + assert env.stream.redis.value == "another-deploy" + + +@pytest.mark.asyncio +async def test_a_lock_lost_before_the_terminal_write_never_records_proof( + operation_env, monkeypatch +): + env = operation_env("grant") + # The first RUN_STATE check is the claim's own start of the Run. + _lose_lock_at(monkeypatch, env, DeployWrite.RUN_STATE, 2) + + result = await _process(env) + + assert env.product.calls() == [("POST", "/users/grant"), ("GET", "/users/access")] + assert result["reason"] == DeployOutcome.DEPLOY_LOCK_LOST.value + [write] = env.api.run_writes + assert write["result"]["deploy_outcome"] == DeployOutcome.DEPLOY_LOCK_LOST.value + assert not _proves_operation(write) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("secrets", "reason"), + [ + pytest.param(None, "capability_unavailable", id="absent"), + pytest.param( + {"USERS_GRANT_CAPABILITY": "not-ciphertext"}, "capability_unreadable", id="bad" + ), + ], +) +@pytest.mark.parametrize("operation", ["grant", "revoke"]) +async def test_an_unusable_stored_capability_fails_closed_and_is_never_replaced( + operation_env, operation, secrets, reason +): + env = operation_env(operation, secrets=secrets, stored=secrets is not None) + + await _process(env) + + assert env.product.requests == [] + _assert_retryable_failure(env, reason) + _assert_deployed_nothing(env) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("operation", "refuse", "access_payload", "reason"), + [ + pytest.param("grant", "/users/grant", None, "grant_rejected", id="grant-refused"), + pytest.param("revoke", "/users/revoke", None, "revoke_rejected", id="revoke-refused"), + pytest.param("grant", "/users/access", None, "access_rejected", id="readback-refused"), + pytest.param( + "grant", + None, + {**QA_IDENTITY, "status": "unknown"}, + "malformed_access", + id="readback-malformed", + ), + pytest.param( + "revoke", + None, + {**QA_IDENTITY, "status": "active"}, + "active", + id="revoke-still-active", + ), + pytest.param( + "grant", + None, + {**QA_IDENTITY, "status": "inactive"}, + "inactive", + id="grant-not-active", + ), + ], +) +async def test_an_unproved_readback_fails_retryably_without_the_capability( + operation_env, operation, refuse, access_payload, reason +): + env = operation_env(operation) + env.product.refuse = refuse + env.product.access_payload = access_payload + + await _process(env) + + _assert_retryable_failure(env, reason) + _assert_credential_safe(env) + _assert_deployed_nothing(env) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "conflict", + [ + pytest.param({"metadata": {USERS_GRANT_INTENT_KEY: "intent-1"}}, id="grant-intent"), + pytest.param( + { + "message": { + "action": DeployAction.STOP.value, + "application_id": TARGET_APPLICATION_ID, + } + }, + id="lifecycle-action", + ), + ], +) +async def test_conflicting_access_metadata_fails_closed_without_deploying(operation_env, conflict): + env = operation_env("grant") + env.api.run_metadata.update(conflict.get("metadata", {})) + env.api.get_users_grant_intent.return_value = SimpleNamespace( + project_id=PROJECT_ID, target_sha=HEAD_SHA, execution_run_id=TASK_ID + ) + + await _process(env, **conflict.get("message", {})) + + assert env.product.requests == [] + assert env.api.allocation_reads == [] + _assert_retryable_failure(env, "does not match") + _assert_deployed_nothing(env) + + +def _lose_lock(env) -> None: + env.stream.redis.value = "another-deploy" + + +def _assert_lock_lost(env, result: dict) -> None: + assert result["reason"] == DeployOutcome.DEPLOY_LOCK_LOST.value + [write] = env.api.run_writes + assert write["result"]["deploy_outcome"] == DeployOutcome.DEPLOY_LOCK_LOST.value + + +def _assert_withdrawn(env, result: dict) -> None: + assert result["status"] == "cancelled" + [write] = env.api.run_writes + assert write["status"] == RunStatus.CANCELLED.value + assert write["result"]["deploy_outcome"] == DeployOutcome.CANCELLED.value + + +@pytest.mark.asyncio +async def test_a_flagged_cleanup_stops_each_earlier_workflow_before_revoking(operation_env): + env = operation_env("revoke") + env.github.runs = {41: "in_progress", 42: "queued"} + + result = await _process(env, fence_active_deploys=True) + + assert result["status"] == "success" + assert sorted(env.github.cancels()) == [41, 42] + events = env.api.events + cancels = [index for index, event in enumerate(events) if event.startswith("cancel:")] + # Every cancellation is the first write after a renewed admission. + assert all(events[index - 1] == "claim" for index in cancels) + assert events[-3:] == ["claim", "/users/revoke", "/users/access"] + assert max(cancels) < events.index("/users/revoke") + assert _proves_operation(env.api.run_writes[0]) + _assert_deployed_nothing(env) + + +@pytest.mark.asyncio +async def test_a_lock_lost_after_listing_cancels_nothing(operation_env): + env = operation_env("revoke") + env.github.runs = {41: "in_progress", 42: "queued"} + env.github.on_list = lambda: _lose_lock(env) + + result = await _process(env, fence_active_deploys=True) + + assert env.github.writes() == [] + assert env.product.requests == [] + _assert_lock_lost(env, result) + + +@pytest.mark.asyncio +async def test_a_lock_lost_between_cancellations_stops_the_next_one(operation_env): + env = operation_env("revoke") + env.github.runs = {41: "in_progress", 42: "queued"} + env.github.on_cancel = lambda run_id: _lose_lock(env) + + result = await _process(env, fence_active_deploys=True) + + assert len(env.github.cancels()) == 1 + assert env.product.requests == [] + _assert_lock_lost(env, result) + + +@pytest.mark.asyncio +async def test_a_withdrawal_between_cancellations_stops_the_next_one_and_the_revoke( + operation_env, +): + env = operation_env("revoke") + env.github.runs = {41: "in_progress", 42: "queued"} + env.github.on_cancel = lambda run_id: env.api.withdraw() + + result = await _process(env, fence_active_deploys=True) + + assert len(env.github.cancels()) == 1 + assert env.product.requests == [] + _assert_withdrawn(env, result) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("deployed", [True, False], ids=["deployed", "undeployed"]) +@pytest.mark.parametrize("source", ["message-flag", "grant-run-evidence"]) +async def test_an_unproven_workflow_stop_fails_the_cleanup_before_any_effect( + operation_env, deployed, source +): + env = operation_env("revoke") + if not deployed: + env.api.allocations[TARGET_APPLICATION_ID] = [] + env.api.history = [_dispatched_grant_run(env)] + env.github.runs = {7: "in_progress"} + env.github.refuse_cancel = True + + await _process(env, fence_active_deploys=source == "message-flag") + + assert env.product.requests == [] + _assert_retryable_failure(env, "could not be proven stopped") + _assert_deployed_nothing(env) + + +@pytest.mark.asyncio +async def test_an_operation_owing_no_stop_starts_no_github_work(operation_env): + env = operation_env("revoke") + env.github.runs = {7: "in_progress"} + + result = await _process(env) + + assert result["status"] == "success" + assert env.github.requests == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("operation", "deployed"), + [("grant", True), ("revoke", True), ("revoke", False)], + ids=["grant", "revoke", "revoke-undeployed"], +) +async def test_a_withdrawal_during_the_target_read_prevents_every_effect( + operation_env, operation, deployed +): + env = operation_env(operation) + if not deployed: + env.api.allocations[TARGET_APPLICATION_ID] = [] + env.api.on_allocation_read = env.api.withdraw + + result = await _process(env, fence_active_deploys=True) + + assert "claim-refused" in env.api.events + assert env.product.requests == [] + assert env.github.writes() == [] + _assert_withdrawn(env, result) + + +def _dispatched_grant_run(env, run_id: str | None = None): + """A grant Run as a withdrawal leaves one that had crossed GitHub dispatch.""" + return _grant_run( + env.api.grant, run_id or env.api.grant.grant_run_id, RunStatus.CANCELLED, claimed=True + ) + + +def _redeliver(env) -> None: + """The next revoke Run of the same record, as a retry publishes it.""" + env.api.run_status = RunStatus.QUEUED + env.api.run_writes.clear() + env.api.events.clear() + env.api.grant_reads = 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("deployed", [True, False], ids=["deployed", "undeployed"]) +@pytest.mark.parametrize("first_attempt", ["unproven-stop", "lock-conflict"]) +async def test_a_cleanup_retry_still_stops_the_earlier_grant_workflow( + operation_env, deployed, first_attempt +): + """The retry arrives without the message flag; the grant Run still owes the stop.""" + env = operation_env("revoke") + if not deployed: + env.api.allocations[TARGET_APPLICATION_ID] = [] + env.api.history = [_dispatched_grant_run(env)] + env.github.runs = {7: "in_progress"} + if first_attempt == "unproven-stop": + env.github.refuse_cancel = True + else: + env.stream.redis.value = "another-deploy" + + first = await _process(env, fence_active_deploys=True) + + assert first["status"] in {"failed", "cancelled"} + assert not any(_proves_operation(write) for write in env.api.run_writes if "result" in write) + assert env.github.runs == {7: "in_progress"} + assert env.product.requests == [] + + _redeliver(env) + env.github.refuse_cancel = False + env.stream.redis.value = None + + result = await _process(env) + + assert result["status"] == "success" + assert env.github.runs == {7: "completed"} + if deployed: + assert env.api.events.index("cancel:7") < env.api.events.index("/users/revoke") + assert _proves_operation(env.api.run_writes[-1]) + + +#: The released grant operation the scheduler replaced; its workflow is still running. +PREDECESSOR_RUN_ID = "temporary-access-grant-a" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("deployed", [True, False], ids=["deployed", "undeployed"]) +@pytest.mark.parametrize("replacement", ["unclaimed", "completed"]) +async def test_a_cleanup_after_a_replaced_grant_still_stops_the_predecessors_workflow( + operation_env, deployed, replacement +): + """Replacement B is the record's grant Run now; admitted A's workflow can still act.""" + env = operation_env("revoke") + if not deployed: + env.api.allocations[TARGET_APPLICATION_ID] = [] + grant = env.api.grant + current = ( + _grant_run(grant, grant.grant_run_id, RunStatus.CANCELLED) + if replacement == "unclaimed" + else _grant_run( + grant, grant.grant_run_id, RunStatus.COMPLETED, claimed=True, succeeded=True + ) + ) + env.api.history = [current, _dispatched_grant_run(env, PREDECESSOR_RUN_ID)] + env.github.runs = {7: "in_progress"} + env.github.refuse_cancel = True + + first = await _process(env) + + assert env.github.cancels() == [7] + assert env.github.runs == {7: "in_progress"} + assert env.product.requests == [] + assert first["status"] == "failed" + _assert_retryable_failure(env, "could not be proven stopped") + + # A restart or retry recomputes the obligation from the same history. + _redeliver(env) + env.github.refuse_cancel = False + + result = await _process(env) + + assert result["status"] == "success" + assert env.github.runs == {7: "completed"} + if deployed: + assert env.api.events.index("cancel:7") < env.api.events.index("/users/revoke") + assert _proves_operation(env.api.run_writes[-1]) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("deployed", [True, False], ids=["deployed", "undeployed"]) +async def test_an_unreadable_run_history_never_proves_cleanup(operation_env, deployed): + env = operation_env("revoke") + if not deployed: + env.api.allocations[TARGET_APPLICATION_ID] = [] + env.api.history_unreadable = True + + result = await _process(env) + + assert result["status"] == "failed" + assert env.product.requests == [] + assert env.github.requests == [] + [write] = env.api.run_writes + assert not _proves_operation(write) + + +@pytest.mark.asyncio +async def test_a_history_without_the_current_grant_run_owes_the_stop(operation_env): + env = operation_env("revoke") + env.api.history = [] + env.github.runs = {7: "in_progress"} + + result = await _process(env) + + assert result["status"] == "success" + assert env.github.cancels() == [7] + assert env.api.events.index("cancel:7") < env.api.events.index("/users/revoke") diff --git a/services/langgraph/tests/unit/test_deploy_worker_proactive.py b/services/langgraph/tests/unit/test_deploy_worker_proactive.py index 7ea5a0e36..cc206687c 100644 --- a/services/langgraph/tests/unit/test_deploy_worker_proactive.py +++ b/services/langgraph/tests/unit/test_deploy_worker_proactive.py @@ -7,18 +7,11 @@ from __future__ import annotations -from datetime import UTC, datetime -from types import SimpleNamespace from unittest.mock import AsyncMock, patch import pytest from shared.contracts.dto.application import ApplicationStatus -from shared.contracts.dto.temporary_access import ( - TemporaryAccessGrantDTO, - TemporaryAccessRevokeReason, - TemporaryAccessStatus, -) from shared.contracts.queues.deploy import DeployTrigger from shared.queues import PO_INPUT_QUEUE, PO_PROACTIVE_QUEUE from tests.unit.factories import make_project, make_repository, make_run, make_run_start @@ -68,15 +61,10 @@ def mock_api(): @pytest.fixture def mock_allocations(): - """Patch allocation lookup (lazy import inside process_deploy_job). - - A product deploy gets or creates allocations; a temporary-access operation - only reads them. Both answer from the same mock here. - """ + """Patch allocation lookup (lazy import inside process_deploy_job).""" mock_fn = AsyncMock(return_value={"server_ip": "1.2.3.4", "port": 8080}) with ( patch("src.allocations.ensure_project_allocations", mock_fn), - patch("src.allocations.existing_application_allocations", mock_fn), patch("src.allocations.AllocationError", Exception), ): yield mock_fn @@ -259,96 +247,6 @@ async def test_deploy_worker_skips_same_sha_for_running_application( ) -@pytest.mark.asyncio -@pytest.mark.parametrize("operation", ["grant", "revoke"]) -async def test_temporary_access_operation_never_takes_the_same_sha_shortcut( - mock_redis, mock_api, mock_allocations, mock_devops_subgraph, operation -): - """A capability redeploy of the running commit still reaches the product.""" - head_sha = "a" * 40 - now = datetime.now(UTC) - grant = TemporaryAccessGrantDTO( - id="tempaccess-qa-1", - project_id="proj-1", - channel="telegram", - external_id="8202532144", - target_application_id=17, - target_base_url="https://exact.example.com", - head_sha=head_sha, - qa_run_id="qa-1", - grant_run_id="deploy-wh-abc" if operation == "grant" else "temporary-access-grant-1", - revoke_run_id="deploy-wh-abc" if operation == "revoke" else None, - revoke_reason=TemporaryAccessRevokeReason.RUN_TERMINAL if operation == "revoke" else None, - qa_message={ - "story_id": "story-1", - "project_id": "proj-1", - "initiating_run_id": "deploy-1", - "telegram_chat_id": "", - "deployed_url": "https://exact.example.com", - "application_id": 17, - "acceptance_criteria": "bot admission", - "run_id": "qa-1", - }, - status=( - TemporaryAccessStatus.GRANTING - if operation == "grant" - else TemporaryAccessStatus.REVOKING - ), - granted_at=now, - created_at=now, - ) - mock_api.get_run = AsyncMock( - return_value=make_run( - run_metadata={ - "temporary_access_grant_id": grant.id, - "temporary_access_operation": operation, - } - ) - ) - mock_api.get_temporary_access_grant = AsyncMock(return_value=grant) - mock_allocations.return_value = {"vps-1:8080": {"application_id": 17, "port": 8080}} - mock_api.get = AsyncMock( - return_value=[{"application_id": 17, "deployed_sha": head_sha, "result": "success"}] - ) - # The grant's recorded target, whose allocations the operation reads. - mock_api.get_application = AsyncMock( - return_value=SimpleNamespace(id=17, server_handle="vps-1", status=ApplicationStatus.RUNNING) - ) - mock_devops_subgraph.ainvoke = AsyncMock( - return_value={ - "deployed_url": "https://exact.example.com", - "application_id": 17, - "deployment_result": {}, - "secret_values": {"USERS_GRANT_CAPABILITY": "capability-value"}, - } - ) - - from src.consumers.deploy import process_deploy_job - - with patch("src.consumers.deploy_result_handler.GeneratedServiceGrantClient") as client: - client.return_value.grant_and_resolve = AsyncMock( - return_value=SimpleNamespace(active=True, failure=None) - ) - client.return_value.revoke_and_resolve = AsyncMock( - return_value=SimpleNamespace(active=False, failure=None) - ) - result = await process_deploy_job(_job(), mock_redis) - - assert result.get("reason") != "already_deployed_same_sha" - mock_devops_subgraph.ainvoke.assert_awaited_once() - applied = getattr(client.return_value, f"{operation}_and_resolve") - applied.assert_awaited_once_with( - channel="telegram", external_id="8202532144", capability="capability-value" - ) - run_results = [ - call.kwargs["json"].get("result") or {} - for call in mock_api.patch.call_args_list - if call.args[0] == "runs/deploy-wh-abc" - ] - assert all(run_result.get("skipped_reason") is None for run_result in run_results) - assert result["status"] == "success" - - @pytest.mark.asyncio async def test_deploy_worker_redeploys_same_sha_after_undeploy( mock_redis, mock_api, mock_allocations, mock_devops_subgraph diff --git a/services/langgraph/tests/unit/test_deploy_workflow_fence.py b/services/langgraph/tests/unit/test_deploy_workflow_fence.py new file mode 100644 index 000000000..96206fbb9 --- /dev/null +++ b/services/langgraph/tests/unit/test_deploy_workflow_fence.py @@ -0,0 +1,84 @@ +"""Stopping earlier deploy workflows checks the deploy lock before every cancellation. + +`fence_active_deploys` runs the real `GitHubAppClient.fence_workflow` against a +GitHub Actions fake served through `respx`; the deploy lock is an in-memory fake. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock + +import httpx +import pytest +import respx + +from shared.clients.github import GitHubAppClient +from src.deploy_fence import DeployFence, DeployFenceLost +from src.subgraphs.devops.deployer import fence_active_deploys + +HOLDER = "deploy-a:1" + + +class Lock: + def __init__(self) -> None: + self.value: str | None = HOLDER + + async def eval(self, script: str, numkeys: int, key: str, token: str) -> int: + return int(self.value == token) + + +class Actions: + """Two unfinished runs; a cancelled run reads completed afterwards.""" + + def __init__(self, lock: Lock) -> None: + self.runs = {41: "in_progress", 42: "queued"} + self.cancelled: list[int] = [] + self.lose_lock_on_cancel = False + self.lock = lock + + def handle(self, request: httpx.Request) -> httpx.Response: + path = request.url.path + if path.endswith("/runs"): + status = request.url.params["status"] + listed = [{"id": i, "status": s} for i, s in self.runs.items() if s == status] + return httpx.Response(200, json={"workflow_runs": listed}) + if path.endswith("/cancel"): + run_id = int(path.split("/")[-2]) + self.cancelled.append(run_id) + self.runs[run_id] = "completed" + if self.lose_lock_on_cancel: + self.lock.value = "deploy-b:1" + return httpx.Response(202) + run_id = int(path.rsplit("/", 1)[-1]) + return httpx.Response(200, json={"id": run_id, "status": self.runs[run_id]}) + + +async def _fence(actions: Actions, lock: Lock) -> list[int]: + github = GitHubAppClient() + github.get_token = AsyncMock(return_value="github-token") + with respx.mock(assert_all_called=False) as mock: + mock.route(host="api.github.com").mock(side_effect=actions.handle) + return await fence_active_deploys( + github, "org", "product", "proj-1", DeployFence(lock, "proj-1", HOLDER) + ) + + +@pytest.mark.asyncio +async def test_a_held_lock_stops_every_unfinished_run(): + lock = Lock() + actions = Actions(lock) + + assert sorted(await _fence(actions, lock)) == [41, 42] + assert sorted(actions.cancelled) == [41, 42] + + +@pytest.mark.asyncio +async def test_a_lock_lost_between_cancellations_refuses_the_next_one_as_lock_lost(): + lock = Lock() + actions = Actions(lock) + actions.lose_lock_on_cancel = True + + with pytest.raises(DeployFenceLost): + await _fence(actions, lock) + + assert len(actions.cancelled) == 1 diff --git a/services/langgraph/tests/unit/test_deployer.py b/services/langgraph/tests/unit/test_deployer.py index 18583fdd1..70f0ce715 100644 --- a/services/langgraph/tests/unit/test_deployer.py +++ b/services/langgraph/tests/unit/test_deployer.py @@ -1030,7 +1030,10 @@ async def test_the_payload_is_replaced_before_earlier_runs_are_stopped( result = await deployer.run({**base_state, "fence_active_deploys": True}) assert order == ["secrets", "fence", "dispatch"] - gh.fence_workflow.assert_awaited_once_with("my-org", "my-repo", "deploy.yml") + gh.fence_workflow.assert_awaited_once() + assert gh.fence_workflow.await_args.args == ("my-org", "my-repo", "deploy.yml") + # Each cancellation is guarded by the deploy lock (test_deploy_workflow_fence.py). + assert callable(gh.fence_workflow.await_args.kwargs["guard"]) assert result["deployment_result"]["status"] == "success" @pytest.mark.asyncio diff --git a/services/scheduler/src/tasks/temporary_access.py b/services/scheduler/src/tasks/temporary_access.py index 816c5171b..2fe5e07e1 100644 --- a/services/scheduler/src/tasks/temporary_access.py +++ b/services/scheduler/src/tasks/temporary_access.py @@ -9,7 +9,6 @@ import httpx import structlog -from shared.contracts.dto.deploy_dispatch import DispatchWithdrawal from shared.contracts.dto.run import RunStatus, RunType from shared.contracts.dto.run_result import QABlocker, QABlockerCategory, QARunResult from shared.contracts.dto.temporary_access import ( @@ -24,6 +23,7 @@ from shared.notifications import notify_admins_best_effort from shared.queues import DEPLOY_QUEUE, QA_QUEUE from shared.redis import RedisStreamClient +from shared.temporary_access_cleanup import grant_cleanup_owes_workflow_stop from .. import startup @@ -231,17 +231,18 @@ async def _publish_operation( grant, run_id: str, operation: str, - *, - fence_active_deploys: bool = False, ) -> None: - """Redeploy the running artifact to write or take away the QA capability. + """Queue one grant or revoke operation Run for the deploy consumer. - The grant records the story commit it belongs to; what has to be redeployed - is the artifact the target is running, and only the target's own deployment - record says which commit that is. Reading it here rather than reusing - `grant.head_sha` is the difference between redeploying the same images and - asking for images of a commit whose CI never published any. + The consumer calls the target's users-core capability directly. The message + also names the commit the target's own deployment record says it runs. Every + revoke re-derives from the record's whole grant Run history whether an earlier + grant workflow must be stopped first, so no retry, replacement or redispatch + loses that obligation. """ + fence_active_deploys = operation == "revoke" and await grant_cleanup_owes_workflow_stop( + api_client, grant + ) deployed_commit_sha = await api_client.latest_deployed_commit_sha(grant.target_application_id) if not deployed_commit_sha: raise RuntimeError( @@ -434,20 +435,11 @@ async def _dispatch_revoke(api_client, redis_client, grant, reason) -> None: ), ) # Close the old grant dispatch only after the durable transition says it is - # obsolete, and before beginning cleanup. A grant that already crossed the - # deploy boundary makes cleanup fenced so Actions cannot later restore it. - withdrawal = await api_client.withdraw_deploy_dispatch( + # obsolete, and before beginning cleanup. + await api_client.withdraw_deploy_dispatch( grant.grant_run_id, "temporary access cleanup superseded grant operation" ) - await _publish_operation( - api_client, - redis_client, - grant, - run_id, - "revoke", - fence_active_deploys=getattr(withdrawal, "outcome", None) - is DispatchWithdrawal.ALREADY_DISPATCHED, - ) + await _publish_operation(api_client, redis_client, grant, run_id, "revoke") async def _settle_revoke(api_client, redis_client, grant, counts, log) -> None: diff --git a/services/scheduler/tests/unit/test_temporary_access_capability.py b/services/scheduler/tests/unit/test_temporary_access_capability.py index 0cdef01aa..e147a23ef 100644 --- a/services/scheduler/tests/unit/test_temporary_access_capability.py +++ b/services/scheduler/tests/unit/test_temporary_access_capability.py @@ -2,14 +2,14 @@ from datetime import UTC, datetime, timedelta from types import SimpleNamespace -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import httpx import pytest from structlog.testing import capture_logs -from shared.contracts.dto.deploy_dispatch import DispatchWithdrawal -from shared.contracts.dto.run import RunStatus +from shared.contracts.dto.deploy_dispatch import DISPATCH_CLAIMED_AT_KEY, DispatchWithdrawal +from shared.contracts.dto.run import RunStatus, RunType from shared.contracts.dto.run_result import DeployRunResult, DeploySkipReason, QABlockerCategory from shared.contracts.dto.temporary_access import ( TemporaryAccessGrantDTO, @@ -74,9 +74,11 @@ def _operation_run( outcome: DeployOutcome | None = None, age_minutes: int = 0, skipped_reason: DeploySkipReason | None = None, + run_metadata: dict | None = None, ): return SimpleNamespace( status=status, + run_metadata=run_metadata or {}, result=( SimpleNamespace(deploy_outcome=outcome, skipped_reason=skipped_reason) if outcome is not None @@ -455,7 +457,13 @@ async def test_cleanup_transitions_then_withdraws_and_fences_an_old_grant_dispat # redeploy has to ask for again. api.latest_deployed_commit_sha = AsyncMock(return_value="e" * 40) api.list_temporary_access_grants_under_watch.return_value = [grant] + # The QA run has ended, so cleanup starts. api.get_run_if_missing_returns_none.return_value = _operation_run(RunStatus.COMPLETED) + # The grant Run as the withdrawal left it: it had crossed the dispatch + # boundary and is now cancelled with its workflow possibly still running. + api.list_project_runs.return_value = [ + _grant_operation_run(grant.grant_run_id, RunStatus.CANCELLED, claimed=True) + ] api.withdraw_deploy_dispatch.return_value = SimpleNamespace( outcome=DispatchWithdrawal.ALREADY_DISPATCHED ) @@ -766,7 +774,7 @@ async def test_revoke_that_found_no_deployment_settles_revoked_without_retry() - api = AsyncMock() api.latest_deployed_commit_sha = AsyncMock(return_value="e" * 40) api.get_run_if_missing_returns_none.return_value = SimpleNamespace( - status=RunStatus.COMPLETED, created_at=datetime.now(UTC) + status=RunStatus.COMPLETED, created_at=datetime.now(UTC), run_metadata={} ) api.withdraw_deploy_dispatch.return_value = SimpleNamespace( outcome=DispatchWithdrawal.ALREADY_DISPATCHED @@ -807,3 +815,164 @@ async def test_revoke_that_found_no_deployment_settles_revoked_without_retry() - assert counts["revoked"] == 1 assert counts["revoke_failed"] == 0 assert counts["escalated"] == 0 + + +def _revoking(**overrides) -> TemporaryAccessGrantDTO: + return _grant( + status=TemporaryAccessStatus.REVOKING, + revoke_reason=TemporaryAccessRevokeReason.RUN_TERMINAL, + revoke_run_id="temporary-access-revoke-1", + revoke_attempts=1, + **overrides, + ) + + +def _grant_operation_run( + run_id: str, status: RunStatus, *, claimed: bool = False, succeeded: bool = False +): + """A grant operation Run of the record `_grant` builds, as the run listing returns it.""" + metadata = { + "temporary_access_grant_id": "tempaccess-qa-1", + "temporary_access_operation": "grant", + } + if claimed: + metadata[DISPATCH_CLAIMED_AT_KEY] = datetime.now(UTC).isoformat() + run = _operation_run( + status, outcome=DeployOutcome.SUCCESS if succeeded else None, run_metadata=metadata + ) + run.id = run_id + return run + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "revoke_run", + [ + pytest.param( + _operation_run(RunStatus.FAILED, outcome=DeployOutcome.OWNER_ACCESS_PROOF_FAILED), + id="unproven-stop-failure-retry", + ), + pytest.param( + _operation_run(RunStatus.CANCELLED, outcome=DeployOutcome.CANCELLED), + id="lock-conflict-redispatch", + ), + ], +) +async def test_every_revoke_retry_keeps_the_earlier_grant_workflow_obligation(revoke_run) -> None: + """A retry is fenced again although the grant Run withdraws as already terminal.""" + from src.tasks.temporary_access import _counts, _settle_revoke + + grant = _revoking() + api = AsyncMock() + api.latest_deployed_commit_sha = AsyncMock(return_value="e" * 40) + api.get_run_if_missing_returns_none = AsyncMock(return_value=revoke_run) + api.list_project_runs = AsyncMock( + return_value=[_grant_operation_run(grant.grant_run_id, RunStatus.CANCELLED, claimed=True)] + ) + api.withdraw_deploy_dispatch.return_value = SimpleNamespace( + outcome=DispatchWithdrawal.ALREADY_TERMINAL + ) + api.update_temporary_access_grant = AsyncMock( + side_effect=lambda grant_id, update: grant.model_copy( + update=update.model_dump(exclude_unset=True) + ) + ) + redis = AsyncMock() + counts = _counts() + + await _settle_revoke(api, redis, grant, counts, MagicMock()) + + [(queue, published)] = [call.args for call in redis.publish_message.await_args_list] + assert queue == DEPLOY_QUEUE + assert published.fence_active_deploys is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("history", "fenced"), + [ + pytest.param( + [ + _grant_operation_run( + "temporary-access-grant-old", RunStatus.COMPLETED, claimed=True, succeeded=True + ) + ], + False, + id="dispatched-and-finished", + ), + pytest.param( + [_grant_operation_run("temporary-access-grant-old", RunStatus.CANCELLED)], + False, + id="withdrawn-before-dispatch", + ), + pytest.param([], True, id="grant-run-missing-from-history"), + ], +) +async def test_a_revoke_is_fenced_unless_the_grant_history_proves_no_workflow_can_act( + history, fenced +) -> None: + from src.tasks.temporary_access import _counts, _redispatch_revoke_without_effect + + grant = _revoking() + api = AsyncMock() + api.latest_deployed_commit_sha = AsyncMock(return_value="e" * 40) + api.list_project_runs = AsyncMock(return_value=history) + redis = AsyncMock() + + await _redispatch_revoke_without_effect(api, redis, grant, _counts()) + + api.list_project_runs.assert_awaited_once_with(grant.project_id, RunType.DEPLOY) + assert redis.publish_message.await_args.args[1].fence_active_deploys is fenced + + +@pytest.mark.asyncio +@pytest.mark.parametrize("replacement", ["unclaimed", "completed"]) +async def test_a_replaced_grant_run_keeps_its_predecessors_workflow_obligation( + replacement, +) -> None: + """A stale admitted grant A is replaced by B; cleanup still owes A's workflow a stop.""" + from src.tasks.temporary_access import _counts, _dispatch_revoke, _retry_grant_operation + + state = {"grant": _grant(grant_run_id="temporary-access-grant-a")} + history = [_grant_operation_run("temporary-access-grant-a", RunStatus.RUNNING, claimed=True)] + + def update(grant_id, change): + state["grant"] = state["grant"].model_copy(update=change.model_dump(exclude_unset=True)) + return state["grant"] + + def withdraw(run_id, reason): + [run] = [run for run in history if run.id == run_id] + if run.status in (RunStatus.COMPLETED, RunStatus.FAILED, RunStatus.CANCELLED): + return SimpleNamespace(outcome=DispatchWithdrawal.ALREADY_TERMINAL) + claimed = DISPATCH_CLAIMED_AT_KEY in run.run_metadata + run.status = RunStatus.CANCELLED + return SimpleNamespace( + outcome=DispatchWithdrawal.ALREADY_DISPATCHED + if claimed + else DispatchWithdrawal.WITHDRAWN + ) + + def create_run(data): + history.append(_grant_operation_run(data["id"], RunStatus.QUEUED)) + + api = AsyncMock() + api.latest_deployed_commit_sha = AsyncMock(return_value="e" * 40) + api.update_temporary_access_grant = AsyncMock(side_effect=update) + api.withdraw_deploy_dispatch = AsyncMock(side_effect=withdraw) + api.create_run = AsyncMock(side_effect=create_run) + api.list_project_runs = AsyncMock(side_effect=lambda *_: list(history)) + redis = AsyncMock() + + await _retry_grant_operation(api, redis, state["grant"], "stale", _counts(), MagicMock()) + replacement_id = state["grant"].grant_run_id + assert replacement_id != "temporary-access-grant-a" + if replacement == "completed": + # The direct grant crossed admission and proved access. + history[-1] = _grant_operation_run( + replacement_id, RunStatus.COMPLETED, claimed=True, succeeded=True + ) + redis.publish_message.reset_mock() + + await _dispatch_revoke(api, redis, state["grant"], TemporaryAccessRevokeReason.RUN_TERMINAL) + + assert redis.publish_message.await_args.args[1].fence_active_deploys is True diff --git a/shared/clients/github/__init__.py b/shared/clients/github/__init__.py index d12c8f4d2..416f98d08 100644 --- a/shared/clients/github/__init__.py +++ b/shared/clients/github/__init__.py @@ -7,6 +7,8 @@ ) from ._base import ( GitHubAppClientBase, + RequestGuard, + RequestGuardRefusedError, WorkflowCancellationUnprovenError, WorkflowCancelledError, WorkflowNotFoundError, @@ -51,6 +53,8 @@ class GitHubAppClient( "UNFINISHED_WORKFLOW_RUN_STATUSES", "WorkflowRunListingIncompleteError", "GitHubAppClient", + "RequestGuard", + "RequestGuardRefusedError", "deploy_pin_tag", "WorkflowCancelledError", "WorkflowCancellationUnprovenError", diff --git a/shared/clients/github/_actions.py b/shared/clients/github/_actions.py index f788509d7..df7e3b2a1 100644 --- a/shared/clients/github/_actions.py +++ b/shared/clients/github/_actions.py @@ -8,6 +8,8 @@ from shared.log_config import get_logger from ._base import ( + RequestGuard, + RequestGuardRefusedError, WorkflowCancellationUnprovenError, WorkflowCancelledError, WorkflowNotFoundError, @@ -481,6 +483,7 @@ async def fence_workflow( *, timeout_seconds: int = 300, poll_interval: int = 5, + guard: RequestGuard | None = None, ) -> list[int]: """Stop every run of *workflow_file* that exists now, and return the ones stopped. @@ -496,6 +499,10 @@ async def fence_workflow( needs a value gone confirms that by reading the deployed service and repeating itself; this only makes the wait shorter. + `guard` is awaited immediately before each cancellation request, and its + own exception is raised unchanged: a caller that may no longer write + stops here, and that is not an unproven cancellation. + Raises: WorkflowCancellationUnprovenError: at least one run could not be proven terminal, so the caller must not treat its own effect as @@ -512,10 +519,12 @@ async def fence_workflow( for run in active: run_id = run["id"] try: - await self.cancel_workflow_run(owner, repo, run_id) + await self.cancel_workflow_run(owner, repo, run_id, guard=guard) await self._wait_for_terminal_workflow_run( owner, repo, run_id, timeout_seconds, poll_interval ) + except RequestGuardRefusedError as refused: + raise refused.error from None except Exception as exc: raise WorkflowCancellationUnprovenError( f"Workflow {workflow_file} run {run_id} in {owner}/{repo} could not be " @@ -807,7 +816,9 @@ async def _wait_for_cancelled_workflow_run( raise WorkflowCancelledError(f"Workflow {workflow_file} cancelled by teardown") await asyncio.sleep(poll_interval) - async def cancel_workflow_run(self, owner: str, repo: str, run_id: int) -> None: + async def cancel_workflow_run( + self, owner: str, repo: str, run_id: int, *, guard: RequestGuard | None = None + ) -> None: """Request GitHub Actions to stop one known workflow run.""" token = await self.get_token(owner, repo) try: @@ -818,6 +829,7 @@ async def cancel_workflow_run(self, owner: str, repo: str, run_id: int) -> None: "Authorization": f"token {token}", "Accept": "application/vnd.github+json", }, + guard=guard, ) except httpx.HTTPStatusError as exc: if exc.response.status_code != httpx.codes.CONFLICT: diff --git a/shared/clients/github/_base.py b/shared/clients/github/_base.py index f589a9a97..1714e47f4 100644 --- a/shared/clients/github/_base.py +++ b/shared/clients/github/_base.py @@ -1,4 +1,5 @@ import asyncio +from collections.abc import Awaitable, Callable from datetime import UTC, datetime, timedelta import os import time @@ -24,6 +25,18 @@ class WorkflowCancellationUnprovenError(RuntimeError): """Raised when teardown cannot prove a GitHub Actions run has stopped.""" +class RequestGuardRefusedError(Exception): + """A caller's write guard refused a request; `error` is the guard's own exception.""" + + def __init__(self, error: Exception) -> None: + super().__init__(str(error)) + self.error = error + + +#: Awaited immediately before each attempt of a guarded write, after every other await. +RequestGuard = Callable[[], Awaitable[None]] + + class GitHubAppClientBase: """Core GitHub App authentication and HTTP request handling.""" @@ -64,16 +77,23 @@ async def _make_request( method: str, url: str, headers: dict[str, str], + *, + guard: RequestGuard | None = None, **kwargs: Any, ) -> httpx.Response: - """Make HTTP request with rate limit handling.""" + """Make HTTP request with rate limit handling. + + A `guard` refusal raises `RequestGuardRefusedError` and sends nothing. + """ if self._http_client is not None: return await self._request_with_client( - self._http_client, method, url, headers, **kwargs + self._http_client, method, url, headers, guard=guard, **kwargs ) async with httpx.AsyncClient() as client: - return await self._request_with_client(client, method, url, headers, **kwargs) + return await self._request_with_client( + client, method, url, headers, guard=guard, **kwargs + ) async def _request_with_client( self, @@ -81,11 +101,18 @@ async def _request_with_client( method: str, url: str, headers: dict[str, str], + *, + guard: RequestGuard | None = None, **kwargs: Any, ) -> httpx.Response: max_retries = 3 for attempt in range(max_retries): + if guard is not None: + try: + await guard() + except Exception as error: + raise RequestGuardRefusedError(error) from error try: resp = await client.request(method, url, headers=headers, **kwargs) except httpx.RequestError: diff --git a/shared/clients/run_api.py b/shared/clients/run_api.py index d69154774..4da764c7d 100644 --- a/shared/clients/run_api.py +++ b/shared/clients/run_api.py @@ -7,7 +7,7 @@ import httpx -from shared.contracts.dto.run import RunDTO, RunUpdate +from shared.contracts.dto.run import RunDTO, RunType, RunUpdate def _run_update_payload(update: RunUpdate | Mapping[str, Any]) -> dict[str, Any]: @@ -28,6 +28,13 @@ async def get_run(self, run_id: str) -> RunDTO: response = await self.request("GET", f"runs/{run_id}") # type: ignore[attr-defined] return RunDTO.model_validate(response.json()) + async def list_project_runs(self, project_id: str, run_type: RunType) -> list[RunDTO]: + """Every Run of one project and type, newest first; the API returns the whole set.""" + response = await self.request( # type: ignore[attr-defined] + "GET", "runs/", params={"project_id": project_id, "run_type": run_type.value} + ) + return [RunDTO.model_validate(row) for row in response.json()] + async def update_run(self, run_id: str, update: RunUpdate | Mapping[str, Any]) -> None: await self.request( # type: ignore[attr-defined] "PATCH", diff --git a/shared/temporary_access_cleanup.py b/shared/temporary_access_cleanup.py new file mode 100644 index 000000000..b462825fc --- /dev/null +++ b/shared/temporary_access_cleanup.py @@ -0,0 +1,54 @@ +"""Whether a temporary-access cleanup must first stop an earlier grant's deploy workflow.""" + +from __future__ import annotations + +from collections.abc import Iterable + +from shared.clients.run_api import RunAPIClientMixin +from shared.contracts.dto.deploy_dispatch import DISPATCH_CLAIMED_AT_KEY +from shared.contracts.dto.run import RunDTO, RunStatus, RunType +from shared.contracts.dto.temporary_access import TemporaryAccessGrantDTO +from shared.contracts.queues.deploy import DeployOutcome + + +def _may_still_act(grant_run: RunDTO) -> bool: + """A Run that crossed the dispatch boundary is outside until it completed SUCCESS.""" + if DISPATCH_CLAIMED_AT_KEY not in grant_run.run_metadata: + return False + return not ( + grant_run.status is RunStatus.COMPLETED + and grant_run.result is not None + and grant_run.result.deploy_outcome is DeployOutcome.SUCCESS + ) + + +def _grant_work_may_still_act( + grant: TemporaryAccessGrantDTO, deploy_runs: Iterable[RunDTO] +) -> bool: + """Whether any grant operation of this record may have left a workflow that can still write. + + Every grant operation Run of the record counts, not only the one it names now: + replacing the operation id stops nothing. A history without the current grant + Run proves nothing, so it owes the stop. + """ + grant_runs = [ + run + for run in deploy_runs + if run.run_metadata.get("temporary_access_grant_id") == grant.id + and run.run_metadata.get("temporary_access_operation") == "grant" + ] + if grant.grant_run_id not in {run.id for run in grant_runs}: + return True + return any(_may_still_act(run) for run in grant_runs) + + +async def grant_cleanup_owes_workflow_stop( + api_client: RunAPIClientMixin, grant: TemporaryAccessGrantDTO +) -> bool: + """Read the project's complete deploy Run history and judge every grant operation in it. + + An unreadable history raises; it is never an empty one. + """ + return _grant_work_may_still_act( + grant, await api_client.list_project_runs(grant.project_id, RunType.DEPLOY) + ) diff --git a/shared/tests/clients/test_github.py b/shared/tests/clients/test_github.py index f23e30825..a84c6f2e6 100644 --- a/shared/tests/clients/test_github.py +++ b/shared/tests/clients/test_github.py @@ -1468,3 +1468,44 @@ async def test_scoped_cache_refreshes_inside_five_minutes(client, mock_jwt): assert await client.get_repo_scoped_token("my-org", "repo-a") == "synthetic-current" assert await client.get_repo_scoped_token("my-org", "repo-a") == "synthetic-current" assert route.call_count == 1 + + +@pytest.mark.asyncio +async def test_fence_guards_each_cancellation_and_raises_its_refusal_unchanged(authed_client): + """The guard runs before every cancel request; its own refusal is not 'unproven'.""" + + class Refused(Exception): + pass + + authed_client.get_token = AsyncMock(return_value="token") + cancelled: list[int] = [] + states = {42: "in_progress", 43: "queued"} + + def handle(request: httpx.Request) -> httpx.Response: + path = request.url.path + if path.endswith("/runs"): + status = request.url.params["status"] + listed = [{"id": i, "status": s} for i, s in states.items() if s == status] + return httpx.Response(200, json={"workflow_runs": listed}) + if path.endswith("/cancel"): + run_id = int(path.split("/")[-2]) + cancelled.append(run_id) + states[run_id] = "completed" + return httpx.Response(202) + run_id = int(path.rsplit("/", 1)[-1]) + return httpx.Response(200, json={"id": run_id, "status": states[run_id]}) + + checks: list[int] = [] + + async def guard() -> None: + checks.append(len(cancelled)) + if cancelled: + raise Refused("lock lost") + + with respx.mock(assert_all_called=False) as mock: + mock.route(host="api.github.com").mock(side_effect=handle) + with pytest.raises(Refused): + await authed_client.fence_workflow("my-org", "my-repo", "deploy.yml", guard=guard) + + assert checks == [0, 1] + assert len(cancelled) == 1 diff --git a/tests/live/pipeline_helpers.py b/tests/live/pipeline_helpers.py index bfe81a892..c8412b108 100644 --- a/tests/live/pipeline_helpers.py +++ b/tests/live/pipeline_helpers.py @@ -5396,9 +5396,8 @@ async def record_stage_notices( async def _project_deploy_runs(api_internal: httpx.AsyncClient, project_id: str) -> list[dict]: """Every deploy Run of this project, not only the ones a story owns. - The capability deploys carry no story — `temporary_access` sends its - `DeployMessage` with `story_id=""` — so `_story_runs` cannot see them, and - they are exactly the deploys whose records have to be accounted for here. + A deploy without a story, such as the owner-grant deploy of a fresh project, + is invisible to `_story_runs`, and its record still has to be accounted for. """ require_unscoped_run_observer(api_internal) response = await api_internal.get( @@ -5444,26 +5443,16 @@ async def wait_service_deployment( """Select the deployment this application is serving, out of a log of deploys. `service_deployments` is an immutable log of deploy attempts - (`shared/models/deployment.py`), and one level-1 run fills it more than - once. Run 35451082771 failed here with `ambiguous deployments for - application 1 … [2, 1]`, and the two rows were written by two different - deploys of the same commit: `deploy-grant-71672573103249b09aaf4cb3dbdf2a54` - at 15:48:00 and `temporary-access-grant-a6a508bec27d` at 15:49:14, the - redeploy that hands QA its temporary capability. Both carry - `service_name=stand-t-477db6f21739461cb11fa8e6f94cc169` — the project's - runtime slug, which is the only service name a deploy record ever holds, - whether the product has one module or two. So "exactly one row per - application" was never about modules: it was an assertion about a product - that deploys once, and this product deploys again whenever a capability is - granted or revoked. - - What the rows cannot say by themselves is which deploy wrote them — the - record carries no run id — so the guard the old assertion existed for is - kept by counting them against the deploys that could have written them. Each - deploy Run of this project writes at most one record and a stop/undeploy Run - writes none, so more records than deploys is the same deploy recorded twice, - and that still fails here by name. Taking the first row instead, or dropping - the count, would have made a genuine duplicate invisible. + (`shared/models/deployment.py`), and one level-1 run can fill it more than + once: every deploy of the same commit, a retry or a fix loop, writes its own + row with the project's runtime slug as `service_name`. A temporary QA access + grant or revoke deploys nothing and writes none. + + The records carry no run id, so a duplicate is caught by counting them + against the deploys that could have written them. Each deploy Run of this + project writes at most one record and a stop/undeploy Run writes none, so + more records than deploys is the same deploy recorded twice, and that fails + here by name. The row that is selected is the newest one, because that is the deploy the application is actually running, and it still has to be a success carrying diff --git a/tests/live/test_pipeline_lifecycle_contract.py b/tests/live/test_pipeline_lifecycle_contract.py index afa342031..62ec76c58 100644 --- a/tests/live/test_pipeline_lifecycle_contract.py +++ b/tests/live/test_pipeline_lifecycle_contract.py @@ -558,14 +558,7 @@ async def test_service_deployment_selects_exact_application_project_and_sha(): @pytest.mark.asyncio async def test_two_deploys_of_one_application_are_a_log_not_an_ambiguity(): - """What run 35451082771 actually had, and what the old assertion called ambiguous. - - The story's own grant deploy wrote one record; the redeploy that hands QA - its temporary capability wrote a second one for the same application, the - same project and the same merged commit — `service_name` is the project's - runtime slug in both, whatever the product's modules are. Two deploys, two - records, and the one the application is serving is the later of them. - """ + """Two deploys of the same commit write two records; the later one is served.""" ctx = {"application_id": 7, "project_id": "project-1", "deploy_head_sha": "abc123"} handler, _ = _deployment_and_runs( [ @@ -574,7 +567,7 @@ async def test_two_deploys_of_one_application_are_a_log_not_an_ambiguity(): ], [ {"id": "deploy-grant-7167257310", "type": "deploy"}, - {"id": "temporary-access-grant-a6a508be", "type": "deploy"}, + {"id": "deploy-retry-7167257310", "type": "deploy"}, ], )