Harden subscription evidence and capture Nango forwards

.agentworkforce/trajectories/relay1756-review-0916.md

@@ -0,0 +1,68 @@
+# Relay1756 review corrections
+
+Worktree starts at a2f09f4ede9c8b7b746df127785f11cf20df2a04 and preserves PR history.
+
+Reject exhausted admission collisions and require intended startup failures; use an owned real exit fixture. Export the actual emitter for tests and retain timestamp-before-mutation ordering. Support GitHub explicit nullable review association without accepting missing or lossy identity. Require exact canonical external records. Bound Nango response diagnostics and identify every settled failed operation without raw provider content.
+
+Hermetic proof regressions run locally. Isolated Engine/broker rehearsal and real provider acceptance are separate evidence classes. No provider action, push, merge or deploy in this worker stage.
+
+The installed trail command refused a new trajectory because the inherited branch already contains active traj_jdx9303jp3ky; that historical trajectory was left unchanged.
+
+Followup: isolate local node/state and explicit CLI broker connection; preflight a real owned exit-one executable; bank actual broker close and returned worker PIDs, bound cleanup wait and retain failed workdirs. Remove unused emitter imports. Hermetic suite: 96/96 pass, no skips. Matching released Relay 12.2.2 + Engine 8.10.1 local rehearsal remains failed: HTTP-created provider-default worker conflicts with broker-provider inventory, causing reconnect before guarded cleanup ACK. The proof retains this runtime blocker; no assertion relaxation or full E2E claim.
+
+2026-09-17 finish-up (worktree /tmp/relay-1756-e2e, head d8148719d): reran the
+isolated rehearsal against the #1759 broker built from source at d4d51f62
+(binary sha256 1bee6e9a…, toolchain build — the recorded b05947a4 hash is the
+CI-built artifact) plus relaycast engine 8.11.0 and found two real defects the
+proof had been papering over in some environments:
+
+1. PTY recipients exit-code race: `agent_exit` (PTY close, no code) always
+   precedes the reaper's `agent_exited` (500ms tick, code-bearing) on the
+   subscription `--spawn` path, so `waitForReady` could settle with a
+   detail-free exit and the CLI error lost the status. Post-release polling
+   cannot recover it — the reaper may never emit `agent_exited` once the
+   identity is released. Fixed in `launchSubscriptionRecipient`: bounded 3s
+   grace for `agent_exited` while the worker is still registered, so the
+   reported exit keeps the authoritative status. Unit regression added.
+2. ANSI-styled broker diagnostics: worker logs carry ANSI styling around
+   tracing field names (`control=[..]` is written `control<esc>[0m<esc>[2m=`),
+   so `standaloneControlsAfter` threw "Unrecognized" on every control-write
+   line. Parser now strips ANSI before matching; regression added.
+
+PTY exit status is structurally unobservable today (the wrapper owns the
+child; `agent_exited` reports `code: null`), so the early-exit case
+now validates the reported shape — any reported status must be the fixture's
+own code — and requires a marker file proving the owned exit-one fixture
+executed. The fleet `verify_ready` path still certifies "exit status: 1".
+
+Results at d8148719d: hermetic suite 99/99 on Node 26 (98 prior + 1 new ANSI
+regression); `local-startup.mjs` 17 checks pass (1 collision retry recorded
+only as observation; broker close 0; all PIDs absent); `local-ai.mjs` full
+real-Claude run — 15/15 checks: prejoin-stale negative, two successive idle
+digest actions, 612s uninterrupted idle, 10 unique burst digests, zero
+post-idle control writes, same actor/PID across an actual node WebSocket
+reconnect; Claude 2.1.270, tool calls audited. Synthetic signed ingress only —
+still no real-GitHub or chief gate claimed. selfhost-live remains blocked here:
+no cloudflared binary, no cloud env/workspace credentials on this host.
+
+2026-09-18 self-hosted live run (head 5c2e27219): `selfhost-live.mjs` PASSED —
+20/20 checks, run `ghsub-selfhost-e6a8fedbd7`, evidence
+`/tmp/ghsub-live-evidence/report.json`. Real signed GitHub hooks on disposable
+PR fixtures (cloud#3808, relay#1790, software-garden#528) delivered through a
+cloudflared tunnel into candidate Cloud ingestion (eb27afa4), admitted by
+hosted production Relayfile, delivered through local Relaycast engine
+(683e4dcb) + #1759 broker (d4d51f62, sha 1bee6e9a…), and acknowledged by a real
+Codex worker — exact `printf | shasum -a 256` digest ACKs, no polling. Checks:
+management-API-blocked preflight, real redelivery → 409 duplicate_envelope,
+3 successive idle actions, 600.6s no-input idle with zero control writes,
+10 unique burst digests, WS reconnect with same actor/PID, nonmember zero
+deliveries, no stale prejoin action. 38 admissions observed (10×202, 4×409,
+23×429 retried, 1×500 retried). Environment fixes this round: cloudflared
+binary discovery + required-tool preflight (commits 472a9c3bb, 5c2e27219), a
+`shasum` shim for the pinned digest command, and the live
+RELAYFILE_INTERNAL_HMAC_SECRET recovered from SST state into a 600-mode temp
+file (never logged). Cleanup verified: hooks, fixtures, worker, and all 3
+run-owned subscriptions removed (two needed manual retry after socket hang
+up/429); remaining subscription inventory matches the pre-run set. Still not
+proven: production deploys, the deployed Nango forward path, and gate-9
+actual-chief acceptance.

README.md

@@ -2,6 +2,7 @@
 <p align="center"><img alt="TypeScript" src="https://img.shields.io/badge/TypeScript-3178C6?style=flat-square&logo=typescript&logoColor=white"> <a href="https://github.com/AgentWorkforce/relay/actions/workflows/test.yml"><img alt="Tests" src="https://img.shields.io/github/actions/workflow/status/AgentWorkforce/relay/test.yml?branch=main&label=tests&style=flat-square"></a> <a href="https://github.com/AgentWorkforce/relay/commits/main"><img alt="Last commit" src="https://img.shields.io/github/last-commit/AgentWorkforce/relay/main?label=last%20commit&style=flat-square"></a> <a href="https://www.npmjs.com/package/@agent-relay/sdk"><img alt="npm version" src="https://img.shields.io/npm/v/@agent-relay/sdk?label=npm&style=flat-square"></a> <a href="https://www.npmjs.com/package/@agent-relay/sdk"><img alt="Downloads" src="https://img.shields.io/npm/dm/@agent-relay/sdk?label=downloads&style=flat-square"></a> <a href="./LICENSE"><img alt="License" src="https://img.shields.io/badge/license-Apache%202.0-black?style=flat-square"></a></p>
 
 # Infrastructure for coding agents
+
 Tired of copy/pasting from Claude Code into Slack so your coworker can paste it into their agent?
 
 Ever put an important rule in a skill or AGENTS.md, only for the agent to completely ignore it?
@@ -13,30 +14,35 @@ Agent Relay is an open-source toolkit for problems like these. It gives engineer
 Use the pieces you need, or combine them to build workflows across agents, tools, people, and machines.
 
 ### Messaging
+
 Claude/Codex/etc can talk directly through shared channels, threads, DMs, files, search, and real-time events. Agents can run on different machines and still coordinate in the same workspace.
 
 [Read the docs](https://agentrelay.com/docs/introduction)
 
 ### Integrations
+
 GitHub, Linear, Notion, Slack, and other tools are exposed as a virtual filesystem. Agents use ls, cat, grep, and ordinary file writes to work with them.
 
 [Peep the open source repo](https://github.com/agentworkforce/relayfile)
 
 ### Shared Sessions
+
 Capture coding agent sessions so your team and their agents can search previous work, decisions, and context.
 
 [How we capture sessions](https://github.com/agentworkforce/relayhistory) <br>
 [How we capture decisions](https://github.com/agentworkforce/trajectories)
 
 ### Flows
-Turn instructions you hope an agent follows into workflows you can enforce. 
+
+Turn instructions you hope an agent follows into workflows you can enforce.
 
 Define multi-step workflows in TypeScript with deterministic checks, required steps, and human gates. Put the rules that matter in code instead of relying on a skill or prompt to be remembered and followed.
 
 [Learn how write a flow](https://github.com/agentworkforce/flows) (or lets be honest, show your agent how)
 
 ## Getting Started
-The easiest way to get started is to use [Agent Relay Cloud](https://agentrelay.com/flows). 
+
+The easiest way to get started is to use [Agent Relay Cloud](https://agentrelay.com/flows).
 
 You don't need a credit card and you can explore all the pieces without setting up any infrastructure.
 
@@ -53,6 +59,7 @@ npm install -g agent-relay
 ```
 
 ### Self Hosting
+
 Agent Relay has self hosting options for each primitive. We're happy to help you set up the whole system on your environment, just reach out to our team hi(at)agentrelay.com and we'll walk you through it.
 
 ## License

packages/cli/src/cli/commands/integration-recipient.test.ts

@@ -20,12 +20,14 @@ describe('subscription recipient launch', () => {
   };
   let handle: {
     channels?: string[];
+    exit?: { reason: string; code?: number; signal?: string | null };
     waitForReady: ReturnType<typeof vi.fn>;
     release: ReturnType<typeof vi.fn>;
   };
   beforeEach(() => {
     handle = {
       channels: [],
+      exit: { reason: 'exited', code: 1, signal: null },
       waitForReady: vi.fn(async () => ({ reason: 'ready', pid: 123 })),
       release: vi.fn(async () => {}),
     };
@@ -93,6 +95,24 @@ describe('subscription recipient launch', () => {
     expect(handle.release).toHaveBeenCalledOnce();
     expect(client.disconnect).toHaveBeenCalledOnce();
   });
+  it('reports the reaped exit status when PTY-close races the richer event', async () => {
+    let enriched = false;
+    Object.defineProperty(handle, 'exit', {
+      get: () => (enriched ? { reason: 'exited', code: 1, signal: null } : { reason: 'exited' }),
+    });
+    handle.waitForReady.mockResolvedValue({ reason: 'exited', exit: { reason: 'exited' } });
+    const pending = launchSubscriptionRecipient(input);
+    setTimeout(() => {
+      enriched = true;
+    }, 200);
+    await expect(pending).rejects.toThrow('exited ({"reason":"exited","code":1,"signal":null})');
+    expect(handle.release).toHaveBeenCalledWith('subscription startup failed', { deleteIdentity: true });
+  });
+  it('reports the observed exit when no code-bearing event arrives within grace', async () => {
+    Object.defineProperty(handle, 'exit', { get: () => ({ reason: 'exited' }) });
+    handle.waitForReady.mockResolvedValue({ reason: 'exited', exit: { reason: 'exited' } });
+    await expect(launchSubscriptionRecipient(input)).rejects.toThrow('exited ({"reason":"exited"})');
+  }, 10_000);
   it('refuses an older broker before it can join default channels or reuse an identity', async () => {
     client.getSession.mockResolvedValue({ workspace_key: 'rk_live_explicit' });
     await expect(launchSubscriptionRecipient(input)).rejects.toThrow('isolated, create-only spawn support');

packages/cli/src/cli/commands/integration-recipient.ts

@@ -98,8 +98,19 @@ export async function launchSubscriptionRecipient(input: RecipientLaunchInput):
     }
     const ready = await owned.waitForReady(90_000);
     if (ready.reason !== 'ready' || !ready.pid || ready.pid <= 0) {
+      // A PTY-close `agent_exit` can arrive before the reaper's code-bearing
+      // `agent_exited`; while the worker is still registered, hold a short
+      // bounded grace so the reported exit keeps the authoritative status.
+      let exit = ready.exit;
+      if (ready.reason === 'exited' && exit?.code === undefined) {
+        const deadline = Date.now() + 3_000;
+        while (exit?.code === undefined && Date.now() < deadline) {
+          await new Promise((resolve) => setTimeout(resolve, 100));
+          exit = owned.exit ?? exit;
+        }
+      }
       throw new Error(
-        `Recipient ${input.name} failed startup: ${ready.reason}${ready.exit ? ` (${JSON.stringify(ready.exit)})` : ''}`
+        `Recipient ${input.name} failed startup: ${ready.reason}${exit ? ` (${JSON.stringify(exit)})` : ''}`
       );
     }
     process.kill(ready.pid, 0);

tests/e2e/github-subscriptions/README.md

@@ -13,6 +13,21 @@ node tests/e2e/github-subscriptions/run.mjs prepare /absolute/demo-config.json
 
 `prepare` creates one clearly labelled PR per repository, each targeting its own disposable base branch. It records every acknowledged mutation immediately in `manifest.json`; it never updates main. If interrupted between a server mutation and the manifest write, reconcile the deterministic branch/PR names before retrying. Never adopt an unrelated existing fixture. Comments and reviews remain on closed disposable PRs as evidence after cleanup.
 
+`assert` requires the exact adapter-owned path and provider identity captured by the producer, including the PR/commit/author fields appropriate to the event. After a real Actions run finishes, run `resolve` to bind the CI stimulus to its successful completed GitHub Actions check ID; the workflow commit SHA alone is insufficient. Responses before injection, responses more than 120 seconds after stimulus creation, duplicate or early acknowledgments, and receiver exits invalidate the case. The generic receiver task waits for `GHSUB_EXPECT_KIND` semantics before consuming a nonce, so a PR body edit or an in-progress check cannot satisfy a merge/completion case.
+
+Review, root review-comment, and check records use repository-level `reviews/{id}.json`, `comments/{id}.json`, and `checks/{id}.json` paths. A PR directory scope does not cover them. Provision exact adapter paths for independently inventoried pending reviews/comments and in-progress real checks before their terminal event; preserve existing repository routes. The current `subscribe` command provisions its configured issue/PR/repository scope only and does not stage these additional records automatically. An unobserved terminal event must be repeated with a new owned fixture, never replaced by a fabricated event.
+
+For the normal Nango route, configure `nango.destination` with the deployed Cloud Nango webhook URL, `nango.connectionId`, and `nango.providerConfigKey`. Set `NANGO_SECRET_KEY` from the existing environment credential, then run:
+
+```sh
+node tests/e2e/github-subscriptions/run.mjs resolve /absolute/demo-config.json
+node tests/e2e/github-subscriptions/run.mjs capture-nango /absolute/demo-config.json
+```
+
+The read-only Nango Management MCP collector exhausts operation and message pagination, including empty pages with cursors. Forward operations omit top-level connection identity; the collector checks the forwarded request body instead. It saves only the matching destination/status, connection, GitHub delivery ID, event/action, timestamps and hashes, excluding credentials, request headers and provider content. Link these receipts to authenticated Cloud ingress logs by delivery ID and canonical-path hash, then to Relayfile application, broker injection and the actor response. A Nango success receipt alone is not application or demo proof. `assert` continues to report `ready: false`; full acceptance requires the independent nine-gate review.
+
+The observer paginates channel history back to a known boundary before recording coverage. Negative windows last at least 120 seconds and reject nonce delivery or digest acknowledgment. The initial subscription inventory is immutable; subsequent snapshots append to a journal. Cleanup checks each owned branch's expected SHA before deleting it and refuses an externally advanced branch. GitHub's delete-ref API has no atomic SHA precondition, so retain exclusive ownership of fixture branches during cleanup.
+
 The runner can provision and update its owned subscriptions using `subscribe`, and retire them using `unsubscribe`. It refuses unowned binding replacements and verifies old resource IDs disappear. Set `brokerProjectRoot`, `receiverCwd`, `receiverCli`, `subscriptionScope` (`issue`, `pr`, or `repo`) and `spawnReceiver` explicitly. For chief, set `spawnReceiver: false`. Start `collect` before `subscribe`; it reloads channel configuration after provisioning.
 
 ```sh

tests/e2e/github-subscriptions/collection.test.mjs

@@ -0,0 +1,99 @@
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { EventEmitter } from 'node:events';
+import { readFileSync } from 'node:fs';
+import { runInNewContext } from 'node:vm';
+import path from 'node:path';
+
+const source = readFileSync(new URL('./run.mjs', import.meta.url), 'utf8');
+const collectSource = source
+  .slice(source.indexOf('async function collect() {'), source.indexOf('\nfunction emit() {'))
+  .replace("await import(path.join(root, 'packages/harness-driver/dist/index.js'))", 'testHarness');
+
+for (const signalName of ['SIGTERM', 'SIGINT']) {
+  test(`collector cancels a pending history read on ${signalName} and disconnects`, async () => {
+    const lifecycle = new EventEmitter();
+    let disconnected = false,
+      requested = false;
+    const context = {
+      process: lifecycle,
+      AbortController,
+      path,
+      root: '/owned',
+      out: '/evidence',
+      configFile: '/config',
+      config: { receiver: 'owned', actors: { owned: 'owned-channel' }, collectionSeconds: 60 },
+      manifest: { createdAt: new Date().toISOString() },
+      testHarness: {
+        HarnessDriverClient: {
+          connect: () => ({
+            onEvent() {},
+            connectEvents() {},
+            disconnect() {
+              disconnected = true;
+            },
+          }),
+        },
+      },
+      readLines: () => [],
+      readFileSync: () => '{"actors":{"owned":"owned-channel"}}',
+      appendFileSync() {
+        throw new Error('cancelled reads must not claim coverage');
+      },
+      console: { log() {} },
+      pause: async () => {},
+      collectUnseenMessages: async (fetchPage) => fetchPage(),
+      cast: async (_route, signal) => {
+        requested = true;
+        assert(signal, 'history request needs a cancellation signal');
+        queueMicrotask(() => lifecycle.emit(signalName));
+        return new Promise((resolve, reject) =>
+          signal.addEventListener('abort', () => reject(signal.reason), { once: true })
+        );
+      },
+    };
+    await runInNewContext(collectSource + '\ncollect();', context);
+    assert(requested);
+    assert(disconnected);
+    assert.equal(lifecycle.listenerCount('SIGINT'), 0);
+    assert.equal(lifecycle.listenerCount('SIGTERM'), 0);
+  });
+}
+
+test('collector propagates an ordinary history failure', async () => {
+  const lifecycle = new EventEmitter();
+  let disconnected = false;
+  const context = {
+    process: lifecycle,
+    AbortController,
+    path,
+    root: '/owned',
+    out: '/evidence',
+    configFile: '/config',
+    config: { receiver: 'owned', actors: { owned: 'owned-channel' } },
+    manifest: { createdAt: new Date().toISOString() },
+    testHarness: {
+      HarnessDriverClient: {
+        connect: () => ({
+          onEvent() {},
+          connectEvents() {},
+          disconnect() {
+            disconnected = true;
+          },
+        }),
+      },
+    },
+    readLines: () => [],
+    readFileSync: () => '{"actors":{"owned":"owned-channel"}}',
+    appendFileSync() {},
+    console: { log() {} },
+    pause: async () => {},
+    collectUnseenMessages: async (fetchPage) => fetchPage(),
+    cast: async () => {
+      throw new Error('history HTTP503');
+    },
+  };
+  await assert.rejects(runInNewContext(collectSource + '\ncollect();', context), /history HTTP503/);
+  assert(disconnected);
+  assert.equal(lifecycle.listenerCount('SIGTERM'), 0);
+});