use anyhow::Result;
use klbr_runtime::{
protocol::{HostRequest, OrientationScope, SessionId},
store::Store,
};
use serde_json::{json, Value};
fn candidate(id: &str, scope: &OrientationScope, parent: Option<&str>) -> Value {
json!({"candidate_id":id,"text":"exact selected prose\nwith spacing intact", "scope":scope,
"parent_revision":parent,"carrier_hash":"test-hash","carrier_role":"system",
"carrier_placement":"leading_system","staged_at_ms":10})
}
fn semantic(candidate: &Value) -> Value {
let mut value = candidate.clone();
value.as_object_mut().unwrap().remove("staged_at_ms");
value
}
async fn stage(store: &Store, session: &SessionId, c: &Value) -> Result<Value> {
store
.stage_orientation_candidate(
session,
c["candidate_id"].as_str().unwrap(),
&serde_json::from_value(c["scope"].clone())?,
semantic(c),
c.clone(),
)
.await
}
async fn adopt(
store: &Store,
session: &SessionId,
c: &Value,
op: &str,
revision_id: &str,
) -> Result<klbr_runtime::store::Event> {
let mut revision = semantic(c);
revision["revision_id"] = json!(revision_id);
revision["operation_id"] = json!(op);
revision["adoption_actor"] = json!({"kind":"agent"});
let receipt = json!({"candidate_id":c["candidate_id"],"revision_id":revision_id,"operation_id":op,
"scope":c["scope"],"carrier_hash":c["carrier_hash"],"carrier_role":c["carrier_role"],
"carrier_placement":c["carrier_placement"],"apply_at":"next_request"});
let request = json!({"candidate":semantic(c),"expected_parent":c["parent_revision"],
"adoption_actor":{"kind":"agent"},"apply_at":"next_request"});
store
.commit_orientation_adoption(
session,
&serde_json::from_value(c["scope"].clone())?,
c["candidate_id"].as_str().unwrap(),
c["parent_revision"].as_str(),
op,
request,
revision,
receipt,
)
.await
}
#[tokio::test]
async fn scoped_selection_races_retries_and_restart_keep_exact_originals() -> Result<()> {
let dir = tempfile::tempdir()?;
let path = dir.path().join("runtime.db");
let session: SessionId = "orientation-test".parse().unwrap();
let store = Store::open(&path)?;
store.ensure_session(&session).await?;
let first = candidate("c1", &OrientationScope::Global, None);
let wrong_scope = OrientationScope::Project {
id: "victim".into(),
};
assert!(store
.stage_orientation_candidate(
&session,
"c1",
&wrong_scope,
semantic(&first),
first.clone()
)
.await
.unwrap_err()
.to_string()
.contains("orientation_scope_mismatch"));
assert!(store.orientation_candidate(&session, "c1").await?.is_none());
assert_eq!(stage(&store, &session, &first).await?, first);
let mut retry = first.clone();
retry["staged_at_ms"] = json!(99);
assert_eq!(stage(&store, &session, &retry).await?, first);
retry["text"] = json!("conflicting prose");
assert!(stage(&store, &session, &retry)
.await
.unwrap_err()
.to_string()
.contains("payload_conflict"));
let original = adopt(&store, &session, &first, "op1", "r1").await?;
let other_scope = candidate(
"c-project",
&OrientationScope::Project { id: "other".into() },
None,
);
stage(&store, &session, &other_scope).await?;
adopt(&store, &session, &other_scope, "op-project", "r-project").await?;
assert_eq!(
store
.current_orientation_revision(&session, &OrientationScope::Global)
.await?
.unwrap()["revision_id"],
"r1"
);
assert!(store
.current_orientation_revision(
&session,
&OrientationScope::Work {
id: "unknown".into()
}
)
.await?
.is_none());
assert!(adopt(&store, &session, &other_scope, "op1", "bad")
.await
.unwrap_err()
.to_string()
.contains("payload_conflict"));
let left = candidate("left", &OrientationScope::Global, Some("r1"));
let right = candidate("right", &OrientationScope::Global, Some("r1"));
stage(&store, &session, &left).await?;
stage(&store, &session, &right).await?;
let (a, b) = tokio::join!(
adopt(&store, &session, &left, "op-left", "r-left"),
adopt(&store, &session, &right, "op-right", "r-right")
);
let loser = match (a, b) {
(Err(error), Ok(_)) | (Ok(_), Err(error)) => error,
_ => panic!("exactly one stale-parent write must lose"),
};
assert!(loser.to_string().contains("stale_parent_conflict"));
let duplicate = adopt(&store, &session, &first, "op1", "unused-new-id").await?;
assert_eq!(duplicate.id, original.id);
assert_eq!(duplicate.payload, original.payload);
let selected = store
.current_orientation_revision(&session, &OrientationScope::Global)
.await?
.unwrap();
let before_failed_commit = store.history(&session, 0, 100).await?;
let blocked = candidate(
"collision",
&OrientationScope::Global,
selected["revision_id"].as_str(),
);
stage(&store, &session, &blocked).await?;
assert!(
adopt(&store, &session, &blocked, "op-collision", "r-project")
.await
.is_err()
);
assert_eq!(
store.history(&session, 0, 100).await?.len(),
before_failed_commit.len(),
"failed revision insert must roll back its event"
);
assert_eq!(
store
.current_orientation_revision(&session, &OrientationScope::Global)
.await?,
Some(selected.clone())
);
drop(store);
let reopened = Store::open(&path)?;
assert_eq!(
reopened
.current_orientation_revision(&session, &OrientationScope::Global)
.await?,
Some(selected)
);
assert_eq!(
reopened.orientation_candidate(&session, "c1").await?,
Some(first)
);
assert_eq!(
reopened
.history(&session, 0, 100)
.await?
.iter()
.filter(|e| e.payload["type"] == "orientation.adopted")
.count(),
3
);
Ok(())
}
#[test]
fn orientation_methods_decode_real_flat_wire_and_reject_unknown_fields() {
let requests = [
json!({"method":"orientation.stage","candidate_id":"c","text":"exact","scope":{"kind":"global"},
"mode":"form","target":"chosen","target_selector":{"kind":"agent"},"carrier_author":{"kind":"model","model_id":"unspecified"}}),
json!({"method":"orientation.current","scope":{"kind":"global"}}),
json!({"method":"orientation.adopt","candidate_id":"c","adoption_actor":{"kind":"agent"},"operation_id":"o"}),
];
for request in requests {
let decoded: HostRequest = serde_json::from_value(request.clone()).unwrap();
assert_eq!(
serde_json::to_value(decoded).unwrap()["method"],
request["method"]
);
let mut invalid = request;
invalid["unexpected"] = json!(true);
assert!(serde_json::from_value::<HostRequest>(invalid).is_err());
}
}