Phase 1: Codex native queue delivery

.agentworkforce/features/manifest.yaml

@@ -1,5 +1,5 @@
 version: '1.1'
-updated: '2026-09-20'
+updated: '2026-09-21'
 
 # Every user-facing feature in this repo, categorized and scored.
 #
@@ -108,9 +108,21 @@ categories:
           accepted route recorded and settled by its own rules, and no
           acknowledgement reported that was not observed. The PTY injector is
           its first backend.
-        location: crates/broker/src/delivery/, crates/broker/src/broker/delivery_verification.rs, crates/broker/src/lib.rs, crates/broker/src/pty_worker.rs, crates/broker/src/runtime/, crates/broker/src/node_control.rs, crates/broker/src/node_delivery_probe.rs, crates/broker/src/worker.rs, crates/broker/tests/, packages/contracts/fixtures/event-fixtures.json, packages/harness-driver/src/protocol.ts, tests/benchmarks/harness.ts, tests/benchmarks/reliability.ts, tests/benchmarks/stress.ts, tests/e2e/unlaunched/, tests/fixtures/delivery-contract-evals.test.ts, tests/fixtures/targeted-feature-verification.test.ts, tests/integration/broker/cli-spawn.test.ts, tests/integration/broker/evals/delivery/observation-ledger.unit.test.ts, tests/integration/broker/infra-failures.test.ts, tests/integration/broker/stress.test.ts, tests/integration/broker/utils/assert-helpers.ts, tests/integration/broker/utils/obligation-conformance.ts, tests/parity/
+        location: crates/broker/src/delivery/, crates/broker/src/broker/delivery_verification.rs, crates/broker/src/lib.rs, crates/broker/src/pty_worker.rs, crates/broker/src/wrap.rs, crates/broker/src/runtime/, crates/broker/src/node_control.rs, crates/broker/src/node_delivery_probe.rs, crates/broker/src/worker.rs, crates/broker/tests/, packages/contracts/fixtures/event-fixtures.json, packages/harness-driver/src/protocol.ts, tests/benchmarks/harness.ts, tests/benchmarks/reliability.ts, tests/benchmarks/stress.ts, tests/e2e/unlaunched/, tests/fixtures/delivery-contract-evals.test.ts, tests/fixtures/targeted-feature-verification.test.ts, tests/integration/broker/cli-spawn.test.ts, tests/integration/broker/evals/delivery/observation-ledger.unit.test.ts, tests/integration/broker/infra-failures.test.ts, tests/integration/broker/stress.test.ts, tests/integration/broker/utils/assert-helpers.ts, tests/integration/broker/utils/obligation-conformance.ts, tests/parity/
         verify_tier: 6
 
+      - id: codex-queue-delivery
+        name: Codex Queue Delivery
+        description: >-
+          Native Codex delivery over the public `codex queue` command. Relay
+          resolves the target thread id, carries a `relay-delivery-id` marker in
+          the message body because Codex assigns its own client id, and settles
+          only by observing that marker in the thread's own session file. A
+          capability probe decides selectability at send time, so an installed
+          Codex without `codex queue` falls back before any write.
+        location: crates/broker/src/delivery/codex_queue.rs, crates/broker/src/codex_thread.rs, crates/broker/src/listen_api.rs, crates/relay-pty/src/codex_session.rs, crates/broker/Cargo.toml, packages/cli/src/cli/agent-relay-mcp.ts, tests/e2e/unlaunched/, tests/fixtures/delivery-contract-evals.codex-queue.test.ts, tests/integration/broker/evals/runner.ts
+        verify_tier: 4
+
   agent-management:
     name: Agent Management
     description: Register and manage agents in the workspace

.agentworkforce/trajectories/active/traj_nd5a2dd4waas/trajectory.json

