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,
}
struct Batch {
arguments: Value,
items: Vec<Value>,
uncertain: bool,
attempted: bool,
}
pub(super) struct Record {
native: Native,
seat: Seat,
person: String,
harness: Value,
output: Output,
conversations: BTreeMap<String, Conversation>,
completed: Vec<Value>,
messages: BTreeMap<(String, String), Value>,
tools: BTreeMap<(String, String), Value>,
batch: Option<Batch>,
dirty: bool,
}
impl Record {
pub fn new(
seat: Seat,
identity: &Identity,
person: String,
harness: Value,
output: 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,
})
}
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 });
}
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.output.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;
let conversation = self.conversation(&id)?;
conversation.seq = conversation
.seq
.checked_add(1)
.ok_or_else(|| anyhow::anyhow!("conversation_sequence_exhausted"))?;
item["seq"] = json!(conversation.seq);
conversation.meta["last_seq"] = json!(conversation.seq);
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.completed.push(item.clone());
}
_ => self.completed.push(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 message(&mut self, item: &Value) -> anyhow::Result<()> {
if item["complete"] != true {
self.output.send(json!({"item":item}))?;
}
let id = item["conversation"]
.as_str()
.ok_or_else(|| anyhow::anyhow!("record_conversation_invalid"))?
.to_owned();
let key = (
id,
format!("{}:{}:{}", item["kind"], item["role"], item["message_id"]),
);
let mut message = self
.messages
.remove(&key)
.unwrap_or_else(|| json!({"text":""}));
let previous = message["text"].as_str().unwrap_or_default().to_owned();
let was_truncated = message["truncated"] == true;
if let Some(fields) = item.as_object() {
for (key, value) in fields {
if key != "text" {
message[key] = value.clone();
}
}
}
if let Some(text) = item.get("text").and_then(Value::as_str) {
message["text"] = json!(if item["replace"] == true {
text.to_owned()
} else {
previous + text
});
}
if was_truncated {
message["truncated"] = json!(true);
}
super::stream::fit(&mut message, self.seat.harness.content_bytes)?;
if item["complete"] == true {
message["replace"] = json!(true);
self.completed.push(message);
} else {
self.messages.insert(key, message);
}
Ok(())
}
fn tool(&mut self, item: &Value) -> anyhow::Result<()> {
let key = (
item["conversation"]
.as_str()
.ok_or_else(|| anyhow::anyhow!("record_conversation_invalid"))?
.to_owned(),
item["tool_call_id"]
.as_str()
.ok_or_else(|| anyhow::anyhow!("record_tool_id_invalid"))?
.to_owned(),
);
let mut tool = self.tools.remove(&key).unwrap_or_else(|| json!({}));
if let Some(fields) = item.as_object() {
for (key, value) in fields {
tool[key] = value.clone();
}
}
super::stream::fit(&mut tool, self.seat.harness.content_bytes)?;
if matches!(tool["status"].as_str(), Some("completed" | "failed")) {
self.completed.push(tool);
} else {
self.output.send(json!({"item":item}))?;
self.tools.insert(key, tool);
}
Ok(())
}
fn finish_messages(&mut self, id: &str) -> anyhow::Result<()> {
let keys: Vec<_> = self
.messages
.keys()
.filter(|(conversation, _)| conversation == id)
.cloned()
.collect();
for key in keys {
if let Some(mut item) = self.messages.remove(&key) {
item["complete"] = json!(true);
item["replace"] = json!(true);
super::stream::fit(&mut item, self.seat.harness.content_bytes)?;
self.completed.push(item);
}
}
let keys: Vec<_> = self
.tools
.keys()
.filter(|(conversation, _)| conversation == id)
.cloned()
.collect();
for key in keys {
if let Some(mut item) = self.tools.remove(&key) {
item["status"] = json!("interrupted");
super::stream::fit(&mut item, self.seat.harness.content_bytes)?;
self.completed.push(item);
}
}
Ok(())
}
fn check_bound(&self) -> anyhow::Result<()> {
let completed = serde_json::to_vec(&self.completed)?.len();
let messages = serde_json::to_vec(&self.messages.values().collect::<Vec<_>>())?.len();
let tools = serde_json::to_vec(&self.tools.values().collect::<Vec<_>>())?.len();
let batch = self
.batch
.as_ref()
.map(|batch| serde_json::to_vec(&batch.arguments).map(|bytes| bytes.len()))
.transpose()?
.unwrap_or(0);
let bytes = completed
.checked_add(messages)
.and_then(|n| n.checked_add(tools))
.and_then(|n| n.checked_add(batch))
.ok_or_else(|| anyhow::anyhow!("record_buffer_bound_overflow"))?;
let limit = self
.seat
.harness
.content_bytes
.checked_mul(self.seat.harness.write_queue)
.ok_or_else(|| anyhow::anyhow!("record_buffer_bound_overflow"))?;
anyhow::ensure!(bytes <= limit, "record_buffer_bytes_exceeded");
Ok(())
}
fn flush(&mut self) -> anyhow::Result<()> {
if self.batch.is_some() && !self.commit()? {
return Ok(());
}
if self.completed.is_empty() {
return Ok(());
}
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 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}));
}
self.batch = Some(Batch {
arguments: json!({"request_id":super::permission::operation()?,"branch":"main","expected":expected,"collection":"_assistant","records":records}),
items,
uncertain: false,
attempted: false,
});
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);
}
}
}
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(&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> {
let batch = self
.batch
.take()
.ok_or_else(|| anyhow::anyhow!("record_batch_absent"))?;
for mut item in batch.items {
item["durable"] = json!(true);
super::stream::fit(&mut item, self.seat.harness.content_bytes)?;
self.output.send(json!({"item":item}))?;
}
self.dirty =
!self.completed.is_empty() || !self.messages.is_empty() || !self.tools.is_empty();
Ok(true)
}
fn failed(&mut self, code: &str) -> anyhow::Result<()> {
let id = self
.conversations
.keys()
.next()
.cloned()
.ok_or_else(|| anyhow::anyhow!("record_failure_without_conversation"))?;
let conversation = self
.conversations
.get_mut(&id)
.ok_or_else(|| anyhow::anyhow!("record_failure_without_conversation"))?;
conversation.seq = conversation
.seq
.checked_add(1)
.ok_or_else(|| anyhow::anyhow!("conversation_sequence_exhausted"))?;
conversation.meta["last_seq"] = json!(conversation.seq);
let mut item = json!({"kind":"turn_failed","source":"record","conversation":id,"seq":conversation.seq,"durable":false,"code":code,"message":"conversation write refused; completed items retained for the next batch"});
super::stream::fit(&mut item, self.seat.harness.content_bytes)?;
self.completed.push(item.clone());
self.dirty = true;
self.output.send(json!({"item":item}))
}
}
fn now() -> anyhow::Result<u64> {
Ok(u64::try_from(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)?
.as_millis(),
)?)
}