diff --git a/crates/webcodex-core/src/workflow_session_contract.rs b/crates/webcodex-core/src/workflow_session_contract.rs index d54f37b75..f55a49b1a 100644 --- a/crates/webcodex-core/src/workflow_session_contract.rs +++ b/crates/webcodex-core/src/workflow_session_contract.rs @@ -10,8 +10,10 @@ pub const MAX_MODEL_VALIDATION_ASSERTION_NAME_CHARS: usize = crate::runner_protocol::VALIDATION_ASSERTION_NAME_MAX_CHARS; pub const TOOL_CALL_RECORDING_SESSION_ID_FIELD: &str = "recording_session_id"; pub const TOOL_CALL_ACK_SESSION_MESSAGE_IDS_FIELD: &str = "ack_session_message_ids"; +pub const TOOL_CALL_ACK_REF_FIELD: &str = "ack_ref"; pub const TOOL_CALL_SESSION_MESSAGE_RESOLUTION_FIELD: &str = "session_message_resolution"; pub const MAX_TOOL_CALL_ACK_MESSAGE_IDS: usize = 8; +pub const MAX_TOOL_CALL_ACK_REF_CHARS: usize = 64; pub const TOOL_EXPECTED_FAILURE_FIELD: &str = "expected_failure"; pub const TOOL_EXPECTED_FAILURE_KIND_FIELD: &str = "expected_failure_kind"; pub const TOOL_RESULT_EXPECTATION_FIELD: &str = "result_expectation"; diff --git a/crates/webcodex-tool-contracts/src/request_schema.rs b/crates/webcodex-tool-contracts/src/request_schema.rs index ad61dda01..b39eb30fb 100644 --- a/crates/webcodex-tool-contracts/src/request_schema.rs +++ b/crates/webcodex-tool-contracts/src/request_schema.rs @@ -297,6 +297,7 @@ mod tests { for wrapper in [ "recording_session_id", "ack_session_message_ids", + "ack_ref", "context_request", "session_message_resolution", "_control", diff --git a/crates/webcodex-tool-contracts/src/tests/input_schemas.rs b/crates/webcodex-tool-contracts/src/tests/input_schemas.rs index fb07107ec..3779a4776 100644 --- a/crates/webcodex-tool-contracts/src/tests/input_schemas.rs +++ b/crates/webcodex-tool-contracts/src/tests/input_schemas.rs @@ -969,6 +969,7 @@ fn code_mode_exec_schema_keeps_authority_outer_bound_and_source_bounded() { "recording_session_id", "ack_session_context_revision", "ack_session_message_ids", + "ack_ref", "context_request", "session_message_resolution", ] @@ -1004,6 +1005,7 @@ fn code_mode_effectful_schema_keeps_authority_outer_bound_and_deadline_explicit( "recording_session_id", "ack_session_context_revision", "ack_session_message_ids", + "ack_ref", "context_request", "session_message_resolution", ] @@ -1033,6 +1035,7 @@ fn code_mode_mutating_schema_keeps_authority_outer_bound_and_mutation_scope_narr "recording_session_id", "ack_session_context_revision", "ack_session_message_ids", + "ack_ref", "context_request", "session_message_resolution", "state_changed", diff --git a/crates/webcodex-workflow-session/src/collaboration_tests.rs b/crates/webcodex-workflow-session/src/collaboration_tests.rs index 07b2f9b99..eda94ae58 100644 --- a/crates/webcodex-workflow-session/src/collaboration_tests.rs +++ b/crates/webcodex-workflow-session/src/collaboration_tests.rs @@ -4,6 +4,7 @@ use super::model::{ MESSAGE_COMPLETION_FINGERPRINT_HEX_CHARS, }; use super::*; +use base64::Engine; use serde_json::Value; fn completion_id(byte: char) -> String { @@ -1310,3 +1311,234 @@ async fn observe_session_messages_duplicate_persisted_message_ids_fail_closed() assert_eq!(delta.messages.len(), 1); assert_eq!(delta.messages[0].message_id, after_restore.message_id); } + +#[test] +fn ack_ref_round_trips_exact_projected_subset_and_replays_after_ack_observation() { + let store = SessionStore::default(); + let session = store.start_session(Some("proj".to_string()), None); + let mut ids = Vec::new(); + for body in ["first", "second", "third", "fourth"] { + ids.push( + store + .post_message_with_ack( + PostSessionMessageInput { + session_id: session.session_id.clone(), + kind: SessionMessageKind::Guidance, + message: body.to_string(), + tags: Vec::new(), + reply_to: None, + priority: SessionMessagePriority::Normal, + }, + true, + ) + .unwrap() + .message_id, + ); + } + + let attention = store.ack_required_messages(&session.session_id, &[]); + let projected = attention + .messages + .iter() + .skip(1) + .take(2) + .map(|message| message.message_id.clone()) + .collect::>(); + let ack_ref = store + .issue_ack_ref(&session.session_id, &projected) + .expect("bounded projected ACK set should mint a ref"); + assert!(ack_ref.starts_with("wc_ack1_")); + assert!(ack_ref.len() <= MAX_TOOL_CALL_ACK_REF_CHARS); + assert_eq!( + store.resolve_ack_ref(&session.session_id, &ack_ref), + Some(projected.clone()) + ); + + let observed = store.observe_message_acks(&session.session_id, &projected); + assert_eq!(observed.accepted_ids, projected); + assert_eq!(observed.first_observed_count, 2); + assert_eq!( + store.resolve_ack_ref(&session.session_id, &ack_ref), + Some(observed.accepted_ids.clone()), + "ACK observability must not stale an otherwise unchanged open ACK set" + ); +} + +#[test] +fn ack_ref_fails_closed_when_membership_changes_or_session_differs() { + let store = SessionStore::default(); + let session = store.start_session(Some("proj".to_string()), None); + let foreign = store.start_session(Some("proj".to_string()), None); + + let first = store + .post_message_with_ack( + PostSessionMessageInput { + session_id: session.session_id.clone(), + kind: SessionMessageKind::Guidance, + message: "first".to_string(), + tags: Vec::new(), + reply_to: None, + priority: SessionMessagePriority::Normal, + }, + true, + ) + .unwrap(); + let initial = vec![first.message_id.clone()]; + let ack_ref = store + .issue_ack_ref(&session.session_id, &initial) + .expect("initial ACK set should mint a ref"); + + assert_eq!( + store.resolve_ack_ref(&foreign.session_id, &ack_ref), + None, + "a Session-bound ref must not resolve in another Session" + ); + assert_eq!( + store.resolve_ack_ref(&session.session_id, "wc_ack1_bad"), + None + ); + + store + .post_message_with_ack( + PostSessionMessageInput { + session_id: session.session_id.clone(), + kind: SessionMessageKind::Question, + message: "new membership".to_string(), + tags: Vec::new(), + reply_to: None, + priority: SessionMessagePriority::High, + }, + true, + ) + .unwrap(); + assert_eq!( + store.resolve_ack_ref(&session.session_id, &ack_ref), + None, + "adding an ACK-required message must stale the old exact-set ref" + ); + + let current = store.ack_required_messages(&session.session_id, &[]); + let current_ids = current + .messages + .iter() + .map(|message| message.message_id.clone()) + .collect::>(); + let current_ref = store + .issue_ack_ref(&session.session_id, ¤t_ids) + .expect("current set should mint a replacement ref"); + assert_ne!(current_ref, ack_ref); + + store + .resolve_message( + &session.session_id, + &first.message_id, + Some("handled".to_string()), + ) + .unwrap(); + assert_eq!( + store.resolve_ack_ref(&session.session_id, ¤t_ref), + None, + "removing an ACK-required open message must stale the prior ref" + ); +} + +#[test] +fn ack_ref_canonicalizes_set_order_and_rejects_unknown_membership() { + let store = SessionStore::default(); + let session = store.start_session(Some("proj".to_string()), None); + for index in 0..17 { + store + .post_message_with_ack( + PostSessionMessageInput { + session_id: session.session_id.clone(), + kind: SessionMessageKind::Guidance, + message: format!("guidance-{index}"), + tags: Vec::new(), + reply_to: None, + priority: SessionMessagePriority::Normal, + }, + true, + ) + .unwrap(); + } + let attention = store.ack_required_messages(&session.session_id, &[]); + let ids = attention + .messages + .iter() + .map(|message| message.message_id.clone()) + .collect::>(); + let canonical = vec![ids[0].clone(), ids[8].clone(), ids[16].clone()]; + let reordered = vec![ids[16].clone(), ids[0].clone(), ids[8].clone()]; + + let canonical_ref = store + .issue_ack_ref(&session.session_id, &canonical) + .expect("canonical subset should mint an ACK ref"); + let reordered_ref = store + .issue_ack_ref(&session.session_id, &reordered) + .expect("ACK set order must not affect identity"); + assert_eq!(reordered_ref, canonical_ref); + assert_eq!( + store.resolve_ack_ref(&session.session_id, &canonical_ref), + Some(canonical) + ); + assert!(store + .issue_ack_ref(&session.session_id, &["wc_msg_not-retained".to_string()]) + .is_none()); + + let encoded = canonical_ref.strip_prefix("wc_ack1_").unwrap(); + let mut payload = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(encoded.as_bytes()) + .unwrap(); + payload.push(0); + let noncanonical = format!( + "wc_ack1_{}", + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(payload) + ); + assert_eq!( + store.resolve_ack_ref(&session.session_id, &noncanonical), + None, + "non-canonical trailing zero membership bytes must fail closed" + ); +} + +#[test] +fn ack_ref_worst_case_retained_set_stays_within_model_facing_bound() { + let store = SessionStore::default(); + let session = store.start_session(Some("proj".to_string()), None); + let mut ids = Vec::new(); + for index in 0..DEFAULT_MAX_MESSAGES_PER_SESSION { + ids.push( + store + .post_message_with_ack( + PostSessionMessageInput { + session_id: session.session_id.clone(), + kind: SessionMessageKind::Guidance, + message: format!("guidance-{index}"), + tags: Vec::new(), + reply_to: None, + priority: SessionMessagePriority::Normal, + }, + true, + ) + .unwrap() + .message_id, + ); + } + + let canonical_ids = store + .ack_required_messages(&session.session_id, &[]) + .messages + .into_iter() + .map(|message| message.message_id) + .collect::>(); + let ack_ref = store + .issue_ack_ref(&session.session_id, &ids) + .expect("the full retained Session message bound should fit in one ACK ref"); + assert!(ack_ref.len() <= MAX_TOOL_CALL_ACK_REF_CHARS); + assert_eq!( + store + .resolve_ack_ref(&session.session_id, &ack_ref) + .expect("full-set ACK ref should resolve"), + canonical_ids + ); +} diff --git a/crates/webcodex-workflow-session/src/lib.rs b/crates/webcodex-workflow-session/src/lib.rs index f51b52e48..536286dd7 100644 --- a/crates/webcodex-workflow-session/src/lib.rs +++ b/crates/webcodex-workflow-session/src/lib.rs @@ -72,11 +72,11 @@ pub use model::{ MAX_MESSAGE_COMPLETION_KEY_CHARS, MAX_MESSAGE_DELIVERY_KEY_CHARS, MAX_MESSAGE_LIST_LIMIT, MAX_MESSAGE_RESOLUTION_CHARS, MAX_MESSAGE_TAGS, MAX_MESSAGE_TAG_CHARS, MAX_MODEL_VALIDATION_ASSERTION_NAME_CHARS, MAX_TOOL_CALL_ACK_MESSAGE_IDS, - SESSION_INBOX_ACK_REQUIRED_ATTENTION_INSTRUCTION, SESSION_INBOX_ACK_REQUIRED_ATTENTION_REASON, - TOOL_ACCEPTED_EXIT_CODES_FIELD, TOOL_ASSERTION_NAME_FIELD, - TOOL_CALL_ACK_SESSION_MESSAGE_IDS_FIELD, TOOL_CALL_RECORDING_SESSION_ID_FIELD, - TOOL_CALL_SESSION_MESSAGE_RESOLUTION_FIELD, TOOL_EXPECTATION_RESULT_UNEXPECTED_FAILURE, - TOOL_RESULT_EXPECTATION_FIELD, + MAX_TOOL_CALL_ACK_REF_CHARS, SESSION_INBOX_ACK_REQUIRED_ATTENTION_INSTRUCTION, + SESSION_INBOX_ACK_REQUIRED_ATTENTION_REASON, TOOL_ACCEPTED_EXIT_CODES_FIELD, + TOOL_ASSERTION_NAME_FIELD, TOOL_CALL_ACK_REF_FIELD, TOOL_CALL_ACK_SESSION_MESSAGE_IDS_FIELD, + TOOL_CALL_RECORDING_SESSION_ID_FIELD, TOOL_CALL_SESSION_MESSAGE_RESOLUTION_FIELD, + TOOL_EXPECTATION_RESULT_UNEXPECTED_FAILURE, TOOL_RESULT_EXPECTATION_FIELD, }; pub use store::SessionStore; pub use util::redact_and_bound_instruction; diff --git a/crates/webcodex-workflow-session/src/messages.rs b/crates/webcodex-workflow-session/src/messages.rs index 3347c2f36..ba416e1a8 100644 --- a/crates/webcodex-workflow-session/src/messages.rs +++ b/crates/webcodex-workflow-session/src/messages.rs @@ -10,7 +10,8 @@ use super::model::{ SessionDiscussionSummary, SessionInboxHint, SessionMessage, SessionMessageDelivery, SessionMessageDeliveryOutcome, SessionMessageError, SessionMessageObservationError, SessionMessageObservationOutcome, SessionMessageStatus, WithdrawSessionMessageOutcome, - DEFAULT_MESSAGE_LIST_LIMIT, MAX_MESSAGE_LIST_LIMIT, MAX_SESSION_MESSAGE_OBSERVATION_TOKEN_LEN, + DEFAULT_MAX_MESSAGES_PER_SESSION, DEFAULT_MESSAGE_LIST_LIMIT, MAX_MESSAGE_LIST_LIMIT, + MAX_SESSION_MESSAGE_OBSERVATION_TOKEN_LEN, MAX_TOOL_CALL_ACK_REF_CHARS, }; use super::query::{build_discussion_summary, build_inbox_hint}; use super::store::SessionStore; @@ -143,6 +144,115 @@ impl SessionStore { outcome } + pub fn issue_ack_ref(&self, session_id: &str, retained_ids: &[String]) -> Option { + if retained_ids.is_empty() { + return None; + } + self.with_record_for_query(session_id, |record, _| { + let open_ids = sorted_open_ack_message_ids(record); + let retained = retained_ids + .iter() + .map(String::as_str) + .collect::>(); + if retained.is_empty() { + return None; + } + + let mut membership = vec![0_u8; open_ids.len().div_ceil(8)]; + let mut matched = 0_usize; + for (index, message_id) in open_ids.iter().enumerate() { + if retained.contains(message_id.as_str()) { + membership[index / 8] |= 1_u8 << (index % 8); + matched += 1; + } + } + if matched != retained.len() { + return None; + } + while membership.last().is_some_and(|byte| *byte == 0) { + membership.pop(); + } + if membership.is_empty() { + return None; + } + + let tag = ack_set_digest( + session_id, + &record.owner_authority_fingerprint, + open_ids.iter().map(String::as_str), + ); + let mut payload = Vec::with_capacity(ACK_REF_TAG_BYTES + membership.len()); + payload.extend_from_slice(&tag[..ACK_REF_TAG_BYTES]); + payload.extend_from_slice(&membership); + let token = format!( + "{ACK_REF_PREFIX}{}", + general_purpose::URL_SAFE_NO_PAD.encode(payload) + ); + (token.len() <= MAX_TOOL_CALL_ACK_REF_CHARS).then_some(token) + }) + .flatten() + } + + pub fn resolve_ack_ref(&self, session_id: &str, ack_ref: &str) -> Option> { + if ack_ref.len() > MAX_TOOL_CALL_ACK_REF_CHARS { + return None; + } + let encoded = ack_ref.strip_prefix(ACK_REF_PREFIX)?; + if encoded.is_empty() || !encoded.is_ascii() { + return None; + } + let payload = general_purpose::URL_SAFE_NO_PAD + .decode(encoded.as_bytes()) + .ok()?; + if payload.len() <= ACK_REF_TAG_BYTES + || payload.len() > ACK_REF_TAG_BYTES + ACK_REF_MAX_MEMBERSHIP_BYTES + { + return None; + } + let (provided_tag, membership) = payload.split_at(ACK_REF_TAG_BYTES); + if membership.last().is_some_and(|byte| *byte == 0) { + return None; + } + + self.with_record_for_query(session_id, |record, _| { + let open_ids = sorted_open_ack_message_ids(record); + let expected_tag = ack_set_digest( + session_id, + &record.owner_authority_fingerprint, + open_ids.iter().map(String::as_str), + ); + if provided_tag != &expected_tag[..ACK_REF_TAG_BYTES] { + return None; + } + + let max_membership_bytes = open_ids.len().div_ceil(8); + if membership.len() > max_membership_bytes { + return None; + } + if membership.len() == max_membership_bytes && !membership.is_empty() { + let used_bits = open_ids.len() % 8; + if used_bits != 0 { + let allowed_mask = ((1_u16 << used_bits) - 1) as u8; + if membership[membership.len() - 1] & !allowed_mask != 0 { + return None; + } + } + } + + let mut resolved = Vec::new(); + for (index, message_id) in open_ids.into_iter().enumerate() { + let Some(byte) = membership.get(index / 8) else { + break; + }; + if byte & (1_u8 << (index % 8)) != 0 { + resolved.push(message_id); + } + } + (!resolved.is_empty()).then_some(resolved) + }) + .flatten() + } + pub fn ack_required_messages( &self, session_id: &str, @@ -484,6 +594,46 @@ fn observation_outcome( }) } +const ACK_REF_PREFIX: &str = "wc_ack1_"; +const ACK_REF_TAG_BYTES: usize = 16; +const ACK_REF_MAX_MEMBERSHIP_BYTES: usize = DEFAULT_MAX_MESSAGES_PER_SESSION.div_ceil(8); + +fn sorted_open_ack_message_ids(record: &super::model::SessionRecord) -> Vec { + let mut messages = record + .messages + .iter() + .filter(|message| message.status == SessionMessageStatus::Open && message.requires_ack) + .map(|message| message.as_ref()) + .collect::>(); + messages.sort_by(|left, right| { + left.created_at + .cmp(&right.created_at) + .then_with(|| left.message_id.cmp(&right.message_id)) + }); + messages + .into_iter() + .map(|message| message.message_id.clone()) + .collect() +} + +fn ack_set_digest<'a>( + session_id: &str, + owner_authority_fingerprint: &str, + message_ids: impl Iterator, +) -> [u8; 32] { + let mut hasher = Sha256::new(); + hasher.update(b"webcodex.session-message-ack-set.v1\0"); + for value in [session_id, owner_authority_fingerprint] { + hasher.update((value.len() as u64).to_be_bytes()); + hasher.update(value.as_bytes()); + } + for message_id in message_ids { + hasher.update((message_id.len() as u64).to_be_bytes()); + hasher.update(message_id.as_bytes()); + } + hasher.finalize().into() +} + const MESSAGE_OBSERVATION_TOKEN_PREFIX: &str = "wsm2_"; const MESSAGE_OBSERVATION_REVISION_BYTES: usize = 8; const MESSAGE_OBSERVATION_TAG_BYTES: usize = 16; diff --git a/crates/webcodex-workflow-session/src/model.rs b/crates/webcodex-workflow-session/src/model.rs index 36974ecdd..21a25f1b2 100644 --- a/crates/webcodex-workflow-session/src/model.rs +++ b/crates/webcodex-workflow-session/src/model.rs @@ -13,9 +13,10 @@ pub use webcodex_core::workflow_session_contract::{ SessionMode, }; pub use webcodex_core::workflow_session_contract::{ - MAX_MODEL_VALIDATION_ASSERTION_NAME_CHARS, MAX_TOOL_CALL_ACK_MESSAGE_IDS, SESSION_ID_PREFIX, + MAX_MODEL_VALIDATION_ASSERTION_NAME_CHARS, MAX_TOOL_CALL_ACK_MESSAGE_IDS, + MAX_TOOL_CALL_ACK_REF_CHARS, SESSION_ID_PREFIX, SESSION_INBOX_ACK_REQUIRED_ATTENTION_INSTRUCTION, SESSION_INBOX_ACK_REQUIRED_ATTENTION_REASON, - TOOL_ACCEPTED_EXIT_CODES_FIELD, TOOL_ASSERTION_NAME_FIELD, + TOOL_ACCEPTED_EXIT_CODES_FIELD, TOOL_ASSERTION_NAME_FIELD, TOOL_CALL_ACK_REF_FIELD, TOOL_CALL_ACK_SESSION_MESSAGE_IDS_FIELD, TOOL_CALL_RECORDING_SESSION_ID_FIELD, TOOL_CALL_SESSION_MESSAGE_RESOLUTION_FIELD, TOOL_EXPECTED_FAILURE_FIELD, TOOL_EXPECTED_FAILURE_KIND_FIELD, TOOL_RESULT_EXPECTATION_FIELD, diff --git a/docs/MCP.md b/docs/MCP.md index 9d4ba5f7b..54043901b 100644 --- a/docs/MCP.md +++ b/docs/MCP.md @@ -365,10 +365,12 @@ are model-visible gateway tools; review/coding catalogs still recommend them. `outputSchema` and projects shorter MCP-specific tool/input descriptions for selection: purpose, nearby tool distinctions, and essential continuation guidance. Repeated Session/context wrapper and audited common-argument copy is shortened -too. Compact discovery omits only the exact opaque-ID regexes on +too. Compact discovery omits the repeated `ack_ref` description while preserving +its exact field and length bound; the full manifest retains the complete Session-only +ACK-set contract. It also omits only the exact opaque-ID regexes on `recording_session_id`, `ack_session_message_ids.items`, and `session_message_resolution.message_id`; their existing parent descriptions keep -the `wc_sess_*` / `wc_msg_*` type hints. Copy the exact returned IDs. +the `wc_sess_*` / `wc_msg_*` type hints. Copy the exact returned values. Business-ID, hash/Git fence and resource-path patterns, all bounds, field names, required fields, enums, object/union shape, annotations, and MCP App/file metadata are preserved. This is discovery presentation only; runtime argument validation diff --git a/docs/agent/manual-window-collaboration.md b/docs/agent/manual-window-collaboration.md index 066ed5ae0..20eb5eb50 100644 --- a/docs/agent/manual-window-collaboration.md +++ b/docs/agent/manual-window-collaboration.md @@ -23,7 +23,7 @@ Peer messages reuse the Session message vocabulary (`note`, `proposal`, `questio Delivery is model-facing and intentionally lightweight: - `requires_ack=false`: WebCodex durably records the message and attempts to piggyback it once in `peer_messages` on the recipient window's next normal model-facing tool result. The persisted first/last projection timestamps and projection count describe a Server projection attempt, not a delivery/read receipt. There is no retry obligation if the model or transport never acts on it. -- `requires_ack=true`: the message is eligible for the same bounded piggyback on later calls whenever the current request omits its id. Echoing the id in `ack_session_message_ids` suppresses it for that one request/response and records the first observed ACK time. ACK proves neither acceptance nor execution, never resolves the message, and never requires a reply. +- `requires_ack=true`: the message is eligible for the same bounded piggyback on later calls whenever the current request omits its id. Echoing the id in `ack_session_message_ids` suppresses it for that one request/response and records the first observed ACK time. Compact Session `ack_ref` does not apply to Peer messages. ACK proves neither acceptance nor execution, never resolves the message, and never requires a reply. Peer transport is deliberately bounded rather than a permanent task queue. Old retained Peer messages and discovery edges may be pruned; `requires_ack` therefore means repeat while retained, not indefinite durable work. ActionAudit activity is discovery input only and never establishes a communication route by itself: a peer route exists only while retained Peer discovery/message state can still resolve it. Use Workflow Session todos and assignment fencing for durable work commitments. @@ -95,7 +95,7 @@ All supplied filters use deterministic AND semantics. `message_id` therefore giv The token is bounded, opaque, bound to the exact Workflow Session, and backed by a durable Session-local monotonic message-observation revision. It is observation state only: it is not authority, an idempotency key, execution identity, an implicit Workflow Session selector, or message-delivery receipt. The same recorder/target authorization fence used by the other collaboration tools applies before any observation result is returned. Token issuance fences the ledger generation containing its revision so a valid token remains usable after Server restart when that Workflow Session can be restored. -Assignment and continuity identities are intentionally separate domains: `assignment_fence` is a semantic todo snapshot, `completion_key` is caller replay identity, `ack_session_message_ids` is request-scoped retained-message proof, an observation token is a generic Session message-state cursor, business `session_id` names the authorized target, `recording_session_id` is Session provenance only, and `wc_peer_*` names a principal-scoped communication endpoint. None substitutes for another or grants Project/Session authority by possession. +Assignment and continuity identities are intentionally separate domains: `assignment_fence` is a semantic todo snapshot, `completion_key` is caller replay identity, `ack_session_message_ids` is exact-ID request-scoped ACK evidence, Session `ack_ref` is compact exact-set ACK evidence, an observation token is a generic Session message-state cursor, business `session_id` names the authorized target, `recording_session_id` is Session provenance only, and `wc_peer_*` names a principal-scoped communication endpoint. None substitutes for another or grants Project/Session authority by possession. Observation tracks real message-state mutation, not deque length. Posts advance it; a resolve advances it only when status/resolution really changes; a new atomic completion advances for the todo resolution and answer creation; exact completion replay and no-op resolve do not advance it. If one retained message changes multiple times between observations, the observer may receive only its latest current state because this primitive is not an event/audit log. @@ -151,7 +151,7 @@ When multiple workers operate on the same source, use normal Git/WebCodex Projec The hosted Runtime Console may post `note`, `guidance`, `question`, and `todo` messages into an exact authorized Workflow Session through the same `post_session_message` kernel path. This is a browser affordance, not a Participant entity, membership record, presence signal, or identity-spoofing surface. The browser route keeps the current collaboration metadata authority policy (`runtime:read`) and still applies the stored Session/project authority fence. -Any Session message may opt into `requires_ack`, independently of kind and priority. A Stateless MCP 2026 caller can echo the visible `wc_msg_*` id in `ack_session_message_ids` on an otherwise ordinary tool call; the same bounded wrapper is also reused for ACK-required Peer messages addressed to the current window. The original tool executes normally whether the ACK is present, missing, unknown, foreign, or stale. A valid ACK suppresses that message body only for the same request/response. If a later request omits the ACK while a Session message remains open, or while a Peer ACK message remains retained, the Server may piggyback it again. The first observed ACK timestamp is observability only; it must never be described as delivered, read, accepted, or currently remembered. Session durable completion still requires normal message resolution. +Any Session message may opt into `requires_ack`, independently of kind and priority. A Stateless MCP 2026 caller may echo visible `wc_msg_*` ids in `ack_session_message_ids`, or reuse the compact `session_attention.ack_ref`. Each returned ref represents the exact Session ACK set explicitly retained across the current exchange: ACK evidence accepted on this request plus ACK-required messages newly projected in this response. As further bounded messages are projected, the next ref carries that retained set forward in one bounded value. The compact ref is Session-only: it is checked against the exact authorized attention Session and its full current open ACK-required membership, so a changed, stale, malformed, foreign-Session, or foreign-principal ref acknowledges nothing and cannot absorb later messages. The exact-ID wrapper remains available and is still the only ACK form for Window Peer messages. The original tool executes normally whether ACK evidence is present, missing, unknown, foreign, or stale. A valid ACK suppresses the represented Session/Peer body only for that request/response. Later omission makes unresolved Session messages or retained Peer ACK messages eligible for bounded re-projection. The first observed ACK timestamp is observability only; it must never be described as delivered, read, accepted, or currently remembered. Session durable completion still requires normal message resolution. ## Bounded payload guidance diff --git a/docs/agent/session-model.md b/docs/agent/session-model.md index a7b851363..052b5f992 100644 --- a/docs/agent/session-model.md +++ b/docs/agent/session-model.md @@ -117,7 +117,9 @@ facts while making an otherwise silent recorder discontinuity observable. One real kernel tool request also receives one trusted runtime-generated logical invocation correlation id. The outer recorder event pair and any inner concrete business-execution event pair inherit that id while retaining independent pair-level `call_id` values. A small recorder/business role discriminator lets Session-local semantic projections deterministically prefer authoritative business execution facts when both pairs land in the same Workflow Session. Raw ledger facts remain intact. Correlation never grants authority and is not a permission identity, retry token, idempotency key, execution identity, lifecycle key, or model-supplied input. If recorder Session `W` and business Session `C` differ, each Session keeps its own one-invocation semantic evidence; correlation is never used for cross-Session global deduplication. Current-v2 ledger events without the additive correlation fields remain uncorrelated and are projected conservatively per event; restore never invents an id or rewrites persisted history. -Stateless MCP 2026 also projects optional `ack_session_message_ids` wrapper metadata, bounded to eight opaque `wc_msg_*` ids. For Session messages, an ACK is request-scoped evidence that the current model context still retains an unresolved ACK-required message in the exact authorized attention Workflow Session: the explicit recorder when present, otherwise the narrow authorized Window-affinity fallback above. For Window Peer messages, the same wrapper may acknowledge an ACK-required message addressed to the current principal-bound ClientWindow even when no Workflow Session recorder exists. The adapter removes ACK metadata before concrete tool parsing; ACK never grants authority, resolves a message, accepts work, or gates the concrete tool effect. Any Session or Peer message kind/priority may request ACK. Accepted ids suppress that body only in the current response; later omission makes an unresolved Session message or retained Peer ACK message eligible for bounded re-projection. Historical ACK state is never used to infer current model-context retention. +Stateless MCP 2026 projects two request-scoped ACK evidence forms. `ack_session_message_ids` remains the bounded exact-ID form and can acknowledge both Session and Window Peer messages. For Session attention, the Server also returns an optional compact `ack_ref`. Each returned ref represents the exact Session ACK set explicitly retained across the current exchange: ACK evidence accepted on this request plus ACK-required messages newly projected in this response. As further bounded messages are projected, the next ref carries that retained set forward in one value. Echoing the ref on a later ordinary tool call expands only after the exact authorized attention Session has been established: the explicit recorder when present, otherwise the narrow authorized Window-affinity fallback above. The ref is Session-only, principal/Session-bound by the existing authority path, and fenced against the full current open ACK-required Session membership; adding or removing membership makes an older ref acknowledge nothing rather than absorb a different set. ACK bookkeeping itself does not change membership, so exact replay remains safe. Peer ACK continues to use explicit `ack_session_message_ids`. + +The adapter removes both ACK forms before concrete tool parsing. Neither grants authority, resolves a message, accepts work, creates sticky Session/Window context, or gates the concrete tool effect. Any Session or Peer message kind/priority may request ACK. Accepted Session evidence suppresses only the represented body/bodies for the current response; later omission makes unresolved Session messages eligible for bounded re-projection with a current `ack_ref`. Retained Peer ACK messages follow the existing exact-ID behavior. Historical ACK state is never used to infer current model-context retention. An ACK-required Session message may persist `first_ack_observed_at`; an ACK-required Peer message persists the analogous window-message timestamp. For Session messages only the first accepted ACK advances message-observation revision; repeated echoes do not create revision churn. These fields mean only that the Server once observed an explicit ACK echo. They are not delivery/read receipts and do not by themselves change business status. `resolve_session_message` remains the durable processed-state transition for Session messages. @@ -137,7 +139,7 @@ with its canonical authority and replay fences; standalone tools remain valid. Executable todo completion is assignment-fenced in 0.4. A worker first calls `get_session_assignment` for the exact coordinator `session_id + message_id`; one atomic store snapshot returns the open todo, every retained direct reply within the bound, and an opaque Session/todo-bound `assignment_fence`. Current `complete_session_message` requests require that exact token as `expected_assignment_fence` together with the independent caller `completion_key`. Assignment-local semantic changes stale the fence before mutation; unrelated Session traffic and ACK bookkeeping do not. A stale result has `state_changed=false` and includes the current assignment plus a fresh durable fence only when that exact current state remains provable. Retention loss or an oversized direct-reply set is non-completable from stale context. -The fence and completion key are different identity domains: the fence proves the semantic assignment snapshot, while the completion key correlates one accepted intent across uncertain retries. `ack_session_message_ids` and message-observation tokens are separate continuity/cursor domains and cannot substitute for either. Business `session_id` remains the independently authorized target; `recording_session_id` remains provenance only and never supplies business guards or execution defaults. Current persisted rows must carry the assignment-history and completion-fence tracking metadata required by the v2 format; restore never invents a missing fence, fingerprint, or tracking state, and a row that cannot prove the current shape is discarded rather than admitted for replay. +The fence and completion key are different identity domains: the fence proves the semantic assignment snapshot, while the completion key correlates one accepted intent across uncertain retries. `ack_session_message_ids` and compact Session `ack_ref` are request-scoped ACK evidence; message-observation tokens are a separate cursor domain. `session_ref` is only a principal-scoped business Session selector and never ACK evidence. None substitutes for assignment/completion identity. Business `session_id` remains the independently authorized target; `recording_session_id` remains provenance only and never supplies business guards or execution defaults. Current persisted rows must carry the assignment-history and completion-fence tracking metadata required by the v2 format; restore never invents a missing fence, fingerprint, or tracking state, and a row that cannot prove the current shape is discarded rather than admitted for replay. ### Explicit handoff recovery and internal snapshot fencing @@ -182,8 +184,10 @@ and without promoting external reports into Session revision or native execution truth. No handoff generation or replacement model-context revision is introduced or returned. -Collaboration is independent: `ack_session_message_ids` still proves that the -current request retains specific ACK-required Session/Peer messages. +Collaboration is independent: `ack_session_message_ids` still proves exact +ACK-required Session/Peer message ids, while `session_attention.ack_ref` is the +compact exact-set form carrying accepted Session ACK evidence plus messages newly +projected in the current exchange. `session_attention` suppression/re-projection, message resolution, assignment fences, completion keys, message-observation tokens, and their durable revision keep their existing semantics. A handoff neither ACKs nor resolves a message and @@ -191,7 +195,7 @@ grants no authority. Stateless MCP 2026 tools also accept an explicit bounded `context_request` wrapper sidecar request. It is independent of collaboration ACKs and handoff recovery and is removed before concrete `ToolCall` parsing. A static canonical material registry authorizes every requested material before its provider is read: `webcodex.workflow` is public; `project.instructions` requires the resolved Project plus `project:read`; `jobs.attention` requires the exact resolved Project plus canonical `runtime:read` and reuses the authorized active-Job summary (at most eight recent Jobs) without selecting a business Session; `skills.catalog` additionally requires the admitted Skill runtime protocol capability; `plugins.catalog` requires the resolved Project plus both `project:read` and `plugin:inspect`; and `memory.bootstrap` requires the admitted Memory protocol capability plus both `project:read` and `memory:read`. Scope or material-capability denial is nonfatal to the main ToolResult and returns a bounded unavailable material without provider content. Unknown material keys are nonfatal and remain open-ended at the MCP schema layer. Sidecar material is projected only after the main tool effect or observation has completed, never grants authority, never retroactively makes requested guidance a precondition of that effect, never records caller-read state, and never infers a Project or model-memory state from a Workflow Session, connection, credential, `Mcp-Session-Id`, or hidden window identity. A model that has lost Project rules or durable Memory guidance must recover `project.instructions` and/or `memory.bootstrap` on an observation call, use `memory_read` when detailed Memory content is needed, reason over that context, and only then issue a later mutation that must obey it. Legacy MCP and generic REST/GPT Actions/OpenAPI do not expose this sidecar request contract. -Project Memory is a separate durable knowledge plane from Workflow Session continuity. `memory_search`/`memory_read` require both `project:read` and `memory:read`; `memory_set`/`memory_delete` require both `project:write` and `memory:manage`, with mutations still passing the independent permission evaluator. Direct shared-key runtime credentials explicitly carry both Memory scopes, while Open Anonymous, ProjectCredential, Project Share, and legacy/default OAuth client scope sets do not gain them from project scopes. A Memory `memory_key` is logical semantic identity, `memory_id` identifies the current incarnation, the internal `definition_hash` identifies canonical model-relevant content, and model-facing `revision` is a generation-bound state ETag/CAS identity; delete and identical recreate therefore produce a different `memory_id` and `revision`. Session events never create or consolidate Memory automatically. `ack_session_message_ids` is limited to ACK-required collaboration messages and never acknowledges Memory. Memory reads/searches may leave bounded metadata-only consequences in Session history, but Memory bodies, summaries, search results, and `memory.bootstrap` projections are not copied into durable Session recovery. Re-registering the same runtime Project id to a different authoritative registered root resolves to a distinct internal Memory scope rather than inheriting the old root's Memory. +Project Memory is a separate durable knowledge plane from Workflow Session continuity. `memory_search`/`memory_read` require both `project:read` and `memory:read`; `memory_set`/`memory_delete` require both `project:write` and `memory:manage`, with mutations still passing the independent permission evaluator. Direct shared-key runtime credentials explicitly carry both Memory scopes, while Open Anonymous, ProjectCredential, Project Share, and legacy/default OAuth client scope sets do not gain them from project scopes. A Memory `memory_key` is logical semantic identity, `memory_id` identifies the current incarnation, the internal `definition_hash` identifies canonical model-relevant content, and model-facing `revision` is a generation-bound state ETag/CAS identity; delete and identical recreate therefore produce a different `memory_id` and `revision`. Session events never create or consolidate Memory automatically. `ack_session_message_ids` and Session `ack_ref` are limited to ACK-required collaboration messages and never acknowledge Memory. Memory reads/searches may leave bounded metadata-only consequences in Session history, but Memory bodies, summaries, search results, and `memory.bootstrap` projections are not copied into durable Session recovery. Re-registering the same runtime Project id to a different authoritative registered root resolves to a distinct internal Memory scope rather than inheriting the old root's Memory. ### Message observation state diff --git a/docs/experiments/code-mode.md b/docs/experiments/code-mode.md index e8cd394f3..20e1b6546 100644 --- a/docs/experiments/code-mode.md +++ b/docs/experiments/code-mode.md @@ -217,6 +217,7 @@ project session_id recording_session_id ack_session_message_ids +ack_ref context_request session_message_resolution expected_failure diff --git a/src/mcp.rs b/src/mcp.rs index 54243ff48..6a39ac1c0 100644 --- a/src/mcp.rs +++ b/src/mcp.rs @@ -63,7 +63,7 @@ use tools::{ add_stateless_workflow_recorder_metadata, mcp_host_file_import_trust_decision_from_state, mcp_host_file_import_trust_from_state, mcp_tools_list_payload_with_compact, mcp_tools_list_payload_with_compact_and_app, mcp_tools_list_payload_with_features_for_auth, - strip_recording_session_id, strip_stateless_ack_session_message_ids, + strip_recording_session_id, strip_stateless_ack_ref, strip_stateless_ack_session_message_ids, strip_stateless_context_request, strip_stateless_session_message_resolution, take_last_mcp_host_file_import_trust_decision, HostFileImportTrustReason, McpToolCallParams, }; diff --git a/src/mcp/discovery.rs b/src/mcp/discovery.rs index e13e26e37..c4a514096 100644 --- a/src/mcp/discovery.rs +++ b/src/mcp/discovery.rs @@ -46,7 +46,8 @@ pub(super) fn compact_tool(tool: &mut Value) { // Nested IDs/resolution have no description; keep their type hints here. for (pointer, description) in [ ("/properties/recording_session_id", "Optional explicit wc_sess_* recorder for one exact Workflow Session; never execution/business authority. If omitted, authorized same-Window affinity may still deliver and ACK Session collaboration without recording."), - ("/properties/ack_session_message_ids", "ACK-required wc_msg_* IDs retained in model context; Session ACK uses explicit recorder or authorized same-Window affinity; never resolves or authorizes."), + ("/properties/ack_session_message_ids", "ACK-required wc_msg_* IDs retained in model context; Session uses recorder/window affinity; never resolves or authorizes."), + ("/properties/ack_ref", "Compact exact Session ACK-set evidence returned in session_attention; Session-only, request-scoped, non-authoritative, and never resolves messages."), ("/properties/session_message_resolution", "Resolve one handled non-todo recorder message by exact wc_msg_*; ACK separately if required. Independent of call success."), ("/properties/context_request", "Post-result sidecar keys; no authority: project.instructions, webcodex.workflow, jobs.attention, skills.catalog, plugins.catalog, memory.bootstrap."), ("/properties/context_request/items", "Context key; unsupported keys are nonfatal."), @@ -58,6 +59,15 @@ pub(super) fn compact_tool(tool: &mut Value) { *copy = description.to_string(); } } + // ACK ref is echoed verbatim from session_attention. Repeating the + // explanation on every compact tool would dominate the token savings. + // Full discovery keeps the safety contract; compact keeps field/bound. + if let Some(property) = schema + .pointer_mut("/properties/ack_ref") + .and_then(Value::as_object_mut) + { + property.remove("description"); + } compact_discovery_validation_annotations(schema); } } diff --git a/src/mcp/tools.rs b/src/mcp/tools.rs index fdebd2cb9..46a62a0f5 100644 --- a/src/mcp/tools.rs +++ b/src/mcp/tools.rs @@ -212,6 +212,7 @@ fn unwrap_adaptive_runtime_gateway_arguments( allowed_wrapper_fields.extend([ crate::tool_runtime::sessions::TOOL_CALL_RECORDING_SESSION_ID_FIELD, crate::tool_runtime::sessions::TOOL_CALL_ACK_SESSION_MESSAGE_IDS_FIELD, + crate::tool_runtime::sessions::TOOL_CALL_ACK_REF_FIELD, crate::tool_runtime::sessions::TOOL_CALL_SESSION_MESSAGE_RESOLUTION_FIELD, crate::tool_runtime::context_projection::TOOL_CALL_CONTEXT_REQUEST_FIELD, crate::tool_runtime::control_sidecar::CONTROL_FIELD, @@ -460,6 +461,14 @@ fn stateless_collaboration_ack_schema() -> Value { }) } +fn stateless_session_ack_ref_schema() -> Value { + json!({ + "type": "string", + "maxLength": crate::tool_runtime::sessions::MAX_TOOL_CALL_ACK_REF_CHARS, + "description": "Compact request-scoped ACK evidence returned as session_attention.ack_ref for the exact retained Session ACK set represented by that exchange. It acknowledges only that Session set, grants no authority, never resolves messages, and does not apply to Peer ACK." + }) +} + fn stateless_session_attention_output_schema() -> Value { json!({ "type": "object", @@ -475,6 +484,11 @@ fn stateless_session_attention_output_schema() -> Value { "enum": ["recording_session", "business_session", "window_affinity"] }, "requires_ack": {"type": "boolean"}, + "ack_ref": { + "type": "string", + "maxLength": crate::tool_runtime::sessions::MAX_TOOL_CALL_ACK_REF_CHARS, + "description": "Exact compact Session ACK set retained across this exchange; echo on a later request while this context is still retained." + }, "messages": { "type": "array", "maxItems": crate::tool_runtime::SESSION_ATTENTION_MAX_MESSAGES, @@ -538,6 +552,10 @@ fn insert_stateless_collaboration_ack_property(properties: &mut serde_json::Map< crate::tool_runtime::sessions::TOOL_CALL_ACK_SESSION_MESSAGE_IDS_FIELD.to_string(), stateless_collaboration_ack_schema(), ); + properties.insert( + crate::tool_runtime::sessions::TOOL_CALL_ACK_REF_FIELD.to_string(), + stateless_session_ack_ref_schema(), + ); } pub(super) fn add_stateless_workflow_recorder_metadata(payload: &mut Value) { @@ -1474,6 +1492,31 @@ pub(super) fn strip_stateless_ack_session_message_ids( Ok(normalized) } +pub(super) fn strip_stateless_ack_ref(arguments: &mut Value) -> Result, String> { + let Some(object) = arguments.as_object_mut() else { + return Ok(None); + }; + match object.remove(crate::tool_runtime::sessions::TOOL_CALL_ACK_REF_FIELD) { + None => Ok(None), + Some(Value::String(value)) => { + let value = value.trim(); + if value.is_empty() + || value.len() > crate::tool_runtime::sessions::MAX_TOOL_CALL_ACK_REF_CHARS + { + return Err(format!( + "field '{}' must be a non-empty bounded string", + crate::tool_runtime::sessions::TOOL_CALL_ACK_REF_FIELD + )); + } + Ok(Some(value.to_string())) + } + Some(_) => Err(format!( + "field '{}' must be a non-empty bounded string", + crate::tool_runtime::sessions::TOOL_CALL_ACK_REF_FIELD + )), + } +} + pub(super) fn strip_stateless_session_message_resolution( arguments: &mut Value, ) -> Result, String> { @@ -1696,6 +1739,20 @@ pub(super) async fn handle_call( } else { Vec::new() }; + let ack_ref = if stateless_2026 { + match strip_stateless_ack_ref(&mut params.arguments) { + Ok(ack_ref) => ack_ref, + Err(message) => { + if let Some(lc) = lifecycle.as_deref() { + lc.dispatch_failed("invalid_arguments"); + lc.dispatch_finished(false, Some(false), "invalid_arguments"); + } + return McpOutcome::BadRequest(rpc_error(id, -32602, message)); + } + } + } else { + None + }; // Strip private control payloads before tracing, canonical argument parsing, // specialized dispatch, and audit. Legacy/hidden adapters reject explicitly. let control = match crate::tool_runtime::control_sidecar::strip_control_sidecars( @@ -2252,6 +2309,7 @@ pub(super) async fn handle_call( ToolInvocationMetadata { control, ack_session_message_ids, + ack_ref, session_message_resolution, context_request, }, diff --git a/src/mcp_tests/http_transport.rs b/src/mcp_tests/http_transport.rs index df706fc02..d66954e62 100644 --- a/src/mcp_tests/http_transport.rs +++ b/src/mcp_tests/http_transport.rs @@ -1183,11 +1183,15 @@ async fn http_mcp_2026_request_scoped_ack_redelivers_until_durable_resolution_bo first["session_attention"]["messages"][0]["message"], "Keep the exact request-scoped ACK contract." ); + let first_ack_ref = first["session_attention"]["ack_ref"] + .as_str() + .expect("projected Session ACK set should expose ack_ref") + .to_string(); let mut ack_args = with_mcp_recording_session(json!({}), &session_id); ack_args.as_object_mut().unwrap().insert( - crate::tool_runtime::sessions::TOOL_CALL_ACK_SESSION_MESSAGE_IDS_FIELD.to_string(), - json!([message_id, message_id]), + crate::tool_runtime::sessions::TOOL_CALL_ACK_REF_FIELD.to_string(), + json!(first_ack_ref), ); let (status, ack_body) = stateless_2026_tool_call(&service, "secret", 223, "list_tools", ack_args, None).await; @@ -1279,6 +1283,30 @@ async fn http_mcp_2026_request_scoped_ack_redelivers_until_durable_resolution_bo .as_str() .unwrap() .to_string(); + + let mut stale_ref_args = with_mcp_recording_session(json!({}), &session_id); + stale_ref_args.as_object_mut().unwrap().insert( + crate::tool_runtime::sessions::TOOL_CALL_ACK_REF_FIELD.to_string(), + json!(first_ack_ref), + ); + let (status, stale_ref_body) = + stateless_2026_tool_call(&service, "secret", 2271, "list_tools", stale_ref_args, None) + .await; + assert_eq!(status, StatusCode::OK, "{stale_ref_body}"); + let stale_ref_output = stateless_tool_output(&stale_ref_body); + assert_eq!( + stale_ref_output["session_attention"]["ack"]["accepted_count"], + 0 + ); + assert_eq!( + stale_ref_output["session_attention"]["messages"][0]["message_id"], + second_message_id + ); + assert_ne!( + stale_ref_output["session_attention"]["ack_ref"], first_ack_ref, + "changed Session ACK membership must replace the old ref" + ); + let mut resolve_with_ack_args = json!({ "session_id": session_id, "message_id": second_message_id, @@ -1458,6 +1486,7 @@ async fn http_mcp_2026_request_scoped_ack_redelivers_until_durable_resolution_bo ) .unwrap(); assert!(!audit.contains("ack_session_message_ids")); + assert!(!audit.contains("ack_ref")); assert!(!audit.contains("__webcodex_stateless_ack_session_message_ids")); assert!(!audit.contains("__webcodex_stateless_session_message_resolution")); assert!(!audit.contains("handled through ordinary list_tools wrapper metadata")); diff --git a/src/mcp_tests/tools.rs b/src/mcp_tests/tools.rs index 96f731004..35739d379 100644 --- a/src/mcp_tests/tools.rs +++ b/src/mcp_tests/tools.rs @@ -651,6 +651,7 @@ fn stateless_workflow_recorder_metadata_adds_protocol_projection() { ); } assert!(input.get("ack_session_message_ids").is_some()); + assert!(input.get("ack_ref").is_some()); } assert!(!read_files_output.contains("session_continuity")); assert!(!read_files_output.contains("session_recovery")); @@ -673,6 +674,9 @@ fn stateless_workflow_recorder_metadata_adds_protocol_projection() { .contains_key(crate::tool_runtime::sessions::TOOL_CALL_RECORDING_SESSION_ID_FIELD)); assert!(!generic_properties .contains_key(crate::tool_runtime::sessions::TOOL_CALL_ACK_SESSION_MESSAGE_IDS_FIELD)); + assert!( + !generic_properties.contains_key(crate::tool_runtime::sessions::TOOL_CALL_ACK_REF_FIELD) + ); assert!(!generic_properties .contains_key(crate::tool_runtime::sessions::TOOL_CALL_SESSION_MESSAGE_RESOLUTION_FIELD)); assert!(!generic_properties.contains_key("ack_session_context_revision")); @@ -717,6 +721,31 @@ fn stateless_ack_wrapper_normalizes_and_is_removed_before_concrete_tool_parsing( assert!(strip_stateless_ack_session_message_ids(&mut oversized).is_err()); } +#[test] +fn stateless_ack_ref_wrapper_is_bounded_and_removed_before_concrete_parsing() { + let mut arguments = json!({ + crate::tool_runtime::sessions::TOOL_CALL_ACK_REF_FIELD: " wc_ack1_example " + }); + let ack_ref = strip_stateless_ack_ref(&mut arguments).unwrap(); + assert_eq!(ack_ref.as_deref(), Some("wc_ack1_example")); + assert!(arguments + .get(crate::tool_runtime::sessions::TOOL_CALL_ACK_REF_FIELD) + .is_none()); + crate::tool_runtime::ToolCall::from_tool_name("list_tools", arguments) + .expect("ACK ref wrapper metadata must be gone before concrete parsing"); + + let mut wrong_type = json!({ + crate::tool_runtime::sessions::TOOL_CALL_ACK_REF_FIELD: ["wc_ack1_example"] + }); + assert!(strip_stateless_ack_ref(&mut wrong_type).is_err()); + + let mut oversized = json!({ + crate::tool_runtime::sessions::TOOL_CALL_ACK_REF_FIELD: + "x".repeat(crate::tool_runtime::sessions::MAX_TOOL_CALL_ACK_REF_CHARS + 1) + }); + assert!(strip_stateless_ack_ref(&mut oversized).is_err()); +} + #[test] fn stateless_message_resolution_wrapper_is_validated_and_removed_before_concrete_parsing() { let mut arguments = json!({ @@ -807,6 +836,7 @@ fn stateless_invocation_metadata_stays_typed_and_business_arguments_stay_clean() "items": [{"path": "src/lib.rs"}], crate::tool_runtime::sessions::TOOL_CALL_RECORDING_SESSION_ID_FIELD: "wc_sess_adapter", crate::tool_runtime::sessions::TOOL_CALL_ACK_SESSION_MESSAGE_IDS_FIELD: ["wc_msg_abcd-efgh_ijklmn"], + crate::tool_runtime::sessions::TOOL_CALL_ACK_REF_FIELD: "wc_ack1_fixture", crate::tool_runtime::sessions::TOOL_CALL_SESSION_MESSAGE_RESOLUTION_FIELD: { "message_id": "wc_msg_abcd-efgh_ijklmn", "resolution": "handled" @@ -815,12 +845,14 @@ fn stateless_invocation_metadata_stays_typed_and_business_arguments_stay_clean() }); let recording_session_id = strip_recording_session_id(&mut arguments).unwrap(); let ack_session_message_ids = strip_stateless_ack_session_message_ids(&mut arguments).unwrap(); + let ack_ref = strip_stateless_ack_ref(&mut arguments).unwrap(); let session_message_resolution = strip_stateless_session_message_resolution(&mut arguments).unwrap(); let context_request = strip_stateless_context_request(&mut arguments).unwrap(); let metadata = crate::tool_runtime::kernel::ToolInvocationMetadata { control: None, ack_session_message_ids, + ack_ref, session_message_resolution, context_request, }; @@ -830,11 +862,13 @@ fn stateless_invocation_metadata_stays_typed_and_business_arguments_stay_clean() metadata.ack_session_message_ids, vec!["wc_msg_abcd-efgh_ijklmn"] ); + assert_eq!(metadata.ack_ref.as_deref(), Some("wc_ack1_fixture")); assert_eq!(metadata.context_request, vec!["webcodex.workflow"]); assert!(metadata.session_message_resolution.is_some()); for field in [ crate::tool_runtime::sessions::TOOL_CALL_RECORDING_SESSION_ID_FIELD, crate::tool_runtime::sessions::TOOL_CALL_ACK_SESSION_MESSAGE_IDS_FIELD, + crate::tool_runtime::sessions::TOOL_CALL_ACK_REF_FIELD, crate::tool_runtime::sessions::TOOL_CALL_SESSION_MESSAGE_RESOLUTION_FIELD, crate::tool_runtime::context_projection::TOOL_CALL_CONTEXT_REQUEST_FIELD, ] { diff --git a/src/tool_runtime/code_mode.rs b/src/tool_runtime/code_mode.rs index 93a9136c0..7c098ee64 100644 --- a/src/tool_runtime/code_mode.rs +++ b/src/tool_runtime/code_mode.rs @@ -667,6 +667,7 @@ mod tests { "session_id", "recording_session_id", "ack_session_message_ids", + "ack_ref", "context_request", "session_message_resolution", "_control", diff --git a/src/tool_runtime/kernel.rs b/src/tool_runtime/kernel.rs index e5704d091..33fc9fdc4 100644 --- a/src/tool_runtime/kernel.rs +++ b/src/tool_runtime/kernel.rs @@ -59,6 +59,7 @@ pub(crate) struct ToolCallRequest { pub(crate) struct ToolInvocationMetadata { pub(crate) control: Option, pub(crate) ack_session_message_ids: Vec, + pub(crate) ack_ref: Option, pub(crate) session_message_resolution: Option, pub(crate) context_request: Vec, } @@ -345,6 +346,7 @@ impl ToolRuntime { return outcome; } } + let ack_ref = invocation_metadata.ack_ref.clone(); let mut recorder_metadata = ToolCallRecorderMetadata::from_business_arguments(&request.arguments); recorder_metadata.ack_session_message_ids = invocation_metadata.ack_session_message_ids; @@ -586,7 +588,16 @@ impl ToolRuntime { }; } } - let recorder_ack_requested = !recorder_metadata.ack_session_message_ids.is_empty(); + let recorder_ack_requested = + !recorder_metadata.ack_session_message_ids.is_empty() || ack_ref.is_some(); + let explicit_session_ack_ids = context.session_id.map(|recorder_session_id| { + session_context::session_ack_message_ids( + &self.sessions, + recorder_session_id, + &recorder_metadata.ack_session_message_ids, + ack_ref.as_deref(), + ) + }); if let Some(recorder_session_id) = context.session_id { recorder_metadata.recording_session_id = Some(recorder_session_id.to_string()); recorder_metadata.recording_session_project = self @@ -640,7 +651,9 @@ impl ToolRuntime { session_context::observe_session_attention_acks( &self.sessions, recorder_session_id, - &recorder_metadata.ack_session_message_ids, + explicit_session_ack_ids + .as_deref() + .unwrap_or(&recorder_metadata.ack_session_message_ids), ) }); if collaboration_session_tool(&request.tool_name) { @@ -1029,10 +1042,16 @@ impl ToolRuntime { // authorized active Session message board for request-scoped ACK and // delivery, but never becomes recorder, business input, execution // context, or durable resolution authority. - let ack = session_context::observe_session_attention_acks( + let session_ack_ids = session_context::session_ack_message_ids( &self.sessions, session_id, &recorder_metadata.ack_session_message_ids, + ack_ref.as_deref(), + ); + let ack = session_context::observe_session_attention_acks( + &self.sessions, + session_id, + &session_ack_ids, ); session_context::add_session_attention_projection( &mut result, diff --git a/src/tool_runtime/orchestration_host.rs b/src/tool_runtime/orchestration_host.rs index bc93de807..6ecf84466 100644 --- a/src/tool_runtime/orchestration_host.rs +++ b/src/tool_runtime/orchestration_host.rs @@ -15,7 +15,7 @@ use tokio::sync::{ Mutex as AsyncMutex, OwnedMutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuard, }; use webcodex_core::workflow_session_contract::{ - TOOL_ACCEPTED_EXIT_CODES_FIELD, TOOL_ASSERTION_NAME_FIELD, + TOOL_ACCEPTED_EXIT_CODES_FIELD, TOOL_ASSERTION_NAME_FIELD, TOOL_CALL_ACK_REF_FIELD, TOOL_CALL_ACK_SESSION_MESSAGE_IDS_FIELD, TOOL_CALL_RECORDING_SESSION_ID_FIELD, TOOL_CALL_SESSION_MESSAGE_RESOLUTION_FIELD, TOOL_EXPECTED_FAILURE_FIELD, TOOL_EXPECTED_FAILURE_KIND_FIELD, TOOL_RESULT_EXPECTATION_FIELD, @@ -145,6 +145,7 @@ const SERVER_OWNED_ARGUMENT_FIELDS: &[&str] = &[ "session_id", TOOL_CALL_RECORDING_SESSION_ID_FIELD, TOOL_CALL_ACK_SESSION_MESSAGE_IDS_FIELD, + TOOL_CALL_ACK_REF_FIELD, TOOL_CALL_SESSION_MESSAGE_RESOLUTION_FIELD, TOOL_CALL_CONTEXT_REQUEST_FIELD, super::control_sidecar::CONTROL_FIELD, diff --git a/src/tool_runtime/session_context.rs b/src/tool_runtime/session_context.rs index 1109ac360..c4cbf778b 100644 --- a/src/tool_runtime/session_context.rs +++ b/src/tool_runtime/session_context.rs @@ -372,6 +372,29 @@ pub(crate) fn observe_session_attention_acks( sessions.observe_message_acks(session_id, ack_message_ids) } +pub(crate) fn session_ack_message_ids( + sessions: &sessions::SessionStore, + session_id: &str, + legacy_ids: &[String], + ack_ref: Option<&str>, +) -> Vec { + let mut ids = legacy_ids.to_vec(); + if let Some(resolved) = + ack_ref.and_then(|ack_ref| sessions.resolve_ack_ref(session_id, ack_ref)) + { + let mut seen = ids + .iter() + .cloned() + .collect::>(); + for message_id in resolved { + if seen.insert(message_id.clone()) { + ids.push(message_id); + } + } + } + ids +} + pub(crate) fn add_session_attention_projection( result: &mut ToolResult, sessions: &sessions::SessionStore, @@ -384,6 +407,7 @@ pub(crate) fn add_session_attention_projection( let unsuppressed_count = attention.messages.len(); let mut remaining_bytes = SESSION_ATTENTION_MAX_BODY_BYTES; let mut messages = Vec::new(); + let mut projected_ids = Vec::new(); let mut body_truncated = false; for message in attention .messages @@ -396,6 +420,7 @@ pub(crate) fn add_session_attention_projection( let (body, truncated) = bound_utf8_bytes(&message.message, remaining_bytes); body_truncated |= truncated; remaining_bytes = remaining_bytes.saturating_sub(body.len()); + projected_ids.push(message.message_id.clone()); messages.push(json!({ "message_id": message.message_id, "kind": message.kind.as_str(), @@ -431,21 +456,25 @@ pub(crate) fn add_session_attention_projection( hint.remove("attention_instruction"); } } - output.insert( - "session_attention".to_string(), - json!({ - "session_id": session_id, - "source": source, - "requires_ack": attention.total_open_requires_ack > 0, - "messages": messages, - "omitted_count": omitted_count, - "truncated": omitted_count > 0, - "ack": { - "accepted_count": ack.accepted_count, - "ignored_count": ack.ignored_count, - } - }), - ); + let mut retained_ack_ids = ack.accepted_ids.clone(); + retained_ack_ids.extend(projected_ids); + let ack_ref = sessions.issue_ack_ref(session_id, &retained_ack_ids); + let mut projection = json!({ + "session_id": session_id, + "source": source, + "requires_ack": attention.total_open_requires_ack > 0, + "messages": messages, + "omitted_count": omitted_count, + "truncated": omitted_count > 0, + "ack": { + "accepted_count": ack.accepted_count, + "ignored_count": ack.ignored_count, + } + }); + if let Some(ack_ref) = ack_ref { + projection["ack_ref"] = json!(ack_ref); + } + output.insert("session_attention".to_string(), projection); result.output = Value::Object(output); } diff --git a/src/tool_runtime/sessions/mod.rs b/src/tool_runtime/sessions/mod.rs index 9873a640c..77f0b548f 100644 --- a/src/tool_runtime/sessions/mod.rs +++ b/src/tool_runtime/sessions/mod.rs @@ -25,11 +25,11 @@ pub(crate) use webcodex_workflow_session::{ MAX_MESSAGE_COMPLETION_KEY_CHARS, MAX_MESSAGE_DELIVERY_KEY_CHARS, MAX_MESSAGE_LIST_LIMIT, MAX_MESSAGE_RESOLUTION_CHARS, MAX_MESSAGE_TAGS, MAX_MESSAGE_TAG_CHARS, MAX_MODEL_VALIDATION_ASSERTION_NAME_CHARS, MAX_TOOL_CALL_ACK_MESSAGE_IDS, - SESSION_INBOX_ACK_REQUIRED_ATTENTION_INSTRUCTION, SESSION_INBOX_ACK_REQUIRED_ATTENTION_REASON, - TOOL_ACCEPTED_EXIT_CODES_FIELD, TOOL_ASSERTION_NAME_FIELD, - TOOL_CALL_ACK_SESSION_MESSAGE_IDS_FIELD, TOOL_CALL_RECORDING_SESSION_ID_FIELD, - TOOL_CALL_SESSION_MESSAGE_RESOLUTION_FIELD, TOOL_EXPECTATION_RESULT_UNEXPECTED_FAILURE, - TOOL_RESULT_EXPECTATION_FIELD, + MAX_TOOL_CALL_ACK_REF_CHARS, SESSION_INBOX_ACK_REQUIRED_ATTENTION_INSTRUCTION, + SESSION_INBOX_ACK_REQUIRED_ATTENTION_REASON, TOOL_ACCEPTED_EXIT_CODES_FIELD, + TOOL_ASSERTION_NAME_FIELD, TOOL_CALL_ACK_REF_FIELD, TOOL_CALL_ACK_SESSION_MESSAGE_IDS_FIELD, + TOOL_CALL_RECORDING_SESSION_ID_FIELD, TOOL_CALL_SESSION_MESSAGE_RESOLUTION_FIELD, + TOOL_EXPECTATION_RESULT_UNEXPECTED_FAILURE, TOOL_RESULT_EXPECTATION_FIELD, }; use super::metadata::{ToolPathHint, ToolRisk}; diff --git a/src/tool_runtime/tests/code_mode.rs b/src/tool_runtime/tests/code_mode.rs index 1bf48b544..af402cd09 100644 --- a/src/tool_runtime/tests/code_mode.rs +++ b/src/tool_runtime/tests/code_mode.rs @@ -1239,6 +1239,7 @@ async fn canonical_orchestration_host_rejects_server_owned_metadata_without_fron ("session_id", json!("wc_sess_0000000000000000")), ("recording_session_id", json!("wc_sess_0000000000000000")), ("ack_session_message_ids", json!([])), + ("ack_ref", json!("wc_ack1_fixture")), ("context_request", json!(["webcodex.workflow"])), ( "session_message_resolution", diff --git a/src/tool_runtime/tests/collaboration.rs b/src/tool_runtime/tests/collaboration.rs index e480e556e..91ec5783e 100644 --- a/src/tool_runtime/tests/collaboration.rs +++ b/src/tool_runtime/tests/collaboration.rs @@ -1245,6 +1245,107 @@ async fn collaboration_cross_project_recorder_fails_closed_before_completion() { ); } +#[tokio::test] +async fn ack_ref_accumulates_retained_and_newly_projected_session_attention() { + let runtime = test_runtime(); + let auth = auth_context(None, true); + let session = start_authorized_unscoped_session(&runtime, "cumulative ACK ref", &auth); + let mut ids = Vec::new(); + for index in 0..6 { + ids.push( + runtime + .sessions + .post_message_with_ack( + PostSessionMessageInput { + session_id: session.session_id.clone(), + kind: SessionMessageKind::Guidance, + message: format!("cumulative-guidance-{index}"), + tags: Vec::new(), + reply_to: None, + priority: SessionMessagePriority::Normal, + }, + true, + ) + .unwrap() + .message_id, + ); + } + + let first = call_with_recorder( + &runtime, + "list_tools", + json!({}), + Some(&session.session_id), + &auth, + None, + ) + .await; + assert!(first.success, "{:?}", first.error); + let first_messages = first.output["session_attention"]["messages"] + .as_array() + .unwrap(); + assert_eq!( + first_messages.len(), + super::super::SESSION_ATTENTION_MAX_MESSAGES + ); + let first_ref = first.output["session_attention"]["ack_ref"] + .as_str() + .expect("first bounded projection should mint ACK ref") + .to_string(); + + let second = call_with_recorder_metadata( + &runtime, + "list_tools", + json!({}), + Some(&session.session_id), + ToolInvocationMetadata { + ack_ref: Some(first_ref.clone()), + ..Default::default() + }, + &auth, + None, + ) + .await; + assert!(second.success, "{:?}", second.error); + assert_eq!( + second.output["session_attention"]["ack"]["accepted_count"], + 3 + ); + let second_messages = second.output["session_attention"]["messages"] + .as_array() + .unwrap(); + assert_eq!(second_messages.len(), 3); + let second_ref = second.output["session_attention"]["ack_ref"] + .as_str() + .expect("retained plus newly projected ACK set should mint replacement ref") + .to_string(); + assert_ne!(second_ref, first_ref); + + let third = call_with_recorder_metadata( + &runtime, + "list_tools", + json!({}), + Some(&session.session_id), + ToolInvocationMetadata { + ack_ref: Some(second_ref.clone()), + ..Default::default() + }, + &auth, + None, + ) + .await; + assert!(third.success, "{:?}", third.error); + assert_eq!( + third.output["session_attention"]["ack"]["accepted_count"], + ids.len() + ); + assert!(third.output["session_attention"]["messages"] + .as_array() + .unwrap() + .is_empty()); + assert_eq!(third.output["session_attention"]["ack_ref"], second_ref); +} + #[tokio::test] async fn foreign_recording_session_is_denied_before_ordinary_tool_recording() { let runtime = test_runtime(); @@ -1261,13 +1362,35 @@ async fn foreign_recording_session_is_denied_before_ordinary_tool_recording() { .await; assert!(started.success, "{:?}", started.error); let session_id = started.output["session_id"].as_str().unwrap().to_string(); + let ack_message = runtime + .sessions + .post_message_with_ack( + PostSessionMessageInput { + session_id: session_id.clone(), + kind: SessionMessageKind::Guidance, + message: "Alice-only ACK evidence".to_string(), + tags: Vec::new(), + reply_to: None, + priority: SessionMessagePriority::High, + }, + true, + ) + .unwrap(); + let ack_ref = runtime + .sessions + .issue_ack_ref(&session_id, std::slice::from_ref(&ack_message.message_id)) + .expect("Alice Session ACK set should mint a ref"); let before = runtime.sessions.summary(&session_id, Some(100)).unwrap(); - let denied = call_with_recorder( + let denied = call_with_recorder_metadata( &runtime, "list_projects", json!({}), Some(&session_id), + ToolInvocationMetadata { + ack_ref: Some(ack_ref), + ..Default::default() + }, &bob, None, ) @@ -1279,6 +1402,20 @@ async fn foreign_recording_session_is_denied_before_ordinary_tool_recording() { assert!(!tool_names(&runtime, &session_id) .iter() .any(|tool| tool == "list_projects")); + let message_after_denial = runtime + .sessions + .list_messages( + &session_id, + sessions::ListSessionMessagesFilter { + message_id: Some(ack_message.message_id), + ..Default::default() + }, + ) + .unwrap(); + assert!( + message_after_denial[0].first_ack_observed_at.is_none(), + "foreign principal must not turn possession of ack_ref into ACK authority" + ); let allowed = call_with_recorder( &runtime, diff --git a/src/tool_runtime/tests/work_on_project.rs b/src/tool_runtime/tests/work_on_project.rs index 64489681d..c6be4903f 100644 --- a/src/tool_runtime/tests/work_on_project.rs +++ b/src/tool_runtime/tests/work_on_project.rs @@ -2202,6 +2202,10 @@ async fn same_window_recorder_gap_is_visible_without_backfilling_session_ledger( unrecorded_result.output["session_attention"]["messages"][0]["message"], "same-window attention without recorder" ); + let affinity_ack_ref = unrecorded_result.output["session_attention"]["ack_ref"] + .as_str() + .expect("Window-affinity Session attention should expose ack_ref") + .to_string(); assert_eq!( runtime .sessions @@ -2223,7 +2227,7 @@ async fn same_window_recorder_gap_is_visible_without_backfilling_session_ledger( window_id, ToolTransport::Mcp, ToolInvocationMetadata { - ack_session_message_ids: vec![guidance.message_id.clone()], + ack_ref: Some(affinity_ack_ref), ..Default::default() }, )