feat(agent): accept messageMetadata on agent.queueMessage

[panghy]

Summary

agent.queueMessage dropped the caller's messageMetadata, so a user answer routed through the explicit queue path lost its question_answers tag and never resolved the pending question set on drain. This PR makes the method carry messageMetadata, and closes the two drain-time gaps review surfaced around it: the shrunk agent:queue:updated was published before the drained row persisted (Q&A wizard flashed back), and nothing linked a persisted drained row to the queue entry it came from.

Workspace: Intent agne-tq-2 ("Dismiss agent Q&A when answer is queued") — intentd half of the cross-component fix; the FE counterpart (intent-hq/cloudlands-fe#2308) routes queued answers through this param and keys on queueInfo.queuedMessageId.

Changes

1. messageMetadata on agent.queueMessage

• Router (crates/intent-transport/src/router.rs): the agent.queueMessage arm reads messageMetadata, strips the reserved fromAgentId / fromAgentName attribution fields (same user-origin front door as agent.sendMessage), and rejects a non-object value with -32602 (messageMetadata must be an object). null / omitted reads as absent.
• Threading: WorkspaceApi::agent_queue_message (intent-core/src/traits.rs), the Services impl (intent-services/src/lib.rs) and agent_queue_message_op (intent-services/src/agent_ops.rs) take message_metadata: Option<Value> and pass it into enqueue_message, so the entry carries it (result queuedMessage, agent.getQueue) and the drain-time persist writes it onto the user row — the answer intake then resolves the pending marker exactly as a direct tagged agent.sendMessage does.
• No catalog / wire-shape change: messageMetadata? already existed on the QueuedMessage shape; it was only ever populated by daemon-internal enqueues.

2. Drain ordering: shrunk agent:queue:updated after the row persists (§6.5)

• Every drain arm published the shrunk queue snapshot BEFORE the drained entry's user row was persisted (and before the question_answers intake cleared the pending marker). The durable agent_queue write-through still runs at dequeue time (crash semantics unchanged); the event now goes out via publish_queue_updated_after_drain_persist only after persist_user succeeds — after the user-row agent:message and, for a tagged answer, the marker-clearing agent:updated. Applied to try_drain_queue, both worker single-entry arms, the batch flush (ONE shrunk snapshot after all rows) and send_queued_message_now. Persist failure keeps the fail-closed path: no shrunk snapshot, the front requeue republishes the restored queue.
• Every publisher, not just the drain arm: an unrelated concurrent mutation (enqueue / edit / remove of another entry) publishing between dequeue and the row persist could still emit a snapshot with the in-flight entry gone. DrainingGuard registers popped entries in a draining_queue_entries overlay that queue_snapshot lists ahead of the live queue (deduped by id / turnId against hand-backs and failure requeues). Every drain arm pops through a *_draining variant and drops the guard right before its settled publish; hand-back / parked / abort paths retire the entry at scope exit.
• Store-only fallback included: the no-AgentManager agent_send_queued_message_now_op (the one agent.sendQueuedMessageNow path outside the manager) followed the old order; it now pops through take_queued_message_draining, persists the durable shrink at dequeue, persists the row + echo + answer intake, then drops the guard and publishes via publish_queue_updated_after_drain_persist. Fail-closed requeue unchanged. No documented exception is needed — the contract text ("every drain arm and agent.sendQueuedMessageNow, every publisher") holds as written.
• Snapshots are read at publish time, under one gate: publish_queue_updated_for took a caller-captured Vec and awaited the write-through persist before publishing it, so a publisher parked on the persist gate could emit a pre-enqueue snapshot omitting an entry that had since been enqueued and started draining (or re-list an already-drained one). Every publisher (publish_queue_updated[_for], publish_queue_updated_after_drain_persist, the poisoned-queue migration) now ends in publish_queue_event, which reads queue_snapshot (overlay included) immediately before the bus write under a new agent_queue_publish_gate; the Vec parameter is gone, so no caller can hand in a stale snapshot. Publication order equals snapshot order.

3. Drain identity link: queueInfo.queuedMessageId

• stamp_queued_message_id: every drained entry's messageMetadata gains queueInfo.queuedMessageId = the entry's own id, next to the existing queuedAt / waitedMs / batchId stamps. Threshold-independent, always rewritten to the entry delivering now (a requeue re-drained under a fresh id re-links), skipped only for persisted: true requeues. queueInfo is daemon-reserved: an absent / null / non-object value is replaced by a fresh object carrying the link, so EVERY drained row names its entry. Applied on all drain arms and on both agent.sendQueuedMessageNow paths (runtime and store-only fallback).
• The user-row agent:message echo lifts the stamp as an additive queuedMessageId? field (omitted when absent). Row ids are unchanged: freshly minted on the three drain arms (never the entry id — there the stamp is the only link), and equal to the entry id on both agent.sendQueuedMessageNow paths as before (runtime and store-only fallback persist under the entry id, so messageId == queuedMessageId there).

Tests

• Unit (intent-services): queue_message_captures_message_metadata_on_entry; stamp coverage incl. non_object_queue_info_is_replaced_by_the_link; draining_overlay_tests — overlay semantics, merged-guard retirement, and two real-drain regressions with the arm parked in the persist retry backoff while unrelated mutations publish (settled path: no snapshot omits the entry before its row echo; rollback path: no flash, no ghost, restored order — both synchronize on the entry actually reaching the draining registry, not a fixed sleep); publisher_suspended_across_persist_never_emits_a_pre_enqueue_snapshot — deterministic stale-publish regression (publisher parked on the persist gate across an enqueue + draining pop; fails against the caller-captured Vec); agent_send_queued_message_now_publishes_shrunk_queue_after_row_persist — store-only fallback wire order (row echo before the first snapshot omitting the entry, no earlier snapshot omits it) plus the row/echo identity link; fails against the previous order.
• WSS e2e e2e_wss_pending_questions.rs:
  • queued_answer_via_queue_message_clears_marker_over_wss — tag lands on the entry (result + getQueue), wire order asserted (user-row agent:message and agent:updated marker clear BEFORE the agent:queue:updated that no longer lists the entry — fails against the previous order), persisted row and echo carry queuedMessageId (row id ≠ entry id), queue empties.
  • queue_message_strips_forged_sender_attribution_over_wss — forged fromAgentId / fromAgentName stripped on the result, agent.getQueue and the MCP ws.agent.getQueue view; the named sibling's ws.agent.removeQueuedMessage is rejected by the ownership guard and the entry stays queued; user-facing agent.removeQueuedMessage removes it.
  • invalid messageMetadata asserts the full JSON-RPC error envelope (jsonrpc "2.0", echoed id, no result, -32602 with a string message).
• WSS e2e e2e_wss_flush_queued_messages.rs: the combined-turn flush asserts both flushed rows carry DISTINCT queueInfo.queuedMessageIds in queue order next to the shared batchId, each row's echo lifts the same id, and the kick-off row/echo carry none.
• WSS e2e e2e_wss_agent_lifecycle.rs: the sub-threshold drain now expects a queueInfo holding only queuedMessageId.

Verification (latest push)

• cargo fmt --check — clean
• cargo clippy --workspace --all-targets -- -D warnings — clean
• cargo nextest run -p intent-services -E "test(/stamp/) or test(/queue_info/) or test(/queued_message_id/) or test(/drain/)" — 88/88 passed
• cargo nextest run -p intentd --test e2e_wss_flush_queued_messages --test e2e_wss_agent_lifecycle -E "test(/flush/) or test(/queued_message_metadata/) or test(/drain/)" — 9/9 passed
• cargo nextest run -p intentd --test e2e_wss_pending_questions — 8/8 passed
• cargo nextest run -p intent-services -E "test(/send_queued_message_now/) or test(/send_now/) or test(/draining/) or test(/stamp/) or test(/queued_message_id/)" — 67/67 passed
• cargo nextest run -p intent-services -p intentd -E "... sendQueuedMessageNow / force_send / last_activity / uds+wss queue ..." — 56/56 passed

Protocol docs

Monorepo docs/protocol/methods/agents.md (agent.queueMessage row, queueInfo.queuedMessageId), docs/protocol/06-events.md (§6.5 drain ordering: the shrunk snapshot follows the row persist, from every publisher) and docs/protocol/versioning.md ("Also within 9.11" additive entries) are updated in intent-hq/intent#4689; presence-detected additive fields, no version bump.

[augmentcode]

^{Deep Code Review Agent🐛}

Review completed with 1 suggestions.

_{Reviewed commit: c87686d}

[augmentcode]

🤖 Augment PR Summary

Summary: This PR preserves caller-supplied message metadata for explicit queued agent messages.

Changes:

• Extends WorkspaceApi::agent_queue_message and the services implementation with optional messageMetadata.
• Threads metadata into queued entries through agent_queue_message_op and existing queue persistence.
• Updates the JSON-RPC router to accept object metadata, treat null or omission as absent, and reject other values with -32602.
• Strips reserved sender-attribution fields at the user-facing queue endpoint.
• Preserves metadata in queue results, queue snapshots, drain-time user rows, and pending-question resolution.
• Updates direct service call sites for the widened method signature.

Tests: Adds service coverage for queued metadata and a WSS regression covering a queued question_answers response, queue inspection, metadata persistence, marker clearing, null handling, and invalid metadata.

_{🤖 Was this summary useful? React with 👍 or 👎}

[augmentcode]

Review completed. 1 suggestion posted.

Comment augment review to trigger a new review at any time.

[panghy]

Drain ordering contract (commit a37531e)

FE review found that every drain arm published the shrunk agent:queue:updated before the drained entry's user row was persisted / the question_answers intake cleared the marker. In that window a client sees marker set + no queued tagged entry + no tagged row, so the Q&A wizard flashes back.

Change. The drain-time publish is split: the durable agent_queue write-through (persist_queue_snapshot) still runs at dequeue time (crash semantics unchanged), and the event goes out through the new Services::publish_queue_updated_after_drain_persist only after persist_user succeeds — i.e. after the user-row agent:message and, for a tagged answer, the marker-clearing agent:updated. Applied to try_drain_queue, both worker single-entry arms, the batch flush (prepare_flush_turn: ONE shrunk snapshot after all rows), and agent.sendQueuedMessageNow (same window; included for a single contract).

New wire order per drained entry: agent:queue:processing → user-row agent:message → (agent:updated marker clear, tagged only) → shrunk agent:queue:updated → stream events. Persist failure keeps the fail-closed path: no shrunk snapshot is published at all; the front requeue republishes the restored queue as before.

Checked for dependents on the old order. turnId correlation is unaffected (carried by agent:queue:processing / agent:message / agent:stream:end, none of which moved relative to each other). The FE's agent:queue:processing handler only promotes send-state by turnId; the queue mirror comes from agent:queue:updated, so the later shrink is exactly what closes the gap. No existing intentd e2e gated on shrink-before-processing/message order (e2e_wss_flush_queued_messages, e2e_wss_agent_turn_correlation, lifecycle queue tests track the signals independently).

Tests. queued_answer_via_queue_message_clears_marker_over_wss now asserts the order and fails against the previous drain code (verified by stashing the two service files: shrunk agent:queue:updated arrived before the drained answer's user-row agent:message).

• cargo fmt --check, cargo clippy --workspace --all-targets -- -D warnings — clean
• e2e_wss_pending_questions + e2e_services_agent_queue + e2e_wss_flush_queued_messages + e2e_wss_agent_turn_correlation — 17/17
• e2e_wss_agent_lifecycle -E 'test(/queue/)' — 11/11
• intent-services -E 'test(/queue/) or test(/drain/) or test(/flush/)' — 213/213

Protocol docs (monorepo workspace branch): docs/protocol/06-events.md §6.5 queue row (drain ordering contract), methods/agents.md (sendQueuedMessageNow row + flush "Events" bullet), versioning.md (appended to the "Also within 9.11" entry).

[panghy]

Drain identity link (commit f499f57)

With the drain reorder the FE briefly sees the persisted user row AND the still-listed queue entry; nothing linked them (drain arms mint a fresh row id; batch-flush echoes share the head turnId).

Change. New stamp_queued_message_id: every drained entry's messageMetadata gains queueInfo.queuedMessageId = the QueuedMessage.id it drained from, next to the existing queuedAt / waitedMs / batchId stamps. Threshold-independent (a sub-threshold drain now carries a queueInfo holding only this key), always rewritten to the entry delivering now (a requeue re-drained under a fresh id re-links), skipped for persisted: true requeues and non-object metadata. Applied on all three drain arms (try_drain_queue, both worker single-entry arms, the prepare_flush_turn batch loop) and on agent.sendQueuedMessageNow for parity (there the row id already equals the entry id, so the stamp is redundant but consistent).

The user-row agent:message echo additionally lifts the stamp as an additive top-level queuedMessageId? (agent_message_event_payload; omitted when absent, never null) — the lean echo carries no metadata, so without this the FE would have to wait for the chat.subscribe delta to match. The row id is unchanged: still freshly minted, never the entry id.

FE matching: event.data.queuedMessageId === queueEntry.id on the agent:message echo, or row.metadata.queueInfo.queuedMessageId === queueEntry.id on conversation reads / chat deltas.

Tests.

• Unit: queued_message_id_stamp_tests (6) — sub-threshold link-only, coexistence with wait + batch stamps, caller metadata preserved, overwrite-to-delivering-entry, persisted skip, non-object skip.
• WSS e2e: queued_answer_via_queue_message_clears_marker_over_wss asserts the stamp on the echo and on the persisted row, and that the row id ≠ entry id. sub_threshold_queued_message_drains_without_annotation_over_wss updated to expect queueInfo == { queuedMessageId } (was: no queueInfo).
• cargo fmt --check / clippy -D warnings clean; 151/151 services queue/drain unit tests; 17/17 e2e_wss_pending_questions + e2e_services_agent_queue + e2e_wss_flush_queued_messages + e2e_wss_agent_turn_correlation; 16/16 lifecycle queue/metadata/annotation tests.

Protocol docs (monorepo workspace branch): methods/agents.md §5.5 dequeue-wait annotation (queueInfo shape + queuedMessageId semantics, flush per-entry rows), 06-events.md §6.5 (agent:message data gains queuedMessageId?; queue-row ordering contract cross-reference), versioning.md 9.11 entry.

[panghy]

Verification Summary

APPROVED — High confidence, scoped to the intentd task and reviewed head 6cbbcc9b54a9fbe74b67210f0363cc893aeb38c0. This is verification approval, not human merge authorization.

Acceptance criteria

• VERIFIED: queueMessage forwards object metadata to result/getQueue and drained rows; omitted/null compatibility, reserved sender stripping, and full -32602 JSON-RPC error envelope covered.
• VERIFIED: tagged queued answers clear pendingQuestionsMessageId on drain; processing/row echo/marker clear precede the settled shrunk snapshot.
• VERIFIED: concurrent publishers retain draining entries and cannot publish stale caller snapshots: all queue events read the current overlay-inclusive snapshot inside the shared gate held through publication. Deterministic pre-enqueue/persist-gate and real-drain settled/rollback regressions pass.
• VERIFIED: queuedMessageId is stamped on every fresh single/worker/batch/send-now drain, including non-object queueInfo normalization; row/echo links and distinct batch entry identities are covered. Already-persisted requeues remain exempt as specified.
• VERIFIED: docs commit 1fa2bc947488442ce97f83cec242a207d1717710 and PR body correctly distinguish normal-drain fresh row IDs from send-now entry IDs. All review threads resolved; conventional headlines checked.

Independent final-head checks

• cargo fmt --check: PASS
• cargo clippy --workspace --all-targets -- -D warnings: PASS
• Targeted pending-questions, services-queue, WSS flush, resume-order, and turn-correlation suites: 22/22 PASS, no skips.
• Selected services queue/drain/flush/stamp/send-now/migration units: 274/274 PASS with --retries 0 (4046 outside-filter tests skipped). Includes the new stale publisher regression and the formerly flaky real-drain regression; no retry.
• make docs-check: PASS (4 docs). Both diff checks: PASS.
• Final-head fast PR CI: SUCCESS, including CI Gate; 0 unresolved review threads.

Explicitly accepted gate split

Coordinator accepts independent targeted tests + green fast PR CI for this review approval. Fast PR CI is not a full-suite pass: coverage-e2e/all are intentionally skipped on pull_request and run on merge_group. The coordinator separately requires the implementor local full nextest run (fail-fast disabled, known-unrelated host flakes isolated) before requesting human merge permission. Merge-group coverage gates landing and ejects on failure. Earlier local MCP probe timeout and pipeline exit-0 masking are recorded limitations, not passing full-suite evidence.

No reviewer code edits. No merge, auto-merge, or merge-queue action taken.