@@ -0,0 +1,46 @@
+{
+  "id": "traj_nd5a2dd4waas",
+  "version": 1,
+  "task": {
+    "title": "Implement native delivery phase 1 with Relayflows"
+  },
+  "status": "active",
+  "startedAt": "2026-09-21T06:13:08.229Z",
+  "agents": [
+    {
+      "name": "default",
+      "role": "lead",
+      "joinedAt": "2026-09-22T22:32:13.924Z"
+    }
+  ],
+  "chapters": [
+    {
+      "id": "chap_60kdytoq4uet",
+      "title": "Work",
+      "agentName": "default",
+      "startedAt": "2026-09-22T22:32:13.924Z",
+      "events": [
+        {
+          "ts": 1790116333925,
+          "type": "decision",
+          "content": "Published Phase 1 as stacked draft PR #1846 on Phase 0 PR #1825: Published Phase 1 as stacked draft PR #1846 on Phase 0 PR #1825",
+          "raw": {
+            "question": "Published Phase 1 as stacked draft PR #1846 on Phase 0 PR #1825",
+            "chosen": "Published Phase 1 as stacked draft PR #1846 on Phase 0 PR #1825",
+            "alternatives": [],
+            "reasoning": "The user requested the work be pushed now. A separate draft branch preserves Phase 0 review scope, exposes the green deterministic evidence, and keeps the pending dual-provider signoff explicit without pushing to main."
+          },
+          "significance": "high"
+        }
+      ]
+    }
+  ],
+  "commits": [],
+  "filesChanged": [],
+  "projectId": "AgentWorkforce/relay",
+  "tags": [],
+  "_trace": {
+    "startRef": "b2b166423560925aa43ddd06fcccd4d78d1afbb1",
+    "endRef": "b2b166423560925aa43ddd06fcccd4d78d1afbb1"
+  }
+}

.gitignore

@@ -106,3 +106,5 @@ tests/integration/broker/evals-reports/
 
 # Relayflows v2 daemon runtime state (journals, sockets, locks)
 .relayflowd/
+.relayflowd-continuation/
+.relayflowd-phase1-signoff-repair/

CHANGELOG.md

@@ -13,9 +13,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Devin CLI is available through Relay PTY, fleet and MCP spawning with isolated worker MCP configuration, preserved approvals, and reliable initial and follow-up message submission.
 - `agent-relay fleet nodes list --pretty` renders the fleet roster as a human-readable table; `agent-relay fleet nodes --pretty` is available as a shorter equivalent, while JSON remains the default.
 
+### Changed
+
+- Codex native delivery appends a `relay-delivery-id` marker to queued messages so Relay can confirm delivery from the Codex thread record without fabricating an acknowledgement.
+- A Codex delivery is acknowledged (and its Relaycast read receipt published) only once the marker appears in the thread rollout as a consumed user item. A message still sitting in Codex's own queue is reported as handed over, so an idle or exited session no longer produces a read receipt for a message nobody read.
+- Setting `manual_flush` on an agent reachable only over a native route (an attached Codex thread) is refused with `manual_flush_unsupported` (HTTP 409); inbound messages for such an agent always drain over the native route instead of parking in a queue the flush path cannot empty.
+
 ### Fixed
 
 - Broker PTY delivery now fails closed when a handoff deadline races an admitted write, preventing fleet retries from injecting the same message twice and retaining cursor-purged siblings as non-redeliverable dead letters.
+- Releasing an agent, or a worker dying, no longer marks a delivery that already reached a durable native route as freely redeliverable: it is dead-lettered in doubt and the withheld engine acknowledgement is recorded on the node delivery probe rather than dropped.
+- A broker restart no longer re-queues a Codex message that a previous broker lifetime already handed to `codex queue`: the accepted route is persisted with the pending delivery and restored into the delivery seam before the first retry.
 - MCP `spawn`, `add_agent`, and direct-message calls now coalesce in-flight JSON-RPC replays and accept an `idempotency_key` for safe retries after a lost response, preventing duplicate workers, false duplicate-name failures, and duplicate DMs while keeping later JSON-RPC ID reuse distinct.
 - `node agent attach --node` now replaces an expired remote terminal session once and reconnects existing local terminal clients, while transient terminal failures retain the bounded resume budget without minting additional sessions.
 

