use super::{config::Seat, door_link::Output, record_client::Native, tools::Identity};
use crate::mcp::{adapter, rpc::ToolCall};
use serde_json::{Value, json};
use std::collections::BTreeMap;
struct Conversation {
meta: Value,
seq: u64,
live_seq: u64,
}
struct Batch {
arguments: Value,
items: Vec<Value>,
uncertain: bool,
attempted: bool,
}
pub(super) struct Record {
native: Native,
seat: Seat,
person: String,
harness: Value,
output: Option<Output>,
conversations: BTreeMap<String, Conversation>,
completed: Vec<Value>,
messages: BTreeMap<(String, String), Value>,
tools: BTreeMap<(String, String), Value>,
batch: Option<Batch>,
dirty: bool,
completed_bytes: usize,
messages_bytes: usize,
tools_bytes: usize,
batch_bytes: usize,
failures: BTreeMap<String, String>,
#[cfg(test)]
measured_bytes: std::cell::Cell<usize>,
}
impl Record {
pub fn new(
seat: Seat,
identity: &Identity,
person: String,
harness: Value,
output: Output,
) -> anyhow::Result<Self> {
Self::build(seat, identity, person, harness, Some(output))
}
fn build(
seat: Seat,
identity: &Identity,
person: String,
harness: Value,
output: Option<Output>,
) -> anyhow::Result<Self> {
Ok(Self {
native: Native::new(&seat, identity)?,
seat,
person,
harness,
output,
conversations: BTreeMap::new(),
completed: Vec::new(),
messages: BTreeMap::new(),
tools: BTreeMap::new(),
batch: None,
dirty: false,
completed_bytes: 0,
messages_bytes: 0,
tools_bytes: 0,
batch_bytes: 0,
failures: BTreeMap::new(),
#[cfg(test)]
measured_bytes: std::cell::Cell::new(0),
})
}
fn send(&self, value: Value) -> anyhow::Result<()> {
self.output
.as_ref()
.map_or_else(|| Ok(()), |output| output.send(value))
}
pub fn startup_failure(
seat: Seat,
identity: &Identity,
person: String,
reason: &str,
) -> anyhow::Result<()> {
let harness = json!({"id":seat.harness.id,"protocol":null});
let mut record = Self::build(seat, identity, person, harness, None)?;
record.accept(json!({"item":{"kind":"ended","conversation":super::permission::operation()?,"exit_status":1,"state":"ended","reason":reason}}))?;
anyhow::ensure!(
record.batch.is_none() && record.completed.is_empty(),
"assistant_record_uncommitted_at_close"
);
Ok(())
}
pub fn run(mut self, mut input: tokio::sync::mpsc::Receiver<Value>) -> anyhow::Result<()> {
while let Some(value) = input.blocking_recv() {
self.accept(value)?;
}
self.flush()?;
anyhow::ensure!(
self.batch.is_none() && self.completed.is_empty(),
"assistant_record_uncommitted_at_close"
);
Ok(())
}
fn conversation(&mut self, id: &str) -> anyhow::Result<&mut Conversation> {
if !self.conversations.contains_key(id) {
if self.conversations.len() >= self.seat.harness.session_list {
anyhow::ensure!(
!self.dirty
&& self.batch.is_none()
&& self.completed.is_empty()
&& self.messages.is_empty()
&& self.tools.is_empty(),
"record_session_limit_with_uncommitted_items"
);
self.conversations.clear();
}
let document=self.native.read("documents.get",&json!({"selection":{"branch":"main"},"collection":"_assistant","document_id":format!("{}/{id}/meta",self.person)})).map_err(|error|anyhow::anyhow!("assistant_record_read_refused: {}",error.code))?;
let now = now()?;
let mut meta = if document["document"].is_null() {
json!({"title":"","harness":self.harness,"created_at_ms":now,"updated_at_ms":now,"turns":0,"state":"live","last_seq":0})
} else {
serde_json::from_str(
document["document"]["raw_json"]
.as_str()
.ok_or_else(|| anyhow::anyhow!("assistant_record_meta_invalid"))?,
)?
};
let seq = match meta.get("last_seq") {
Some(value) => value
.as_u64()
.ok_or_else(|| anyhow::anyhow!("assistant_record_meta_sequence_invalid"))?,
None => super::record_restore::sequence(
&mut self.native,
&self.person,
id,
&document["selection"]["version"],
)?,
};
meta["last_seq"] = json!(seq);
self.conversations.insert(
id.to_owned(),
Conversation {
meta,
seq,
live_seq: seq,
},
);
}
self.conversations
.get_mut(id)
.ok_or_else(|| anyhow::anyhow!("assistant_record_conversation_absent"))
}
fn accept(&mut self, mut value: Value) -> anyhow::Result<()> {
let Some(item) = value.get_mut("item") else {
return self.send(value);
};
let id = item["conversation"]
.as_str()
.filter(|id| !id.is_empty())
.ok_or_else(|| anyhow::anyhow!("record_conversation_invalid"))?
.to_owned();
let limit = self.seat.harness.title_chars;
item["seq"] = json!(self.next_sequence(&id)?);
if item["replayed"] == true {
self.mark_live(item)?;
return self.send(value);
}
let conversation = self.conversation(&id)?;
conversation.meta["updated_at_ms"] = json!(now()?);
conversation.meta["state"] = json!("live");
let kind = item["kind"]
.as_str()
.ok_or_else(|| anyhow::anyhow!("record_kind_invalid"))?
.to_owned();
if kind == "request" && conversation.meta["title"] == "" {
conversation.meta["title"] = json!(
item["text"]
.as_str()
.unwrap_or_default()
.chars()
.take(limit)
.collect::<String>()
);
}
if matches!(kind.as_str(), "turn_ended" | "turn_failed") {
conversation.meta["turns"] = json!(
conversation.meta["turns"]
.as_u64()
.ok_or_else(|| anyhow::anyhow!("record_turn_count_invalid"))?
.checked_add(1)
.ok_or_else(|| anyhow::anyhow!("record_turn_count_exhausted"))?
);
}
if kind == "ended" {
conversation.meta["state"] = item["state"].clone();
conversation.meta["exit_status"] = item["exit_status"].clone();
}
self.dirty = true;
match kind.as_str() {
"text" | "thought" => self.message(item)?,
"tool_call" => self.tool(item)?,
"turn_ended" | "turn_failed" | "ended" => {
self.finish_messages(&id)?;
self.complete(item.clone())?;
}
_ => self.complete(item.clone())?,
}
self.check_bound()?;
if matches!(
kind.as_str(),
"prompt_accepted"
| "prompt_queued"
| "permission_ask"
| "permission_answered"
| "turn_ended"
| "turn_failed"
| "ended"
| "session_failed"
| "sessions"
| "config_options"
| "config_failed"
) || (kind == "available_commands"
&& !self
.completed
.iter()
.any(|item| item["kind"] == "session_opened"))
{
self.flush()?;
}
Ok(())
}
fn flush(&mut self) -> anyhow::Result<()> {
if self.batch.is_some() && !self.commit()? {
return Ok(());
}
if self.completed.is_empty() {
return Ok(());
}
self.resequence_completed()?;
self.completed.sort_by(|a, b| {
a["conversation"]
.as_str()
.cmp(&b["conversation"].as_str())
.then_with(|| a["seq"].as_u64().cmp(&b["seq"].as_u64()))
});
let expected=match self.native.read("documents.get",&json!({"selection":{"branch":"main"},"collection":"_assistant","document_id":format!("{}/head",self.person)})) {Ok(answer)=>answer["selection"]["version"].clone(),Err(error)=>{self.failed(&error.code)?;return Ok(());}};
let items = std::mem::take(&mut self.completed);
let item_bytes = std::mem::take(&mut self.completed_bytes);
let mut records = Vec::new();
for item in &items {
let mut item = item.clone();
item["durable"] = json!(true);
records.push(json!({"document_id":format!("{}/{}/{:020}",self.person,item["conversation"].as_str().ok_or_else(||anyhow::anyhow!("record_conversation_invalid"))?,item["seq"].as_u64().ok_or_else(||anyhow::anyhow!("record_sequence_invalid"))?),"document":item}));
}
for (id, conversation) in &self.conversations {
records.push(json!({"document_id":format!("{}/{id}/meta",self.person),"document":conversation.meta}));
}
let arguments = json!({"request_id":super::permission::operation()?,"branch":"main","expected":expected,"collection":"_assistant","records":records});
self.batch_bytes = self
.measured(&arguments)?
.checked_add(item_bytes)
.ok_or_else(|| anyhow::anyhow!("record_buffer_bound_overflow"))?;
self.batch = Some(Batch {
arguments,
items,
uncertain: false,
attempted: false,
});
self.check_bound()?;
self.commit()?;
Ok(())
}
fn commit(&mut self) -> anyhow::Result<bool> {
let batch = self
.batch
.as_mut()
.ok_or_else(|| anyhow::anyhow!("record_batch_absent"))?;
if batch.attempted {
let receipt = self.native.read(
"mutation.receipt",
&json!({"branch":"main","request_id":batch.arguments["request_id"]}),
);
match receipt {
Ok(value)
if value["branch"] == "main"
&& value["request_id"] == batch.arguments["request_id"] =>
{
return self.committed();
}
Ok(value) if value.is_null() => {
if !batch.uncertain {
match self.native.read("documents.get",&json!({"selection":{"branch":"main"},"collection":"_assistant","document_id":format!("{}/head",self.person)})) {
Ok(answer)=>batch.arguments["expected"]=answer["selection"]["version"].clone(),
Err(error)=>{self.failed(&error.code)?;return Ok(false);}
}
}
}
Ok(_) => anyhow::bail!("record_receipt_invalid"),
Err(error) => {
self.failed(&error.code)?;
return Ok(false);
}
}
}
self.batch_bytes = serde_json::to_vec(&batch.arguments)?
.len()
.checked_add(serde_json::to_vec(&batch.items)?.len())
.ok_or_else(|| anyhow::anyhow!("record_buffer_bound_overflow"))?;
let call = ToolCall {
name: "put_records".to_owned(),
arguments: serde_json::value::to_raw_value(&batch.arguments)?,
metadata: None,
};
batch.attempted = true;
let result = adapter::dispatch_assistant_record(&mut self.native, &call)?;
if result.is_error {
let error: Value = serde_json::from_str(result.structured_content.get())?;
let code = error["code"]
.as_str()
.ok_or_else(|| anyhow::anyhow!("record_refusal_invalid"))?;
batch.uncertain = code.contains("outcome_unknown") || code.contains("complete_receipt");
self.failed(code)?;
return Ok(false);
}
self.committed()
}
fn committed(&mut self) -> anyhow::Result<bool> {
if self.stale_completion() {
return self.promote_committed();
}
let batch = self
.batch
.take()
.ok_or_else(|| anyhow::anyhow!("record_batch_absent"))?;
self.batch_bytes = 0;
for mut item in batch.items {
if let Some(id) = item["conversation"].as_str() {
self.failures.remove(id);
}
item["durable"] = json!(true);
super::stream::fit(&mut item, self.seat.harness.content_bytes)?;
self.send(json!({"item":item}))?;
}
self.dirty =
!self.completed.is_empty() || !self.messages.is_empty() || !self.tools.is_empty();
Ok(true)
}
}
fn now() -> anyhow::Result<u64> {
Ok(u64::try_from(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)?
.as_millis(),
)?)
}
#[cfg(test)]
#[path = "record_review_tests.rs"]
mod review_tests;
#[path = "record_failures.rs"]
mod failures;
#[path = "record_items.rs"]
mod items;
#[path = "record_promote.rs"]
mod promote;