From 81408c0a6eef766adc2737fbd92afcf9aa5a060a Mon Sep 17 00:00:00 2001 From: Proactive Runtime Bot Date: Tue, 11 Aug 2026 01:42:34 +0200 Subject: [PATCH 1/4] test(broker): obligation-lifecycle conformance fixture (#1474) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an executable conformance fixture for the obligation lifecycle: an obligation is discharged only when the obligating author, or its named discharge delegate, confirms it was answered — not on read, not on a timer, not on the recipient's belief that it replied. This adds no implementation. It is the test that proves the mechanism is missing and that discriminates between an implementation of it and one that simply never discharges anything. Four arms plus a control: - Arm A (must-fire): delivered, read, a non-answering reply, recipient reacts `done`, recipient reacts `seen` — the obligation must still return, and the return must take a model turn at the recipient. Fails on main, by design. - Arm B (must-not-fire): the author reacts `done` — it must not return. Passes on main trivially, because nothing ever returns. A alone is satisfied by a host that never discharges; B alone by a host that does nothing. The pair is the deliverable. - Arm C (pending): no signals — returns at t, 2t, 3t at equal intervals (no backoff), then escalation observed at a different recipient. - Arm D (pending): arm A run with read state set and unset, compared byte for byte, so read-independence is observable rather than promised. - Control: RELAY_OBLIGATION_BOOMERANG=0 must turn arms A and C red. A suite that stays green with the mechanism removed passes vacuously. Every message crosses the production send path — the same createAgentClient(...).dm(...) call the send_dm MCP tool makes, against the real broker binary. No test-only constructor and no fake host. The model-turn assertion sits behind one pluggable helper, assertRecipientTookTurn, with an implementation per delivery path. The native (AI-SDK) path asserts turn.settled and is proof. The PTY path has no model-turn signal a test can consume, so it uses the weakest honest substitute — the agent subsequently emitted a message — and returns it typed as a proxy so no caller can mistake it for proof. Gated behind RELAY_OBLIGATION_CONFORMANCE=1, following the existing RELAY_INTEGRATION_REAL_CLI idiom, so it does not run in normal CI. Refs #1474 Co-Authored-By: Claude Opus 5 --- .../broker/obligation-conformance.test.ts | 495 +++++++++++ tests/integration/broker/tsconfig.json | 2 + .../broker/utils/obligation-conformance.ts | 775 ++++++++++++++++++ 3 files changed, 1272 insertions(+) create mode 100644 tests/integration/broker/obligation-conformance.test.ts create mode 100644 tests/integration/broker/utils/obligation-conformance.ts diff --git a/tests/integration/broker/obligation-conformance.test.ts b/tests/integration/broker/obligation-conformance.test.ts new file mode 100644 index 000000000..0eff77a85 --- /dev/null +++ b/tests/integration/broker/obligation-conformance.test.ts @@ -0,0 +1,495 @@ +/** + * Obligation-lifecycle conformance fixture. + * + * The defect: a message can be delivered, injected, READ, and never answered, + * and nothing in the system can tell that state apart from an answered one. + * The settled fix: **an obligation is discharged only when the obligating + * author (or its named discharge delegate) confirms it was answered** — not on + * read, not on a timer, not on the recipient's belief that it replied. + * + * This file does not implement that. It is the test that proves it is missing + * and that discriminates between an implementation of it and an implementation + * that merely never discharges anything. + * + * ── Why the arms come in a pair ───────────────────────────────────────────── + * + * "Fails before the change" proves novelty, not relevance: every test of a + * feature that does not exist fails. What discriminates is the pair. + * + * Arm A (must-fire) delivered + read + a non-answering reply + the + * recipient reacts `done` + reacts `seen` + * -> the obligation MUST still return. + * FAILS on unmodified main. + * + * Arm B (must-not-fire) the AUTHOR reacts `done` naming the recipient + * -> it MUST NOT return. + * PASSES on unmodified main, trivially, because + * nothing ever returns. + * + * A alone is satisfied by a host that never discharges anything. B alone is + * satisfied by today's code doing nothing. Neither is worth anything on its + * own. A-fails / B-passes is the expected and correct pre-implementation state. + * + * Arm C (pending) no signals at all -> returns at t, 2t, 3t at EQUAL + * intervals (no backoff), then an escalation observed + * at a DIFFERENT recipient. + * + * Arm D (pending) arm A run twice, once with read state set and once + * unset -> byte-identical behaviour. This makes + * read-independence observable instead of a promise. + * + * Control the same suite with the boomerang mechanism disabled + * must turn arms A and C RED. A suite that stays green + * with the feature removed is passing vacuously, which + * is the same failure shape as the bug. + * + * ── Running it ────────────────────────────────────────────────────────────── + * + * npx tsc -p tests/integration/broker/tsconfig.json + * cd tests/integration/broker + * + * # conformance run (arms A and C are expected RED on main) + * RELAY_OBLIGATION_CONFORMANCE=1 \ + * RELAY_OBLIGATION_PATH=native \ + * RELAY_OBLIGATION_MODEL=openai/gpt-4o-mini \ + * node --test dist/obligation-conformance.test.js + * + * # control run (mechanism disabled — arms A and C must be RED) + * RELAY_OBLIGATION_CONFORMANCE=1 \ + * RELAY_OBLIGATION_PATH=native \ + * RELAY_OBLIGATION_MODEL=openai/gpt-4o-mini \ + * RELAY_OBLIGATION_BOOMERANG=0 \ + * node --test dist/obligation-conformance.test.js + * + * Without RELAY_OBLIGATION_CONFORMANCE=1 every arm skips, so this never runs in + * normal CI. It is expected to fail on unmodified main, by design. + * + * Requires: the agent-relay-broker binary (AGENT_RELAY_BIN or target/debug), + * and a Relaycast workspace key (RELAY_API_KEY, or one is minted). + */ +import assert from 'node:assert/strict'; +import test, { type TestContext } from 'node:test'; + +import { + type ArmTranscript, + type ConformanceContext, + type TurnEvidence, + assertEvidenceIsProof, + assertNoReturn, + assertRecipientTookTurn, + boomerangDisabled, + clientFor, + ConformancePreconditionError, + emitSignal, + readReactions, + recipientHasReadReceipt, + sendObligatingDm, + serializeTranscript, + skipUnlessConformance, + startConformanceContext, + waitForDelivery, + waitForReturn, + watchForTurn, + BOOMERANG_FLAG, + SIGNAL_GLYPH, +} from './utils/obligation-conformance.js'; +import { sleep } from './utils/cli-helpers.js'; + +const QUESTION = 'Is the release gate open? I am holding until you rule.'; + +const DECLARATION = { + blocks: 'halted', + defaultAction: 'Keep the release paused', +} as const; + +/** + * Run an arm body, or — when the suite is running as the control with the + * mechanism disabled — assert that it goes RED. + * + * Note honestly what this does and does not prove today. With no mechanism at + * all, the arm is red under both settings, so the control does not yet + * discriminate. It becomes the load-bearing check the moment boomerang exists: + * from then on the conformance run must be green and this run must stay red, + * and any implementation that ignores the toggle fails here. + */ +async function runOrExpectRed(name: string, body: () => Promise): Promise { + if (!boomerangDisabled()) { + await body(); + return; + } + await assert.rejects( + body, + (error: unknown) => { + // A precondition failure is not the red the control is looking for. If + // the message never arrived, the arm never got as far as testing the + // mechanism, and calling that "correctly failed" would be reporting a + // result nobody observed. + if (error instanceof ConformancePreconditionError) throw error; + return true; + }, + `control: with ${BOOMERANG_FLAG}=0 arm ${name} must fail. It did not, which means the ` + + `arm passes without the mechanism it is supposed to be testing.` + ); +} + +// ══════════════════════════════════════════════════════════════════════════════ +// ARM A — must-fire +// ══════════════════════════════════════════════════════════════════════════════ + +test( + 'obligation arm A: read, replied-to, and recipient-signalled `done` — the obligation MUST still return', + { timeout: 300_000 }, + async (t: TestContext) => { + if (skipUnlessConformance(t)) return; + + const ctx = await startConformanceContext({ label: 'a' }); + try { + await runOrExpectRed('A', () => armA(ctx, { setReadState: true })); + } finally { + await ctx.stop(); + } + } +); + +interface ArmARun { + transcript: ArmTranscript; + /** + * Whether the recipient's read receipt was actually set. This is arm D's + * independent variable and is deliberately NOT part of the transcript, which + * is the dependent one. + */ + readStateObserved: boolean; +} + +/** + * Arm A body. Returns the normalised transcript so arm D can compare two runs. + * + * Every message here crosses the production send path: + * `createAgentClient({ agentToken }).dm(...)` is the exact call the + * `mcp__agent-relay__send_dm` tool makes, and the broker under test is the real + * binary spawned by BrokerHarness. There is no test-only constructor and no + * fake host anywhere in this flow. + */ +async function armA(ctx: ConformanceContext, options: { setReadState: boolean }): Promise { + const { harness, author, recipient, path, intervalMs } = ctx; + + harness.clearEvents(); + const since = harness.getEvents().length; + + // 1. The author sends an obligating event through the production send path. + const obligationId = await sendObligatingDm(author, recipient.name, QUESTION, DECLARATION); + + // 2. It is delivered and injected at the recipient. A failure here is a + // delivery-path failure and says nothing about the obligation lifecycle; + // waitForDelivery's error message says so explicitly. + await waitForDelivery(harness, recipient.name, { since, timeoutMs: 60_000 }); + + // 3. Read state. The point of the arm is that this must not matter. + // + // On the PTY path the broker sets it automatically anyway: delivery is + // confirmed by scanning recipient stdout for the injected string, the + // worker acks, and the read-ack follows from that match with no model in + // the loop. The native runtime reaches the same place by a shorter route — + // the sidecar acks the `deliver_relay` frame and the broker marks it read + // off that ack. The "unset" variant used by arm D is therefore only as + // unset as the substrate permits, which is what arm D checks first. + if (options.setReadState) { + await clientFor(recipient).markRead(obligationId); + } + + // 4. The recipient posts a reply that does NOT answer the question, through + // the production send path. + await clientFor(recipient).dm(author.name, 'Got it — looking at this shortly.'); + + // 5. The recipient emits `done` and then `seen`. + // + // Per the protocol a recipient's `done` is a claim that it answered: + // evidence, not a discharge. A party's signal never states anything about + // another party's work. Neither of these may close the obligation. + await emitSignal(recipient, obligationId, 'done'); + await emitSignal(recipient, obligationId, 'seen'); + + const readStateObserved = await recipientHasReadReceipt(author, obligationId, recipient.name); + + // 6. The obligation MUST still return, and the return MUST take a model turn + // at the recipient. + // Arm A records no interval bucket. Interval spacing is arm C's + // assertion; carrying a wall-clock-derived number in arm A's transcript + // would make arm D's byte-for-byte comparison fail on scheduling jitter + // rather than on anything to do with read state. + const returns: ArmTranscript['returns'] = []; + const evidence: TurnEvidence[] = []; + + const watch = await watchForTurn(harness, recipient.name, path); + const observed = await waitForReturn(harness, recipient.name, obligationId, { + since, + timeoutMs: intervalMs * 3, + }); + returns.push({ index: 1, via: observed.via }); + + const turn = await assertRecipientTookTurn(harness, recipient.name, watch, { + timeoutMs: 120_000, + path, + }); + assertEvidenceIsProof(turn, path); + evidence.push(turn); + + return { + transcript: { arm: 'A', returns, turnEvidence: evidence.map((e) => e.kind), escalations: 0 }, + readStateObserved, + }; +} + +// ══════════════════════════════════════════════════════════════════════════════ +// ARM B — must-not-fire +// ══════════════════════════════════════════════════════════════════════════════ + +test( + 'obligation arm B: the AUTHOR reacts `done` naming the recipient — it MUST NOT return', + { timeout: 300_000 }, + async (t: TestContext) => { + if (skipUnlessConformance(t)) return; + + const ctx = await startConformanceContext({ label: 'b' }); + try { + const { harness, author, recipient, intervalMs } = ctx; + harness.clearEvents(); + const since = harness.getEvents().length; + + const obligationId = await sendObligatingDm(author, recipient.name, QUESTION, DECLARATION); + await waitForDelivery(harness, recipient.name, { since, timeoutMs: 60_000 }); + + // The author discharges. + // + // SUBSTRATE GAP, stated rather than papered over: the spec requires a + // discharging reaction to name, in `recipient`, the recipient it + // discharges, and says a reaction naming no recipient discharges + // nothing. The reaction record here is + // `{ id, message_id, agent_id, emoji, created_at }` — there is no + // `recipient` field to set. So this call cannot express the naming the + // spec requires; it is the closest the substrate allows, and the arm is + // correspondingly weaker than the spec until the field exists. + await emitSignal(author, obligationId, 'done'); + + // Arm B is deliberately NOT wrapped in runOrExpectRed. It must hold with + // the mechanism enabled and with it disabled alike — a control run that + // turned B red would mean the control had broken the must-not-fire half + // of the pair rather than removed the mechanism. + await assertNoReturn(harness, recipient.name, obligationId, { + since, + windowMs: intervalMs * 3 + 2_000, + }); + + // Said plainly so no one reads a green B as evidence of anything: + // on unmodified main this passes because nothing ever returns. It is + // only meaningful paired with arm A. + } finally { + await ctx.stop(); + } + } +); + +// ══════════════════════════════════════════════════════════════════════════════ +// ARM C — PENDING IMPLEMENTATION +// ══════════════════════════════════════════════════════════════════════════════ + +test( + 'obligation arm C [pending-implementation]: no signals — returns at t, 2t, 3t at EQUAL intervals, then escalation at a different recipient', + { timeout: 600_000 }, + async (t: TestContext) => { + if (skipUnlessConformance(t)) return; + + const ctx = await startConformanceContext({ label: 'c', withEscalationTarget: true }); + try { + await runOrExpectRed('C', () => armC(ctx)); + } finally { + await ctx.stop(); + } + } +); + +/** + * PENDING IMPLEMENTATION. + * + * Two things about this arm are worth stating. + * + * First, it asserts equal intervals, not merely three returns. Backing off + * makes an unanswered obligation quieter over time, which is the failure being + * fixed, so a backoff must fail this arm rather than slip through it. + * + * Second, it sleeps through real wall-clock time, which is a defect in the arm + * and not a property of the design. The broker's existing Scheduler takes `now` + * as a parameter instead of reading the clock internally + * (crates/broker/src/scheduler.rs), which is exactly what makes it testable. + * Boomerang state should do the same; when it does, this arm should drive the + * clock instead of waiting on it, and RELAY_OBLIGATION_INTERVAL_MS becomes + * unnecessary. + */ +async function armC(ctx: ConformanceContext): Promise { + const { harness, author, recipient, escalationTarget, intervalMs } = ctx; + + harness.clearEvents(); + const since = harness.getEvents().length; + const sentAt = Date.now(); + + const obligationId = await sendObligatingDm(author, recipient.name, QUESTION, DECLARATION); + await waitForDelivery(harness, recipient.name, { since, timeoutMs: 60_000 }); + + // No signals at all. Nobody reads it, nobody reacts, nobody replies. + + const returns: ArmTranscript['returns'] = []; + for (let index = 1; index <= 3; index += 1) { + const observed = await waitForReturn(harness, recipient.name, obligationId, { + since, + timeoutMs: intervalMs * (index + 1), + }); + returns.push({ + index, + bucket: Math.round((Date.now() - sentAt) / intervalMs), + via: observed.via, + }); + } + + // Equal intervals: buckets must be 1, 2, 3. Any backoff pushes the later + // buckets out and fails here. + assert.deepEqual( + returns.map((entry) => entry.bucket), + [1, 2, 3], + `returns must arrive at t, 2t and 3t at equal intervals (no backoff). ` + + `Observed buckets: ${JSON.stringify(returns.map((entry) => entry.bucket))}` + ); + + // Escalation is observed at a DIFFERENT recipient. Each rung is itself an + // obligating event addressed to one recipient, so the ladder is recursive and + // an escalation target that fails silently is caught one rung up. + const escalation = await waitForReturn(harness, escalationTarget.name, obligationId, { + since, + timeoutMs: intervalMs * 2, + }); + assert.ok(escalation.via, 'escalation must reach a recipient other than the original'); + + return { arm: 'C', returns, turnEvidence: [], escalations: 1 }; +} + +// ══════════════════════════════════════════════════════════════════════════════ +// ARM D — PENDING IMPLEMENTATION +// ══════════════════════════════════════════════════════════════════════════════ + +test( + 'obligation arm D [pending-implementation]: arm A with read state set and unset — byte-identical behaviour', + { timeout: 600_000 }, + async (t: TestContext) => { + if (skipUnlessConformance(t)) return; + + // Arm D is not wrapped in runOrExpectRed: it is a statement about the shape + // of the behaviour, not about whether the mechanism fires. With the + // mechanism disabled both runs are empty and identical, which is true but + // uninformative — the two guards below are what stop that from reading as + // a pass. + const withRead = await startConformanceContext({ label: 'd-read' }); + let readRun: ArmARun; + try { + readRun = await armA(withRead, { setReadState: true }); + } finally { + await withRead.stop(); + } + + const withoutRead = await startConformanceContext({ label: 'd-unread' }); + let unreadRun: ArmARun; + try { + unreadRun = await armA(withoutRead, { setReadState: false }); + } finally { + await withoutRead.stop(); + } + + // Guard 1 — did the independent variable actually vary? + // + // Usually it will not, and that is a substrate finding rather than a bug in + // the arm. The broker marks an inbound message read on the recipient's + // behalf as soon as the delivery is acked, on both runtimes, so "read state + // unset" is not a state a test can currently put the system into. Skipping + // the explicit markRead call does not make the message unread. + // + // The arm fails loudly here rather than comparing two runs that were + // secretly the same run. Asserting read-independence from two identical + // read states would be exactly the substitution of a well-formed signal for + // an unverified fact that this whole issue is about. + assert.notEqual( + readRun.readStateObserved, + unreadRun.readStateObserved, + 'arm D could not vary read state: the recipient read receipt was ' + + `${readRun.readStateObserved} in both runs. The broker sets it automatically off the ` + + 'delivery ack, so read-independence is not observable on this substrate yet. Making it ' + + 'observable needs a way to suppress the automatic read-ack for one delivery — until then ' + + 'this arm asserts nothing and must not be reported as passing.' + ); + + // Guard 2 — equality alone is vacuously satisfiable: two runs that both + // observed nothing are byte-identical. Require the runs to have observed + // the conformant behaviour before comparing them. + assert.ok( + readRun.transcript.returns.length >= 1, + 'arm D compares two runs of arm A; with no return observed there is nothing to compare' + ); + + assert.equal( + serializeTranscript(readRun.transcript), + serializeTranscript(unreadRun.transcript), + 'read state must not change behaviour. Read state MUST NOT be an input to discharge ' + + 'anywhere in the implementation, including as an optimisation.' + ); + } +); + +// ══════════════════════════════════════════════════════════════════════════════ +// Substrate gaps — these PASS today and are here to keep the gaps visible. +// ══════════════════════════════════════════════════════════════════════════════ + +test( + 'obligation substrate: a reaction cannot name the recipient it discharges', + { timeout: 120_000 }, + async (t: TestContext) => { + if (skipUnlessConformance(t)) return; + + const ctx = await startConformanceContext({ label: 'gap' }); + try { + const { author, recipient } = ctx; + const obligationId = await sendObligatingDm(author, recipient.name, QUESTION, DECLARATION); + + // The author reacts `done`; so does the recipient. The spec makes these + // two acts categorically different — one discharges, one is merely + // evidence — and the store makes them indistinguishable apart from + // `agent_id`. Neither carries a `recipient`, so neither is a well-formed + // discharging reaction under the spec. + await emitSignal(author, obligationId, 'done'); + await emitSignal(recipient, obligationId, 'done'); + await sleep(1_000); + + const stored = await readReactions(ctx.apiKey, obligationId); + const done = stored.filter((entry) => entry.emoji === SIGNAL_GLYPH.done); + assert.ok(done.length > 0, 'the `done` reactions should be stored'); + + for (const entry of done) { + assert.equal( + 'recipient' in entry, + false, + 'if a reaction has grown a `recipient` field, arm B can finally assert what the spec ' + + 'requires — that a discharging reaction names the recipient it discharges — and ' + + 'this test should be deleted' + ); + } + + // And the author's `done` is indistinguishable from the recipient's + // except by actor, which is the entire author/recipient asymmetry the + // design rests on being absent from the record. + const shapes = new Set(done.map((entry) => JSON.stringify(Object.keys(entry).sort()))); + assert.equal( + shapes.size, + 1, + 'the author and recipient `done` records should currently have identical shapes' + ); + } finally { + await ctx.stop(); + } + } +); diff --git a/tests/integration/broker/tsconfig.json b/tests/integration/broker/tsconfig.json index ac646ba46..0b6c16bec 100644 --- a/tests/integration/broker/tsconfig.json +++ b/tests/integration/broker/tsconfig.json @@ -17,6 +17,8 @@ "@agent-relay/sdk/*": ["packages/sdk/dist/*"], "@agent-relay/harness-driver": ["packages/harness-driver/dist/index.d.ts"], "@agent-relay/harness-driver/*": ["packages/harness-driver/dist/*"], + "@agent-relay/harnesses": ["packages/harnesses/dist/index.d.ts"], + "@agent-relay/harnesses/*": ["packages/harnesses/dist/*"], "@agent-relay/config": ["packages/config/dist/index.d.ts"], "@agent-relay/config/*": ["packages/config/dist/*"], "@agent-relay/utils": ["packages/utils/dist/index.d.ts"], diff --git a/tests/integration/broker/utils/obligation-conformance.ts b/tests/integration/broker/utils/obligation-conformance.ts new file mode 100644 index 000000000..b83097521 --- /dev/null +++ b/tests/integration/broker/utils/obligation-conformance.ts @@ -0,0 +1,775 @@ +/** + * Support code for the obligation-lifecycle conformance fixture. + * + * The fixture proves one property: **an obligation is discharged only when the + * obligating author (or its named discharge delegate) confirms it was + * answered** — not on read, not on the recipient's belief that it replied, and + * not on a timer. The defect being guarded against is that a message can be + * delivered, injected, read and never answered, and nothing in the system can + * tell that state apart from an answered one. + * + * Everything here is deliberately written so that the *assertions* are honest + * on today's substrate even though the *mechanism* does not exist yet. Where + * something cannot be asserted truthfully it is named as a proxy or left + * pending, in code, rather than weakened until it passes. + * + * ── What exists today, and what does not ──────────────────────────────────── + * + * Exists: + * - The production send path: `createAgentClient({ agentToken }).dm(...)`, + * which is the exact call `mcp__agent-relay__send_dm` makes + * (packages/cli/src/cli/mcp/messaging-tools.ts, `send_dm` handler -> + * `getAgentClient(as).dm(to, text, ...)`). + * - Reactions: `.react(messageId, emoji)` / `.unreact(...)`. They store + * `{ id, message_id, agent_id, emoji, created_at }` and carry no + * `recipient` field and no tombstone. + * - Read state: `.markRead(messageId)`. On the PTY path the broker also sets + * it automatically after echo verification — see + * crates/broker/src/runtime/delivery.rs, whose own doc comment says a + * read-ack means "delivered to the recipient location", not proof that a + * model turn cognitively processed the message. + * - A real model-turn signal, but only on the native (AI-SDK) runtime: + * `turn.settled` (packages/harnesses/src/ai-sdk/harness-host.ts), surfaced + * to a driver client through `getAgentEventHistory`. + * + * Does not exist: + * - Any `obligation` object on the wire. `send_dm` has no field for it, so + * this fixture carries it as a text envelope (see `OBLIGATION_MARKER`) and + * that is a stand-in, not the protocol shape. + * - Any semantic on a reaction. A `done` from the author and a `done` from + * the recipient are byte-identical records; neither names a recipient. + * - Boomerang: nothing re-surfaces an unanswered message. Two near misses, + * both worth knowing about before anyone builds this: + * * crates/broker/src/scheduler.rs is not wired into anything. Its only + * references are its own `#[cfg(test)]` module, so it coalesces nothing + * in production today. Its shape is still the right one to copy — + * `push`/`drain_ready` take `now` as a parameter instead of reading the + * clock, which is exactly what would let arm C assert interval spacing + * without sleeping through real time. + * * crates/broker/src/runtime/maintenance.rs runs a 500ms sweep that + * retries deliveries past `next_retry_at`. That is a *transport* retry + * keyed on the pending map: a delivery leaves the map the moment the + * worker acks, so once a message is injected nothing re-surfaces it. + * It is the natural hook for boomerang, and it is not boomerang. + * - Any organisational edge, so `dischargeDelegate` and the escalation ladder + * have no one to resolve to. Arm C therefore names its escalation target + * explicitly rather than resolving it. + * + * ── Prerequisite worth checking before reading any result ─────────────────── + * + * Every send here is Relaycast-mediated. The broker has no local-injection + * shortcut even for a worker running in its own process (see the comment in + * crates/broker/src/runtime/api.rs on the `/api/send` path), so an inbound + * message only reaches a worker if the broker is a live delivery node that + * Relaycast can route back to. Where that is not the case, nothing is delivered + * and no arm can say anything about obligations. `waitForDelivery` raises a + * ConformancePreconditionError for exactly that case rather than letting it be + * misread as a boomerang finding. + */ +import assert from 'node:assert/strict'; +import type { TestContext } from 'node:test'; + +import type { BrokerEvent } from '@agent-relay/harness-driver'; +import { createNativeHarnessLaunch } from '@agent-relay/harnesses'; +import { createAgentClient, createWorkspaceClient } from '@agent-relay/sdk'; +import { RelayCast } from '@relaycast/sdk'; + +import { BrokerHarness, checkPrerequisites, ensureApiKey, uniqueSuffix } from './broker-harness.js'; +import { sleep } from './cli-helpers.js'; + +// ── Environment gating ─────────────────────────────────────────────────────── +// +// The repo gates expensive integration fixtures behind `std::env::var`-style +// opt-in flags (`RELAY_INTEGRATION_REAL_CLI=1` in mcp-injection.test.ts). This +// fixture follows the same idiom with its own flag so it never runs in normal +// CI: it is expected to FAIL on unmodified main, by design. + +/** Opt-in gate. Without it every arm skips. */ +export const CONFORMANCE_FLAG = 'RELAY_OBLIGATION_CONFORMANCE'; + +/** + * The control toggle. `0` means "run the suite with the boomerang mechanism + * disabled"; arms A and C must then go RED. It is exported to the broker + * process as well as read here, so that a future broker-side implementation can + * honour it with the repo's usual `std::env::var("RELAY_OBLIGATION_BOOMERANG")` + * check without the fixture changing shape. + * + * A suite that stays green with the mechanism removed is passing vacuously, + * which is the same failure shape as the bug itself. + */ +export const BOOMERANG_FLAG = 'RELAY_OBLIGATION_BOOMERANG'; + +/** Which substrate the arms run against. See `ConformancePath`. */ +export const PATH_FLAG = 'RELAY_OBLIGATION_PATH'; + +/** Flat return interval, in ms. Also exported to the broker process. */ +export const INTERVAL_FLAG = 'RELAY_OBLIGATION_INTERVAL_MS'; + +/** Model id for the native path (for example `openai/gpt-4o-mini`). */ +export const MODEL_FLAG = 'RELAY_OBLIGATION_MODEL'; + +/** CLI name for the PTY path (for example `claude`). */ +export const CLI_FLAG = 'RELAY_OBLIGATION_CLI'; + +/** AI-SDK adapter backing the native path. */ +const NATIVE_ADAPTER = 'pi'; + +/** + * `native` + * A real model behind the AI-SDK sidecar. `turn.settled` is emitted when + * model generation actually completes, so "the recipient took a model turn" + * is PROVABLE here. This is the only path on which the spec's assertion can + * be made truthfully today, which is why it is the default. + * + * `pty` + * The default production runtime. Delivery is confirmed by echo verification + * and read state is then set by the broker with no model in the loop. There + * is no model-turn signal to assert against, so this path uses a PROXY — + * see `assertRecipientTookTurn`. + * + * `native-fixture` + * The scripted sidecar at tests/integration/broker/fixtures/native-sidecar.js. + * Its `turn.settled` frames are real protocol frames but the turn is a + * script, not a model. Useful only for exercising the fixture's own wiring + * without model credentials; it is NEVER counted as proof. + */ +export type ConformancePath = 'native' | 'pty' | 'native-fixture'; + +export function conformancePath(): ConformancePath { + const raw = (process.env[PATH_FLAG] ?? 'native').trim(); + if (raw === 'native' || raw === 'pty' || raw === 'native-fixture') return raw; + throw new Error(`${PATH_FLAG} must be one of native | pty | native-fixture (got "${raw}")`); +} + +/** True when the suite is running as the control: mechanism disabled. */ +export function boomerangDisabled(): boolean { + return (process.env[BOOMERANG_FLAG] ?? '1').trim() === '0'; +} + +export function returnIntervalMs(): number { + const raw = process.env[INTERVAL_FLAG]; + const parsed = raw === undefined ? Number.NaN : Number.parseInt(raw, 10); + return Number.isFinite(parsed) && parsed > 0 ? parsed : 5_000; +} + +/** + * Skip unless the fixture is explicitly opted into and its prerequisites exist. + * Returns true when the caller should return immediately. + */ +export function skipUnlessConformance(t: TestContext): boolean { + if (process.env[CONFORMANCE_FLAG] !== '1') { + t.skip(`Set ${CONFORMANCE_FLAG}=1 to run the obligation-lifecycle conformance fixture`); + return true; + } + const missing = checkPrerequisites(); + if (missing) { + t.skip(missing); + return true; + } + if (conformancePath() === 'native' && !process.env[MODEL_FLAG]) { + // Deliberately a skip and not a silent downgrade to the scripted sidecar. + // The spec's assertion is on a *model* turn; running it against a script + // and reporting green would be exactly the substitution this issue exists + // to stop. + t.skip( + `${PATH_FLAG}=native requires ${MODEL_FLAG} (a real model id) so the model-turn assertion is provable` + ); + return true; + } + return false; +} + +// ── The obligating event ───────────────────────────────────────────────────── + +/** + * PLACEHOLDER, and marked as one on purpose. + * + * The protocol puts the obligation in a structured `obligation` object on the + * event. Nothing on this wire carries it: `send_dm` takes `{ to, text, mode, + * attachments }` and no more. Until the field exists the fixture encodes the + * object into the message text behind a marker so the arms can be written + * against the real send path today. + * + * When the wire field lands, this envelope is deleted and the object moves into + * the `dm(...)` options. The arms do not change. + */ +export const OBLIGATION_MARKER = '@@c2a-obligation@@'; + +/** The knock a boomerang return is expected to carry. Does not exist yet. */ +export const RETURN_MARKER = '@@c2a-obligation-return@@'; + +export interface ObligationDeclaration { + blocks: 'halted' | 'degraded' | 'none'; + defaultAction: string; + dischargeDelegate?: string; +} + +export function obligatingText(question: string, declaration: ObligationDeclaration): string { + return `${question}\n${OBLIGATION_MARKER}${JSON.stringify(declaration)}`; +} + +// ── Identities ─────────────────────────────────────────────────────────────── + +export interface ConformanceIdentity { + name: string; + /** Agent token (`at_live_...`), the credential every production call uses. */ + token: string; +} + +/** + * Register an identity and keep its token. + * + * The recipient's token is minted here and handed to the broker at spawn time + * (`spawnPty`/`spawnHeadless` accept `agentToken`). That matters: it is the only + * way the fixture can emit the recipient's own read-marks and reactions through + * the production path instead of inventing a test-only side door. Letting the + * broker mint the token would rotate this one and lock the fixture out. + */ +export async function registerIdentity(apiKey: string, name: string): Promise { + const workspace = createWorkspaceClient({ workspaceKey: apiKey }); + const registration = await workspace.agents.registerOrRotate({ name }); + const token = registration.token; + assert.ok(typeof token === 'string' && token.length > 0, `no agent token minted for "${name}"`); + return { name: registration.name ?? name, token }; +} + +export function clientFor(identity: ConformanceIdentity) { + // The exact factory `mcp__agent-relay__send_dm` builds its client with. + return createAgentClient({ agentToken: identity.token }); +} + +/** Send an obligating DM through the production send path. */ +export async function sendObligatingDm( + author: ConformanceIdentity, + recipient: string, + question: string, + declaration: ObligationDeclaration +): Promise { + const response = (await clientFor(author).dm(recipient, obligatingText(question, declaration))) as Record< + string, + unknown + >; + const messageId = extractMessageId(response); + assert.ok(messageId, `send_dm returned no message id: ${JSON.stringify(response)}`); + return messageId; +} + +function extractMessageId(response: Record): string | undefined { + const direct = response.id ?? response.message_id ?? response.messageId; + if (typeof direct === 'string') return direct; + const nested = response.message; + if (nested && typeof nested === 'object') { + const id = (nested as Record).id; + if (typeof id === 'string') return id; + } + return undefined; +} + +// ── Signals ────────────────────────────────────────────────────────────────── +// +// The protocol names signals, not glyphs. The store carries glyphs, so the map +// below is the fixture's own and is the third substrate gap: nothing on the +// wire distinguishes `done` from any other emoji, and no reaction can name the +// recipient it discharges. A discharging reaction that names no recipient +// discharges nothing, so on today's substrate *no* reaction is well-formed +// enough to discharge — which is a gap, not a pass. + +export const SIGNAL_GLYPH = { + seen: '👀', + agree: '👍', + working: '🔧', + queued: '🕐', + claimed: '✋', + done: '✅', + declined: '🙅', + blocked: '🚧', + unclear: '❓', +} as const; + +export type Signal = keyof typeof SIGNAL_GLYPH; + +export async function emitSignal( + actor: ConformanceIdentity, + messageId: string, + signal: Signal +): Promise { + await clientFor(actor).react(messageId, SIGNAL_GLYPH[signal]); +} + +/** + * Read the stored reaction records for a message. + * + * Read back rather than re-written: reactions are unique per + * (message, agent, emoji), so probing by reacting again fails at the store and + * would mask what is being inspected. + */ +export async function readReactions( + apiKey: string, + messageId: string +): Promise>> { + const relay = new RelayCast({ apiKey }); + const groups = (await relay.messages.reactions(messageId)) as unknown as Array>; + return groups; +} + +// ── Model-turn evidence ────────────────────────────────────────────────────── + +/** + * Evidence that the recipient took a turn. + * + * The distinction between the two kinds is the whole point of the fixture and + * must never be collapsed. `proof` is a signal emitted when model generation + * completed. `proxy` is a downstream side effect that is *consistent with* a + * turn having happened and is not the same claim — it is the same class of + * inference as "the message was echoed to stdout, therefore it was read", which + * is the substitution this issue exists to stop. + */ +export type TurnEvidence = + | { kind: 'proof'; source: 'native:turn.settled'; turnId: string } + | { kind: 'proxy'; source: 'pty:agent-emitted-message'; detail: string } + | { kind: 'scripted'; source: 'native-fixture:turn.settled'; turnId: string }; + +export interface TurnWatch { + /** Sequence cursor (native) or event index (pty) captured before the stimulus. */ + baseline: number; +} + +/** Capture the cursor a later `assertRecipientTookTurn` measures from. */ +export async function watchForTurn( + harness: BrokerHarness, + recipient: string, + path: ConformancePath +): Promise { + if (path === 'pty') return { baseline: harness.getEvents().length }; + const history = await harness.client.getAgentEventHistory(recipient, 0); + return { baseline: history.high_water_sequence }; +} + +/** + * The single pluggable assertion the spec requires: "each assertion is on + * whether the recipient model took a turn, not on whether an event was + * emitted." + * + * There is one implementation per delivery path and the arms call only this + * helper, so that when a PTY model-turn signal is added later the arms do not + * get rewritten — only the `pty` branch below does. + * + * @returns evidence, whose `kind` says how strong the claim is. Callers that + * need proof must check it; the helper will not silently upgrade a proxy. + */ +export async function assertRecipientTookTurn( + harness: BrokerHarness, + recipient: string, + watch: TurnWatch, + options: { timeoutMs: number; path: ConformancePath } +): Promise { + const { timeoutMs, path } = options; + + if (path === 'native' || path === 'native-fixture') { + // PROOF (native) — `turn.settled` is emitted by the AI-SDK harness host + // once model generation actually completes + // (packages/harnesses/src/ai-sdk/harness-host.ts). The broker forwards the + // sidecar's canonical agent-event stream verbatim + // (crates/broker/src/runtime/worker_events.rs, `agent_event` branch) and + // the replay buffer keys it per agent, so a `turn.settled` with a sequence + // above the pre-stimulus high-water mark is a turn that happened after the + // stimulus and not a replay of an earlier one. + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const history = await harness.client.getAgentEventHistory(recipient, watch.baseline); + const settled = history.events.find((entry) => entry.event.kind === 'turn.settled'); + if (settled) { + const turnId = String((settled.event as Record).turnId ?? 'unknown'); + return path === 'native' + ? { kind: 'proof', source: 'native:turn.settled', turnId } + : { kind: 'scripted', source: 'native-fixture:turn.settled', turnId }; + } + await sleep(250); + } + throw new Error( + `recipient "${recipient}" did not take a model turn within ${timeoutMs}ms ` + + `(no turn.settled above sequence ${watch.baseline})` + ); + } + + // PROXY, NOT PROOF — read this before trusting it. + // + // The PTY path has no model-turn signal a test can consume. The broker does + // emit `turn.started`/`turn.settled` for PTY workers + // (crates/broker/src/runtime/worker_events.rs, `publish_pty_busy` / + // `publish_pty_idle`) but it labels them `fidelity: "inferred"` in its own + // capability report, derives them from stdout busy/idle boundaries, and + // publishes them only to the hosted event stream — they are not on the local + // agent-event history a driver client can read, which `getAgentEventHistory` + // serves and which only ever holds native sidecar frames + // (crates/broker/src/replay_buffer.rs keys it on `agent_event`). So the + // strongest thing + // available here is that the agent subsequently emitted a relay message of + // its own, which is evidence and not proof. It is returned as `kind: 'proxy'` + // so no caller can mistake it for the spec's assertion, and it is the branch + // to replace the day a real PTY turn signal exists. + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const emitted = harness + .getEvents() + .slice(watch.baseline) + .find( + (event: BrokerEvent) => + event.kind === 'relay_inbound' && (event as { from: string }).from === recipient + ); + if (emitted) { + return { + kind: 'proxy', + source: 'pty:agent-emitted-message', + detail: `relay_inbound from ${recipient}`, + }; + } + await sleep(250); + } + throw new Error( + `recipient "${recipient}" emitted no message within ${timeoutMs}ms ` + + `(PROXY for a model turn on the PTY path — see assertRecipientTookTurn)` + ); +} + +/** Assert the evidence is strong enough for the path being exercised. */ +export function assertEvidenceIsProof(evidence: TurnEvidence, path: ConformancePath): void { + if (path === 'native') { + assert.equal( + evidence.kind, + 'proof', + `${PATH_FLAG}=native must produce model-turn proof, got ${evidence.kind} (${evidence.source})` + ); + } +} + +// ── Observing a boomerang return ───────────────────────────────────────────── + +/** + * Raised when the fixture could not put the system into the state an arm + * assumes — most often, the obligating event never reached the recipient at all. + * + * It is a distinct type so the control run cannot count it as the failure it + * was looking for. A control that accepts *any* exception as "the arm went red" + * would report success when the suite never got as far as testing anything, + * which is the same substitution of a well-formed signal for an unverified fact + * that this fixture exists to catch. + */ +export class ConformancePreconditionError extends Error { + readonly precondition = true; + + constructor(message: string) { + super(`PRECONDITION FAILED, not an obligation finding: ${message}`); + this.name = 'ConformancePreconditionError'; + } +} + +export interface ObservedReturn { + /** 1-based ordinal of this return. */ + index: number; + /** Interval bucket: `round(elapsedSinceObligation / intervalMs)`. */ + bucket: number; + /** Broker event kind the return was observed on. */ + via: string; +} + +function injectionAt(event: BrokerEvent, recipient: string): { body: string; id: string } | null { + if (event.kind === 'relay_inbound') { + const inbound = event as { target: string; body: string; event_id: string }; + if (inbound.target !== recipient) return null; + return { body: inbound.body ?? '', id: inbound.event_id }; + } + if (event.kind === 'worker_stream') { + const stream = event as { name: string; chunk: string }; + if (stream.name !== recipient) return null; + return { body: stream.chunk ?? '', id: '' }; + } + return null; +} + +/** + * Wait for the obligation to return to its recipient. + * + * A return is an injection at the recipient that is not the original delivery: + * it must reference the obligating event and must carry the knock marker. The + * spec is explicit that the return "MUST be injected" and that a host does not + * satisfy it by writing to a store the model reads on its own initiative, which + * is why this watches the injection stream rather than the message store. + * + * Nothing produces such an injection today. This function is expected to time + * out on unmodified main, and that timeout is the finding. + */ +export async function waitForReturn( + harness: BrokerHarness, + recipient: string, + obligationId: string, + options: { since: number; timeoutMs: number } +): Promise<{ via: string }> { + const deadline = Date.now() + options.timeoutMs; + while (Date.now() < deadline) { + const hit = harness + .getEvents() + .slice(options.since) + .find((event) => { + const injection = injectionAt(event, recipient); + if (!injection) return false; + if (injection.id === obligationId) return false; // the original delivery + return injection.body.includes(RETURN_MARKER) && injection.body.includes(obligationId); + }); + if (hit) return { via: hit.kind }; + await sleep(250); + } + throw new Error( + `no boomerang return for obligation ${obligationId} reached "${recipient}" within ` + + `${options.timeoutMs}ms — the obligation was delivered, read and left unanswered, and ` + + `nothing brought it back` + ); +} + +/** + * Wait until the obligating event actually reaches the recipient. + * + * This is deliberately separate from `waitForReturn`, and its failure message + * is deliberately different. "The obligation never came back" and "the message + * never arrived in the first place" are different findings, and an arm that + * reported the first when the second happened would be doing exactly what this + * issue is about: letting a well-formed signal stand in for a fact nobody + * verified. + * + * Delivery is watched across every kind the broker emits for an inbound + * message, because the shape differs by runtime. A PTY delivery runs + * `relay_inbound` -> `delivery_queued` -> `delivery_injected` -> `delivery_ack` + * -> `message_delivery_confirmed` -> `delivery_verified` -> `delivery_read_ack`; + * a native one is thinner, because the sidecar acks the `deliver_relay` frame + * directly and never sends the queued/injected/verified frames. + */ +export async function waitForDelivery( + harness: BrokerHarness, + recipient: string, + options: { since: number; timeoutMs: number } +): Promise { + const kinds = new Set([ + 'relay_inbound', + 'delivery_queued', + 'delivery_injected', + 'delivery_ack', + 'message_delivery_confirmed', + 'delivery_verified', + 'delivery_read_ack', + ]); + const deadline = Date.now() + options.timeoutMs; + while (Date.now() < deadline) { + const hit = harness + .getEvents() + .slice(options.since) + .find((event) => { + if (!kinds.has(event.kind)) return false; + const named = event as unknown as { name?: string; target?: string }; + return named.name === recipient || named.target === recipient; + }); + if (hit) return hit.kind; + await sleep(250); + } + throw new ConformancePreconditionError( + `the obligating event was accepted by the authoritative log but no delivery of it reached ` + + `"${recipient}" within ${options.timeoutMs}ms. The arm cannot say anything about the ` + + `obligation lifecycle because the message never arrived. Check that the broker is enrolled ` + + `as a live delivery node and that the recipient is routable before reading any other ` + + `result from this run.` + ); +} + +/** Assert no return arrives for the whole window. Used by arm B. */ +export async function assertNoReturn( + harness: BrokerHarness, + recipient: string, + obligationId: string, + options: { since: number; windowMs: number } +): Promise { + await sleep(options.windowMs); + const hit = harness + .getEvents() + .slice(options.since) + .find((event) => { + const injection = injectionAt(event, recipient); + if (!injection) return false; + if (injection.id === obligationId) return false; + return injection.body.includes(RETURN_MARKER) && injection.body.includes(obligationId); + }); + assert.equal( + hit, + undefined, + `obligation ${obligationId} returned to "${recipient}" after its author discharged it` + ); +} + +// ── Arm transcripts (arm D) ────────────────────────────────────────────────── + +/** + * A normalised record of what an arm observed, with every volatile field + * (ids, wall-clock timestamps, agent names) removed, so two runs can be + * compared byte for byte. + */ +export interface ArmTranscript { + arm: string; + /** + * `bucket` is present only where the arm asserts interval spacing (arm C). + * A wall-clock-derived number in an arm that does not assert spacing would + * make arm D's byte-for-byte comparison fail on scheduling jitter instead of + * on read state. + */ + returns: Array<{ index: number; via: string; bucket?: number }>; + turnEvidence: Array; + escalations: number; +} + +/** Deterministic, key-order-independent serialisation for a byte-for-byte compare. */ +export function serializeTranscript(transcript: ArmTranscript): string { + const canonical = (value: unknown): unknown => { + if (Array.isArray(value)) return value.map(canonical); + if (value && typeof value === 'object') { + return Object.fromEntries( + Object.entries(value as Record) + .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)) + .map(([key, entry]) => [key, canonical(entry)]) + ); + } + return value; + }; + return JSON.stringify(canonical(transcript)); +} + +/** + * Whether the recipient's read receipt is set on the obligating event. + * + * Used by arm D to check that its independent variable actually varied. It + * usually will not: the broker marks the message read on the recipient's + * behalf as soon as the delivery is acked + * (crates/broker/src/runtime/worker_events.rs, `delivery_ack` arm -> + * `mark_delivery_read_ack` -> `mark_read_as_agent`), and that happens on the + * native runtime as well as the PTY one. Read state is set by infrastructure, + * not by attention — which is the premise of the whole design, and also the + * reason arm D cannot be asserted honestly yet. + */ +export async function recipientHasReadReceipt( + reader: ConformanceIdentity, + messageId: string, + recipientName: string +): Promise { + const readers = (await clientFor(reader).readers(messageId)) as Array>; + return readers.some( + (entry) => + entry.agent_id === recipientName || entry.agentId === recipientName || entry.name === recipientName + ); +} + +// ── Fixture context ────────────────────────────────────────────────────────── + +export interface ConformanceContext { + harness: BrokerHarness; + apiKey: string; + path: ConformancePath; + intervalMs: number; + author: ConformanceIdentity; + recipient: ConformanceIdentity; + /** Second recipient, so arm C can observe escalation at a *different* one. */ + escalationTarget: ConformanceIdentity; + stop(): Promise; +} + +/** + * Boot a broker, register the three identities, and spawn the recipient (and, + * for arm C, the escalation target) on the selected path. + * + * The broker is started with the boomerang and interval flags in its + * environment. They do nothing today — there is no mechanism to configure — but + * wiring them now is what makes the control command real rather than a + * description of one, and it follows the `std::env::var` toggle idiom the + * broker already uses (for example `RELAY_INJECT_RATE_MS` in + * crates/broker/src/pty_worker.rs). + */ +export async function startConformanceContext(options: { + label: string; + withEscalationTarget?: boolean; +}): Promise { + const path = conformancePath(); + const intervalMs = returnIntervalMs(); + const apiKey = await ensureApiKey(); + const suffix = uniqueSuffix(); + + const harness = new BrokerHarness({ + env: { + ...process.env, + [BOOMERANG_FLAG]: boomerangDisabled() ? '0' : '1', + [INTERVAL_FLAG]: String(intervalMs), + }, + }); + await harness.start(); + + const author = await registerIdentity(apiKey, `obl-author-${options.label}-${suffix}`); + const recipient = await registerIdentity(apiKey, `obl-recipient-${options.label}-${suffix}`); + const escalationTarget = await registerIdentity(apiKey, `obl-escalation-${options.label}-${suffix}`); + + const spawn = async (identity: ConformanceIdentity) => { + if (path === 'pty') { + await harness.client.spawnPty({ + name: identity.name, + cli: process.env[CLI_FLAG] ?? 'claude', + channels: ['general'], + agentToken: identity.token, + }); + return; + } + if (path === 'native') { + // Built by the production launch builder rather than hand-rolled here, + // so the sidecar contract this fixture exercises is the one the CLI + // ships (`agent spawn --runtime native` goes through the same + // call in packages/cli/src/cli/lib/client-factory.ts). + const { transport: _transport, ...launch } = createNativeHarnessLaunch(NATIVE_ADAPTER, { + name: identity.name, + channels: ['general'], + model: process.env[MODEL_FLAG], + }); + await harness.client.spawnHeadless({ ...launch, agentToken: identity.token }); + return; + } + // native-fixture: scripted sidecar, wiring validation only. + const fixture = new URL('../fixtures/native-sidecar.js', import.meta.url).pathname; + await harness.client.spawnHeadless({ + name: identity.name, + cli: 'codex', + channels: ['general'], + agentToken: identity.token, + harnessConfig: { + runtime: 'native', + command: process.execPath, + args: [fixture], + sessionId: `obl-${uniqueSuffix()}`, + metadata: { + runtimeKind: 'native', + nativeHarnessProtocolVersion: 1, + nativeHarnessCapabilities: { activeInput: true }, + }, + }, + }); + }; + + await spawn(recipient); + if (options.withEscalationTarget) await spawn(escalationTarget); + + // Give the recipient time to come up before the obligating event is sent. + await sleep(path === 'pty' ? 12_000 : 5_000); + + return { + harness, + apiKey, + path, + intervalMs, + author, + recipient, + escalationTarget, + async stop() { + await harness.stop(); + }, + }; +} From 15fe7d65ce6eea0b2d697519e04a79c4d9529bbd Mon Sep 17 00:00:00 2001 From: c2a-lead-0811b Date: Tue, 11 Aug 2026 11:54:17 +0200 Subject: [PATCH 2/4] fix(conformance): resolve CodeQL alert and three fixture correctness bugs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four fixes identified in reviewer sweeps: 1. Replace Math.random() with crypto.randomBytes in uniqueSuffix() and BrokerHarness brokerName default (broker-harness.ts). Math.random() is non-cryptographic; CodeQL flagged the sessionId use as insecure randomness (GHAS check run 93632914840). No behaviour change — this is test isolation only. 2. Advance the `since` cursor after each observed return in armC (obligation-conformance.test.ts). The cursor was fixed at the pre-send snapshot, so all three waitForReturn calls rediscovered return #1. Buckets were effectively [1,1,1] and the equal-interval assertion passed vacuously. Now `since` advances to getEvents().length after each match, so iterations 2 and 3 wait for distinct events. 3. Fix recipientHasReadReceipt to compare entry.agentName (the string name) instead of entry.agentId (an opaque identifier) against recipientName (obligation-conformance.ts). The old predicate always returned false, making arm D's independent-variable guard unverifiable. 4. Emit turn.settled from the native sidecar after turn.finished (native-sidecar.ts). The AI-SDK harness emits turn.started → turn.finished → turn.settled; the sidecar was stopping at turn.finished, so assertRecipientTookTurn on the native-fixture path always timed out waiting for turn.settled. The sidecar fix is in the fixture, not the assertion: turn.settled is the correct "turn complete" signal per the relay SDK contract (packages/harnesses/src/ai-sdk/harness-host.ts). --- tests/integration/broker/fixtures/native-sidecar.ts | 4 ++++ tests/integration/broker/obligation-conformance.test.ts | 5 ++++- tests/integration/broker/utils/broker-harness.ts | 5 +++-- tests/integration/broker/utils/obligation-conformance.ts | 4 +++- 4 files changed, 14 insertions(+), 4 deletions(-) diff --git a/tests/integration/broker/fixtures/native-sidecar.ts b/tests/integration/broker/fixtures/native-sidecar.ts index ea43f53eb..d998343c6 100644 --- a/tests/integration/broker/fixtures/native-sidecar.ts +++ b/tests/integration/broker/fixtures/native-sidecar.ts @@ -95,6 +95,10 @@ for await (const line of lines) { event('text.delta', { messageId: 'fixture-message', delta: `echo:${String(payload.text ?? '')}` }); event('text.finished', { messageId: 'fixture-message' }); event('turn.finished', { turnId: 'fixture-turn' }); + // turn.settled is the definitive "turn complete" signal emitted by the + // AI-SDK harness after control.done resolves. The sidecar must emit it so + // assertRecipientTookTurn can observe completion on the native-fixture path. + event('turn.settled', { turnId: 'fixture-turn' }); event('activity.changed', { activity: 'idle', previousActivity: 'thinking', diff --git a/tests/integration/broker/obligation-conformance.test.ts b/tests/integration/broker/obligation-conformance.test.ts index 0eff77a85..f446348ef 100644 --- a/tests/integration/broker/obligation-conformance.test.ts +++ b/tests/integration/broker/obligation-conformance.test.ts @@ -329,7 +329,7 @@ async function armC(ctx: ConformanceContext): Promise { const { harness, author, recipient, escalationTarget, intervalMs } = ctx; harness.clearEvents(); - const since = harness.getEvents().length; + let since = harness.getEvents().length; const sentAt = Date.now(); const obligationId = await sendObligatingDm(author, recipient.name, QUESTION, DECLARATION); @@ -343,6 +343,9 @@ async function armC(ctx: ConformanceContext): Promise { since, timeoutMs: intervalMs * (index + 1), }); + // Advance the cursor past this return so the next iteration waits for a + // distinct event rather than re-discovering the same one. + since = harness.getEvents().length; returns.push({ index, bucket: Math.round((Date.now() - sentAt) / intervalMs), diff --git a/tests/integration/broker/utils/broker-harness.ts b/tests/integration/broker/utils/broker-harness.ts index d310c1a54..246af2022 100644 --- a/tests/integration/broker/utils/broker-harness.ts +++ b/tests/integration/broker/utils/broker-harness.ts @@ -5,6 +5,7 @@ * Provides helpers to start/stop the broker, spawn/release agents, * send messages, and wait for specific broker events. */ +import { randomBytes } from 'node:crypto'; import fs from 'node:fs'; import path from 'node:path'; @@ -89,7 +90,7 @@ export class BrokerHarness { binaryArgs: options.binaryArgs ?? {}, brokerName: options.brokerName ?? - `test-harness-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 6)}`, + `test-harness-${Date.now().toString(36)}-${randomBytes(2).toString('hex')}`, channels: options.channels ?? ['general'], cwd: options.cwd ?? process.cwd(), requestTimeoutMs: options.requestTimeoutMs ?? 10_000, @@ -338,5 +339,5 @@ export function checkPrerequisites(): string | null { * Generate a unique name suffix for test isolation. */ export function uniqueSuffix(): string { - return `${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 6)}`; + return `${Date.now().toString(36)}-${randomBytes(2).toString('hex')}`; } diff --git a/tests/integration/broker/utils/obligation-conformance.ts b/tests/integration/broker/utils/obligation-conformance.ts index b83097521..814d48ef0 100644 --- a/tests/integration/broker/utils/obligation-conformance.ts +++ b/tests/integration/broker/utils/obligation-conformance.ts @@ -659,7 +659,9 @@ export async function recipientHasReadReceipt( const readers = (await clientFor(reader).readers(messageId)) as Array>; return readers.some( (entry) => - entry.agent_id === recipientName || entry.agentId === recipientName || entry.name === recipientName + // readers() returns normalized receipts with agentName (the string name) + // and agentId (an opaque identifier). Compare by name only. + entry.agentName === recipientName ); } From 467ab0e40422fb55e3f9d3a0b99c61bb85d48b6b Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Tue, 11 Aug 2026 09:55:33 +0000 Subject: [PATCH 3/4] style: auto-format with Prettier --- tests/integration/broker/utils/broker-harness.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tests/integration/broker/utils/broker-harness.ts b/tests/integration/broker/utils/broker-harness.ts index 246af2022..4672fd556 100644 --- a/tests/integration/broker/utils/broker-harness.ts +++ b/tests/integration/broker/utils/broker-harness.ts @@ -89,8 +89,7 @@ export class BrokerHarness { binaryPath: options.binaryPath ?? resolveBinaryPath(), binaryArgs: options.binaryArgs ?? {}, brokerName: - options.brokerName ?? - `test-harness-${Date.now().toString(36)}-${randomBytes(2).toString('hex')}`, + options.brokerName ?? `test-harness-${Date.now().toString(36)}-${randomBytes(2).toString('hex')}`, channels: options.channels ?? ['general'], cwd: options.cwd ?? process.cwd(), requestTimeoutMs: options.requestTimeoutMs ?? 10_000, From 8502323b2be92c1ac085a9fbf087720f616b66da Mon Sep 17 00:00:00 2001 From: c2a-lead-0811b Date: Tue, 11 Aug 2026 12:07:04 +0200 Subject: [PATCH 4/4] fix(conformance): tighten delivery precondition, reactions assertion, and turn baseline capture MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bug 4 — waitForDelivery: remove relay_inbound and delivery_queued from the accepted-kinds set. Both fire before the message is injected into the recipient worker (relay_inbound is the broker's first sight of the message from Relaycast; delivery_queued is the internal queue slot). Accepting either let arms proceed to read/reply/signal steps before the recipient actually had the message, which is the same substitution of a well-formed signal for an unverified fact that this fixture exists to catch. Minimum signal is now delivery_injected (PTY) / delivery_ack (native). Bug 5 — readReactions: relay.messages.reactions() returns ReactionGroup[] where each group bundles all agents who reacted with the same emoji into { emoji, count, agents: string[] }. The substrate gap test's shapes assertion — that author and recipient done-reaction records have identical key shapes — was trivially true with one group entry per emoji. Flatten into per-actor records ({ emoji, agent_name }) so the assertion requires two entries and the shape comparison is meaningful. Bug 6 — armA turn baseline race: watchForTurn on the native path reads the high-water sequence from getAgentEventHistory. It was called right before waitForReturn, after several network-bound steps (markRead, dm, emitSignal ×2, recipientHasReadReceipt). If RELAY_OBLIGATION_INTERVAL_MS is short, the boomerang return can fire and the recipient sidecar can emit turn.settled before watchForTurn captures the baseline, placing the boomerang turn's sequence <= baseline and making assertRecipientTookTurn miss it. Move the call to immediately after waitForDelivery: after the initial delivery turn completes, before any boomerang stimulus. Co-Authored-By: Claude Sonnet 4.6 --- .../broker/obligation-conformance.test.ts | 13 ++++++++- .../broker/utils/obligation-conformance.ts | 29 ++++++++++++++++--- 2 files changed, 37 insertions(+), 5 deletions(-) diff --git a/tests/integration/broker/obligation-conformance.test.ts b/tests/integration/broker/obligation-conformance.test.ts index f446348ef..4d1f7b078 100644 --- a/tests/integration/broker/obligation-conformance.test.ts +++ b/tests/integration/broker/obligation-conformance.test.ts @@ -184,6 +184,18 @@ async function armA(ctx: ConformanceContext, options: { setReadState: boolean }) // waitForDelivery's error message says so explicitly. await waitForDelivery(harness, recipient.name, { since, timeoutMs: 60_000 }); + // Capture the turn baseline NOW — after the initial delivery is confirmed but + // before the steps below (read/reply/signal) that could span several network + // roundtrips. On the native path, watchForTurn reads the high-water sequence + // from the agent event history. If the boomerang fires during steps 3-5 + // (which is possible when RELAY_OBLIGATION_INTERVAL_MS is short), capturing + // the baseline late means the boomerang turn's turn.settled already has a + // sequence <= the baseline and assertRecipientTookTurn never finds it. + // + // Capturing here places the baseline after the initial delivery turn (so we + // don't mistake it for the boomerang turn) and before any boomerang stimulus. + const watch = await watchForTurn(harness, recipient.name, path); + // 3. Read state. The point of the arm is that this must not matter. // // On the PTY path the broker sets it automatically anyway: delivery is @@ -220,7 +232,6 @@ async function armA(ctx: ConformanceContext, options: { setReadState: boolean }) const returns: ArmTranscript['returns'] = []; const evidence: TurnEvidence[] = []; - const watch = await watchForTurn(harness, recipient.name, path); const observed = await waitForReturn(harness, recipient.name, obligationId, { since, timeoutMs: intervalMs * 3, diff --git a/tests/integration/broker/utils/obligation-conformance.ts b/tests/integration/broker/utils/obligation-conformance.ts index 814d48ef0..8ac0bafb9 100644 --- a/tests/integration/broker/utils/obligation-conformance.ts +++ b/tests/integration/broker/utils/obligation-conformance.ts @@ -308,8 +308,21 @@ export async function readReactions( messageId: string ): Promise>> { const relay = new RelayCast({ apiKey }); - const groups = (await relay.messages.reactions(messageId)) as unknown as Array>; - return groups; + // relay.messages.reactions() returns ReactionGroup[] — each group has + // { emoji: string, count: number, agents: string[] } where `agents` is a list + // of agent name strings. The substrate gap test needs per-actor records so it + // can (a) assert two distinct reactors are stored and (b) compare their key + // shapes. Flatten into one record per agent name, carrying the emoji from the + // group, so the shape assertion in the substrate gap test is meaningful rather + // than trivially true with a single group entry. + const groups = (await relay.messages.reactions(messageId)) as unknown as Array<{ + emoji: string; + count: number; + agents: string[]; + }>; + return groups.flatMap((group) => + group.agents.map((agentName) => ({ emoji: group.emoji, agent_name: agentName })) + ); } // ── Model-turn evidence ────────────────────────────────────────────────────── @@ -548,9 +561,17 @@ export async function waitForDelivery( recipient: string, options: { since: number; timeoutMs: number } ): Promise { + // `relay_inbound` and `delivery_queued` are pre-injection signals: they fire + // when the broker receives the message from Relaycast but before it has been + // injected into the recipient worker. Accepting them here would let the arm + // proceed to read/reply/signal steps before the message actually reached the + // recipient, which is the same substitution of a well-formed signal for an + // unverified fact that this whole issue is about. + // + // PTY path: delivery_injected (stdin written) -> delivery_ack -> ... + // Native path: delivery_ack (sidecar acked deliver_relay frame) only — + // the sidecar never sends queued/injected/verified frames. const kinds = new Set([ - 'relay_inbound', - 'delivery_queued', 'delivery_injected', 'delivery_ack', 'message_delivery_confirmed',