crates/broker/Cargo.toml

@@ -8,10 +8,18 @@ autobins = false
 name = "relay_broker"
 path = "src/lib.rs"
 
+[features]
+default = ["seam-probe"]
+seam-probe = []
+
 [[bin]]
 name = "agent-relay-broker"
 path = "src/main.rs"
 
+[[test]]
+name = "delivery_seam_invariants"
+required-features = ["seam-probe"]
+
 [dependencies]
 anyhow = "1.0"
 axum = { version = "0.8", features = ["ws"] }

crates/broker/src/broker/delivery_verification.rs

@@ -8,7 +8,7 @@ use serde_json::{json, Value};
 
 use crate::{
     ids::{DeliveryId, EventId, MessageTarget, RequestId, WorkspaceAlias, WorkspaceId},
-    util::ansi::strip_ansi,
+    util::ansi::{floor_char_boundary, strip_ansi},
     worker::detection::ActivityDetector,
 };
 
@@ -193,15 +193,46 @@ impl VerificationOutput {
 
     /// Retained output read after the supplied producer sequence.
     pub(crate) fn since(&self, boundary: u64) -> Cow<'_, str> {
+        self.since_with_overlap(boundary, 0)
+    }
+
+    /// Retained output read after the supplied producer sequence, plus a
+    /// bounded suffix from the boundary segment.
+    ///
+    /// Unix PTY reads can straddle a verified-write admission: bytes already
+    /// readable before the write keep the boundary sequence, while bytes from
+    /// the just-submitted echo may be coalesced into the same kernel read or
+    /// the immediately following one. Echo verification needs enough context
+    /// to reconstruct a match that begins just before the fresh segment and
+    /// ends after it, without accepting a complete stale match. Callers pass
+    /// `expected.len() - 1`, so a full expected echo wholly before the boundary
+    /// is still one byte too long to match.
+    pub(crate) fn since_with_overlap(&self, boundary: u64, overlap_bytes: usize) -> Cow<'_, str> {
+        self.since_with_overlap_and_fresh_start(boundary, overlap_bytes)
+            .0
+    }
+
+    fn since_with_overlap_and_fresh_start(
+        &self,
+        boundary: u64,
+        overlap_bytes: usize,
+    ) -> (Cow<'_, str>, usize) {
         let Some(segment) = self
             .segments
             .iter()
             .find(|segment| segment.sequence > boundary)
         else {
-            return Cow::Borrowed("");
+            return (Cow::Borrowed(""), 0);
         };
-        let start = segment.start_offset.max(self.base_offset) - self.base_offset;
-        String::from_utf8_lossy(&self.buffer[start..])
+        let fresh_start = segment.start_offset.max(self.base_offset);
+        let start_offset = fresh_start
+            .saturating_sub(overlap_bytes)
+            .max(self.base_offset);
+        let start = start_offset - self.base_offset;
+        (
+            String::from_utf8_lossy(&self.buffer[start..]),
+            fresh_start.saturating_sub(start_offset),
+        )
     }
 
     pub(crate) fn retained(&self) -> Cow<'_, str> {
@@ -214,8 +245,15 @@ pub(crate) fn pending_verification_echo_seen(
     output: &VerificationOutput,
     verification: &PendingVerification,
 ) -> bool {
-    let observed = output.since(verification.output_boundary);
-    check_echo_in_output(&observed, &verification.expected_echo)
+    let fresh = output.since(verification.output_boundary);
+    if full_echo_match(&fresh, &verification.expected_echo) {
+        return true;
+    }
+    let (straddling, fresh_start) = output.since_with_overlap_and_fresh_start(
+        verification.output_boundary,
+        verification.expected_echo.len().saturating_sub(1),
+    );
+    full_echo_match_ending_after(&straddling, &verification.expected_echo, fresh_start)
 }
 
 /// Return a verification whose echo arrived before the PTY write ack was
