diff --git a/.mergequeue.example.toml b/.mergequeue.example.toml index 50c3815..81b93d4 100644 --- a/.mergequeue.example.toml +++ b/.mergequeue.example.toml @@ -49,6 +49,7 @@ merge_to_live_target_minutes = 10 auto_promote = true intent_scope = "head" # safest: a trusted source agent refreshes replacement heads pause_on_failure = true +# Receives best-effort events, including retryable thread-deployed messages. # webhook_url_env = "DEPLOYBOT_WEBHOOK_URL" # [[pipeline.verifications]] diff --git a/README.md b/README.md index f939246..d2437a3 100644 --- a/README.md +++ b/README.md @@ -149,9 +149,10 @@ they cannot create the original per-pull-request deploy intent. ## Delivery controller -`deploybot status` reports active metadata-only agent threads, every PR stage, -deploy requests, queue order, overlaps, exact-`main` CI, deployment, and pipeline -pause state. It never stores prompts, transcripts, source, or credentials. +`deploybot status` reports active metadata-only agent threads, pending native +notifications, every PR stage, deploy requests, queue order, overlaps, +exact-`main` CI, deployment, and pipeline pause state. It never stores prompts, +transcripts, source, or credentials. `deploybot react` promotes ready intent, skips blockers, drains independent work, and creates integration PRs when configured. A conflict produces a repair @@ -171,6 +172,28 @@ and emits a new wake-up event. `follow` tracks newer cumulative `main` revisions until exact CI, deployment, and optional HTTP checks pass. A CI or deploy failure can pause further merges until `deploybot unpause`. +At merge time, DeployBot records a non-expiring notification obligation. At +exact-main verification, it promotes every contained obligation to `pending`, +moves the matching source thread to `deployed` when that thread has not moved +on, and returns a stable `thread_notifications` payload for each one. +The provider adapter posts the supplied message into that native thread; for +Codex it wakes the thread with `send_message_to_thread`. The source thread then +acknowledges delivery and becomes `completed`: + +```bash +deploybot thread acknowledge --provider codex --thread-id "$CODEX_THREAD_ID" \ + --notification-id "$DEPLOYBOT_NOTIFICATION_ID" +``` + +DeployBot does not treat a registry comment as user notification. If native +delivery fails, an independent outbox entry stays visible under pending +`notifications`, even if the source thread starts new work, and the same +`notification_id` can be retried. When `pipeline.webhook_url_env` is configured, +the provider-neutral webhook also receives the `thread-deployed` payload. +Source adapters attach a native thread heartbeat before returning from an +asynchronous deploy request, so the originating thread can retrieve, acknowledge, +and display the final notification even when no interactive coordinator remains. + ```toml [pipeline] ci_workflows = ["CI"] @@ -251,6 +274,7 @@ deploybot plan --json deploybot doctor --json deploybot inspect [PR] --json deploybot thread update --provider CLIENT --thread-id ID --phase PHASE [metadata] +deploybot thread acknowledge --provider CLIENT --thread-id ID --notification-id ID deploybot request [PR] [--provider CLIENT] [--thread-id ID] [--thread-url URL] deploybot cancel-request [PR] deploybot refresh-request [PR] diff --git a/adapters/claude-code/skills/deploybot/SKILL.md b/adapters/claude-code/skills/deploybot/SKILL.md index 4a4fbed..05aa2c3 100644 --- a/adapters/claude-code/skills/deploybot/SKILL.md +++ b/adapters/claude-code/skills/deploybot/SKILL.md @@ -70,3 +70,29 @@ cumulative base heads until CI, deployment, and configured health checks verify. Use `diagnose`/`deploybot doctor` for setup drift and `delivery_metrics` for p50, p95, and slow-stage evidence. A failed cumulative CI or deployment pauses the controller; only a designated coordinator may unpause after recovery. + +## Notify Source Threads + +After exact-main verification, `follow_release` returns one +`thread_notifications` entry per source thread and records that thread as +`deployed`. Deliver every entry's `message` into the recorded native thread so +the user can see completion by looking at that thread. In Codex, use the app's +`send_message_to_thread` tool to wake that thread with the supplied message and +instruct it to make the supplied message user-visible, then call +`acknowledge_thread_deployment` without doing more code work. Pass that entry's +`notification_id` to the acknowledgement. If already operating in the source +thread, show the message first, acknowledge it, and finish with the same status. + +Treat `notification_id` as the idempotency key. Never acknowledge a thread +before its native-thread delivery succeeds, and never substitute a registry +comment for the native message. If native delivery is unavailable or fails, +leave the notification `pending`; a later coordinator or the provider-neutral +`thread-deployed` webhook can retry it. + +The source thread that calls `request_deployment` also owns a durable wake-up. +If it will stop running before verification, attach the provider's native +thread heartbeat or follow-up monitor before returning. In Codex, use a thread +heartbeat automation. On wake, read `pipeline_status`; once this thread is +listed under pending `notifications`, first show its supplied message to the +user, then acknowledge it and remove the heartbeat. Do not use a tight polling +loop. diff --git a/adapters/claude-code/skills/manage-merge-queue/SKILL.md b/adapters/claude-code/skills/manage-merge-queue/SKILL.md index ebbe387..0f7bcdf 100644 --- a/adapters/claude-code/skills/manage-merge-queue/SKILL.md +++ b/adapters/claude-code/skills/manage-merge-queue/SKILL.md @@ -19,3 +19,13 @@ honor dependencies, route overlap or cumulative validation through `create_integration_pull_request`, return repair packets to the source thread, and use `resume_pull_request` after fresh review. Finish with `follow_release`; a failed CI or deployment pauses the pipeline until verified recovery. + +When `follow_release` returns `thread_notifications`, send each supplied +message to its native source thread. The source thread calls +`acknowledge_thread_deployment` with the matching `notification_id` and reports +completion to the user. Leave failed notifications `pending` so they remain +retryable. + +Before a requesting source thread stops running, attach a native follow-up +monitor that checks `pipeline_status` and wakes it to report and acknowledge its +matching pending notification. diff --git a/adapters/codex/agent-merge-queue/skills/deploybot/SKILL.md b/adapters/codex/agent-merge-queue/skills/deploybot/SKILL.md index 4a4fbed..05aa2c3 100644 --- a/adapters/codex/agent-merge-queue/skills/deploybot/SKILL.md +++ b/adapters/codex/agent-merge-queue/skills/deploybot/SKILL.md @@ -70,3 +70,29 @@ cumulative base heads until CI, deployment, and configured health checks verify. Use `diagnose`/`deploybot doctor` for setup drift and `delivery_metrics` for p50, p95, and slow-stage evidence. A failed cumulative CI or deployment pauses the controller; only a designated coordinator may unpause after recovery. + +## Notify Source Threads + +After exact-main verification, `follow_release` returns one +`thread_notifications` entry per source thread and records that thread as +`deployed`. Deliver every entry's `message` into the recorded native thread so +the user can see completion by looking at that thread. In Codex, use the app's +`send_message_to_thread` tool to wake that thread with the supplied message and +instruct it to make the supplied message user-visible, then call +`acknowledge_thread_deployment` without doing more code work. Pass that entry's +`notification_id` to the acknowledgement. If already operating in the source +thread, show the message first, acknowledge it, and finish with the same status. + +Treat `notification_id` as the idempotency key. Never acknowledge a thread +before its native-thread delivery succeeds, and never substitute a registry +comment for the native message. If native delivery is unavailable or fails, +leave the notification `pending`; a later coordinator or the provider-neutral +`thread-deployed` webhook can retry it. + +The source thread that calls `request_deployment` also owns a durable wake-up. +If it will stop running before verification, attach the provider's native +thread heartbeat or follow-up monitor before returning. In Codex, use a thread +heartbeat automation. On wake, read `pipeline_status`; once this thread is +listed under pending `notifications`, first show its supplied message to the +user, then acknowledge it and remove the heartbeat. Do not use a tight polling +loop. diff --git a/adapters/codex/agent-merge-queue/skills/manage-merge-queue/SKILL.md b/adapters/codex/agent-merge-queue/skills/manage-merge-queue/SKILL.md index 0f8cc29..c4b96b8 100644 --- a/adapters/codex/agent-merge-queue/skills/manage-merge-queue/SKILL.md +++ b/adapters/codex/agent-merge-queue/skills/manage-merge-queue/SKILL.md @@ -20,3 +20,13 @@ dependencies, and use `create_integration_pull_request` for overlaps or a cumulative batch gate. Return repair packets to their source thread and use `resume_pull_request` after fresh review. Finish with `follow_release`; a failed CI or deployment pauses the pipeline until verified recovery. + +When `follow_release` returns `thread_notifications`, send each supplied +message to its native source thread. In Codex use `send_message_to_thread`; +the source thread calls `acknowledge_thread_deployment` with the matching +`notification_id` and reports completion to the user. Leave failed +notifications `pending` so they remain retryable. + +Before a requesting source thread stops running, attach a native thread +heartbeat that checks `pipeline_status` and wakes it to report and acknowledge +its matching pending notification. diff --git a/adapters/cursor/.cursor/rules/deploybot.mdc b/adapters/cursor/.cursor/rules/deploybot.mdc index 1143334..f103c5f 100644 --- a/adapters/cursor/.cursor/rules/deploybot.mdc +++ b/adapters/cursor/.cursor/rules/deploybot.mdc @@ -14,3 +14,12 @@ exact heads, blockers, overlap, CI, and deployment. Only the user's exact the stable Cursor thread ID, never prompts or transcripts. Refresh intent only after replacement-head review. Only the coordinator may react, integrate, drain, follow, pause, or resume repaired work. + +After exact-main verification, deliver each returned `thread_notifications` +message into its native source thread, then call +`acknowledge_thread_deployment` with the matching `notification_id`. Leave a +failed notification `pending` so a later coordinator can retry it; a registry +comment is not a user notification. + +Before the requesting source thread stops, attach a native follow-up monitor +that wakes it when `pipeline_status` lists its pending notification. diff --git a/adapters/cursor/AGENTS.md b/adapters/cursor/AGENTS.md index 17c3125..6f48787 100644 --- a/adapters/cursor/AGENTS.md +++ b/adapters/cursor/AGENTS.md @@ -15,3 +15,11 @@ Never poll, merge an unlabeled PR, or absorb unrelated work. Let the event worke promote fresh exact heads, use one integration PR for overlaps or cumulative validation, return repair packets to the source thread, atomically resume after fresh review, and follow cumulative `main` through verified deployment. + +For each verified `thread_notifications` entry, post its message back to the +native source thread and only then call `acknowledge_thread_deployment`. Leave +failed notifications `pending` for a later retry, and pass the matching +`notification_id` when acknowledging. + +Before a requesting source thread stops, attach a native follow-up monitor that +wakes it when `pipeline_status` lists its pending notification. diff --git a/docs/reference.md b/docs/reference.md index e88a8d9..0a73562 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -23,7 +23,7 @@ DeployBot resolves the pull request for the current branch. | `deploybot init [--force]` | Write a safe starter policy. Existing files are preserved unless `--force` is supplied. | | `deploybot ensure-labels` | Create or refresh the configured queue, blocked, intent, pause, and registry labels. | | `deploybot doctor [--json]` | Check authentication, policy, labels, actors, checks, workflows, and branch protection without changing repository state. | -| `deploybot status [--json]` | Read active thread metadata, PR stages, deploy intent, queue state, exact-main CI, deployment, and pipeline control state. | +| `deploybot status [--json]` | Read active thread metadata, pending native notifications, PR stages, deploy intent, queue state, exact-main CI, deployment, and pipeline control state. | | `deploybot plan [--json]` | Read the ordered queue, dependencies, blockers, and source-overlap groups. | | `deploybot inspect [PR] [--json]` | Evaluate one exact PR head without granting merge authority. | | `deploybot metrics [--limit N] [--json]` | Summarize p50, p95, and maximum delivery timings for recent merged PRs. The default limit is 25. | @@ -36,7 +36,8 @@ batch marker. | Command | Purpose | | --- | --- | -| `deploybot thread update --provider CLIENT --thread-id ID --phase PHASE [--title TEXT] [--branch NAME] [--pr NUMBER] [--url URL]` | Publish metadata-only thread state. Valid phases are `working`, `pr-draft`, `pr-review`, `ready`, `deploy-requested`, `queued`, `merged`, `blocked`, `completed`, and `abandoned`. | +| `deploybot thread update --provider CLIENT --thread-id ID --phase PHASE [--title TEXT] [--branch NAME] [--pr NUMBER] [--url URL]` | Publish metadata-only thread state. Valid client-published phases are `working`, `pr-draft`, `pr-review`, `ready`, `deploy-requested`, `queued`, `merged`, `blocked`, `completed`, and `abandoned`; `deployed` is controller-owned. | +| `deploybot thread acknowledge --provider CLIENT --thread-id ID --notification-id ID` | Mark the matching `deployed` notification `completed` only after its native-thread message is delivered. Repeated acknowledgement is safe; stale IDs are rejected. | | `deploybot request [PR] [--provider CLIENT] [--thread-id ID] [--thread-url URL]` | Record the user's durable deploy intent for the current exact head, even while gates are pending. | | `deploybot cancel-request [PR]` | Cancel an unmerged durable deploy request. | | `deploybot refresh-request [PR]` | Bind existing user intent to a freshly reviewed replacement head. | @@ -97,6 +98,7 @@ arguments shown below are the tool-specific arguments. | `follow_release` | `follow --json` | optional `timeout_seconds` | | `delivery_metrics` | `metrics --json` | optional `limit` | | `update_agent_thread` | `thread update` | `provider`, `thread_id`, `phase`; optional `pull_request`, `title`, `branch`, `url` | +| `acknowledge_thread_deployment` | `thread acknowledge` | `provider`, `thread_id`, `notification_id` | | `block_pull_request` | `block PR` | `pull_request`, `reason` | | `unblock_pull_request` | `unblock PR` | `pull_request` | | `resume_pull_request` | `resume PR` | `pull_request` | @@ -156,7 +158,7 @@ Provider fields are: | `pause_label` | `"deploybot-paused"` | | `registry_label` | `"deploybot-registry"` | | `registry_title` | `"DeployBot delivery registry"` | -| `thread_active_hours` | Positive integer; default 72. | +| `thread_active_hours` | Positive integer; default 72. Notification obligations and pending messages use their own non-expiring outbox. | | `ci_workflows` | Workflow names followed as exact-main CI. Default: `["CI"]`. | | `deploy_workflows` | Deployment workflow names. Default: `["Deploy"]`. | | `batch_settle_seconds` | Non-negative window for coalescing near-ready deploy requests before freezing a batch. Default: 15. | @@ -167,7 +169,7 @@ Provider fields are: | `auto_promote` | Default `true`. | | `intent_scope` | Currently must be `"head"`. | | `pause_on_failure` | Default `true`. | -| `webhook_url_env` | Optional environment-variable name containing a best-effort event webhook URL. GitHub remains authoritative if notification fails. | +| `webhook_url_env` | Optional environment-variable name containing a best-effort event webhook URL. It receives retryable `thread-deployed` payloads after exact-main verification. GitHub remains authoritative if notification fails. | | `verifications` | Optional array of post-deployment HTTP verification tables. Default: empty. | | `name` | Required health-check display name in each `pipeline.verifications` entry. | | `url` | Required URL checked after deployment in each `pipeline.verifications` entry. | @@ -207,3 +209,11 @@ list aligned with `pipeline.ci_workflows`. webhook URL. - The composite Action maps `github.token` to `GH_TOKEN`; callers do not need to create a separate token when its documented permissions are sufficient. + +The `thread-deployed` event contains `notification_id`, `repository`, +`provider`, `thread_id`, `main_sha`, and a user-facing `message`, plus available +thread, pull-request, CI, and deployment URLs. Consumers must deduplicate on +`notification_id`, deliver `message` into the native provider thread, and call +`acknowledge_thread_deployment` only after that succeeds. Until acknowledgement, +the independent outbox record remains `pending` even if thread lifecycle moves +on, and later release followers can retry the same notification. diff --git a/skills/deploybot/SKILL.md b/skills/deploybot/SKILL.md index 4a4fbed..05aa2c3 100644 --- a/skills/deploybot/SKILL.md +++ b/skills/deploybot/SKILL.md @@ -70,3 +70,29 @@ cumulative base heads until CI, deployment, and configured health checks verify. Use `diagnose`/`deploybot doctor` for setup drift and `delivery_metrics` for p50, p95, and slow-stage evidence. A failed cumulative CI or deployment pauses the controller; only a designated coordinator may unpause after recovery. + +## Notify Source Threads + +After exact-main verification, `follow_release` returns one +`thread_notifications` entry per source thread and records that thread as +`deployed`. Deliver every entry's `message` into the recorded native thread so +the user can see completion by looking at that thread. In Codex, use the app's +`send_message_to_thread` tool to wake that thread with the supplied message and +instruct it to make the supplied message user-visible, then call +`acknowledge_thread_deployment` without doing more code work. Pass that entry's +`notification_id` to the acknowledgement. If already operating in the source +thread, show the message first, acknowledge it, and finish with the same status. + +Treat `notification_id` as the idempotency key. Never acknowledge a thread +before its native-thread delivery succeeds, and never substitute a registry +comment for the native message. If native delivery is unavailable or fails, +leave the notification `pending`; a later coordinator or the provider-neutral +`thread-deployed` webhook can retry it. + +The source thread that calls `request_deployment` also owns a durable wake-up. +If it will stop running before verification, attach the provider's native +thread heartbeat or follow-up monitor before returning. In Codex, use a thread +heartbeat automation. On wake, read `pipeline_status`; once this thread is +listed under pending `notifications`, first show its supplied message to the +user, then acknowledge it and remove the heartbeat. Do not use a tight polling +loop. diff --git a/src/agent_merge_queue/cli.py b/src/agent_merge_queue/cli.py index 77bbb17..38229a8 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -34,11 +34,14 @@ INTEGRATION_MARKER, REPAIR_MARKER, THREAD_PHASES, + DeploymentNotificationRecord, ThreadRecord, control_body, + deployment_notification_body, integration_body, intent_body, latest_intent, + latest_deployment_notifications, latest_payload, latest_thread_records, parse_time, @@ -1115,7 +1118,28 @@ def record_thread(self, record: ThreadRecord) -> None: raise QueueError("could not create DeployBot registry") self.issue_comment(number, thread_record_body(record)) - def thread_records(self) -> list[dict[str, Any]]: + def record_deployment_notification( + self, record: DeploymentNotificationRecord + ) -> None: + number = self.registry_issue_number(create=True) + if number is None: # pragma: no cover - create owns this invariant. + raise QueueError("could not create DeployBot registry") + self.issue_comment(number, deployment_notification_body(record)) + + def deployment_notifications( + self, *, include_delivered: bool = False + ) -> list[dict[str, Any]]: + trusted = self.trusted_logins | self.coordinator_logins + return [ + record.as_dict() + for record in latest_deployment_notifications( + self.registry_comments(), + trusted, + include_delivered=include_delivered, + ) + ] + + def thread_records(self, *, include_terminal: bool = False) -> list[dict[str, Any]]: trusted = self.trusted_logins | self.coordinator_logins return [ record.as_dict() @@ -1123,6 +1147,7 @@ def thread_records(self) -> list[dict[str, Any]]: self.registry_comments(), trusted, active_hours=self.config.pipeline.thread_active_hours, + include_terminal=include_terminal, ) ] @@ -1559,6 +1584,19 @@ def dependency_is_merged(self, number: int) -> bool: ) return comparison.get("status") in {"ahead", "identical"} + def pull_merge_sha(self, number: int) -> str | None: + value = self._json( + "pr", + "view", + str(number), + "--repo", + self.repository, + "--json", + "mergeCommit,state", + ) + merge_sha = str((value.get("mergeCommit") or {}).get("oid") or "") + return merge_sha if value.get("state") == "MERGED" and merge_sha else None + def externally_integrated_merge( self, number: int, expected_head: str ) -> str | None: @@ -1804,6 +1842,7 @@ def pipeline_status(client: GitHub) -> dict[str, Any]: "repository": client.repository, "control": client.pipeline_control(), "threads": client.thread_records(), + "notifications": client.deployment_notifications(), "pull_requests": stages, "queue": [entry_dict(entry) for entry in queued], "overlap_groups": overlap_groups(queued), @@ -1820,6 +1859,7 @@ def print_pipeline_status(value: dict[str, Any], *, json_output: bool) -> None: print( "threads: " f"{len(value['threads'])} active; " + f"notifications: {len(value.get('notifications') or [])} pending; " f"deploy requests: {len(stages['deploy_requested'])}; " f"queue: {len(value['queue'])}; " f"release: {value['release']['state']}" @@ -1862,6 +1902,138 @@ def command_thread_update( print(f"recorded {provider} thread {thread_id} as {phase}") +def thread_notification_id( + *, + repository: str, + provider: str, + thread_id: str, + merge_sha: str, + pull_request: int | None, +) -> str: + identifier = hashlib.sha256( + f"{repository}:{provider.lower()}:{thread_id}:{merge_sha}:{pull_request}".encode() + ).hexdigest()[:24] + return f"thread-deployed:{identifier}" + + +def record_deployment_notification_obligation( + client: GitHub, + *, + provider: str, + thread_id: str, + merge_sha: str, + pull_request: int | None, + thread_url: str | None, + updated_at: str | None = None, +) -> DeploymentNotificationRecord: + record = DeploymentNotificationRecord( + notification_id=thread_notification_id( + repository=client.repository, + provider=provider, + thread_id=thread_id, + merge_sha=merge_sha, + pull_request=pull_request, + ), + provider=provider, + thread_id=thread_id, + state="awaiting-verification", + updated_at=updated_at or utc_now(), + repository=client.repository, + merge_sha=merge_sha, + pull_request=pull_request, + thread_url=thread_url, + ) + client.record_deployment_notification(record) + return record + + +def command_thread_acknowledge( + client: GitHub, + *, + provider: str, + thread_id: str, + notification_id: str, +) -> dict[str, Any]: + notification = next( + ( + value + for value in client.deployment_notifications(include_delivered=True) + if str(value.get("notification_id") or "") == notification_id + ), + None, + ) + if notification is None: + raise QueueError(f"DeployBot notification {notification_id} was not found") + if ( + str(notification.get("provider") or "").lower() != provider.lower() + or str(notification.get("thread_id") or "") != thread_id + ): + raise QueueError( + f"notification does not belong to DeployBot thread {provider}/{thread_id}" + ) + if notification.get("state") == "awaiting-verification": + raise QueueError(f"DeployBot notification {notification_id} is not deployed") + already_delivered = notification.get("state") == "delivered" + delivered = DeploymentNotificationRecord( + notification_id=notification_id, + provider=str(notification["provider"]), + thread_id=str(notification["thread_id"]), + state="delivered", + updated_at=( + str(notification["updated_at"]) if already_delivered else utc_now() + ), + repository=str(notification["repository"]), + main_sha=str(notification["main_sha"]), + message=str(notification["message"]), + merge_sha=str(notification["merge_sha"]), + pull_request=notification.get("pull_request"), + thread_url=str(notification.get("thread_url") or "") or None, + ci_url=str(notification.get("ci_url") or "") or None, + deployment_url=str(notification.get("deployment_url") or "") or None, + ) + if not already_delivered: + client.record_deployment_notification(delivered) + current_thread = next( + ( + value + for value in client.thread_records(include_terminal=True) + if str(value.get("provider") or "").lower() == provider.lower() + and str(value.get("thread_id") or "") == thread_id + ), + None, + ) + if ( + current_thread + and current_thread.get("phase") in {"merged", "deployed"} + and ( + current_thread.get("phase") == "merged" + or current_thread.get("deployed_sha") == notification.get("main_sha") + ) + and current_thread.get("merge_sha") == notification.get("merge_sha") + ): + client.record_thread( + ThreadRecord( + provider=str(current_thread["provider"]), + thread_id=str(current_thread["thread_id"]), + phase="completed", + updated_at=delivered.updated_at, + title=str(current_thread.get("title") or "") or None, + branch=str(current_thread.get("branch") or "") or None, + pull_request=current_thread.get("pull_request"), + url=str(current_thread.get("url") or "") or None, + merge_sha=delivered.merge_sha, + deployed_sha=delivered.main_sha, + ci_url=delivered.ci_url, + deployment_url=delivered.deployment_url, + ) + ) + if already_delivered: + print(f"thread notification already acknowledged for {provider}/{thread_id}") + else: + print(f"acknowledged thread notification for {provider}/{thread_id}") + return delivered.as_dict() + + def command_request( client: GitHub, selector: str | None, @@ -2089,6 +2261,15 @@ def reconcile_externally_merged_threads(client: GitHub) -> list[dict[str, Any]]: if not merge_sha: continue updated_at = utc_now() + record_deployment_notification_obligation( + client, + provider=str(record["provider"]), + thread_id=str(record["thread_id"]), + merge_sha=merge_sha, + pull_request=number, + thread_url=str(record.get("url") or "") or None, + updated_at=updated_at, + ) client.record_thread( ThreadRecord( provider=str(record["provider"]), @@ -2099,6 +2280,7 @@ def reconcile_externally_merged_threads(client: GitHub) -> list[dict[str, Any]]: branch=str(record.get("branch") or "") or None, pull_request=number, url=str(record.get("url") or "") or None, + merge_sha=merge_sha, ) ) value = { @@ -2983,14 +3165,25 @@ def command_merge( coordinator_logins(client), ) if intent and intent.get("provider") and intent.get("thread_id"): + updated_at = utc_now() + record_deployment_notification_obligation( + client, + provider=str(intent["provider"]), + thread_id=str(intent["thread_id"]), + merge_sha=merge_sha, + pull_request=number, + thread_url=str(intent.get("thread_url") or "") or None, + updated_at=updated_at, + ) client.record_thread( ThreadRecord( provider=str(intent["provider"]), thread_id=str(intent["thread_id"]), phase="merged", - updated_at=utc_now(), + updated_at=updated_at, pull_request=number, url=str(intent.get("thread_url") or "") or None, + merge_sha=merge_sha, ) ) notify( @@ -3015,14 +3208,25 @@ def command_merge( and source_intent.get("provider") and source_intent.get("thread_id") ): + updated_at = utc_now() + record_deployment_notification_obligation( + client, + provider=str(source_intent["provider"]), + thread_id=str(source_intent["thread_id"]), + merge_sha=merge_sha, + pull_request=int(source_number), + thread_url=(str(source_intent.get("thread_url") or "") or None), + updated_at=updated_at, + ) client.record_thread( ThreadRecord( provider=str(source_intent["provider"]), thread_id=str(source_intent["thread_id"]), phase="merged", - updated_at=utc_now(), + updated_at=updated_at, pull_request=int(source_number), url=str(source_intent.get("thread_url") or "") or None, + merge_sha=merge_sha, ) ) if emit: @@ -3219,6 +3423,83 @@ def command_integrate(client: GitHub, *, all_entries: bool) -> dict[str, Any]: return result +def thread_deployment_notification( + *, + repository: str, + record: dict[str, Any], + release: dict[str, Any], +) -> dict[str, Any]: + main_sha = str(record.get("deployed_sha") or release["main_sha"]) + provider = str(record["provider"]) + thread_id = str(record["thread_id"]) + pull_request = record.get("pull_request") + subject = f"PR #{pull_request}" if pull_request is not None else "Your change" + same_release = str(release["main_sha"]) == main_sha + ci_url = str(record.get("ci_url") or "") + deployment_url = str(record.get("deployment_url") or "") + if same_release: + ci_url = ci_url or str((release.get("latest_ci") or {}).get("url") or "") + deployment_url = deployment_url or str( + (release.get("latest_deploy") or {}).get("url") or "" + ) + message = ( + f"Deployed: {subject} is live from exact main `{main_sha}`. " + "CI, deployment, and configured health checks passed." + ) + if deployment_url: + message += f" [Deployment run]({deployment_url})" + notification: dict[str, Any] = { + "notification_id": thread_notification_id( + repository=repository, + provider=provider, + thread_id=thread_id, + merge_sha=str(record.get("merge_sha") or ""), + pull_request=pull_request, + ), + "repository": repository, + "provider": provider, + "thread_id": thread_id, + "main_sha": main_sha, + "message": message, + } + optional = { + "thread_url": record.get("url"), + "pull_request": pull_request, + "merge_sha": record.get("merge_sha"), + "ci_url": ci_url or None, + "deployment_url": deployment_url or None, + } + notification.update( + {key: value for key, value in optional.items() if value is not None} + ) + return notification + + +def record_pending_deployment_notification( + client: GitHub, + notification: dict[str, Any], + *, + updated_at: str | None = None, +) -> DeploymentNotificationRecord: + record = DeploymentNotificationRecord( + notification_id=str(notification["notification_id"]), + provider=str(notification["provider"]), + thread_id=str(notification["thread_id"]), + state="pending", + updated_at=updated_at or utc_now(), + repository=str(notification["repository"]), + merge_sha=str(notification["merge_sha"]), + main_sha=str(notification["main_sha"]), + message=str(notification["message"]), + pull_request=notification.get("pull_request"), + thread_url=str(notification.get("thread_url") or "") or None, + ci_url=str(notification.get("ci_url") or "") or None, + deployment_url=str(notification.get("deployment_url") or "") or None, + ) + client.record_deployment_notification(record) + return record + + def command_follow( client: GitHub, *, @@ -3248,21 +3529,124 @@ def command_follow( f"{result['state']} on {result['main_sha']}", ) if result["state"] == "verified": - for record in client.thread_records(): - if record.get("phase") != "merged": + existing_notifications = client.deployment_notifications(include_delivered=True) + if not isinstance(existing_notifications, list): + existing_notifications = [] + notification_states = { + str(value["notification_id"]): str(value["state"]) + for value in existing_notifications + } + pending_notifications = { + str(value["notification_id"]): { + key: item + for key, item in value.items() + if key not in {"state", "updated_at"} and item is not None + } + for value in existing_notifications + if value.get("state") == "pending" + } + for obligation in existing_notifications: + if obligation.get("state") != "awaiting-verification": continue - client.record_thread( - ThreadRecord( - provider=str(record["provider"]), - thread_id=str(record["thread_id"]), - phase="completed", - updated_at=utc_now(), - title=str(record.get("title") or "") or None, - branch=str(record.get("branch") or "") or None, - pull_request=record.get("pull_request"), - url=str(record.get("url") or "") or None, - ) + merge_sha = str(obligation.get("merge_sha") or "") + if not merge_sha or not client.is_ancestor( + merge_sha, str(result["main_sha"]) + ): + continue + deployed_record = { + "provider": obligation["provider"], + "thread_id": obligation["thread_id"], + "pull_request": obligation.get("pull_request"), + "url": obligation.get("thread_url"), + "merge_sha": merge_sha, + "deployed_sha": result["main_sha"], + "ci_url": (result.get("latest_ci") or {}).get("url"), + "deployment_url": (result.get("latest_deploy") or {}).get("url"), + } + notification = thread_deployment_notification( + repository=client.repository, + record=deployed_record, + release=result, ) + notification_id = str(notification["notification_id"]) + record_pending_deployment_notification(client, notification) + pending_notifications[notification_id] = notification + notification_states[notification_id] = "pending" + for record in client.thread_records(include_terminal=True): + if record.get("phase") not in {"merged", "deployed"}: + continue + merge_sha = str(record.get("merge_sha") or "") + if record.get("phase") == "merged": + if not merge_sha and record.get("pull_request") is not None: + merge_sha = str( + client.pull_merge_sha(int(record["pull_request"])) or "" + ) + if not merge_sha or not client.is_ancestor( + merge_sha, str(result["main_sha"]) + ): + # The record may have arrived after follow_release captured + # its verified revision. Leave it merged for the next release. + continue + else: + deployed_sha = str(record.get("deployed_sha") or "") + if ( + not merge_sha + or not deployed_sha + or not client.is_ancestor(merge_sha, deployed_sha) + ): + continue + deployed_sha = str(record.get("deployed_sha") or result["main_sha"]) + same_release = deployed_sha == str(result["main_sha"]) + ci_url = str(record.get("ci_url") or "") or ( + str((result.get("latest_ci") or {}).get("url") or "") + if same_release + else "" + ) + deployment_url = str(record.get("deployment_url") or "") or ( + str((result.get("latest_deploy") or {}).get("url") or "") + if same_release + else "" + ) + deployed_record = { + **record, + "merge_sha": merge_sha, + "deployed_sha": deployed_sha, + "ci_url": ci_url or None, + "deployment_url": deployment_url or None, + } + notification = thread_deployment_notification( + repository=client.repository, + record=deployed_record, + release=result, + ) + notification_id = str(notification["notification_id"]) + if notification_id not in notification_states: + record_pending_deployment_notification(client, notification) + pending_notifications[notification_id] = notification + notification_states[notification_id] = "pending" + if notification_id in pending_notifications and ( + record.get("phase") == "merged" or not record.get("deployed_sha") + ): + client.record_thread( + ThreadRecord( + provider=str(record["provider"]), + thread_id=str(record["thread_id"]), + phase="deployed", + updated_at=utc_now(), + title=str(record.get("title") or "") or None, + branch=str(record.get("branch") or "") or None, + pull_request=record.get("pull_request"), + url=str(record.get("url") or "") or None, + merge_sha=merge_sha, + deployed_sha=deployed_sha, + ci_url=ci_url or None, + deployment_url=deployment_url or None, + ) + ) + notifications = list(pending_notifications.values()) + for notification in notifications: + notify(client.config.pipeline, "thread-deployed", notification) + result = {**result, "thread_notifications": notifications} notify( client.config.pipeline, "verified", @@ -3290,9 +3674,16 @@ def release_follow_needed(client: GitHub) -> bool: return current.get("latest_ci") is not None if current["state"] == "verified": # Verification can finish just before the original worker is replaced. - # Only revisit it when merged thread records still need completion. - return any( - record.get("phase") == "merged" for record in client.thread_records() + # Revisit it while a merge needs an outbox entry or a native-thread + # notification is still awaiting acknowledgement. + notifications = client.deployment_notifications(include_delivered=True) + has_open_notification = isinstance(notifications, list) and any( + value.get("state") in {"awaiting-verification", "pending"} + for value in notifications + ) + return has_open_notification or any( + record.get("phase") == "merged" + for record in client.thread_records(include_terminal=True) ) return True @@ -3713,11 +4104,21 @@ def build_parser() -> argparse.ArgumentParser: thread_update = thread_commands.add_parser("update") thread_update.add_argument("--provider", required=True) thread_update.add_argument("--thread-id", required=True) - thread_update.add_argument("--phase", required=True, choices=sorted(THREAD_PHASES)) + thread_update.add_argument( + "--phase", + required=True, + choices=sorted(THREAD_PHASES - {"deployed"}), + ) thread_update.add_argument("--title") thread_update.add_argument("--branch") thread_update.add_argument("--pr", type=int, dest="pull_request") thread_update.add_argument("--url") + thread_acknowledge = thread_commands.add_parser( + "acknowledge", help="mark a delivered native-thread notification complete" + ) + thread_acknowledge.add_argument("--provider", required=True) + thread_acknowledge.add_argument("--thread-id", required=True) + thread_acknowledge.add_argument("--notification-id", required=True) freeze = subparsers.add_parser("freeze", help="persist one exact queue pass") freeze.add_argument("--json", action="store_true", dest="json_output") drain = subparsers.add_parser( @@ -3812,16 +4213,24 @@ def main(argv: list[str] | None = None) -> int: pipeline_status(client), json_output=arguments.json_output ) elif arguments.command == "thread": - command_thread_update( - client, - provider=arguments.provider, - thread_id=arguments.thread_id, - phase=arguments.phase, - title=arguments.title, - branch=arguments.branch, - pull_request=arguments.pull_request, - url=arguments.url, - ) + if arguments.thread_command == "acknowledge": + command_thread_acknowledge( + client, + provider=arguments.provider, + thread_id=arguments.thread_id, + notification_id=arguments.notification_id, + ) + else: + command_thread_update( + client, + provider=arguments.provider, + thread_id=arguments.thread_id, + phase=arguments.phase, + title=arguments.title, + branch=arguments.branch, + pull_request=arguments.pull_request, + url=arguments.url, + ) elif arguments.command == "freeze": command_freeze(client, json_output=arguments.json_output) elif arguments.command == "drain": diff --git a/src/agent_merge_queue/config.py b/src/agent_merge_queue/config.py index 9e76e51..416d6db 100644 --- a/src/agent_merge_queue/config.py +++ b/src/agent_merge_queue/config.py @@ -127,6 +127,7 @@ class QueueConfig: auto_promote = true intent_scope = "head" pause_on_failure = true +# Receives best-effort events, including retryable thread-deployed messages. # webhook_url_env = "DEPLOYBOT_WEBHOOK_URL" [integration] diff --git a/src/agent_merge_queue/mcp_server.py b/src/agent_merge_queue/mcp_server.py index f2aeac6..5ed3517 100644 --- a/src/agent_merge_queue/mcp_server.py +++ b/src/agent_merge_queue/mcp_server.py @@ -231,6 +231,29 @@ def update_agent_thread( return _run("thread", *arguments, repository=repository, config=config) +@mcp.tool() +def acknowledge_thread_deployment( + provider: str, + thread_id: str, + notification_id: str, + repository: str | None = None, + config: str | None = None, +) -> str: + """Acknowledge only after the deployed message reaches the native thread.""" + return _run( + "thread", + "acknowledge", + "--provider", + provider, + "--thread-id", + thread_id, + "--notification-id", + notification_id, + repository=repository, + config=config, + ) + + @mcp.tool() def block_pull_request( pull_request: str, diff --git a/src/agent_merge_queue/records.py b/src/agent_merge_queue/records.py index f17bd20..4f838b8 100644 --- a/src/agent_merge_queue/records.py +++ b/src/agent_merge_queue/records.py @@ -14,6 +14,7 @@ REPAIR_PREFIX = "deploybot-repair:v1" CONTROL_PREFIX = "deploybot-control:v1" INTEGRATION_PREFIX = "deploybot-integration:v1" +NOTIFICATION_PREFIX = "deploybot-notification:v1" THREAD_PHASES = { "working", @@ -23,6 +24,7 @@ "deploy-requested", "queued", "merged", + "deployed", "blocked", "completed", "abandoned", @@ -62,6 +64,9 @@ def _marker(prefix: str, prose: str) -> re.Pattern[str]: INTEGRATION_MARKER = _marker( INTEGRATION_PREFIX, "Recorded DeployBot integration pull request." ) +NOTIFICATION_MARKER = _marker( + NOTIFICATION_PREFIX, "Recorded DeployBot thread notification." +) def marker_body(prefix: str, payload: dict[str, Any], prose: str) -> str: @@ -117,10 +122,41 @@ class ThreadRecord: branch: str | None = None pull_request: int | None = None url: str | None = None + merge_sha: str | None = None + deployed_sha: str | None = None + ci_url: str | None = None + deployment_url: str | None = None + + def as_dict(self) -> dict[str, Any]: + return asdict(self) + + +@dataclass(frozen=True) +class DeploymentNotificationRecord: + notification_id: str + provider: str + thread_id: str + state: str + updated_at: str + repository: str + merge_sha: str + main_sha: str | None = None + message: str | None = None + pull_request: int | None = None + thread_url: str | None = None + ci_url: str | None = None + deployment_url: str | None = None def as_dict(self) -> dict[str, Any]: return asdict(self) + def payload(self) -> dict[str, Any]: + return { + key: value + for key, value in self.as_dict().items() + if key not in {"state", "updated_at"} and value is not None + } + def thread_record_body(record: ThreadRecord) -> str: if record.phase not in THREAD_PHASES: @@ -131,15 +167,122 @@ def thread_record_body(record: ThreadRecord) -> str: return marker_body(THREAD_PREFIX, payload, "Recorded DeployBot thread metadata.") +def deployment_notification_body(record: DeploymentNotificationRecord) -> str: + if record.state not in {"awaiting-verification", "pending", "delivered"}: + raise ValueError(f"unsupported notification state: {record.state}") + required = ( + record.notification_id, + record.provider, + record.thread_id, + record.repository, + record.merge_sha, + ) + if any(not value.strip() for value in required): + raise ValueError("deployment notification fields are required") + if record.state != "awaiting-verification" and ( + not record.main_sha or not record.message + ): + raise ValueError("delivered notification content is required") + payload = {"schema": 1, **record.as_dict()} + return marker_body( + NOTIFICATION_PREFIX, + payload, + "Recorded DeployBot thread notification.", + ) + + +def latest_deployment_notifications( + comments: Iterable[dict[str, Any]], + trusted_logins: Iterable[str], + *, + include_delivered: bool = False, +) -> list[DeploymentNotificationRecord]: + trusted = {value.lower() for value in trusted_logins} + latest: dict[str, tuple[tuple[str, int, int], DeploymentNotificationRecord]] = {} + for index, comment in enumerate(comments): + if comment_login(comment) not in trusted: + continue + value = _payload(str(comment.get("body") or ""), NOTIFICATION_MARKER) + if value is None or value.get("state") not in { + "awaiting-verification", + "pending", + "delivered", + }: + continue + try: + record = DeploymentNotificationRecord( + notification_id=str(value["notification_id"]), + provider=str(value["provider"]), + thread_id=str(value["thread_id"]), + state=str(value["state"]), + updated_at=str(value["updated_at"]), + repository=str(value["repository"]), + merge_sha=str(value["merge_sha"]), + main_sha=str(value["main_sha"]) if value.get("main_sha") else None, + message=str(value["message"]) if value.get("message") else None, + pull_request=( + int(value["pull_request"]) + if value.get("pull_request") is not None + else None + ), + thread_url=( + str(value["thread_url"]) if value.get("thread_url") else None + ), + ci_url=str(value["ci_url"]) if value.get("ci_url") else None, + deployment_url=( + str(value["deployment_url"]) + if value.get("deployment_url") + else None + ), + ) + except (KeyError, TypeError, ValueError): + continue + required = ( + record.notification_id, + record.provider, + record.thread_id, + record.repository, + record.merge_sha, + ) + content_missing = record.state != "awaiting-verification" and ( + not record.main_sha or not record.message + ) + if ( + any(not item.strip() for item in required) + or content_missing + or parse_time(record.updated_at) is None + ): + continue + candidate = (_comment_key(comment, index), record) + previous = latest.get(record.notification_id) + if previous is None: + latest[record.notification_id] = candidate + else: + ranks = {"awaiting-verification": 0, "pending": 1, "delivered": 2} + previous_rank = ranks[previous[1].state] + candidate_rank = ranks[record.state] + if candidate_rank > previous_rank or ( + candidate_rank == previous_rank and candidate[0] > previous[0] + ): + latest[record.notification_id] = candidate + records = [record for _, record in latest.values()] + if not include_delivered: + records = [record for record in records if record.state == "pending"] + return sorted(records, key=lambda value: value.updated_at, reverse=True) + + def latest_thread_records( comments: Iterable[dict[str, Any]], trusted_logins: Iterable[str], *, active_hours: int, + include_terminal: bool = False, now: datetime | None = None, ) -> list[ThreadRecord]: trusted = {value.lower() for value in trusted_logins} - latest: dict[tuple[str, str], tuple[tuple[str, int, int], ThreadRecord]] = {} + candidates: dict[ + tuple[str, str], list[tuple[tuple[str, int, int], ThreadRecord]] + ] = {} current = now or datetime.now(timezone.utc) for index, comment in enumerate(comments): if comment_login(comment) not in trusted: @@ -161,6 +304,16 @@ def latest_thread_records( else None ), url=str(value["url"]) if value.get("url") else None, + merge_sha=str(value["merge_sha"]) if value.get("merge_sha") else None, + deployed_sha=( + str(value["deployed_sha"]) if value.get("deployed_sha") else None + ), + ci_url=str(value["ci_url"]) if value.get("ci_url") else None, + deployment_url=( + str(value["deployment_url"]) + if value.get("deployment_url") + else None + ), ) except (KeyError, TypeError, ValueError): continue @@ -169,8 +322,55 @@ def latest_thread_records( continue key = (record.provider.lower(), record.thread_id) candidate = (_comment_key(comment, index), record) - if key not in latest or candidate[0] > latest[key][0]: - latest[key] = candidate + candidates.setdefault(key, []).append(candidate) + latest: dict[tuple[str, str], tuple[tuple[str, int, int], ThreadRecord]] = {} + for key, values in candidates.items(): + state: tuple[tuple[str, int, int], ThreadRecord] | None = None + for candidate in sorted(values, key=lambda item: item[0]): + record = candidate[1] + if record.phase == "deployed" and state: + current_record = state[1] + matches_current_merge = ( + bool(record.merge_sha) + and current_record.phase in {"merged", "deployed"} + and current_record.merge_sha == record.merge_sha + ) + matches_legacy_merge = ( + current_record.phase == "merged" + and not current_record.merge_sha + and current_record.pull_request == record.pull_request + ) + if not (matches_current_merge or matches_legacy_merge): + # A newer source-thread state won the append race. Keep it + # authoritative instead of applying this late transition. + continue + if record.phase == "completed" and state: + current_record = state[1] + acknowledges_current_deployment = ( + bool(record.deployed_sha) + and current_record.phase in {"deployed", "completed"} + and current_record.deployed_sha == record.deployed_sha + ) + acknowledges_current_merge = ( + bool(record.merge_sha) + and current_record.phase == "merged" + and current_record.merge_sha == record.merge_sha + ) + if ( + current_record.phase == "deployed" + and not acknowledges_current_deployment + ) or ( + record.deployed_sha + and not ( + acknowledges_current_deployment or acknowledges_current_merge + ) + ): + # An acknowledgement for an older deployment must not + # overwrite a newer state appended during its API race. + continue + state = candidate + if state is not None: + latest[key] = state terminal = {"completed", "abandoned"} result = [] for _, record in latest.values(): @@ -178,7 +378,9 @@ def latest_thread_records( age_hours = ( (current - timestamp).total_seconds() / 3600 if timestamp else 999999 ) - if record.phase not in terminal and age_hours <= active_hours: + if include_terminal or ( + record.phase not in terminal and age_hours <= active_hours + ): result.append(record) return sorted( result, key=lambda value: (value.updated_at, value.provider), reverse=True diff --git a/tests/test_cli.py b/tests/test_cli.py index bd7e515..9606f16 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -3,7 +3,7 @@ import io import json import unittest -from contextlib import redirect_stdout +from contextlib import redirect_stderr, redirect_stdout from unittest.mock import Mock, call, patch from agent_merge_queue.cli import ( @@ -20,12 +20,14 @@ command_drain, command_enqueue, command_integrate, + command_follow, command_merge, command_promote, command_react, command_refresh_request, command_request, command_resume, + command_thread_acknowledge, command_unblock, completed_batch_ids, entries_in_batch, @@ -49,6 +51,7 @@ settle_integration_checks, should_settle_batch, structured_dependencies, + thread_notification_id, ) from agent_merge_queue.config import parse_config from agent_merge_queue.records import integration_body, intent_body @@ -91,6 +94,34 @@ def entry(number: int, *paths: str, state: str = "ready") -> QueueEntry: return value +def deployment_notification( + *, main_sha: str, state: str = "pending", merge_sha: str = "m" * 40 +) -> dict[str, object]: + repository = "example/repo" + provider = "codex" + thread_id = "thread-42" + return { + "notification_id": thread_notification_id( + repository=repository, + provider=provider, + thread_id=thread_id, + merge_sha=merge_sha, + pull_request=42, + ), + "provider": provider, + "thread_id": thread_id, + "state": state, + "updated_at": "2026-06-20T00:05:00Z", + "repository": repository, + "main_sha": main_sha, + "message": f"Deployed on {main_sha}", + "merge_sha": merge_sha, + "pull_request": 42, + "ci_url": "https://example.test/ci/1", + "deployment_url": "https://example.test/deploy/2", + } + + class QueueCoreTest(unittest.TestCase): def test_ensure_labels_exist_is_race_safe(self) -> None: client = object.__new__(GitHub) @@ -250,6 +281,489 @@ def test_reconciles_an_externally_merged_requested_thread(self) -> None: record = client.record_thread.call_args.args[0] self.assertEqual(record.phase, "merged") self.assertEqual(record.pull_request, 42) + self.assertEqual(record.merge_sha, merge_sha) + obligation = client.record_deployment_notification.call_args.args[0] + self.assertEqual(obligation.state, "awaiting-verification") + self.assertEqual(obligation.merge_sha, merge_sha) + + def test_verified_release_creates_native_thread_notification(self) -> None: + sha = "a" * 40 + release = { + "state": "verified", + "main_sha": sha, + "latest_ci": {"url": "https://example.test/ci/1"}, + "latest_deploy": {"url": "https://example.test/deploy/2"}, + "verifications": [], + } + client = Mock() + client.config = CONFIG + client.repository = "example/repo" + client.is_ancestor.return_value = True + client.deployment_notifications.return_value = [] + client.thread_records.return_value = [ + { + "phase": "merged", + "provider": "codex", + "thread_id": "thread-42", + "pull_request": 42, + "url": "codex://thread/thread-42", + "merge_sha": "m" * 40, + } + ] + + with ( + patch("agent_merge_queue.cli.follow_release", return_value=release), + patch("agent_merge_queue.cli.notify") as notify, + patch( + "agent_merge_queue.cli.utc_now", + return_value="2026-06-20T00:05:00Z", + ), + ): + result = command_follow( + client, + timeout_seconds=10, + poll_seconds=1, + json_output=True, + emit=False, + ) + + notification = result["thread_notifications"][0] + self.assertEqual(notification["provider"], "codex") + self.assertEqual(notification["thread_id"], "thread-42") + self.assertEqual(notification["pull_request"], 42) + self.assertEqual(notification["main_sha"], sha) + self.assertEqual( + notification["deployment_url"], "https://example.test/deploy/2" + ) + self.assertIn("PR #42 is live", notification["message"]) + self.assertTrue(notification["notification_id"].startswith("thread-deployed:")) + deployed = client.record_thread.call_args.args[0] + outbox = client.record_deployment_notification.call_args.args[0] + self.assertEqual(outbox.state, "pending") + self.assertEqual(outbox.notification_id, notification["notification_id"]) + self.assertEqual(deployed.phase, "deployed") + self.assertEqual(deployed.merge_sha, "m" * 40) + self.assertEqual(deployed.deployed_sha, sha) + self.assertEqual(deployed.ci_url, "https://example.test/ci/1") + self.assertEqual(deployed.deployment_url, "https://example.test/deploy/2") + client.is_ancestor.assert_called_once_with("m" * 40, sha) + notify.assert_any_call(CONFIG.pipeline, "thread-deployed", notification) + + def test_verified_release_promotes_obligation_after_thread_moves_on(self) -> None: + sha = "a" * 40 + merge_sha = "m" * 40 + obligation = deployment_notification(main_sha=sha, merge_sha=merge_sha) + obligation.update( + { + "state": "awaiting-verification", + "main_sha": None, + "message": None, + "ci_url": None, + "deployment_url": None, + } + ) + client = Mock() + client.config = CONFIG + client.repository = "example/repo" + client.is_ancestor.return_value = True + client.deployment_notifications.return_value = [obligation] + client.thread_records.return_value = [ + { + "phase": "working", + "provider": "codex", + "thread_id": "thread-42", + } + ] + release = { + "state": "verified", + "main_sha": sha, + "latest_ci": {"url": "https://example.test/ci/1"}, + "latest_deploy": {"url": "https://example.test/deploy/2"}, + "verifications": [], + } + + with ( + patch("agent_merge_queue.cli.follow_release", return_value=release), + patch("agent_merge_queue.cli.notify"), + ): + result = command_follow( + client, + timeout_seconds=10, + poll_seconds=1, + json_output=True, + emit=False, + ) + + notification = result["thread_notifications"][0] + self.assertEqual(notification["notification_id"], obligation["notification_id"]) + self.assertEqual(notification["main_sha"], sha) + pending = client.record_deployment_notification.call_args.args[0] + self.assertEqual(pending.state, "pending") + self.assertEqual(pending.merge_sha, merge_sha) + client.record_thread.assert_not_called() + + def test_verified_release_does_not_notify_uncontained_merge(self) -> None: + sha = "a" * 40 + client = Mock() + client.config = CONFIG + client.repository = "example/repo" + client.is_ancestor.return_value = False + client.deployment_notifications.return_value = [] + client.thread_records.return_value = [ + { + "phase": "merged", + "provider": "codex", + "thread_id": "thread-42", + "pull_request": 42, + "merge_sha": "m" * 40, + } + ] + release = { + "state": "verified", + "main_sha": sha, + "latest_ci": None, + "latest_deploy": None, + "verifications": [], + } + + with ( + patch("agent_merge_queue.cli.follow_release", return_value=release), + patch("agent_merge_queue.cli.notify") as notify, + ): + result = command_follow( + client, + timeout_seconds=10, + poll_seconds=1, + json_output=True, + emit=False, + ) + + self.assertEqual(result["thread_notifications"], []) + client.record_thread.assert_not_called() + client.record_deployment_notification.assert_not_called() + client.is_ancestor.assert_called_once_with("m" * 40, sha) + self.assertEqual(notify.call_count, 1) + self.assertEqual(notify.call_args.args[1], "verified") + + def test_verified_release_resolves_legacy_merged_record(self) -> None: + sha = "a" * 40 + merge_sha = "m" * 40 + client = Mock() + client.config = CONFIG + client.repository = "example/repo" + client.pull_merge_sha.return_value = merge_sha + client.is_ancestor.return_value = True + client.deployment_notifications.return_value = [] + client.thread_records.return_value = [ + { + "phase": "merged", + "provider": "codex", + "thread_id": "legacy-thread", + "pull_request": 42, + } + ] + release = { + "state": "verified", + "main_sha": sha, + "latest_ci": None, + "latest_deploy": None, + "verifications": [], + } + + with ( + patch("agent_merge_queue.cli.follow_release", return_value=release), + patch("agent_merge_queue.cli.notify"), + ): + result = command_follow( + client, + timeout_seconds=10, + poll_seconds=1, + json_output=True, + emit=False, + ) + + self.assertEqual(result["thread_notifications"][0]["merge_sha"], merge_sha) + self.assertEqual(client.record_thread.call_args.args[0].merge_sha, merge_sha) + client.pull_merge_sha.assert_called_once_with(42) + client.is_ancestor.assert_called_once_with(merge_sha, sha) + + def test_pending_thread_notification_retries_with_stable_identity(self) -> None: + deployed_sha = "a" * 40 + current_sha = "b" * 40 + release = { + "state": "verified", + "main_sha": current_sha, + "latest_ci": {"url": "https://example.test/ci/new"}, + "latest_deploy": {"url": "https://example.test/deploy/new"}, + "verifications": [], + } + client = Mock() + client.config = CONFIG + client.repository = "example/repo" + pending = deployment_notification(main_sha=deployed_sha) + pending["ci_url"] = "https://example.test/ci/original" + pending["deployment_url"] = "https://example.test/deploy/original" + client.deployment_notifications.return_value = [pending] + client.thread_records.return_value = [ + { + "phase": "working", + "provider": "codex", + "thread_id": "thread-42", + } + ] + with ( + patch("agent_merge_queue.cli.follow_release", return_value=release), + patch("agent_merge_queue.cli.notify"), + ): + first = command_follow( + client, + timeout_seconds=10, + poll_seconds=1, + json_output=True, + emit=False, + ) + second = command_follow( + client, + timeout_seconds=10, + poll_seconds=1, + json_output=True, + emit=False, + ) + + self.assertEqual( + first["thread_notifications"][0]["notification_id"], + second["thread_notifications"][0]["notification_id"], + ) + self.assertEqual(first["thread_notifications"][0]["main_sha"], deployed_sha) + self.assertEqual( + first["thread_notifications"][0]["ci_url"], + "https://example.test/ci/original", + ) + self.assertEqual( + first["thread_notifications"][0]["deployment_url"], + "https://example.test/deploy/original", + ) + self.assertNotIn("/new", first["thread_notifications"][0]["message"]) + client.record_thread.assert_not_called() + client.record_deployment_notification.assert_not_called() + client.is_ancestor.assert_not_called() + + def test_delivered_notification_is_not_reopened_by_release_follow(self) -> None: + sha = "a" * 40 + merge_sha = "m" * 40 + client = Mock() + client.config = CONFIG + client.repository = "example/repo" + client.is_ancestor.return_value = True + client.deployment_notifications.return_value = [ + deployment_notification( + main_sha=sha, + merge_sha=merge_sha, + state="delivered", + ) + ] + client.thread_records.return_value = [ + { + "phase": "merged", + "provider": "codex", + "thread_id": "thread-42", + "pull_request": 42, + "merge_sha": merge_sha, + } + ] + release = { + "state": "verified", + "main_sha": sha, + "latest_ci": None, + "latest_deploy": None, + "verifications": [], + } + + with ( + patch("agent_merge_queue.cli.follow_release", return_value=release), + patch("agent_merge_queue.cli.notify") as notify, + ): + result = command_follow( + client, + timeout_seconds=10, + poll_seconds=1, + json_output=True, + emit=False, + ) + + self.assertEqual(result["thread_notifications"], []) + client.record_deployment_notification.assert_not_called() + client.record_thread.assert_not_called() + self.assertEqual(notify.call_count, 1) + self.assertEqual(notify.call_args.args[1], "verified") + + def test_clients_cannot_publish_controller_owned_deployed_phase(self) -> None: + with ( + redirect_stderr(io.StringIO()), + self.assertRaises(SystemExit), + ): + main( + [ + "thread", + "update", + "--provider", + "codex", + "--thread-id", + "thread-42", + "--phase", + "deployed", + ] + ) + + def test_unverified_notification_obligation_cannot_be_acknowledged(self) -> None: + notification = deployment_notification(main_sha="a" * 40) + notification.update( + { + "state": "awaiting-verification", + "main_sha": None, + "message": None, + } + ) + client = Mock() + client.deployment_notifications.return_value = [notification] + + with self.assertRaisesRegex(QueueError, "is not deployed"): + command_thread_acknowledge( + client, + provider="codex", + thread_id="thread-42", + notification_id=str(notification["notification_id"]), + ) + + client.record_deployment_notification.assert_not_called() + + def test_acknowledging_native_thread_message_completes_record(self) -> None: + sha = "a" * 40 + merge_sha = "m" * 40 + client = Mock() + client.repository = "example/repo" + notification = deployment_notification(main_sha=sha, merge_sha=merge_sha) + client.deployment_notifications.return_value = [notification] + client.thread_records.return_value = [ + { + "phase": "deployed", + "provider": "codex", + "thread_id": "thread-42", + "pull_request": 42, + "merge_sha": merge_sha, + "deployed_sha": sha, + "ci_url": "https://example.test/ci/1", + "deployment_url": "https://example.test/deploy/2", + } + ] + with patch( + "agent_merge_queue.cli.utc_now", + return_value="2026-06-20T00:06:00Z", + ): + result = command_thread_acknowledge( + client, + provider="codex", + thread_id="thread-42", + notification_id=str(notification["notification_id"]), + ) + + client.deployment_notifications.assert_called_once_with(include_delivered=True) + client.thread_records.assert_called_once_with(include_terminal=True) + delivered = client.record_deployment_notification.call_args.args[0] + self.assertEqual(delivered.state, "delivered") + self.assertEqual(delivered.notification_id, notification["notification_id"]) + completed = client.record_thread.call_args.args[0] + self.assertEqual(completed.phase, "completed") + self.assertEqual(completed.deployed_sha, sha) + self.assertEqual(completed.ci_url, "https://example.test/ci/1") + self.assertEqual(completed.deployment_url, "https://example.test/deploy/2") + self.assertEqual(result["state"], "delivered") + + def test_acknowledgement_completes_merge_when_phase_write_was_interrupted( + self, + ) -> None: + sha = "a" * 40 + merge_sha = "m" * 40 + client = Mock() + notification = deployment_notification(main_sha=sha, merge_sha=merge_sha) + client.deployment_notifications.return_value = [notification] + client.thread_records.return_value = [ + { + "phase": "merged", + "provider": "codex", + "thread_id": "thread-42", + "pull_request": 42, + "merge_sha": merge_sha, + } + ] + + command_thread_acknowledge( + client, + provider="codex", + thread_id="thread-42", + notification_id=str(notification["notification_id"]), + ) + + self.assertEqual(client.record_thread.call_args.args[0].phase, "completed") + self.assertEqual( + client.record_thread.call_args.args[0].deployed_sha, + sha, + ) + + def test_thread_acknowledgement_is_idempotent(self) -> None: + sha = "a" * 40 + client = Mock() + client.repository = "example/repo" + notification = deployment_notification(main_sha=sha, state="delivered") + client.deployment_notifications.return_value = [notification] + client.thread_records.return_value = [ + { + "phase": "completed", + "provider": "codex", + "thread_id": "thread-42", + "deployed_sha": sha, + } + ] + + result = command_thread_acknowledge( + client, + provider="codex", + thread_id="thread-42", + notification_id=str(notification["notification_id"]), + ) + + self.assertEqual(result["notification_id"], notification["notification_id"]) + self.assertEqual(result["state"], "delivered") + client.record_deployment_notification.assert_not_called() + client.record_thread.assert_not_called() + + def test_old_acknowledgement_does_not_complete_newer_thread_state(self) -> None: + old_sha = "a" * 40 + new_sha = "b" * 40 + client = Mock() + client.repository = "example/repo" + notification = deployment_notification(main_sha=old_sha) + client.deployment_notifications.return_value = [notification] + client.thread_records.return_value = [ + { + "phase": "deployed", + "provider": "codex", + "thread_id": "thread-42", + "merge_sha": "n" * 40, + "deployed_sha": new_sha, + } + ] + + result = command_thread_acknowledge( + client, + provider="codex", + thread_id="thread-42", + notification_id=str(notification["notification_id"]), + ) + + self.assertEqual(result["state"], "delivered") + client.record_deployment_notification.assert_called_once() + client.record_thread.assert_not_called() def test_overlap_hold_includes_transitive_ready_members(self) -> None: config = parse_config( diff --git a/tests/test_mcp.py b/tests/test_mcp.py index 8800f64..68f00e9 100644 --- a/tests/test_mcp.py +++ b/tests/test_mcp.py @@ -8,9 +8,10 @@ MCP_AVAILABLE = importlib.util.find_spec("mcp") is not None if MCP_AVAILABLE: - from agent_merge_queue.mcp_server import _run + from agent_merge_queue.mcp_server import _run, acknowledge_thread_deployment else: _run = None + acknowledge_thread_deployment = None @unittest.skipUnless(MCP_AVAILABLE, "the optional MCP extra is not installed") @@ -38,6 +39,31 @@ def test_other_mcp_commands_still_fail_closed(self) -> None: ): _run("drain", "--json") + def test_thread_acknowledgement_binds_notification_identity(self) -> None: + with patch( + "agent_merge_queue.mcp_server._run", return_value="acknowledged" + ) as run: + result = acknowledge_thread_deployment( + "codex", + "thread-42", + "thread-deployed:abc", + repository="example/repo", + ) + + self.assertEqual(result, "acknowledged") + run.assert_called_once_with( + "thread", + "acknowledge", + "--provider", + "codex", + "--thread-id", + "thread-42", + "--notification-id", + "thread-deployed:abc", + repository="example/repo", + config=None, + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_records.py b/tests/test_records.py index e8afdd4..cee298f 100644 --- a/tests/test_records.py +++ b/tests/test_records.py @@ -4,8 +4,11 @@ from datetime import datetime, timezone from agent_merge_queue.records import ( + DeploymentNotificationRecord, ThreadRecord, + deployment_notification_body, intent_body, + latest_deployment_notifications, latest_intent, latest_thread_records, thread_record_body, @@ -81,6 +84,345 @@ def test_terminal_and_stale_threads_are_not_active(self) -> None: ) self.assertEqual(values, []) + def test_pending_notification_survives_thread_activity_ttl(self) -> None: + deployed = ThreadRecord( + provider="codex", + thread_id="pending-message", + phase="deployed", + updated_at="2026-06-10T01:00:00Z", + pull_request=42, + deployed_sha="a" * 40, + ci_url="https://example.test/ci/1", + deployment_url="https://example.test/deploy/2", + ) + notification = DeploymentNotificationRecord( + notification_id="thread-deployed:one", + provider="codex", + thread_id="pending-message", + state="pending", + updated_at="2026-06-10T01:00:01Z", + repository="example/repo", + merge_sha="m" * 40, + main_sha="a" * 40, + message="Deployed", + pull_request=42, + ) + comments = [ + comment("trusted", thread_record_body(deployed), deployed.updated_at, 1), + comment( + "trusted", + deployment_notification_body(notification), + notification.updated_at, + 2, + ), + ] + + active = latest_thread_records( + comments, + {"trusted"}, + active_hours=72, + now=datetime(2026, 6, 20, 2, tzinfo=timezone.utc), + ) + notifications = latest_deployment_notifications( + comments, + {"trusted"}, + ) + + self.assertEqual(active, []) + self.assertEqual(len(notifications), 1) + self.assertEqual(notifications[0].notification_id, "thread-deployed:one") + + def test_delivered_notification_is_terminal_in_outbox(self) -> None: + pending = DeploymentNotificationRecord( + notification_id="thread-deployed:one", + provider="codex", + thread_id="thread-1", + state="pending", + updated_at="2026-06-20T00:00:00Z", + repository="example/repo", + merge_sha="m" * 40, + main_sha="a" * 40, + message="Deployed", + ) + awaiting = DeploymentNotificationRecord( + **{ + **pending.as_dict(), + "state": "awaiting-verification", + "updated_at": "2026-06-19T23:59:00Z", + "main_sha": None, + "message": None, + } + ) + delivered = DeploymentNotificationRecord( + **{ + **pending.as_dict(), + "state": "delivered", + "updated_at": "2026-06-20T00:01:00Z", + } + ) + late_pending = DeploymentNotificationRecord( + **{ + **pending.as_dict(), + "updated_at": "2026-06-20T00:02:00Z", + } + ) + late_awaiting = DeploymentNotificationRecord( + **{ + **awaiting.as_dict(), + "updated_at": "2026-06-20T00:03:00Z", + } + ) + comments = [ + comment( + "trusted", + deployment_notification_body(awaiting), + awaiting.updated_at, + 1, + ), + comment( + "trusted", + deployment_notification_body(pending), + pending.updated_at, + 2, + ), + comment( + "trusted", + deployment_notification_body(delivered), + delivered.updated_at, + 3, + ), + comment( + "trusted", + deployment_notification_body(late_pending), + late_pending.updated_at, + 4, + ), + comment( + "trusted", + deployment_notification_body(late_awaiting), + late_awaiting.updated_at, + 5, + ), + ] + + self.assertEqual(latest_deployment_notifications(comments, {"trusted"}), []) + all_notifications = latest_deployment_notifications( + comments, {"trusted"}, include_delivered=True + ) + self.assertEqual(len(all_notifications), 1) + self.assertEqual(all_notifications[0].state, "delivered") + + def test_stale_ack_comment_cannot_hide_newer_thread_state(self) -> None: + first_merged = ThreadRecord( + provider="codex", + thread_id="thread-1", + phase="merged", + updated_at="2026-06-19T23:59:00Z", + pull_request=1, + merge_sha="1" * 40, + ) + first = ThreadRecord( + provider="codex", + thread_id="thread-1", + phase="deployed", + updated_at="2026-06-20T00:00:00Z", + pull_request=1, + merge_sha="1" * 40, + deployed_sha="a" * 40, + ) + newer_merged = ThreadRecord( + provider="codex", + thread_id="thread-1", + phase="merged", + updated_at="2026-06-20T00:00:30Z", + pull_request=2, + merge_sha="2" * 40, + ) + newer = ThreadRecord( + provider="codex", + thread_id="thread-1", + phase="deployed", + updated_at="2026-06-20T00:01:00Z", + pull_request=2, + merge_sha="2" * 40, + deployed_sha="b" * 40, + ) + delayed_ack = ThreadRecord( + provider="codex", + thread_id="thread-1", + phase="completed", + updated_at="2026-06-20T00:02:00Z", + deployed_sha="a" * 40, + ) + unbound_completion = ThreadRecord( + provider="codex", + thread_id="thread-1", + phase="completed", + updated_at="2026-06-20T00:03:00Z", + ) + + values = latest_thread_records( + [ + comment( + "trusted", + thread_record_body(first_merged), + first_merged.updated_at, + 1, + ), + comment("trusted", thread_record_body(first), first.updated_at, 2), + comment( + "trusted", + thread_record_body(newer_merged), + newer_merged.updated_at, + 3, + ), + comment("trusted", thread_record_body(newer), newer.updated_at, 4), + comment( + "trusted", + thread_record_body(delayed_ack), + delayed_ack.updated_at, + 5, + ), + comment( + "trusted", + thread_record_body(unbound_completion), + unbound_completion.updated_at, + 6, + ), + ], + {"trusted"}, + active_hours=72, + now=datetime(2026, 6, 20, 3, tzinfo=timezone.utc), + ) + + self.assertEqual(len(values), 1) + self.assertEqual(values[0].phase, "deployed") + self.assertEqual(values[0].deployed_sha, "b" * 40) + + def test_late_deployed_transition_cannot_hide_newer_merge(self) -> None: + first_merge = ThreadRecord( + provider="codex", + thread_id="thread-1", + phase="merged", + updated_at="2026-06-20T00:00:00Z", + pull_request=1, + merge_sha="1" * 40, + ) + newer_merge = ThreadRecord( + provider="codex", + thread_id="thread-1", + phase="merged", + updated_at="2026-06-20T00:01:00Z", + pull_request=2, + merge_sha="2" * 40, + ) + late_transition = ThreadRecord( + provider="codex", + thread_id="thread-1", + phase="deployed", + updated_at="2026-06-20T00:02:00Z", + pull_request=1, + merge_sha="1" * 40, + deployed_sha="a" * 40, + ) + + values = latest_thread_records( + [ + comment( + "trusted", + thread_record_body(first_merge), + first_merge.updated_at, + 1, + ), + comment( + "trusted", + thread_record_body(newer_merge), + newer_merge.updated_at, + 2, + ), + comment( + "trusted", + thread_record_body(late_transition), + late_transition.updated_at, + 3, + ), + ], + {"trusted"}, + active_hours=72, + now=datetime(2026, 6, 20, 3, tzinfo=timezone.utc), + ) + + self.assertEqual(len(values), 1) + self.assertEqual(values[0].phase, "merged") + self.assertEqual(values[0].merge_sha, "2" * 40) + + def test_resolved_sha_upgrades_legacy_merged_record(self) -> None: + legacy = ThreadRecord( + provider="codex", + thread_id="thread-1", + phase="merged", + updated_at="2026-06-20T00:00:00Z", + pull_request=42, + ) + deployed = ThreadRecord( + provider="codex", + thread_id="thread-1", + phase="deployed", + updated_at="2026-06-20T00:01:00Z", + pull_request=42, + merge_sha="m" * 40, + deployed_sha="a" * 40, + ) + + values = latest_thread_records( + [ + comment("trusted", thread_record_body(legacy), legacy.updated_at, 1), + comment( + "trusted", thread_record_body(deployed), deployed.updated_at, 2 + ), + ], + {"trusted"}, + active_hours=72, + now=datetime(2026, 6, 20, 2, tzinfo=timezone.utc), + ) + + self.assertEqual(len(values), 1) + self.assertEqual(values[0].phase, "deployed") + self.assertEqual(values[0].merge_sha, "m" * 40) + + def test_ack_can_complete_merge_when_deployed_phase_write_was_lost(self) -> None: + merged = ThreadRecord( + provider="codex", + thread_id="thread-1", + phase="merged", + updated_at="2026-06-20T00:00:00Z", + pull_request=42, + merge_sha="m" * 40, + ) + completed = ThreadRecord( + provider="codex", + thread_id="thread-1", + phase="completed", + updated_at="2026-06-20T00:01:00Z", + pull_request=42, + merge_sha="m" * 40, + deployed_sha="a" * 40, + ) + + values = latest_thread_records( + [ + comment("trusted", thread_record_body(merged), merged.updated_at, 1), + comment( + "trusted", thread_record_body(completed), completed.updated_at, 2 + ), + ], + {"trusted"}, + active_hours=72, + now=datetime(2026, 6, 20, 2, tzinfo=timezone.utc), + ) + + self.assertEqual(values, []) + def test_deploy_intent_survives_head_change_but_cancellation_wins(self) -> None: requested = intent_body( intent_id="intent", diff --git a/tests/test_skill.py b/tests/test_skill.py index 9651ba9..7e4101c 100644 --- a/tests/test_skill.py +++ b/tests/test_skill.py @@ -34,6 +34,10 @@ def test_status_guidance_is_read_only(self) -> None: self.assertIn("request_deployment", skill) self.assertIn("resume_pull_request", skill) self.assertIn("follow_release", skill) + self.assertIn("thread_notifications", skill) + self.assertIn("send_message_to_thread", skill) + self.assertIn("acknowledge_thread_deployment", skill) + self.assertIn("heartbeat automation", skill) self.assertIn("Never publish prompts, transcripts", skill) self.assertIn("Never call `freeze_queue` merely to view status", skill) self.assertIn("exact `deploy` instruction", skill) @@ -45,6 +49,8 @@ def test_cursor_adapter_exposes_status_workflow(self) -> None: self.assertIn("queue_plan", rule) self.assertIn("pipeline_status", rule) self.assertIn("request_deployment", rule) + self.assertIn("thread_notifications", rule) + self.assertIn("acknowledge_thread_deployment", rule) self.assertIn("never freeze a", rule) self.assertIn("queue merely to inspect it", rule)