use std::collections::{BTreeMap, HashSet};
use std::sync::{Arc, RwLock};
use chrono::Utc;
use tempfile::tempdir;
use crate::error::OmInferenceSource;
use crate::fs::LocalContextFs;
use crate::index::InMemoryIndex;
use crate::state::SqliteStateStore;
use super::*;
fn msg(id: &str, role: &str, text: &str) -> OmObserverMessageCandidate {
OmObserverMessageCandidate {
id: id.to_string(),
role: role.to_string(),
text: text.to_string(),
created_at: Utc::now(),
source_thread_id: Some("s-test".to_string()),
source_session_id: Some("s-test".to_string()),
}
}
fn msg_with_thread(
id: &str,
role: &str,
text: &str,
source_thread_id: &str,
source_session_id: &str,
) -> OmObserverMessageCandidate {
OmObserverMessageCandidate {
id: id.to_string(),
role: role.to_string(),
text: text.to_string(),
created_at: Utc::now(),
source_thread_id: Some(source_thread_id.to_string()),
source_session_id: Some(source_session_id.to_string()),
}
}
fn observer_config(mode: OmObserverMode, model_enabled: bool) -> OmObserverConfig {
let limits = crate::config::OmRuntimeLimitsConfig::default();
OmObserverConfig {
mode,
model_enabled,
llm: OmObserverLlmConfig {
endpoint: "http://127.0.0.1:11434/api/chat".to_string(),
model: "qwen2.5:7b-instruct".to_string(),
timeout_ms: DEFAULT_OM_OBSERVER_LLM_TIMEOUT_MS,
max_output_tokens: DEFAULT_OM_OBSERVER_LLM_MAX_OUTPUT_TOKENS,
temperature_milli: DEFAULT_OM_OBSERVER_LLM_TEMPERATURE_MILLI,
strict: false,
max_chars_per_message: DEFAULT_OM_OBSERVER_LLM_MAX_CHARS_PER_MESSAGE,
max_input_tokens: DEFAULT_OM_OBSERVER_LLM_MAX_INPUT_TOKENS,
},
text_budget: OmObserverTextBudget {
observation_max_chars: limits.observation_max_chars,
active_observations_max_chars: limits.observer_active_observations_max_chars,
other_conversation_max_part_chars: limits.observer_other_conversation_max_part_chars,
},
}
}
#[test]
fn parse_observer_response_value_reads_object_payload() {
let payload = serde_json::json!({
"observations": "[user] OAuth token refresh",
"observed_message_ids": ["m1", "m2"],
"observation_token_count": 8,
"usage": {"input_tokens": 12, "output_tokens": 5}
});
let known_ids = vec!["m1".to_string(), "m2".to_string()];
let parsed = parse_observer_response_value(
&payload,
&known_ids,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect("parse");
assert_eq!(parsed.observations, "[user] OAuth token refresh");
assert_eq!(
parsed.observed_message_ids,
vec!["m1".to_string(), "m2".to_string()]
);
assert_eq!(parsed.observation_token_count, 8);
assert_eq!(parsed.current_task, None);
assert_eq!(parsed.suggested_response, None);
assert_eq!(parsed.usage.input_tokens, 12);
assert_eq!(parsed.usage.output_tokens, 5);
}
#[test]
fn parse_observer_response_value_preserves_observation_lines() {
let payload = serde_json::json!({
"observations": " line-a \n\n line-b ",
"observed_message_ids": ["m1"],
"current_task": "Primary: debug auth",
"suggested_response": "Ask user to confirm"
});
let known_ids = vec!["m1".to_string()];
let parsed = parse_observer_response_value(
&payload,
&known_ids,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect("parse");
assert_eq!(parsed.observations, "line-a\nline-b");
assert_eq!(parsed.current_task.as_deref(), Some("Primary: debug auth"));
assert_eq!(
parsed.suggested_response.as_deref(),
Some("Ask user to confirm")
);
}
#[test]
fn parse_observer_response_value_accepts_mastra_alias_fields() {
let payload = serde_json::json!({
"observations": "line-a",
"observedMessageIds": ["m1"],
"observationTokenCount": 7,
"usage": {"inputTokens": 12, "outputTokens": 5},
"currentTask": "Primary: debug auth",
"suggestedContinuation": "Ask user to confirm"
});
let known_ids = vec!["m1".to_string()];
let parsed = parse_observer_response_value(
&payload,
&known_ids,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect("parse");
assert_eq!(parsed.observations, "line-a");
assert_eq!(parsed.observed_message_ids, vec!["m1".to_string()]);
assert_eq!(parsed.observation_token_count, 7);
assert_eq!(parsed.usage.input_tokens, 12);
assert_eq!(parsed.usage.output_tokens, 5);
assert_eq!(parsed.current_task.as_deref(), Some("Primary: debug auth"));
assert_eq!(
parsed.suggested_response.as_deref(),
Some("Ask user to confirm")
);
}
#[test]
fn parse_llm_observer_response_accepts_embedded_json_content() {
let payload = serde_json::json!({
"message": {
"content": "```json\n{\"header\":{\"contract_name\":\"axiomsync.om.prompt\",\"contract_version\":\"2.0.0\",\"protocol_version\":\"om-v2\"},\"observations\":\"[assistant] keep signal\",\"observed_message_ids\":[\"m9\"]}\n```"
}
});
let known_ids = vec!["m9".to_string()];
let parsed = parse_llm_observer_response(
&payload,
&known_ids,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect("parse");
assert_eq!(parsed.observations, "[assistant] keep signal");
assert_eq!(parsed.observed_message_ids, vec!["m9".to_string()]);
}
#[test]
fn parse_llm_observer_response_rejects_missing_contract_header_for_json_schema() {
let payload = serde_json::json!({
"observations": "line-a",
"observed_message_ids": ["m9"]
});
let known_ids = vec!["m9".to_string()];
let err = parse_llm_observer_response(
&payload,
&known_ids,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect_err("must reject missing contract header for known schema");
match err {
AxiomError::OmInference {
inference_source,
kind,
..
} => {
assert_eq!(inference_source, OmInferenceSource::Observer);
assert_eq!(kind, OmInferenceFailureKind::Schema);
}
other => panic!("unexpected error type: {other}"),
}
}
#[test]
fn parse_llm_observer_response_accepts_xml_observations_content() {
let payload = serde_json::json!({
"message": {
"content": "<contract-name>axiomsync.om.prompt</contract-name>\n<contract-version>2.0.0</contract-version>\n<protocol-version>om-v2</protocol-version>\n<observations>\n* 🔴 (09:15) User prefers direct answers\n* 🟡 (09:20) User asked about auth flow\n</observations>\n<current-task>\nPrimary: debug auth\n</current-task>\n<suggested-response>\nAsk user to confirm\n</suggested-response>"
}
});
let known_ids = vec!["m7".to_string(), "m8".to_string()];
let parsed = parse_llm_observer_response(
&payload,
&known_ids,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect("parse");
assert!(parsed.observations.contains("User prefers direct answers"));
assert!(parsed.observations.contains("User asked about auth flow"));
assert_eq!(parsed.observed_message_ids, known_ids);
assert_eq!(parsed.current_task.as_deref(), Some("Primary: debug auth"));
assert_eq!(
parsed.suggested_response.as_deref(),
Some("Ask user to confirm")
);
}
#[test]
fn parse_llm_observer_response_accepts_list_items_without_xml_tags() {
let payload = serde_json::json!({
"message": {
"content": "<contract-name>axiomsync.om.prompt</contract-name>\n<contract-version>2.0.0</contract-version>\n<protocol-version>om-v2</protocol-version>\n* 🔴 (09:15) User prefers direct answers\n- 🟡 (09:20) User asked about auth flow\n1. 🟢 (09:22) Assistant suggested follow-up"
}
});
let known_ids = vec!["m7".to_string()];
let parsed = parse_llm_observer_response(
&payload,
&known_ids,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect("parse");
assert!(parsed.observations.contains("User prefers direct answers"));
assert!(parsed.observations.contains("User asked about auth flow"));
assert!(
parsed
.observations
.contains("Assistant suggested follow-up")
);
assert_eq!(parsed.observed_message_ids, known_ids);
}
#[test]
fn parse_llm_observer_response_rejects_xml_without_contract_marker() {
let payload = serde_json::json!({
"message": {
"content": "<observations>\nline-a\nline-b\n</observations>"
}
});
let known_ids = vec!["m7".to_string()];
let err = parse_llm_observer_response(
&payload,
&known_ids,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect_err("must reject xml fallback without contract marker");
match err {
AxiomError::OmInference {
inference_source,
kind,
..
} => {
assert_eq!(inference_source, OmInferenceSource::Observer);
assert_eq!(kind, OmInferenceFailureKind::Schema);
}
other => panic!("unexpected error type: {other}"),
}
}
#[test]
fn parse_llm_observer_response_rejects_xml_without_protocol_marker() {
let payload = serde_json::json!({
"message": {
"content": "<contract-name>axiomsync.om.prompt</contract-name>\n<contract-version>2.0.0</contract-version>\n<observations>\nline-a\nline-b\n</observations>"
}
});
let known_ids = vec!["m7".to_string()];
let err = parse_llm_observer_response(
&payload,
&known_ids,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect_err("must reject xml fallback without protocol marker");
match err {
AxiomError::OmInference {
inference_source,
kind,
..
} => {
assert_eq!(inference_source, OmInferenceSource::Observer);
assert_eq!(kind, OmInferenceFailureKind::Schema);
}
other => panic!("unexpected error type: {other}"),
}
}
#[test]
fn parse_llm_observer_response_rejects_marker_like_plain_text_without_structured_contract() {
let payload = serde_json::json!({
"message": {
"content": "notes: contract_name axiomsync.om.prompt contract_version 2.0.0 protocol_version om-v2\n- line-a\n- line-b"
}
});
let known_ids = vec!["m7".to_string()];
let err = parse_llm_observer_response(
&payload,
&known_ids,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect_err("must reject non-structured contract marker text");
assert!(matches!(
err,
AxiomError::OmInference {
inference_source: OmInferenceSource::Observer,
kind: OmInferenceFailureKind::Schema,
..
}
));
}
#[test]
fn parse_llm_observer_response_returns_schema_taxonomy_for_invalid_payload() {
let payload = serde_json::json!({"unexpected": "shape"});
let known_ids = vec!["m9".to_string()];
let err = parse_llm_observer_response(
&payload,
&known_ids,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect_err("must fail");
match err {
AxiomError::OmInference {
inference_source,
kind,
..
} => {
assert_eq!(inference_source, OmInferenceSource::Observer);
assert_eq!(kind, OmInferenceFailureKind::Schema);
}
other => panic!("unexpected error type: {other}"),
}
}
#[test]
fn parse_llm_observer_response_rejects_contract_version_mismatch() {
let payload = serde_json::json!({
"observations": "line-a",
"observed_message_ids": ["m9"],
"header": {
"contract_name": "axiomsync.om.prompt",
"contract_version": "9.9.9",
"protocol_version": "om-v2"
}
});
let known_ids = vec!["m9".to_string()];
let err = parse_llm_observer_response(
&payload,
&known_ids,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect_err("must reject contract version mismatch");
match err {
AxiomError::OmInference {
inference_source,
kind,
..
} => {
assert_eq!(inference_source, OmInferenceSource::Observer);
assert_eq!(kind, OmInferenceFailureKind::Schema);
}
other => panic!("unexpected error type: {other}"),
}
}
#[test]
fn parse_llm_observer_response_rejects_protocol_version_mismatch() {
let payload = serde_json::json!({
"observations": "line-a",
"observed_message_ids": ["m9"],
"header": {
"contract_name": "axiomsync.om.prompt",
"contract_version": "2.0.0",
"protocol_version": "om-v999"
}
});
let known_ids = vec!["m9".to_string()];
let err = parse_llm_observer_response(
&payload,
&known_ids,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect_err("must reject protocol version mismatch");
match err {
AxiomError::OmInference {
inference_source,
kind,
..
} => {
assert_eq!(inference_source, OmInferenceSource::Observer);
assert_eq!(kind, OmInferenceFailureKind::Schema);
}
other => panic!("unexpected error type: {other}"),
}
}
#[test]
fn parse_llm_observer_response_rejects_xml_protocol_version_mismatch_marker() {
let payload = serde_json::json!({
"message": {
"content": "<contract-name>axiomsync.om.prompt</contract-name>\n<contract-version>2.0.0</contract-version>\n<protocol-version>om-v999</protocol-version>\n<observations>\n- keep signal\n</observations>"
}
});
let known_ids = vec!["m9".to_string()];
let err = parse_llm_observer_response(
&payload,
&known_ids,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect_err("must reject xml contract marker with protocol mismatch");
match err {
AxiomError::OmInference {
inference_source,
kind,
message,
} => {
assert_eq!(inference_source, OmInferenceSource::Observer);
assert_eq!(kind, OmInferenceFailureKind::Schema);
assert!(
message.contains("missing contract marker"),
"unexpected message: {message}"
);
}
other => panic!("unexpected error type: {other}"),
}
}
#[test]
fn normalize_observation_text_preserves_lines_and_removes_blanks() {
assert_eq!(
normalize_observation_text(" a \n\n b \n c "),
"a\nb\nc".to_string()
);
}
#[test]
fn parse_memory_section_xml_returns_inner_observations_text() {
let content = "<observations>\nline-a\nline-b\n</observations>\n<current-task>x</current-task>";
let parsed = crate::om::parse_memory_section_xml(content, crate::om::OmParseMode::Strict);
assert_eq!(parsed.observations, "line-a\nline-b");
}
#[test]
fn parse_memory_section_xml_joins_multiple_observation_blocks() {
let content =
"<observations>\nline-a\n</observations>\nnoise\n<observations>\nline-b\n</observations>";
let parsed = crate::om::parse_memory_section_xml(content, crate::om::OmParseMode::Strict);
assert_eq!(parsed.observations, "line-a\nline-b");
}
#[test]
fn parse_memory_section_xml_ignores_inline_tag_mentions() {
let content = "literal <observations>inline</observations> mention\n<observations>\nline-a\n</observations>";
let parsed = crate::om::parse_memory_section_xml(content, crate::om::OmParseMode::Strict);
assert_eq!(parsed.observations, "line-a");
}
#[test]
fn build_observer_thread_messages_groups_and_sorts_by_thread_and_time() {
let base = Utc::now();
let candidates = vec![
OmObserverMessageCandidate {
id: "m-b2".to_string(),
role: "assistant".to_string(),
text: "b2".to_string(),
created_at: base + chrono::Duration::milliseconds(2),
source_thread_id: Some("thread-b".to_string()),
source_session_id: Some("thread-b".to_string()),
},
OmObserverMessageCandidate {
id: "m-a1".to_string(),
role: "user".to_string(),
text: "a1".to_string(),
created_at: base + chrono::Duration::milliseconds(1),
source_thread_id: Some("thread-a".to_string()),
source_session_id: Some("thread-a".to_string()),
},
OmObserverMessageCandidate {
id: "m-b1".to_string(),
role: "user".to_string(),
text: "b1".to_string(),
created_at: base,
source_thread_id: Some("thread-b".to_string()),
source_session_id: Some("thread-b".to_string()),
},
];
let grouped =
build_observer_thread_messages(&candidates, OmScope::Resource, "resource:r1", "fallback");
assert_eq!(grouped.len(), 2);
assert_eq!(grouped[0].thread_id, "thread-a");
assert!(grouped[0].message_history.contains("a1"));
assert_eq!(grouped[1].thread_id, "thread-b");
let b_history = &grouped[1].message_history;
let b1_idx = b_history.find("b1").expect("b1");
let b2_idx = b_history.find("b2").expect("b2");
assert!(b1_idx < b2_idx);
}
#[test]
fn parse_llm_multi_thread_observer_response_aggregates_primary_thread_metadata() {
let payload = serde_json::json!({
"message": {
"content": "<contract-name>axiomsync.om.prompt</contract-name>\n<contract-version>2.0.0</contract-version>\n<protocol-version>om-v2</protocol-version>\n<observations>\n<thread id=\"s-main\">\nDate: Dec 4, 2025\n* 🔴 (14:30) User prefers direct answers\n<current-task>Primary: implement auth</current-task>\n<suggested-response>Continue auth changes</suggested-response>\n</thread>\n<thread id=\"s-peer\">\n* 🟡 (15:00) Peer session context\n<current-task>Primary: peer task</current-task>\n</thread>\n</observations>"
}
});
let known_ids = vec!["m1".to_string(), "m2".to_string()];
let known_ids_by_thread = BTreeMap::from([
("s-main".to_string(), vec!["m1".to_string()]),
("s-peer".to_string(), vec!["m2".to_string()]),
]);
let parsed = parse_llm_multi_thread_observer_response(
&payload,
"s-main",
&known_ids,
&known_ids_by_thread,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect("parse result")
.expect("multi-thread parse");
assert!(
parsed
.response
.observations
.contains("<thread id=\"s-main\">")
);
assert!(
parsed
.response
.observations
.contains("<thread id=\"s-peer\">")
);
assert_eq!(
parsed.response.current_task.as_deref(),
Some("Primary: implement auth")
);
assert_eq!(
parsed.response.suggested_response.as_deref(),
Some("Continue auth changes")
);
assert_eq!(parsed.response.observed_message_ids, known_ids);
assert_eq!(parsed.thread_states.len(), 2);
assert!(
parsed
.thread_states
.iter()
.any(|state| state.thread_id == "s-main"
&& state.current_task.as_deref() == Some("Primary: implement auth"))
);
}
#[test]
fn parse_llm_multi_thread_observer_response_rejects_contract_version_mismatch() {
let payload = serde_json::json!({
"header": {
"contract_name": "axiomsync.om.prompt",
"contract_version": "9.9.9",
"protocol_version": "om-v2"
},
"message": {
"content": "<contract-name>axiomsync.om.prompt</contract-name>\n<contract-version>2.0.0</contract-version>\n<protocol-version>om-v2</protocol-version>\n<observations>\n<thread id=\"s-main\">\n* 🔴 (14:30) User prefers direct answers\n</thread>\n</observations>"
}
});
let known_ids = vec!["m1".to_string()];
let known_ids_by_thread = BTreeMap::from([("s-main".to_string(), vec!["m1".to_string()])]);
let err = parse_llm_multi_thread_observer_response(
&payload,
"s-main",
&known_ids,
&known_ids_by_thread,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect_err("must reject contract version mismatch");
match err {
AxiomError::OmInference {
inference_source,
kind,
..
} => {
assert_eq!(inference_source, OmInferenceSource::Observer);
assert_eq!(kind, OmInferenceFailureKind::Schema);
}
other => panic!("unexpected error type: {other}"),
}
}
#[test]
fn parse_llm_multi_thread_observer_response_rejects_xml_without_contract_marker() {
let payload = serde_json::json!({
"message": {
"content": "<observations>\n<thread id=\"s-main\">\n* 🔴 (14:30) User prefers direct answers\n</thread>\n</observations>"
}
});
let known_ids = vec!["m1".to_string()];
let known_ids_by_thread = BTreeMap::from([("s-main".to_string(), vec!["m1".to_string()])]);
let err = parse_llm_multi_thread_observer_response(
&payload,
"s-main",
&known_ids,
&known_ids_by_thread,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect_err("must reject missing content marker");
assert!(matches!(
err,
AxiomError::OmInference {
inference_source: OmInferenceSource::Observer,
kind: OmInferenceFailureKind::Schema,
..
}
));
}
#[test]
fn parse_llm_multi_thread_observer_response_limits_observed_ids_to_present_threads() {
let payload = serde_json::json!({
"message": {
"content": "<contract-name>axiomsync.om.prompt</contract-name>\n<contract-version>2.0.0</contract-version>\n<protocol-version>om-v2</protocol-version>\n<observations>\n<thread id=\"s-main\">\n* 🔴 (14:30) User prefers direct answers\n</thread>\n</observations>"
}
});
let known_ids = vec!["m1".to_string(), "m2".to_string()];
let known_ids_by_thread = BTreeMap::from([
("s-main".to_string(), vec!["m1".to_string()]),
("s-peer".to_string(), vec!["m2".to_string()]),
]);
let parsed = parse_llm_multi_thread_observer_response(
&payload,
"s-main",
&known_ids,
&known_ids_by_thread,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect("parse")
.expect("response");
assert_eq!(parsed.response.observed_message_ids, vec!["m1".to_string()]);
}
#[test]
fn parse_llm_multi_thread_observer_response_falls_back_to_primary_thread_ids_when_unmapped() {
let payload = serde_json::json!({
"message": {
"content": "<contract-name>axiomsync.om.prompt</contract-name>\n<contract-version>2.0.0</contract-version>\n<protocol-version>om-v2</protocol-version>\n<observations>\n<thread id=\"hallucinated-thread\">\n* 🔴 (14:30) User prefers direct answers\n</thread>\n</observations>"
}
});
let known_ids = vec!["m1".to_string(), "m2".to_string()];
let known_ids_by_thread = BTreeMap::from([
("s-main".to_string(), vec!["m1".to_string()]),
("s-peer".to_string(), vec!["m2".to_string()]),
]);
let parsed = parse_llm_multi_thread_observer_response(
&payload,
"s-main",
&known_ids,
&known_ids_by_thread,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect("parse")
.expect("response");
assert_eq!(parsed.response.observed_message_ids, vec!["m1".to_string()]);
}
#[test]
fn build_observer_thread_messages_uses_scope_thread_key_for_thread_scope() {
let base = Utc::now();
let candidates = vec![
OmObserverMessageCandidate {
id: "m-a".to_string(),
role: "user".to_string(),
text: "from session a".to_string(),
created_at: base,
source_thread_id: Some("s-a".to_string()),
source_session_id: Some("s-a".to_string()),
},
OmObserverMessageCandidate {
id: "m-b".to_string(),
role: "assistant".to_string(),
text: "from session b".to_string(),
created_at: base + chrono::Duration::milliseconds(1),
source_thread_id: Some("s-b".to_string()),
source_session_id: Some("s-b".to_string()),
},
];
let grouped =
build_observer_thread_messages(&candidates, OmScope::Thread, "thread:t-1", "fallback");
assert_eq!(grouped.len(), 1);
assert_eq!(grouped[0].thread_id, "t-1");
assert!(grouped[0].message_history.contains("from session a"));
assert!(grouped[0].message_history.contains("from session b"));
}
#[test]
fn chunk_observer_thread_batches_splits_by_token_budget_and_keeps_order() {
let threads = vec![
OmObserverThreadMessages {
thread_id: "t-a".to_string(),
message_history: "a".repeat(80),
},
OmObserverThreadMessages {
thread_id: "t-b".to_string(),
message_history: "b".repeat(80),
},
OmObserverThreadMessages {
thread_id: "t-c".to_string(),
message_history: "c".repeat(80),
},
];
let batches = chunk_observer_thread_batches(&threads, 30);
assert_eq!(batches.len(), 3);
assert_eq!(batches[0][0].thread_id, "t-a");
assert_eq!(batches[1][0].thread_id, "t-b");
assert_eq!(batches[2][0].thread_id, "t-c");
}
#[test]
fn collect_known_ids_for_thread_batch_respects_batch_membership() {
let batch = vec![
OmObserverThreadMessages {
thread_id: "thread-a".to_string(),
message_history: "a".to_string(),
},
OmObserverThreadMessages {
thread_id: "thread-c".to_string(),
message_history: "c".to_string(),
},
];
let known_ids = BTreeMap::from([
(
"thread-a".to_string(),
vec!["m1".to_string(), "m2".to_string()],
),
("thread-b".to_string(), vec!["m3".to_string()]),
("thread-c".to_string(), vec!["m4".to_string()]),
]);
let collected = collect_known_ids_for_thread_batch(&batch, &known_ids);
assert_eq!(
collected,
vec!["m1".to_string(), "m2".to_string(), "m4".to_string()]
);
}
#[test]
fn build_observer_batch_tasks_filters_empty_known_ids_and_preserves_batch_index() {
let thread_batches = vec![
vec![OmObserverThreadMessages {
thread_id: "thread-a".to_string(),
message_history: "a".to_string(),
}],
vec![OmObserverThreadMessages {
thread_id: "thread-missing".to_string(),
message_history: "x".to_string(),
}],
vec![OmObserverThreadMessages {
thread_id: "thread-c".to_string(),
message_history: "c".to_string(),
}],
];
let known_ids = BTreeMap::from([
("thread-a".to_string(), vec!["m1".to_string()]),
(
"thread-c".to_string(),
vec!["m3".to_string(), "m4".to_string()],
),
]);
let tasks = build_observer_batch_tasks(thread_batches, &known_ids);
assert_eq!(tasks.len(), 2);
assert_eq!(tasks[0].index, 0);
assert_eq!(tasks[0].known_ids, vec!["m1".to_string()]);
assert_eq!(
tasks[0].known_ids_by_thread,
BTreeMap::from([("thread-a".to_string(), vec!["m1".to_string()])])
);
assert_eq!(tasks[1].index, 2);
assert_eq!(tasks[1].known_ids, vec!["m3".to_string(), "m4".to_string()]);
assert_eq!(
tasks[1].known_ids_by_thread,
BTreeMap::from([(
"thread-c".to_string(),
vec!["m3".to_string(), "m4".to_string()],
)])
);
}
#[test]
fn select_messages_for_observer_llm_enforces_bounds() {
let selected = vec![
msg("m1", "user", &"a".repeat(200)),
msg("m2", "assistant", &"b".repeat(200)),
msg("m3", "user", &"c".repeat(200)),
];
let bounded = select_messages_for_observer_llm(&selected, 12, 40);
assert!(!bounded.is_empty());
assert!(bounded.len() <= selected.len());
assert!(bounded.iter().all(|item| item.text.chars().count() <= 12));
let total_tokens = bounded.iter().fold(0u32, |sum, item| {
sum.saturating_add(estimate_text_tokens(&item.id))
.saturating_add(estimate_text_tokens(&item.role))
.saturating_add(estimate_text_tokens(&item.text))
.saturating_add(8)
});
assert!(total_tokens <= 40);
}
#[test]
fn build_observation_chunk_uses_latest_selected_candidate_boundary() {
let base = Utc::now();
let selected = vec![
OmObserverMessageCandidate {
id: "m-old".to_string(),
role: "user".to_string(),
text: "older text".to_string(),
created_at: base,
source_thread_id: Some("s-a".to_string()),
source_session_id: Some("s-a".to_string()),
},
OmObserverMessageCandidate {
id: "m-new".to_string(),
role: "assistant".to_string(),
text: "newer text".to_string(),
created_at: base + chrono::Duration::milliseconds(1),
source_thread_id: Some("s-b".to_string()),
source_session_id: Some("s-b".to_string()),
},
];
let chunk = build_observation_chunk(
"record-1",
&selected,
&[],
Utc::now(),
"[assistant] summarized observation",
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
)
.expect("chunk");
assert_eq!(chunk.cycle_id, "observer_sync:m-new");
assert_eq!(chunk.last_observed_at, selected[1].created_at);
assert_eq!(
chunk.message_ids,
vec!["m-old".to_string(), "m-new".to_string()]
);
assert!(chunk.message_tokens > 0);
}
#[test]
fn record_with_buffered_observation_context_combines_active_and_buffered_text() {
let mut record = new_session_om_record("s-buffered", "session:s-buffered", Utc::now());
record.active_observations = "active-a".to_string();
let buffered_chunks = vec![
OmObservationChunk {
id: "c1".to_string(),
record_id: record.id.clone(),
seq: 1,
cycle_id: "cy1".to_string(),
observations: "buffered-b".to_string(),
token_count: 1,
message_tokens: 1,
message_ids: vec!["m1".to_string()],
last_observed_at: Utc::now(),
created_at: Utc::now(),
},
OmObservationChunk {
id: "c2".to_string(),
record_id: record.id.clone(),
seq: 2,
cycle_id: "cy2".to_string(),
observations: "buffered-c".to_string(),
token_count: 1,
message_tokens: 1,
message_ids: vec!["m2".to_string()],
last_observed_at: Utc::now(),
created_at: Utc::now(),
},
];
let combined = record_with_buffered_observation_context(
&record,
&buffered_chunks,
crate::config::OmRuntimeLimitsConfig::default().observer_active_observations_max_chars,
);
assert!(combined.active_observations.contains("active-a"));
assert!(combined.active_observations.contains("buffered-b"));
assert!(combined.active_observations.contains("buffered-c"));
assert!(
combined
.active_observations
.contains("--- BUFFERED (pending activation) ---")
);
}
#[test]
fn split_pending_and_other_conversation_candidates_partitions_by_session() {
let selected = vec![
OmObserverMessageCandidate {
id: "m-local".to_string(),
role: "user".to_string(),
text: "local".to_string(),
created_at: Utc::now(),
source_thread_id: Some("s-local".to_string()),
source_session_id: Some("s-local".to_string()),
},
OmObserverMessageCandidate {
id: "m-peer".to_string(),
role: "assistant".to_string(),
text: "peer".to_string(),
created_at: Utc::now(),
source_thread_id: Some("s-peer".to_string()),
source_session_id: Some("s-peer".to_string()),
},
];
let (pending, others) =
split_pending_and_other_conversation_candidates(&selected, Some("s-local"));
assert_eq!(pending.len(), 1);
assert_eq!(pending[0].id, "m-local");
assert_eq!(others.len(), 1);
assert_eq!(others[0].id, "m-peer");
}
#[test]
fn select_observed_message_candidates_filters_to_response_ids() {
let selected = vec![
OmObserverMessageCandidate {
id: "m1".to_string(),
role: "user".to_string(),
text: "a".to_string(),
created_at: Utc::now(),
source_thread_id: Some("s-local".to_string()),
source_session_id: Some("s-local".to_string()),
},
OmObserverMessageCandidate {
id: "m2".to_string(),
role: "assistant".to_string(),
text: "b".to_string(),
created_at: Utc::now(),
source_thread_id: Some("s-local".to_string()),
source_session_id: Some("s-local".to_string()),
},
];
let filtered =
select_observed_message_candidates(&selected, &["m2".to_string(), "unknown".to_string()]);
assert_eq!(filtered.len(), 1);
assert_eq!(filtered[0].id, "m2");
}
#[test]
fn observer_model_feature_flag_off_forces_deterministic_output() {
let selected = vec![msg("m1", "user", "important oauth detail")];
let now = Utc::now();
let record = new_session_om_record("s-flag", "session:s-flag", now);
let config = observer_config(OmObserverMode::Llm, false);
let resolved = resolve_observer_response_with_config(
&record,
"session:s-flag",
&selected,
"s-flag",
crate::om::DEFAULT_OBSERVER_MAX_TOKENS_PER_BATCH,
false,
&config,
)
.expect("resolve");
let expected = deterministic_observer_response(
&record,
&selected,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
);
assert_eq!(resolved.response.observations, expected.observations);
assert_eq!(resolved.selected_messages.len(), 1);
assert_eq!(resolved.selected_messages[0].id, "m1");
}
#[test]
fn deterministic_observer_response_avoids_exact_active_duplicates() {
let selected = vec![
msg("m1", "user", "same detail"),
msg("m2", "assistant", "new detail"),
];
let mut record = new_session_om_record("s-dedupe", "session:s-dedupe", Utc::now());
record.active_observations = "[user] same detail".to_string();
let response = deterministic_observer_response(
&record,
&selected,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
);
assert!(!response.observations.contains("[user] same detail"));
assert!(response.observations.contains("[assistant] new detail"));
assert_eq!(
response.observed_message_ids,
vec!["m1".to_string(), "m2".to_string()]
);
}
#[test]
fn deterministic_observer_response_avoids_duplicates_present_in_buffered_context() {
let selected = vec![
msg("m1", "user", "same detail"),
msg("m2", "assistant", "new detail"),
];
let mut record =
new_session_om_record("s-buffer-dedupe", "session:s-buffer-dedupe", Utc::now());
let buffered_chunks = vec![OmObservationChunk {
id: "c1".to_string(),
record_id: record.id.clone(),
seq: 1,
cycle_id: "cy1".to_string(),
observations: "[user] same detail".to_string(),
token_count: 1,
message_tokens: 1,
message_ids: vec!["m1".to_string()],
last_observed_at: Utc::now(),
created_at: Utc::now(),
}];
record = record_with_buffered_observation_context(
&record,
&buffered_chunks,
crate::config::OmRuntimeLimitsConfig::default().observer_active_observations_max_chars,
);
let response = deterministic_observer_response(
&record,
&selected,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
);
assert!(!response.observations.contains("[user] same detail"));
assert!(response.observations.contains("[assistant] new detail"));
}
#[test]
fn deterministic_fallback_emits_current_task() {
let selected = vec![msg(
"m1",
"user",
"Please fix AXIOMSYNC_RERANKER handling in src/client/search/mod.rs and verify tests.",
)];
let record = new_session_om_record("s-fallback-task", "session:s-fallback-task", Utc::now());
let response = deterministic_observer_response(
&record,
&selected,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
);
assert_eq!(
response.current_task.as_deref(),
Some(
"Primary: Please fix AXIOMSYNC_RERANKER handling in src/client/search/mod.rs and verify tests."
)
);
assert!(
response
.suggested_response
.as_deref()
.is_some_and(|value| value.contains("AXIOMSYNC_RERANKER")),
"suggested response should preserve identifier context: {:?}",
response.suggested_response
);
}
#[test]
fn deterministic_fallback_populates_thread_states_for_resource_scope() {
let selected = vec![
msg_with_thread(
"m1",
"user",
"Please fix queue replay for worker-a and verify AXIOMSYNC_RERANKER stability.",
"thread-a",
"s-fallback-threads",
),
msg_with_thread(
"m2",
"tool",
"error E401 at worker-a while applying AXIOMSYNC_RERANKER patch",
"thread-a",
"s-fallback-threads",
),
msg_with_thread(
"m3",
"user",
"Please investigate E409 in worker-b and patch serde_json::from_str handling.",
"thread-b",
"s-fallback-threads",
),
msg_with_thread(
"m4",
"tool",
"error E409 at worker-b while running serde_json::from_str flow",
"thread-b",
"s-fallback-threads",
),
];
let mut record = new_session_om_record(
"s-fallback-threads",
"resource:r-fallback-threads",
Utc::now(),
);
record.scope = OmScope::Resource;
record.scope_key = "resource:r-fallback-threads".to_string();
record.session_id = Some("s-fallback-threads".to_string());
record.resource_id = Some("r-fallback-threads".to_string());
record.thread_id = None;
let config = observer_config(OmObserverMode::Llm, false);
let resolved = resolve_observer_response_with_config(
&record,
&record.scope_key,
&selected,
"s-fallback-threads",
4096,
false,
&config,
)
.expect("resolve observer response");
assert_eq!(resolved.thread_states.len(), 2);
assert!(
resolved
.thread_states
.iter()
.any(|state| state.thread_id == "thread-a"),
"thread-a deterministic state must be emitted"
);
assert!(
resolved
.thread_states
.iter()
.any(|state| state.thread_id == "thread-b"),
"thread-b deterministic state must be emitted"
);
assert!(
resolved
.thread_states
.iter()
.any(|state| state.current_task.is_some() || state.suggested_response.is_some()),
"at least one deterministic thread state must carry continuation hints"
);
}
#[test]
fn deterministic_fallback_preserves_error_context_identifiers() {
let selected = vec![
msg(
"m1",
"user",
"Investigate the queue replay failure and patch serde_json::from_str path handling.",
),
msg(
"m2",
"tool",
"error: serde_json::from_str failed at src/session/om.rs:518 with code E1002",
),
];
let record = new_session_om_record(
"s-fallback-identifiers",
"session:s-fallback-identifiers",
Utc::now(),
);
let response = deterministic_observer_response(
&record,
&selected,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
);
assert_eq!(
response.current_task.as_deref(),
Some(
"Primary: Investigate the queue replay failure and patch serde_json::from_str path handling."
)
);
assert!(
response
.suggested_response
.as_deref()
.is_some_and(|value| value.contains("serde_json::from_str")),
"error identifiers should survive deterministic fallback: {:?}",
response.suggested_response
);
}
#[test]
fn deterministic_fallback_suppresses_low_confidence_suggested_response() {
let selected = vec![msg("m1", "user", "Thanks!")];
let record = new_session_om_record(
"s-fallback-low-confidence",
"session:s-fallback-low-confidence",
Utc::now(),
);
let response = deterministic_observer_response(
&record,
&selected,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
);
assert_eq!(response.current_task, None);
assert_eq!(response.suggested_response, None);
}
#[test]
fn deterministic_fallback_handles_non_english_task_and_error_signal() {
let selected = vec![
msg(
"m1",
"user",
"请修复队列回放并更新src/session/om.rs，然后验证E409错误已消失。",
),
msg(
"m2",
"tool",
"错误E409发生在src/session/om.rs:518，操作失败。",
),
];
let record = new_session_om_record(
"s-fallback-non-english",
"session:s-fallback-non-english",
Utc::now(),
);
let response = deterministic_observer_response(
&record,
&selected,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
);
assert!(
response
.current_task
.as_deref()
.is_some_and(|value| value.starts_with("Primary: 请修复队列回放")),
"non-English task sentence should be preserved: {:?}",
response.current_task
);
assert!(
response
.suggested_response
.as_deref()
.is_some_and(|value| value.contains("E409")),
"error signal identifier should be preserved from non-English blocking text: {:?}",
response.suggested_response
);
}
#[test]
fn deterministic_fallback_extracts_identifiers_without_whitespace() {
let selected = vec![
msg(
"m1",
"user",
"Please fix queue replay handling in worker:queue and keep AXIOMSYNC_RERANKER stable.",
),
msg("m2", "tool", "错误E409发生在worker:queue，操作失败"),
];
let record = new_session_om_record(
"s-fallback-no-whitespace",
"session:s-fallback-no-whitespace",
Utc::now(),
);
let response = deterministic_observer_response(
&record,
&selected,
crate::config::OmRuntimeLimitsConfig::default().observation_max_chars,
);
assert!(
response
.suggested_response
.as_deref()
.is_some_and(|value| value.contains("E409") && value.contains("worker:queue")),
"identifier extraction should work even when error text has no whitespace: {:?}",
response.suggested_response
);
}
#[test]
fn observer_rollout_profile_overrides_model_enabled_flag() {
assert!(resolve_observer_model_enabled(false, Some("observer_only")));
assert!(resolve_observer_model_enabled(false, Some("full_model")));
assert!(!resolve_observer_model_enabled(true, Some("baseline")));
assert!(resolve_observer_model_enabled(true, Some("unknown")));
assert!(!resolve_observer_model_enabled(false, Some("unknown")));
}
#[test]
fn collect_observer_messages_for_resource_scope_includes_peer_sessions() {
let temp = tempdir().expect("tempdir");
let fs = LocalContextFs::new(temp.path());
fs.initialize().expect("init failed");
let state = SqliteStateStore::open(temp.path().join("state.db")).expect("state open");
let index = Arc::new(RwLock::new(InMemoryIndex::new()));
let session_a = Session::new("s-resource-a", fs.clone(), state.clone(), index.clone());
let session_b = Session::new("s-resource-b", fs, state.clone(), index);
session_a.load().expect("load a");
session_b.load().expect("load b");
let a_msg = session_a
.add_message("user", "resource shared note from-a")
.expect("append a");
let _b_msg = session_b
.add_message("user", "resource shared note from-b")
.expect("append b");
state
.upsert_om_scope_session("resource:r-1", "s-resource-a")
.expect("scope map a");
state
.upsert_om_scope_session("resource:r-1", "s-resource-b")
.expect("scope map b");
let observed = HashSet::from([a_msg.id.clone()]);
let selected = session_a
.collect_observer_messages_for_scope(
OmScope::Resource,
"resource:r-1",
None,
&observed,
8,
false,
)
.expect("collect");
assert!(
selected.iter().any(|item| item.text.contains("from-b")),
"expected peer-session message to be included"
);
assert!(
selected.iter().all(|item| item.id != a_msg.id),
"observed ids should be filtered"
);
}
#[test]
fn collect_observer_messages_for_thread_scope_includes_peer_sessions() {
let temp = tempdir().expect("tempdir");
let fs = LocalContextFs::new(temp.path());
fs.initialize().expect("init failed");
let state = SqliteStateStore::open(temp.path().join("state.db")).expect("state open");
let index = Arc::new(RwLock::new(InMemoryIndex::new()));
let session_a = Session::new("s-thread-a", fs.clone(), state.clone(), index.clone());
let session_b = Session::new("s-thread-b", fs, state.clone(), index);
session_a.load().expect("load a");
session_b.load().expect("load b");
let a_msg = session_a
.add_message("user", "thread shared note from-a")
.expect("append a");
let _b_msg = session_b
.add_message("assistant", "thread shared note from-b")
.expect("append b");
state
.upsert_om_scope_session("thread:t-1", "s-thread-a")
.expect("scope map a");
state
.upsert_om_scope_session("thread:t-1", "s-thread-b")
.expect("scope map b");
let observed = HashSet::from([a_msg.id.clone()]);
let selected = session_a
.collect_observer_messages_for_scope(
OmScope::Thread,
"thread:t-1",
None,
&observed,
8,
false,
)
.expect("collect");
assert!(
selected.iter().any(|item| item.text.contains("from-b")),
"expected peer-session message to be included for thread scope"
);
assert!(
selected.iter().all(|item| item.id != a_msg.id),
"observed ids should be filtered"
);
}
#[test]
fn collect_observer_messages_for_scope_prefers_unobserved_since_cursor() {
let temp = tempdir().expect("tempdir");
let fs = LocalContextFs::new(temp.path());
fs.initialize().expect("init failed");
let state = SqliteStateStore::open(temp.path().join("state.db")).expect("state open");
let index = Arc::new(RwLock::new(InMemoryIndex::new()));
let session_a = Session::new(
"s-resource-cursor-a",
fs.clone(),
state.clone(),
index.clone(),
);
let session_b = Session::new("s-resource-cursor-b", fs, state.clone(), index);
session_a.load().expect("load a");
session_b.load().expect("load b");
let old = session_b
.add_message("user", "resource old peer message")
.expect("append old");
std::thread::sleep(std::time::Duration::from_millis(2));
let new = session_b
.add_message("user", "resource new peer message")
.expect("append new");
state
.upsert_om_scope_session("resource:r-cursor", "s-resource-cursor-a")
.expect("scope map a");
state
.upsert_om_scope_session("resource:r-cursor", "s-resource-cursor-b")
.expect("scope map b");
let selected = session_a
.collect_observer_messages_for_scope(
OmScope::Resource,
"resource:r-cursor",
Some(old.created_at),
&HashSet::new(),
16,
false,
)
.expect("collect");
assert!(
selected.iter().any(|item| item.id == new.id),
"new peer message must remain after cursor filter"
);
assert!(
selected.iter().all(|item| item.id != old.id),
"old peer message must be filtered by last_observed_at"
);
}
#[test]
fn resolve_om_scope_binding_defaults_to_session_scope() {
let resolved = resolve_om_scope_binding("s-1", None, None, None).expect("resolve");
assert_eq!(resolved.scope, OmScope::Session);
assert_eq!(resolved.scope_key, "session:s-1");
assert_eq!(resolved.session_id.as_deref(), Some("s-1"));
assert_eq!(resolved.thread_id, None);
assert_eq!(resolved.resource_id, None);
}
#[test]
fn resolve_om_scope_binding_resolves_thread_scope() {
let resolved =
resolve_om_scope_binding("s-1", Some("thread"), Some("t-1"), Some("r-1")).expect("resolve");
assert_eq!(resolved.scope, OmScope::Thread);
assert_eq!(resolved.scope_key, "thread:t-1");
assert_eq!(resolved.session_id, None);
assert_eq!(resolved.thread_id.as_deref(), Some("t-1"));
assert_eq!(resolved.resource_id.as_deref(), Some("r-1"));
}
#[test]
fn resolve_om_scope_binding_explicit_resolves_session_scope() {
let resolved = resolve_om_scope_binding_explicit(
"s-explicit-session",
OmScope::Session,
Some("ignored-thread"),
Some("ignored-resource"),
)
.expect("resolve");
assert_eq!(resolved.scope, OmScope::Session);
assert_eq!(resolved.scope_key, "session:s-explicit-session");
assert_eq!(resolved.session_id.as_deref(), Some("s-explicit-session"));
assert_eq!(resolved.thread_id, None);
assert_eq!(resolved.resource_id, None);
}
#[test]
fn resolve_om_scope_binding_explicit_resolves_thread_scope() {
let resolved = resolve_om_scope_binding_explicit(
"s-explicit-thread",
OmScope::Thread,
Some("thread-explicit"),
Some("resource-explicit"),
)
.expect("resolve");
assert_eq!(resolved.scope, OmScope::Thread);
assert_eq!(resolved.scope_key, "thread:thread-explicit");
assert_eq!(resolved.session_id, None);
assert_eq!(resolved.thread_id.as_deref(), Some("thread-explicit"));
assert_eq!(resolved.resource_id.as_deref(), Some("resource-explicit"));
}
#[test]
fn resolve_om_scope_binding_explicit_resolves_resource_scope() {
let resolved = resolve_om_scope_binding_explicit(
"s-explicit-resource",
OmScope::Resource,
Some("thread-explicit"),
Some("resource-explicit"),
)
.expect("resolve");
assert_eq!(resolved.scope, OmScope::Resource);
assert_eq!(resolved.scope_key, "resource:resource-explicit");
assert_eq!(resolved.session_id, None);
assert_eq!(resolved.thread_id, None);
assert_eq!(resolved.resource_id.as_deref(), Some("resource-explicit"));
}
#[test]
fn resolve_om_scope_binding_rejects_invalid_scope() {
let err = resolve_om_scope_binding("s-1", Some("invalid"), Some("t-1"), Some("r-1"))
.expect_err("must fail");
assert!(err.to_string().contains(ENV_OM_SCOPE));
}
#[test]
fn resolve_om_scope_binding_rejects_thread_scope_without_thread_id() {
let err =
resolve_om_scope_binding("s-1", Some("thread"), None, Some("r-1")).expect_err("must fail");
assert!(err.to_string().contains("thread_id"));
}
#[test]
fn runtime_config_rejects_share_budget_when_async_buffering_enabled() {
let env = RuntimeOmEnv {
share_token_budget: Some("true".to_string()),
..RuntimeOmEnv::default()
};
let err = resolve_runtime_om_config(&env, OmScope::Session).expect_err("must reject");
match err {
AxiomError::Validation(message) => {
assert!(message.contains("shareTokenBudget"));
assert!(message.contains("async buffering"));
}
other => panic!("unexpected error: {other}"),
}
}
#[test]
fn runtime_config_allows_share_budget_when_buffering_is_disabled() {
let env = RuntimeOmEnv {
share_token_budget: Some("true".to_string()),
buffer_tokens: Some("disabled".to_string()),
..RuntimeOmEnv::default()
};
let resolved = resolve_runtime_om_config(&env, OmScope::Session).expect("resolve");
assert!(resolved.share_token_budget);
assert!(resolved.async_buffering_disabled);
assert_eq!(resolved.observation.total_budget, Some(70_000));
assert_eq!(resolved.observation.buffer_tokens, None);
assert_eq!(resolved.observation.buffer_activation, None);
assert_eq!(resolved.reflection.buffer_activation, None);
assert_eq!(resolved.reflection.block_after, None);
}
#[test]
fn runtime_config_rejects_invalid_buffer_tokens_env_value() {
let env = RuntimeOmEnv {
buffer_tokens: Some("abc".to_string()),
..RuntimeOmEnv::default()
};
let err = resolve_runtime_om_config(&env, OmScope::Session).expect_err("must reject");
match err {
AxiomError::Validation(message) => {
assert!(message.contains(ENV_OM_BUFFER_TOKENS));
assert!(message.contains("got: abc"));
}
other => panic!("unexpected error: {other}"),
}
}
#[test]
fn runtime_config_resource_scope_disables_async_by_default() {
let env = RuntimeOmEnv::default();
let resolved = resolve_runtime_om_config(&env, OmScope::Resource).expect("resolve");
assert!(resolved.async_buffering_disabled);
assert_eq!(resolved.observation.buffer_tokens, None);
assert_eq!(resolved.observation.buffer_activation, None);
assert_eq!(resolved.observation.block_after, None);
assert_eq!(
resolved.observation.max_tokens_per_batch,
crate::om::DEFAULT_OBSERVER_MAX_TOKENS_PER_BATCH
);
assert_eq!(resolved.reflection.buffer_activation, None);
assert_eq!(resolved.reflection.block_after, None);
}
#[test]
fn runtime_config_resource_scope_rejects_explicit_async_buffering() {
let env = RuntimeOmEnv {
buffer_tokens: Some("0.2".to_string()),
..RuntimeOmEnv::default()
};
let err = resolve_runtime_om_config(&env, OmScope::Resource).expect_err("must reject");
match err {
AxiomError::Validation(message) => {
assert!(message.contains("resource scope"));
assert!(message.contains("async buffering"));
}
other => panic!("unexpected error: {other}"),
}
}
#[test]
fn runtime_config_resource_scope_share_budget_allowed_without_explicit_async() {
let env = RuntimeOmEnv {
share_token_budget: Some("true".to_string()),
..RuntimeOmEnv::default()
};
let resolved = resolve_runtime_om_config(&env, OmScope::Resource).expect("resolve");
assert!(resolved.share_token_budget);
assert!(resolved.async_buffering_disabled);
assert_eq!(resolved.observation.total_budget, Some(70_000));
}
#[test]
fn runtime_config_resource_scope_share_budget_rejected_when_async_explicitly_enabled() {
let env = RuntimeOmEnv {
share_token_budget: Some("true".to_string()),
buffer_tokens: Some("0.2".to_string()),
..RuntimeOmEnv::default()
};
let err = resolve_runtime_om_config(&env, OmScope::Resource).expect_err("must reject");
match err {
AxiomError::Validation(message) => {
assert!(message.contains("resource scope"));
assert!(message.contains("async buffering"));
}
other => panic!("unexpected error: {other}"),
}
}
#[test]
fn runtime_config_allows_overriding_max_tokens_per_batch() {
let env = RuntimeOmEnv {
observer_max_tokens_per_batch: Some("4096".to_string()),
..RuntimeOmEnv::default()
};
let resolved = resolve_runtime_om_config(&env, OmScope::Session).expect("resolve");
assert_eq!(resolved.observation.max_tokens_per_batch, 4096);
}
#[test]
fn runtime_config_session_scope_parses_ratio_buffer_tokens_without_behavior_change() {
let env = RuntimeOmEnv {
buffer_tokens: Some("0.2".to_string()),
..RuntimeOmEnv::default()
};
let resolved = resolve_runtime_om_config(&env, OmScope::Session).expect("resolve");
assert!(!resolved.async_buffering_disabled);
assert_eq!(resolved.observation.buffer_tokens, Some(6_000));
assert_eq!(resolved.observation.buffer_activation, Some(0.8));
assert_eq!(resolved.observation.block_after, Some(36_000));
}
#[test]
fn om_enabled_parser_defaults_to_true_and_accepts_false_tokens() {
assert!(parse_env_enabled_default_true(None));
assert!(parse_env_enabled_default_true(Some("")));
assert!(parse_env_enabled_default_true(Some("true")));
assert!(parse_env_enabled_default_true(Some("1")));
assert!(!parse_env_enabled_default_true(Some("false")));
assert!(!parse_env_enabled_default_true(Some("0")));
assert!(!parse_env_enabled_default_true(Some("off")));
assert!(!parse_env_enabled_default_true(Some("disabled")));
}