@@ -346,13 +384,44 @@ pub(crate) fn verification_timeout_frames(
 }
 
 /// Check if the expected echo string appears in PTY output (after stripping ANSI).
+#[cfg(test)]
 pub(crate) fn check_echo_in_output(output: &str, expected: &str) -> bool {
+    full_echo_match(output, expected)
+}
+
+fn full_echo_match(output: &str, expected: &str) -> bool {
+    full_echo_match_ending_after(output, expected, 0)
+}
+
+fn full_echo_match_ending_after(output: &str, expected: &str, min_end: usize) -> bool {
     let clean = strip_ansi(output);
-    if clean.contains(expected) {
+    let clean_min_end =
+        strip_ansi(&output[..floor_char_boundary(output, min_end.min(output.len()))]).len();
+    if contains_match_ending_after(&clean, expected, clean_min_end) {
         return true;
     }
     let normalize = |value: &str| value.replace("\r\n", "\n");
-    normalize(&clean).contains(&normalize(expected))
+    let clean_min_end = normalize(&clean[..floor_char_boundary(&clean, clean_min_end)]).len();
+    let clean = normalize(&clean);
+    let expected = normalize(expected);
+    if contains_match_ending_after(&clean, &expected, clean_min_end) {
+        return true;
+    }
+    // PTYs can report visual soft wraps as line feeds when a long logical line
+    // crosses the terminal width. The delivery still landed if the full
+    // character sequence appears with only those line feeds inserted.
+    let without_lf = |value: &str| value.chars().filter(|ch| *ch != '\n').collect::<String>();
+    let clean_min_end = without_lf(&clean[..floor_char_boundary(&clean, clean_min_end)]).len();
+    contains_match_ending_after(&without_lf(&clean), &without_lf(&expected), clean_min_end)
+}
+
+fn contains_match_ending_after(haystack: &str, needle: &str, min_end: usize) -> bool {
+    if needle.is_empty() {
+        return false;
+    }
+    haystack
+        .match_indices(needle)
+        .any(|(start, _)| start.saturating_add(needle.len()) > min_end)
 }
 
 pub(crate) fn current_timestamp_ms() -> u64 {
@@ -446,6 +515,31 @@ mod tests {
         assert!(check_echo_in_output(output, expected));
     }
 
+    #[test]
+    fn check_echo_tolerates_terminal_soft_wraps() {
+        let output = "Relay message from Alice [workspace-1234567890 / evt_12345678901234\n567890]: hello world";
+        let expected =
+            "Relay message from Alice [workspace-1234567890 / evt_12345678901234567890]: hello world";
+
+        assert!(check_echo_in_output(output, expected));
+    }
+
+    #[test]
+    fn check_echo_rejects_interleaved_reminder_even_with_intact_relay_line() {
+        let expected = "<system-reminder>\nAgent Relay MCP tools are available for replies.\n</system-reminder>\nRelay message from Alice [evt_1]: hello world";
+        let output = "<system-reminder>\nAgent Relay MCP tools are available for replies.\nRelay message from Alice [evt_1]: hello world";
+
+        assert!(!check_echo_in_output(output, expected));
+    }
+
+    #[test]
+    fn check_echo_rejects_partial_wrapped_injection() {
+        let expected = "<system-reminder>\nAgent Relay MCP tools are available for replies.\n</system-reminder>\nRelay message from Alice [evt_1]: hello world";
+        let output = "<system-reminder>\nAgent Relay MCP tools are available for replies.\n</system-reminder>\nRelay message from Alice [evt_1]: hello";
+
+        assert!(!check_echo_in_output(output, expected));
+    }
+
     #[test]
     fn check_echo_does_not_turn_bare_cr_into_line_break() {
         assert!(!check_echo_in_output("foo\rbar", "foo\nbar"));
@@ -507,6 +601,74 @@ mod tests {
         assert!(pending_verification_echo_seen(&output, &verification));
     }
 
+    #[test]
+    fn normalized_full_echo_cannot_match_stale_overlap() {
+        let expected = format!(
+            "<system-reminder>\nAgent Relay MCP tools are available for replies.\n</system-reminder>\n{}",
+            "Relay message from Alice [evt_repeat]: same body"
+        );
+        let mut output = VerificationOutput::default();
+        output.push_str(&expected.replace('\n', ""));
+        let output_boundary = output.boundary();
+        let verification = PendingVerification {
+            delivery_id: "delivery-repeat".into(),
+            event_id: "evt-repeat".into(),
+            expected_echo: expected,
+            output_boundary,
+            injected_at: Instant::now(),
+            attempts: 1,
+            max_attempts: 1,
+            request_id: None,
+            workspace_id: None,
+            workspace_alias: None,
+            from: "Alice".to_string(),
+            body: "same body".to_string(),
+            target: "Worker".into(),
+        };
+
+        output.push_str("\nfresh prompt only\n");
+        assert!(
+            !pending_verification_echo_seen(&output, &verification),
+            "a stale normalized echo inside the overlap must not verify a later write"
+        );
+        output.push_str(&verification.expected_echo);
+        assert!(
+            pending_verification_echo_seen(&output, &verification),
+            "a full echo observed after the submission boundary still verifies"
+        );
+    }
+
+    #[test]
+    fn relay_line_alone_never_verifies_a_wrapped_injection() {
+        let relay_line = "Relay message from Alice [evt_real]: same body";
+        let expected = format!(
+            "<system-reminder>\nAgent Relay MCP tools are available for replies.\n</system-reminder>\n{relay_line}"
+        );
+        let mut output = VerificationOutput::default();
+        let output_boundary = output.boundary();
+        let verification = PendingVerification {
+            delivery_id: "delivery-real".into(),
+            event_id: "evt-real".into(),
+            expected_echo: expected,
+            output_boundary,
+            injected_at: Instant::now(),
+            attempts: 1,
+            max_attempts: 1,
+            request_id: None,
+            workspace_id: None,
+            workspace_alias: None,
+            from: "Alice".to_string(),
+            body: "same body".to_string(),
+            target: "Worker".into(),
+        };
+
+        output.push_str(relay_line);
+        assert!(
+            !pending_verification_echo_seen(&output, &verification),
+            "real harnesses must echo the full submitted payload, not only the Relay line"
+        );
+    }
+
     #[test]
     fn verification_ignores_matching_output_queued_before_write_submission() {
         let expected = "Relay message from Alice [evt-queued]: same body";
@@ -546,6 +708,44 @@ mod tests {
         );
     }
 
+    #[test]
+    fn verification_reconstructs_echo_split_across_the_boundary() {
+        let expected =
+            "Relay message from Alice [workspace-1234567890 / evt-1234567890]: wrapped body";
+        let split_at = expected
+            .find("evt-")
+            .expect("expected text contains split marker")
+            + 4;
+        let mut output = VerificationOutput::default();
+        output.push_output(1, &expected.as_bytes()[..split_at]);
+        let output_boundary = output.boundary();
+        let verification = PendingVerification {
+            delivery_id: "delivery-split".into(),
+            event_id: "evt-split".into(),
+            expected_echo: expected.to_string(),
+            output_boundary,
+            injected_at: Instant::now(),
+            attempts: 1,
+            max_attempts: 1,
+            request_id: None,
+            workspace_id: None,
+            workspace_alias: None,
+            from: "Alice".to_string(),
+            body: "wrapped body".to_string(),
+            target: "Worker".into(),
+        };
+
+        assert!(
+            !pending_verification_echo_seen(&output, &verification),
+            "a complete pre-boundary echo must not verify"
+        );
+        output.push_output(2, &expected.as_bytes()[split_at..]);
+        assert!(
+            pending_verification_echo_seen(&output, &verification),
+            "a match that starts in the boundary segment and finishes after it must verify"
+        );
+    }
+
     #[test]
     fn verification_offsets_remain_monotonic_when_the_tail_is_trimmed() {
         let mut output = VerificationOutput::default();