From c92549f2227aa7ca3636c91ba6729e5097cf84d6 Mon Sep 17 00:00:00 2001 From: jiashuoz <39790535+jiashuoz@users.noreply.github.com> Date: Sat, 5 Sep 2026 12:48:01 -0700 Subject: [PATCH 1/3] fix(sendingpolicy): lock the agent FOR NO KEY UPDATE in the accept path The v1.9.0 staging conformance gate failed on eight parallel HITL holds: SQLSTATE 40P01. Each accept transaction inserts its message first, which takes a FOR KEY SHARE lock on the agent row through the foreign key and a row lock on account_usage through the storage trigger, then prepares its operation, which locked the agent FOR UPDATE. FOR UPDATE conflicts with KEY SHARE, so two concurrent sends waited on each other. The direct send path (PrepareExternalTx) has the identical shape and deadlocks the same way under parallel sends; staging simply never ran that case. FOR NO KEY UPDATE keeps every ordering the gate needs (callers serialize against each other and against any update or delete of the row) and does not conflict with a foreign-key share. Same change for the webhook row. Two regression tests reproduce the deadlock deterministically at the gate and through the API, and both fail with the old lock. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01AjfGxvXW6fNKWGFHuo68yX --- internal/e2e/sending_concurrency_e2e_test.go | 46 ++++++++++ internal/sendingpolicy/operations.go | 20 ++++- .../sendingpolicy/store_integration_test.go | 85 +++++++++++++++++++ 3 files changed, 148 insertions(+), 3 deletions(-) create mode 100644 internal/e2e/sending_concurrency_e2e_test.go diff --git a/internal/e2e/sending_concurrency_e2e_test.go b/internal/e2e/sending_concurrency_e2e_test.go new file mode 100644 index 000000000..416dab70d --- /dev/null +++ b/internal/e2e/sending_concurrency_e2e_test.go @@ -0,0 +1,46 @@ +//go:build integration + +package e2e_test + +import ( + "fmt" + "strings" + "sync" + "testing" + + "github.com/tokencanopy/e2a/internal/testutil" +) + +// TestParallelSendsFromOneAgentAllAccept: eight concurrent sends from one +// agent must all be accepted. Each accept transaction inserts the message and +// then prepares its sending operation under the gate; the v1.9.0 staging +// conformance gate caught the two steps deadlocking against each other +// (SQLSTATE 40P01 → 500) when the gate locked the agent FOR UPDATE. +func TestParallelSendsFromOneAgentAllAccept(t *testing.T) { + pool := testutil.TestDB(t) + ts := testutil.TestServer(t, pool, testutil.WithOutboundSMTP("127.0.0.1", 1025, "test.e2a.dev")) + _, key, agent := setupDomainAndAgent(t, ts, "agent@conc.example.com", "conc.example.com", "", "") + + const n = 8 + var wg sync.WaitGroup + results := make([][]byte, n) + statuses := make([]int, n) + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + statuses[i], results[i] = authedJSON(t, "POST", sendURL(ts.HTTPServer.URL, agent.EmailAddress()), key.PlaintextKey, + fmt.Sprintf(`{"to":["alice@example.com"],"subject":"parallel %d","text":"parallel send #%d"}`, i, i)) + }(i) + } + wg.Wait() + + for i := range statuses { + if statuses[i] != 200 && statuses[i] != 202 { + t.Errorf("send %d: status=%d body=%s", i, statuses[i], results[i]) + } + if !strings.Contains(string(results[i]), `"message_id":"msg_`) { + t.Errorf("send %d: no message id in %s", i, results[i]) + } + } +} diff --git a/internal/sendingpolicy/operations.go b/internal/sendingpolicy/operations.go index e0781a498..3f79aafa3 100644 --- a/internal/sendingpolicy/operations.go +++ b/internal/sendingpolicy/operations.go @@ -172,6 +172,17 @@ func (m *Module) PrepareExternalTx(ctx context.Context, tx pgx.Tx, messageID str // Agent before message, matching migration 113 and the irreversible // deletion path: deletion locks an agent and then its messages, so taking // them in the other order here would deadlock against a concurrent purge. + // + // FOR NO KEY UPDATE, not FOR UPDATE. This runs inside the accept + // transaction AFTER the message insert, and every concurrent insert of a + // message for the same agent holds a FOR KEY SHARE lock on the agent row + // through its foreign key. FOR UPDATE conflicts with KEY SHARE, so two + // parallel sends deadlocked: each held its own insert's share lock plus + // the account_usage row its storage trigger took, and each waited for + // the other's agent lock (seen as SQLSTATE 40P01 on staging, v1.9.0). + // NO KEY UPDATE still serializes gate callers against each other and + // against any update or delete of the agent, which is all the ordering + // this needs, and it does not conflict with a foreign-key share. var agentID string err := tx.QueryRow(ctx, `SELECT agent_id FROM messages WHERE id = $1 AND direction = 'outbound'`, messageID, @@ -185,7 +196,7 @@ func (m *Module) PrepareExternalTx(ctx context.Context, tx pgx.Tx, messageID str var userID string err = tx.QueryRow(ctx, - `SELECT user_id FROM agent_identities WHERE id = $1 FOR UPDATE`, agentID, + `SELECT user_id FROM agent_identities WHERE id = $1 FOR NO KEY UPDATE`, agentID, ).Scan(&userID) if errors.Is(err, pgx.ErrNoRows) { return "", OperationRef{}, ErrSourceUnavailable @@ -295,7 +306,7 @@ func (m *Module) PrepareNotificationTx(ctx context.Context, tx pgx.Tx, ref Notif // names another episode is detectably stale. var warnedAt, disabledAt *time.Time err = tx.QueryRow(ctx, - `SELECT user_id, warn_notified_at, auto_disabled_at FROM webhooks WHERE id = $1 FOR UPDATE`, ref.id, + `SELECT user_id, warn_notified_at, auto_disabled_at FROM webhooks WHERE id = $1 FOR NO KEY UPDATE`, ref.id, ).Scan(&userID, &warnedAt, &disabledAt) if errors.Is(err, pgx.ErrNoRows) { err = ErrSourceUnavailable @@ -358,7 +369,10 @@ func lockHITLSourceOwner(ctx context.Context, tx pgx.Tx, messageID string) (stri var userID string err = tx.QueryRow(ctx, - `SELECT user_id FROM agent_identities WHERE id = $1 FOR UPDATE`, agentID, + // NO KEY UPDATE for the same reason as PrepareExternalTx: the hold's + // accept transaction inserted the message first, and concurrent + // inserts hold the agent row FOR KEY SHARE. + `SELECT user_id FROM agent_identities WHERE id = $1 FOR NO KEY UPDATE`, agentID, ).Scan(&userID) if errors.Is(err, pgx.ErrNoRows) { return "", ErrSourceUnavailable diff --git a/internal/sendingpolicy/store_integration_test.go b/internal/sendingpolicy/store_integration_test.go index d2a53f432..2292c4e4d 100644 --- a/internal/sendingpolicy/store_integration_test.go +++ b/internal/sendingpolicy/store_integration_test.go @@ -1222,3 +1222,88 @@ func (f *fixture) tryTx(fn func(tx pgx.Tx) error) error { } return tx.Commit(f.ctx) } + +// TestPrepareDoesNotDeadlockAgainstConcurrentInsert reproduces the v1.9.0 +// staging failure: two accept transactions for the same agent each insert +// their message (taking a FOR KEY SHARE lock on the agent row through the +// foreign key, and the account_usage row through the storage trigger) and +// then prepare their operation. With the agent locked FOR UPDATE the second +// insert waits on the first's account_usage row while the first's prepare +// waits on the second's key share — SQLSTATE 40P01. The gate's NO KEY UPDATE +// lock lets the first prepare proceed. +func TestPrepareDoesNotDeadlockAgainstConcurrentInsert(t *testing.T) { + for name, prepare := range map[string]func(g sendingpolicy.Gate, tx pgx.Tx, messageID string) error{ + "external": func(g sendingpolicy.Gate, tx pgx.Tx, messageID string) error { + _, _, err := g.PrepareExternalTx(context.Background(), tx, messageID) + return err + }, + "hitl notification": func(g sendingpolicy.Gate, tx pgx.Tx, messageID string) error { + _, err := g.PrepareNotificationTx(context.Background(), tx, sendingpolicy.NewHITLNotificationRef(messageID)) + return err + }, + } { + t.Run(name, func(t *testing.T) { + f := newFixture(t) + g := f.gate(sendingpolicy.DisabledPolicy()) + user := f.user("standard") + agent := f.agent(user) + status := "sent" + if name == "hitl notification" { + status = "pending_review" + } + insert := func(tx pgx.Tx, id string) error { + _, err := tx.Exec(f.ctx, + `INSERT INTO messages (id, agent_id, direction, to_recipients, sent_as, status, body_text) + VALUES ($1, $2, 'outbound', ARRAY['rcpt@example.test'], 'relay', $3, 'x')`, + id, agent, status) + return err + } + + txA, err := f.pool.Begin(f.ctx) + if err != nil { + t.Fatal(err) + } + defer func() { _ = txA.Rollback(f.ctx) }() + if err := insert(txA, "msg_lock_a"); err != nil { + t.Fatalf("insert A: %v", err) + } + + // B inserts concurrently: it takes its key share on the agent, then + // its storage-trigger upsert blocks on A's account_usage row. + bDone := make(chan error, 1) + go func() { + txB, err := f.pool.Begin(f.ctx) + if err != nil { + bDone <- err + return + } + defer func() { _ = txB.Rollback(f.ctx) }() + if err := insert(txB, "msg_lock_b"); err != nil { + bDone <- fmt.Errorf("insert B: %w", err) + return + } + if err := prepare(g, txB, "msg_lock_b"); err != nil { + bDone <- fmt.Errorf("prepare B: %w", err) + return + } + bDone <- txB.Commit(f.ctx) + }() + time.Sleep(300 * time.Millisecond) // let B reach its blocked insert + + if err := prepare(g, txA, "msg_lock_a"); err != nil { + t.Fatalf("prepare A must not deadlock against B's in-flight insert: %v", err) + } + if err := txA.Commit(f.ctx); err != nil { + t.Fatalf("commit A: %v", err) + } + select { + case err := <-bDone: + if err != nil { + t.Fatalf("B: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("B never completed after A committed") + } + }) + } +} From bd19183ec0804b51736a5eb4d824231ef8e8a2aa Mon Sep 17 00:00:00 2001 From: jiashuoz <39790535+jiashuoz@users.noreply.github.com> Date: Sat, 5 Sep 2026 13:04:51 -0700 Subject: [PATCH 2/3] test(sendingpolicy): force the deadlock interleaving instead of timing it Review of the lock-order fix: the gate test's 300ms sleep could let A finish before B ever blocked, passing vacuously against the bug. B now reports its backend pid and A waits until pg_stat_activity shows it blocked on a lock. The e2e test no longer calls t.Fatal from worker goroutines, and the PrepareExternalTx ordering comment now describes the function rather than every caller. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01AjfGxvXW6fNKWGFHuo68yX --- internal/e2e/sending_concurrency_e2e_test.go | 43 +++++++++--- internal/sendingpolicy/operations.go | 9 ++- .../sendingpolicy/store_integration_test.go | 66 ++++++++++++++----- 3 files changed, 91 insertions(+), 27 deletions(-) diff --git a/internal/e2e/sending_concurrency_e2e_test.go b/internal/e2e/sending_concurrency_e2e_test.go index 416dab70d..00ea7cdf8 100644 --- a/internal/e2e/sending_concurrency_e2e_test.go +++ b/internal/e2e/sending_concurrency_e2e_test.go @@ -4,6 +4,8 @@ package e2e_test import ( "fmt" + "io" + "net/http" "strings" "sync" "testing" @@ -22,25 +24,48 @@ func TestParallelSendsFromOneAgentAllAccept(t *testing.T) { _, key, agent := setupDomainAndAgent(t, ts, "agent@conc.example.com", "conc.example.com", "", "") const n = 8 + type result struct { + status int + body []byte + err error + } + results := make([]result, n) var wg sync.WaitGroup - results := make([][]byte, n) - statuses := make([]int, n) for i := 0; i < n; i++ { wg.Add(1) go func(i int) { defer wg.Done() - statuses[i], results[i] = authedJSON(t, "POST", sendURL(ts.HTTPServer.URL, agent.EmailAddress()), key.PlaintextKey, - fmt.Sprintf(`{"to":["alice@example.com"],"subject":"parallel %d","text":"parallel send #%d"}`, i, i)) + // No t.Fatal from a worker goroutine: collect and assert after Wait. + body := fmt.Sprintf(`{"to":["alice@example.com"],"subject":"parallel %d","text":"parallel send #%d"}`, i, i) + req, err := http.NewRequest("POST", sendURL(ts.HTTPServer.URL, agent.EmailAddress()), strings.NewReader(body)) + if err != nil { + results[i].err = err + return + } + req.Header.Set("Authorization", "Bearer "+key.PlaintextKey) + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + if err != nil { + results[i].err = err + return + } + defer resp.Body.Close() + out, _ := io.ReadAll(resp.Body) + results[i] = result{status: resp.StatusCode, body: out} }(i) } wg.Wait() - for i := range statuses { - if statuses[i] != 200 && statuses[i] != 202 { - t.Errorf("send %d: status=%d body=%s", i, statuses[i], results[i]) + for i, r := range results { + if r.err != nil { + t.Errorf("send %d: %v", i, r.err) + continue + } + if r.status != 200 && r.status != 202 { + t.Errorf("send %d: status=%d body=%s", i, r.status, r.body) } - if !strings.Contains(string(results[i]), `"message_id":"msg_`) { - t.Errorf("send %d: no message id in %s", i, results[i]) + if !strings.Contains(string(r.body), `"message_id":"msg_`) { + t.Errorf("send %d: no message id in %s", i, r.body) } } } diff --git a/internal/sendingpolicy/operations.go b/internal/sendingpolicy/operations.go index 3f79aafa3..d2e2cb52a 100644 --- a/internal/sendingpolicy/operations.go +++ b/internal/sendingpolicy/operations.go @@ -169,9 +169,12 @@ func (m *Module) PrepareExternalTx(ctx context.Context, tx pgx.Tx, messageID str return "", OperationRef{}, ErrSourceUnavailable } - // Agent before message, matching migration 113 and the irreversible - // deletion path: deletion locks an agent and then its messages, so taking - // them in the other order here would deadlock against a concurrent purge. + // Within this function: agent before message, matching migration 113 and + // the irreversible deletion path (which locks an agent and then its + // messages). The enclosing accept transaction may already hold a message + // row — an approval updates the held message first, a reply locks its + // parent — so the order is a property of this function, not a guarantee + // about every caller. // // FOR NO KEY UPDATE, not FOR UPDATE. This runs inside the accept // transaction AFTER the message insert, and every concurrent insert of a diff --git a/internal/sendingpolicy/store_integration_test.go b/internal/sendingpolicy/store_integration_test.go index 2292c4e4d..3dc8aaeb3 100644 --- a/internal/sendingpolicy/store_integration_test.go +++ b/internal/sendingpolicy/store_integration_test.go @@ -1231,31 +1231,36 @@ func (f *fixture) tryTx(fn func(tx pgx.Tx) error) error { // insert waits on the first's account_usage row while the first's prepare // waits on the second's key share — SQLSTATE 40P01. The gate's NO KEY UPDATE // lock lets the first prepare proceed. +// +// The interleaving is forced, not timed: B reports its backend pid and A +// waits until pg_stat_activity shows that backend blocked on a lock before +// preparing, so the test cannot pass vacuously by A finishing first. func TestPrepareDoesNotDeadlockAgainstConcurrentInsert(t *testing.T) { - for name, prepare := range map[string]func(g sendingpolicy.Gate, tx pgx.Tx, messageID string) error{ - "external": func(g sendingpolicy.Gate, tx pgx.Tx, messageID string) error { + cases := []struct { + name string + status string + prepare func(g sendingpolicy.Gate, tx pgx.Tx, messageID string) error + }{ + {"external", "sent", func(g sendingpolicy.Gate, tx pgx.Tx, messageID string) error { _, _, err := g.PrepareExternalTx(context.Background(), tx, messageID) return err - }, - "hitl notification": func(g sendingpolicy.Gate, tx pgx.Tx, messageID string) error { + }}, + {"hitl notification", "pending_review", func(g sendingpolicy.Gate, tx pgx.Tx, messageID string) error { _, err := g.PrepareNotificationTx(context.Background(), tx, sendingpolicy.NewHITLNotificationRef(messageID)) return err - }, - } { - t.Run(name, func(t *testing.T) { + }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { f := newFixture(t) g := f.gate(sendingpolicy.DisabledPolicy()) user := f.user("standard") agent := f.agent(user) - status := "sent" - if name == "hitl notification" { - status = "pending_review" - } insert := func(tx pgx.Tx, id string) error { _, err := tx.Exec(f.ctx, `INSERT INTO messages (id, agent_id, direction, to_recipients, sent_as, status, body_text) VALUES ($1, $2, 'outbound', ARRAY['rcpt@example.test'], 'relay', $3, 'x')`, - id, agent, status) + id, agent, tc.status) return err } @@ -1270,6 +1275,7 @@ func TestPrepareDoesNotDeadlockAgainstConcurrentInsert(t *testing.T) { // B inserts concurrently: it takes its key share on the agent, then // its storage-trigger upsert blocks on A's account_usage row. + bPID := make(chan int, 1) bDone := make(chan error, 1) go func() { txB, err := f.pool.Begin(f.ctx) @@ -1278,19 +1284,49 @@ func TestPrepareDoesNotDeadlockAgainstConcurrentInsert(t *testing.T) { return } defer func() { _ = txB.Rollback(f.ctx) }() + var pid int + if err := txB.QueryRow(f.ctx, `SELECT pg_backend_pid()`).Scan(&pid); err != nil { + bDone <- err + return + } + bPID <- pid if err := insert(txB, "msg_lock_b"); err != nil { bDone <- fmt.Errorf("insert B: %w", err) return } - if err := prepare(g, txB, "msg_lock_b"); err != nil { + if err := tc.prepare(g, txB, "msg_lock_b"); err != nil { bDone <- fmt.Errorf("prepare B: %w", err) return } bDone <- txB.Commit(f.ctx) }() - time.Sleep(300 * time.Millisecond) // let B reach its blocked insert - if err := prepare(g, txA, "msg_lock_a"); err != nil { + var pid int + select { + case pid = <-bPID: + case err := <-bDone: + t.Fatalf("B ended before starting: %v", err) + case <-time.After(10 * time.Second): + t.Fatal("B never reported its backend") + } + deadline := time.Now().Add(10 * time.Second) + for { + var blocked bool + if err := f.pool.QueryRow(f.ctx, + `SELECT EXISTS (SELECT 1 FROM pg_stat_activity WHERE pid = $1 AND wait_event_type = 'Lock')`, pid, + ).Scan(&blocked); err != nil { + t.Fatal(err) + } + if blocked { + break + } + if time.Now().After(deadline) { + t.Fatal("B never blocked on A's insert; the interleaving this test needs did not happen") + } + time.Sleep(20 * time.Millisecond) + } + + if err := tc.prepare(g, txA, "msg_lock_a"); err != nil { t.Fatalf("prepare A must not deadlock against B's in-flight insert: %v", err) } if err := txA.Commit(f.ctx); err != nil { From 88913164b69147f41ff747362b0216f7e6aa9b60 Mon Sep 17 00:00:00 2001 From: jiashuoz <39790535+jiashuoz@users.noreply.github.com> Date: Sat, 5 Sep 2026 14:26:33 -0700 Subject: [PATCH 3/3] test(conformance): cover parallel direct sends and record the lock rule The staging gate only sent in parallel from a HITL agent; the direct accept path has the same insert-then-lock shape and carries almost all traffic. Add the eight-parallel-direct-sends case, write the accept transaction's lock order into the pipeline design doc, and record the FOR KEY SHARE / FOR UPDATE rule in AGENTS.md. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01AjfGxvXW6fNKWGFHuo68yX --- AGENTS.md | 13 +++++++ docs/design/async-message-pipeline.md | 16 +++++++++ tests/e2e-prod/suites/03-concurrency.test.ts | 37 ++++++++++++++++++++ 3 files changed, 66 insertions(+) diff --git a/AGENTS.md b/AGENTS.md index b92244fdd..4f972639f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -408,6 +408,19 @@ manually on every API change even though the template won't remind you. bundled drive-by cleanup. CI must be green. - **Coverage floors** only move up (see Testing strategy). - **Postgres**: local dev runs on port **5433** (not 5432) via docker compose. +- **Row locks in multi-statement transactions**: an `INSERT` holds + `FOR KEY SHARE` on every row it references by foreign key until commit, and + `FOR UPDATE` conflicts with that. So a `SELECT … FOR UPDATE` on a parent row + taken *after* inserting a child in the same transaction deadlocks against a + concurrent insert for the same parent (v1.9.0: the accept transaction + inserted the message, then the gate locked the agent `FOR UPDATE`; two + parallel sends → SQLSTATE 40P01). Lock the parent `FOR NO KEY UPDATE` + (excludes updates, deletes and other lockers, coexists with key shares), or + lock it before the insert. The accept transaction's full lock order is in + `docs/design/async-message-pipeline.md`; any new lock on that path must be + checked against it, and any parallel-write path needs a concurrency test + (see `TestPrepareDoesNotDeadlockAgainstConcurrentInsert` for the + deterministic two-transaction shape). - The Mailpit service in `docker-compose.yaml` is local-dev only — production deployments must drop it and point `E2A_OUTBOUND_SMTP_*` at a real relay. diff --git a/docs/design/async-message-pipeline.md b/docs/design/async-message-pipeline.md index 89c4f08f0..a2547fd9e 100644 --- a/docs/design/async-message-pipeline.md +++ b/docs/design/async-message-pipeline.md @@ -345,6 +345,22 @@ a paused account's message job is also left unstamped for the worker's hold path. The workers resolve legacy jobs themselves, so the command is a convenience for a clean cutover, not a prerequisite. +**Lock order of the accept transaction.** Every lock the accept path takes, +in order, so the next change can check itself against it: the message +insert takes `FOR KEY SHARE` on the agent row (foreign key) and an +exclusive lock on the account's `account_usage` row (storage trigger); the +gate's Prepare then takes `FOR NO KEY UPDATE` on the agent, `FOR UPDATE` on +the message, and the `account_sending_controls` upsert (which holds `KEY +SHARE` on the user); then the operation insert, the River job insert, and +the message's own stamp. The gate's agent lock is `NO KEY UPDATE` and must +stay that way: `FOR UPDATE` conflicts with the key share every concurrent +insert for the same agent already holds, and v1.9.0 deadlocked two parallel +sends exactly there. The rule generalizes: a `FOR UPDATE` taken after an +`INSERT` that references the locked row by foreign key, in the same +transaction, deadlocks under concurrency. An approval and a reply hold a +message row before the gate runs, so "agent before message" is a property +of Prepare itself, not of every caller. + Two consequences worth knowing. Notification and feedback mail now cross the same submitter as customer mail, so it carries `X-SES-CONFIGURATION-SET` and SES publishes delivery feedback for it; none of it correlates to a diff --git a/tests/e2e-prod/suites/03-concurrency.test.ts b/tests/e2e-prod/suites/03-concurrency.test.ts index dee86f0d2..6946b185a 100644 --- a/tests/e2e-prod/suites/03-concurrency.test.ts +++ b/tests/e2e-prod/suites/03-concurrency.test.ts @@ -196,6 +196,43 @@ test("concurrency: parallel DELETE of the same agent is idempotent under content } }); +test("concurrency: 8 parallel sends from a normal agent — all accepted (no 5xx, no duplicates)", async () => { + // The accept transaction inserts the message and then prepares its sending + // operation under the gate; v1.9.0 deadlocked those two steps against each + // other (SQLSTATE 40P01 → 500) for parallel sends from one agent. The HITL + // case below caught it on the hold path; this covers the direct path, which + // has the same shape and carries almost all production traffic. + const slug = uniqueSlug("sendconc"); + const c = await client.post<{ email: string }>("/v1/agents", { + body: { email: `${slug}@${client.env.sharedDomain}`, name: "send-conc" }, + }); + assert.equal(c.status, 201); + const email = c.body!.email; + track("agent", email); + + const N = 8; + const sends = await Promise.all( + Array.from({ length: N }, (_, i) => + burst.post<{ message_id: string; status: string }>(`/v1/agents/${encodeURIComponent(email)}/messages`, { + body: { + to: [SINK_EMAIL], + subject: `parallel direct ${i}`, + text: `parallel direct send #${i}`, + }, + }), + ), + ); + + const ids = new Set(); + for (const r of sends) { + assert.ok(r.status === 202 || r.status === 200, `parallel direct send: status ${r.status}, body: ${r.raw.slice(0, 200)}`); + assert.ok(r.body?.message_id?.startsWith("msg_"), `message_id present and prefixed`); + ids.add(r.body!.message_id); + } + assert.equal(ids.size, N, `expected ${N} distinct message_ids, got ${ids.size}`); + info(SUITE, "parallel-direct-sends", `${N} parallel direct sends accepted with ${ids.size} distinct ids`); +}); + test("concurrency: 8 parallel sends from HITL agent — all queue (no dropped/duplicated)", async () => { const slug = uniqueSlug("hitlconc"); const c = await client.post<{ email: string }>("/v1/agents", {