diff --git a/CHANGELOG.md b/CHANGELOG.md index a33c7ee67..9cf89eb1c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,8 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- `agent-relay node agent attach --node ` now opens an authenticated terminal session for physical and Daytona fleet nodes, preserving view, drive, and passthrough modes. -- `agent-relay node agent attach --ssh-host ` now provides an explicit SSH fallback for physical fleet nodes without exporting the remote broker or its API key. +- Default-enabled obligation/boomerang lifecycle: obligating DMs (containing `@@c2a-obligation@@`) trigger automatic re-injection to the recipient every 500 ms (up to 3 times) until the author reacts with ✅. Controlled via `RELAY_OBLIGATION_BOOMERANG=0` to disable and `RELAY_OBLIGATION_INTERVAL_MS=` to configure the return interval. +- `agent-relay local agent attach --node ` now opens an authenticated terminal session for physical and Daytona fleet nodes, preserving view, drive, and passthrough modes. +- `agent-relay local agent attach --ssh-host ` now provides an explicit SSH fallback for physical fleet nodes without exporting the remote broker or its API key. - Spawned agents now stamp a `Session-Id:` git trailer on commits when the dispatcher supplies a session reference, enabling auditors to trace each commit back to the session that produced it. - `agent-relay node agent attach` now distinguishes between "agent does not exist" and "agent is running on a different fleet node": when a 404 resolves to a workspace-registered agent with a fleet placement, the error names the node (`agent 'X' is registered on node 'finn-mini'; cross-node attach is not yet supported`) instead of the indistinguishable "no agent named 'X'". diff --git a/crates/broker/src/lib.rs b/crates/broker/src/lib.rs index f7e9938c2..46b75fcd6 100644 --- a/crates/broker/src/lib.rs +++ b/crates/broker/src/lib.rs @@ -27,6 +27,8 @@ pub(crate) mod listen_api; #[allow(dead_code)] pub(crate) mod metrics; pub(crate) mod node_control; +#[allow(dead_code)] +pub(crate) mod obligation; pub(crate) mod priorities; pub(crate) mod pty_worker; #[allow(dead_code)] diff --git a/crates/broker/src/obligation.rs b/crates/broker/src/obligation.rs new file mode 100644 index 000000000..83d885c24 --- /dev/null +++ b/crates/broker/src/obligation.rs @@ -0,0 +1,328 @@ +//! Obligation lifecycle and boomerang for relay#1474. +//! +//! ## Design +//! +//! A message whose body contains `OBLIGATION_MARKER` registers an +//! [`ObligationRecord`] keyed on the message ID. The record tracks: +//! - which agent sent the message (the *author*) +//! - which agent received it (the *recipient*) +//! - when the next boomerang return should fire +//! +//! **Clearing rule (load-bearing):** only the author can discharge an +//! obligation, by reacting with `✅` (`DONE_EMOJI`) on their own message. +//! A recipient `✅` reaction does NOT discharge — the broker checks +//! `reactor == author` before clearing the flag. +//! +//! **Boomerang delivery:** the maintenance tick drains obligations whose +//! `next_fire_at` has passed, then re-injects a knock message into the +//! recipient worker. The injected body carries `RETURN_MARKER` and the +//! original message ID so the conformance fixture can detect it. +//! +//! **Toggle:** `RELAY_OBLIGATION_BOOMERANG` — any value other than `"0"` +//! (and the absent case) enables the feature. Set to `"0"` to suppress all +//! boomerang behaviour. The env var is re-read on every call so the test +//! control arm can set it before spawning the broker process. + +use std::{ + collections::HashMap, + time::{Duration, Instant}, +}; + +// ── Public constants ────────────────────────────────────────────────────────── + +/// Embedded in the message body of an obligating DM by the sender. +pub const OBLIGATION_MARKER: &str = "@@c2a-obligation@@"; + +/// Embedded in the body of every boomerang re-injection so the conformance +/// fixture's `waitForReturn` can detect it. +pub const RETURN_MARKER: &str = "@@c2a-obligation-return@@"; + +/// The emoji that, when reacted by the *author*, discharges the obligation. +pub const DONE_EMOJI: &str = "✅"; + +/// Env var that gates the whole feature. Any value other than `"0"` (and +/// absence) enables it. +pub const BOOMERANG_FLAG: &str = "RELAY_OBLIGATION_BOOMERANG"; + +/// Env var that sets the flat boomerang return interval in milliseconds. +/// Defaults to 500 ms when absent or unparseable. +pub const INTERVAL_FLAG: &str = "RELAY_OBLIGATION_INTERVAL_MS"; + +// ── Feature gate ───────────────────────────────────────────────────────────── + +/// Returns `true` when `RELAY_OBLIGATION_BOOMERANG` is unset or any value +/// other than `"0"`. +pub fn boomerang_enabled() -> bool { + std::env::var(BOOMERANG_FLAG) + .map(|v| v.trim() != "0") + .unwrap_or(true) +} + +/// Returns the configured return interval. Re-read each call so a test that +/// sets the env var after process start (unlikely but possible via +/// `harness.env`) picks it up. +pub fn interval_ms() -> u64 { + std::env::var(INTERVAL_FLAG) + .ok() + .and_then(|v| v.trim().parse::().ok()) + .filter(|&v| v > 0) + .unwrap_or(500) +} + +// ── Record ──────────────────────────────────────────────────────────────────── + +#[derive(Debug, Clone)] +pub(crate) struct ObligationRecord { + /// Message ID of the obligating message (also the store key). + pub message_id: String, + /// Agent name of the sender — the only party who can discharge. + pub author: String, + /// Agent name of the recipient — where boomerang returns are injected. + pub recipient: String, + /// When the obligation was first registered (for GC). + pub registered_at: Instant, + /// When the next boomerang return should fire. + pub next_fire_at: Instant, + /// How many returns have been injected so far. + pub fire_count: u32, + /// `true` once the author reacts ✅. + pub discharged: bool, + /// `true` once the obligation has fired the maximum number of times (3). + /// Exhausted obligations are not drained again. + pub exhausted: bool, +} + +// ── Store ───────────────────────────────────────────────────────────────────── + +/// In-memory store of outstanding obligation records. Lives on +/// [`crate::runtime::event_loop::BrokerRuntime`] and is swept by the +/// 500 ms maintenance tick. +#[derive(Debug, Default)] +pub(crate) struct ObligationStore { + records: HashMap, +} + +impl ObligationStore { + /// Register a new obligation. Idempotent: registering the same message ID + /// twice is a no-op (the first registration wins). + pub fn register( + &mut self, + message_id: String, + author: String, + recipient: String, + interval: Duration, + ) { + if self.records.contains_key(&message_id) { + return; + } + let now = Instant::now(); + self.records.insert( + message_id.clone(), + ObligationRecord { + message_id, + author, + recipient, + registered_at: now, + next_fire_at: now + interval, + fire_count: 0, + discharged: false, + exhausted: false, + }, + ); + } + + /// Attempt to discharge the obligation identified by `message_id`. + /// + /// The discharge succeeds only when `reactor` is the obligation's author. + /// Returns `true` when the record was found and marked discharged. + pub fn try_discharge(&mut self, message_id: &str, reactor: &str) -> bool { + if let Some(record) = self.records.get_mut(message_id) { + if !record.discharged && !record.exhausted && record.author == reactor { + record.discharged = true; + return true; + } + } + false + } + + /// Collect all obligations that are due for a boomerang return, advance + /// their `next_fire_at` by `interval`, and return the + /// `(message_id, recipient)` pairs to inject. + /// + /// Discharged obligations are silently skipped. + pub fn drain_due(&mut self, now: Instant, interval: Duration) -> Vec<(String, String)> { + let mut due = Vec::new(); + for record in self.records.values_mut() { + if record.discharged || record.exhausted || record.next_fire_at > now { + continue; + } + due.push((record.message_id.clone(), record.recipient.clone())); + record.fire_count += 1; + if record.fire_count >= 3 { + record.exhausted = true; + tracing::warn!( + message_id = %record.message_id, + recipient = %record.recipient, + "obligation exhausted after max fires; no more boomerang returns will be sent" + ); + } else { + record.next_fire_at = now + interval; + } + } + due + } + + /// Remove discharged records older than one hour to bound memory growth. + pub fn gc(&mut self, now: Instant) { + const MAX_DISCHARGED_AGE: Duration = Duration::from_secs(3600); + self.records.retain(|_, r| { + (!r.discharged && !r.exhausted) + || now.duration_since(r.registered_at) < MAX_DISCHARGED_AGE + }); + } +} + +// ── Helpers ─────────────────────────────────────────────────────────────────── + +/// Returns `true` when the message body contains the obligation marker. +#[inline] +pub fn is_obligating(body: &str) -> bool { + body.contains(OBLIGATION_MARKER) +} + +/// Build the boomerang knock body that is injected at the recipient. +/// +/// The body must contain both `RETURN_MARKER` and `original_message_id` so +/// `waitForReturn` in the conformance fixture can detect it. +pub fn build_return_body(original_message_id: &str) -> String { + format!( + "{RETURN_MARKER}{original_message_id}\n\ + Your attention is still required. \ + Obligating message id: {original_message_id}" + ) +} + +// ── Unit tests ──────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + fn store_with_obligation(interval: Duration) -> (ObligationStore, Instant) { + let mut store = ObligationStore::default(); + let now = Instant::now(); + store.register("msg-1".into(), "alice".into(), "bob".into(), interval); + (store, now) + } + + #[test] + fn register_idempotent() { + let mut store = ObligationStore::default(); + let interval = Duration::from_secs(5); + store.register("msg-1".into(), "alice".into(), "bob".into(), interval); + store.register("msg-1".into(), "alice2".into(), "bob2".into(), interval); + // Second registration must not overwrite the first. + assert_eq!(store.records["msg-1"].author, "alice"); + } + + #[test] + fn author_discharges_obligation() { + let interval = Duration::from_secs(5); + let (mut store, _now) = store_with_obligation(interval); + assert!(store.try_discharge("msg-1", "alice")); + assert!(store.records["msg-1"].discharged); + } + + #[test] + fn recipient_cannot_discharge() { + let interval = Duration::from_secs(5); + let (mut store, _now) = store_with_obligation(interval); + // "bob" is the recipient, not the author. + assert!(!store.try_discharge("msg-1", "bob")); + assert!(!store.records["msg-1"].discharged); + } + + #[test] + fn unknown_message_id_discharge_is_noop() { + let interval = Duration::from_secs(5); + let (mut store, _now) = store_with_obligation(interval); + assert!(!store.try_discharge("nonexistent", "alice")); + } + + #[test] + fn drain_due_fires_at_interval() { + let interval = Duration::from_millis(100); + let (mut store, _now) = store_with_obligation(interval); + // Nothing due immediately. + let due = store.drain_due(Instant::now(), interval); + assert!(due.is_empty()); + // Past interval: now due. + let future = Instant::now() + interval + Duration::from_millis(50); + let due = store.drain_due(future, interval); + assert_eq!(due.len(), 1); + assert_eq!(due[0].0, "msg-1"); + assert_eq!(due[0].1, "bob"); + // Same instant: not due again (next_fire_at advanced). + let due2 = store.drain_due(future, interval); + assert!(due2.is_empty()); + } + + #[test] + fn discharged_obligation_not_drained() { + let interval = Duration::from_millis(100); + let (mut store, _now) = store_with_obligation(interval); + store.try_discharge("msg-1", "alice"); + let future = Instant::now() + interval + Duration::from_millis(50); + let due = store.drain_due(future, interval); + assert!(due.is_empty()); + } + + #[test] + fn build_return_body_contains_markers() { + let body = build_return_body("msg-99"); + assert!(body.contains(RETURN_MARKER)); + assert!(body.contains("msg-99")); + } + + #[test] + fn is_obligating_detects_marker() { + assert!(is_obligating("hello\n@@c2a-obligation@@{}")); + assert!(!is_obligating("plain message")); + } + + #[test] + fn obligation_exhausts_after_max_fires() { + let interval = Duration::from_millis(100); + let (mut store, _now) = store_with_obligation(interval); + // Fire 3 times; after the 3rd fire the obligation must be exhausted. + for i in 0..3u32 { + let t = + Instant::now() + interval * (i + 1) + Duration::from_millis(50 * (i + 1) as u64); + let due = store.drain_due(t, interval); + assert_eq!(due.len(), 1, "fire {} must still drain", i); + } + // After 3 fires, obligation is exhausted; 4th drain returns empty. + let t4 = Instant::now() + interval * 10; + let due = store.drain_due(t4, interval); + assert!(due.is_empty(), "exhausted obligation must not drain again"); + } + + #[test] + fn gc_removes_old_discharged_records() { + let mut store = ObligationStore::default(); + let interval = Duration::from_secs(5); + store.register("msg-old".into(), "alice".into(), "bob".into(), interval); + // Manually discharge and backdate. + { + let r = store.records.get_mut("msg-old").unwrap(); + r.discharged = true; + r.registered_at = Instant::now() - Duration::from_secs(7200); + } + // Active obligation stays. + store.register("msg-active".into(), "carol".into(), "dave".into(), interval); + store.gc(Instant::now()); + assert!(!store.records.contains_key("msg-old")); + assert!(store.records.contains_key("msg-active")); + } +} diff --git a/crates/broker/src/runtime/event_loop.rs b/crates/broker/src/runtime/event_loop.rs index 1bb2d1997..de76bde2b 100644 --- a/crates/broker/src/runtime/event_loop.rs +++ b/crates/broker/src/runtime/event_loop.rs @@ -260,6 +260,7 @@ pub(crate) struct BrokerRuntime { #[cfg(windows)] pub(super) sigterm: tokio::signal::windows::CtrlShutdown, pub(super) telemetry: TelemetryClient, + pub(super) obligation_store: crate::obligation::ObligationStore, } enum RuntimeEvent { diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index bf74c6e99..2ae572614 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -585,6 +585,20 @@ impl BrokerRuntime { } match queue_result.outcome { InboundQueueOutcome::Queued => { + // P2-3: register obligation only when the message is actually queued + // (not on RejectedFull or WorkerMissing where delivery may not occur). + if crate::obligation::boomerang_enabled() + && crate::obligation::is_obligating(&fields.body) + { + let interval = + std::time::Duration::from_millis(crate::obligation::interval_ms()); + self.obligation_store.register( + deliver.msg_id.to_string(), + fields.from.clone(), + deliver.agent.to_string(), + interval, + ); + } tracing::info!( target = "relay_broker::fleet", agent = %deliver.agent, @@ -614,6 +628,19 @@ impl BrokerRuntime { Ok(FleetDeliverySurfaceOutcome::HoldForManualFlush) } InboundQueueOutcome::DrainNow(to_drain) => { + // P2-3: register obligation only when the message will actually drain. + if crate::obligation::boomerang_enabled() + && crate::obligation::is_obligating(&fields.body) + { + let interval = + std::time::Duration::from_millis(crate::obligation::interval_ms()); + self.obligation_store.register( + deliver.msg_id.to_string(), + fields.from.clone(), + deliver.agent.to_string(), + interval, + ); + } // Mirrors the HTTP send path: drain may surface older // backlog alongside the message this specific `deliver` // frame is for. Only a failure injecting THIS delivery's @@ -671,6 +698,31 @@ impl BrokerRuntime { payload_type = %payload_type, "acking node receipt/reaction delivery without PTY surfacing (deferred)" ); + // Check for ✅ reaction by author to discharge an obligation. + // Read from payload.data first (v5 envelope), then flat payload as fallback. + let data = deliver.payload.get("data"); + let get_field = |field: &str| -> Option<&str> { + data.and_then(|d| d.get(field)) + .or_else(|| deliver.payload.get(field)) + .and_then(|v| v.as_str()) + }; + let action = get_field("action").unwrap_or(""); + let emoji = get_field("emoji").unwrap_or(""); + let agent_name = get_field("agent_name").unwrap_or(""); + let message_id = get_field("message_id").unwrap_or(""); + if action == "added" + && emoji == crate::obligation::DONE_EMOJI + && !message_id.is_empty() + && !agent_name.is_empty() + && self.obligation_store.try_discharge(message_id, agent_name) + { + tracing::info!( + target = "relay_broker::obligation", + message_id = %message_id, + reactor = %agent_name, + "obligation discharged via ✅ reaction" + ); + } Ok(FleetDeliverySurfaceOutcome::Acknowledge) } FleetDeliverySurfacing::AckUnknown => { diff --git a/crates/broker/src/runtime/init.rs b/crates/broker/src/runtime/init.rs index dd89cfc27..b3a655093 100644 --- a/crates/broker/src/runtime/init.rs +++ b/crates/broker/src/runtime/init.rs @@ -721,6 +721,7 @@ pub(crate) async fn run_init(cmd: InitCommand, telemetry: TelemetryClient) -> Re lease_check, sigterm, telemetry, + obligation_store: crate::obligation::ObligationStore::default(), }; runtime.run().await diff --git a/crates/broker/src/runtime/maintenance.rs b/crates/broker/src/runtime/maintenance.rs index 208042fa5..0805599c2 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -685,5 +685,61 @@ impl BrokerRuntime { // Pending deliveries are persisted by the event loop whenever the // map is mutated (see `BrokerRuntime::flush_pending_deliveries`), // so no tick-time snapshot is needed here. + + // Obligation boomerang: GC is unconditional; drain only when feature is + // enabled. The obligation store lives on BrokerRuntime and must be + // accessed through `self` here (the local-reference reborrow at the top + // of this function does not cover it). + self.obligation_store.gc(now); + if crate::obligation::boomerang_enabled() { + let interval = std::time::Duration::from_millis(crate::obligation::interval_ms()); + let due = self.obligation_store.drain_due(now, interval); + for (msg_id, recipient) in due { + let boomerang_body = crate::obligation::build_return_body(&msg_id); + let delivery_id = DeliveryId::new(uuid::Uuid::new_v4().to_string()); + let event_id = EventId::new(uuid::Uuid::new_v4().to_string()); + let relay_delivery = crate::protocol::RelayDelivery { + delivery_id, + event_id: event_id.clone(), + workspace_id: self.default_workspace_id.clone(), + workspace_alias: self.default_workspace.workspace_alias.clone(), + from: "broker".to_string(), + target: crate::ids::MessageTarget::new(recipient.clone()), + body: boomerang_body.clone(), + thread_id: None, + priority: Some(2), + injection_mode: crate::protocol::MessageInjectionMode::Wait, + }; + tracing::info!( + target = "relay_broker::obligation", + recipient = %recipient, + obligation_msg_id = %msg_id, + "injecting boomerang return to recipient" + ); + if let Err(error) = self.workers.deliver(&recipient, relay_delivery).await { + tracing::warn!( + target = "relay_broker::obligation", + recipient = %recipient, + obligation_msg_id = %msg_id, + error = %error, + "failed to inject boomerang return" + ); + } else { + // Emit a relay_inbound event so waitForReturn in the harness + // driver can detect the boomerang injection. + let _ = send_event( + &self.sdk_out_tx, + serde_json::json!({ + "kind": "relay_inbound", + "target": recipient, + "obligation_msg_id": msg_id, + "event_id": event_id.as_str(), + "body": boomerang_body, + }), + ) + .await; + } + } + } } } diff --git a/packages/cli/src/cli/lib/attach-fleet-node.ts b/packages/cli/src/cli/lib/attach-fleet-node.ts index 7e02cc27d..3dabf9b91 100644 --- a/packages/cli/src/cli/lib/attach-fleet-node.ts +++ b/packages/cli/src/cli/lib/attach-fleet-node.ts @@ -106,7 +106,10 @@ function readBody(request: IncomingMessage): Promise> { } function asWsUrl(value: string): string { - return value.replace(/^http/i, 'ws'); + const lower = value.toLowerCase(); + if (lower.startsWith('https://')) return 'wss://' + value.slice(8); + if (lower.startsWith('http://')) return 'ws://' + value.slice(7); + return value; } function safeNodePath(node: string): string { diff --git a/tests/integration/broker/obligation-conformance.test.ts b/tests/integration/broker/obligation-conformance.test.ts new file mode 100644 index 000000000..4d1f7b078 --- /dev/null +++ b/tests/integration/broker/obligation-conformance.test.ts @@ -0,0 +1,509 @@ +/** + * 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 }); + + // 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 + // 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 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(); + let 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), + }); + // 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), + 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/utils/obligation-conformance.ts b/tests/integration/broker/utils/obligation-conformance.ts new file mode 100644 index 000000000..82fb7855b --- /dev/null +++ b/tests/integration/broker/utils/obligation-conformance.ts @@ -0,0 +1,800 @@ +/** + * 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`. + * - Boomerang (`crates/broker/src/obligation.rs`, `ObligationStore`): wired + * into `crates/broker/src/runtime/maintenance.rs`. Obligating DMs + * (containing OBLIGATION_MARKER) register an ObligationRecord; the 500 ms + * maintenance tick drains due obligations and re-injects a knock carrying + * RETURN_MARKER to the recipient (up to 3 times). A ✅ reaction from the + * author discharges the obligation. Toggle: `RELAY_OBLIGATION_BOOMERANG=0` + * disables all boomerang; `RELAY_OBLIGATION_INTERVAL_MS=` sets the + * return interval. After successful injection the broker emits a + * `relay_inbound` event carrying `obligation_msg_id`, which `waitForReturn` + * observes on the native path. + * + * Does not exist (deferred): + * - 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. + * - 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. + * - Discharge by delegate: current implementation enforces author-only + * discharge per the load-bearing clearing rule in obligation.rs. + * + * ── 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 }); + // 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 ────────────────────────────────────────────────────── + +/** + * 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. + * + * The boomerang injection is produced by `crates/broker/src/runtime/maintenance.rs` + * draining the ObligationStore (see `crates/broker/src/obligation.rs`). The broker + * emits a `relay_inbound` event when injecting boomerang returns, which this function + * detects. With RELAY_OBLIGATION_BOOMERANG=0, the injection is suppressed and this + * function will time out, which is the expected finding on a disabled or unmodified broker. + */ +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 { + // `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([ + '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) => + // readers() returns normalized receipts with agentName (the string name) + // and agentId (an opaque identifier). Compare by name only. + entry.agentName === 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(); + }, + }; +}