//! Kennedy's complete logical session lifecycle and agent orchestration.
#![forbid(unsafe_code)]
mod context;
mod services;
use chrono::{Datelike, Timelike};
pub use kcode_telegram_session_coordinator::validate_file_name as validate_delivery_file_name;
pub use services::{Api as Service, LocalServices as Capabilities};
/// Application-selected primary model facts used mechanically by a session.
#[derive(Clone, Debug)]
pub struct RuntimeModel {
pub model: String,
pub reasoning_effort: String,
pub context_window_tokens: u64,
}
impl RuntimeModel {
pub fn from_intelligence(runtime: kcode_intelligence_router::RuntimeModel) -> Self {
Self {
model: runtime.model,
reasoning_effort: runtime.reasoning_effort,
context_window_tokens: runtime.context_window_tokens,
}
}
fn attribution(&self) -> String {
format!("{}-{}", self.model, self.reasoning_effort)
}
}
use std::{
collections::{BTreeMap, HashMap, HashSet},
future::Future,
time::Duration,
};
use anyhow::Context as _;
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
use chrono::{DateTime, Utc};
use kcode_commit_session::{CommitReceipt, CommitRequest, PlannedNode};
use kcode_dev_tools::{
ATTACH_OBJECT_WEB_LIB_TOOL, CALL_RUST_BIN_TOOL, RUST_BIN_TOOLS, RUST_LIB_TOOLS, WEB_LIB_TOOLS,
WRITE_RUST_BIN_TOOL, WRITE_RUST_LIB_TOOL, WRITE_WEB_LIB_TOOL, proposed_write_snapshot,
};
use kcode_dev_tools_chatend::{
FreeformWrite, SourceSnapshot, apply_snapshot, prepare_freeform_write, source_box_id,
};
use kcode_history_ingress_context::{
Outcome as HistoryIngressContextOutcome, RecoveryOutcome as ContextRecoveryOutcome,
};
use kcode_kweb_context::{
Context as KwebContext, Node as KwebNode, NodeDraft, StagedCreate as KwebStagedCreate,
};
use kcode_kweb_db::{NodeId, ObjectId};
use kcode_server_object_envelopes::{StoredFile, encode_file, sanitize_file_name};
use kcode_session_history::{
NewSession, Session as HistorySession,
chatend::{
BoxContent, BoxId, BoxOwner, EventId, EventKind, ObjectMetadata, PendingId, Representation,
SessionKind, SessionMetadata,
},
};
use kcode_speech_classification::KTOOLS as SPEECH_CLASSIFICATION_TOOLS;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use uuid::Uuid;
use context::{load_durable_batch, node_from_value};
const AGENT_LOOP_ROUND_LIMIT: u64 = 100;
const BROWSER_CONVERSATION_REQUEST_TIMEOUT: Duration = Duration::from_secs(90 * 60);
const HISTORY_INGRESS_REQUEST_TIMEOUT: Duration = Duration::from_secs(90 * 60);
const WAKEUP_REQUEST_TIMEOUT: Duration = Duration::from_secs(90 * 60);
const MIN_NODE_SHORT_NAME_CHARACTERS: usize = 4;
const MAX_NODE_SHORT_NAME_CHARACTERS: usize = 50;
const MAX_NODE_SHORT_DESCRIPTION_CHARACTERS: usize = 200;
const MAX_NODE_LONG_DESCRIPTION_CHARACTERS: usize = 5_000;
const MAX_MEDIA_ENRICHMENT_BYTES: u64 = 20 * 1024 * 1024;
const KWEB_TOOL_INSTANCE: &str = "kweb";
const CONTEXT_OVERFLOW_WARNING_BOX_NAME: &str = "Context overflow warning";
const CONTEXT_OVERFLOW_WARNING: &str = "Context size was exceeded, some context has been dehydrated. The session is now at risk of destabilizing, please perform any cleanup tasks and end the session";
const INGRESS_FORCE_COMMIT_NOTE: &str = "ingress_force_commit";
const SUBAGENT_CONTEXT_NODE_LIMIT: usize = 64;
const BOX_TEXT_OBJECT_SOURCE: &str = "kennedy-box-text";
const BOX_TEXT_MEDIA_TYPE: &str = "text/plain; charset=utf-8";
const SLOW_TOOL_THRESHOLD: Duration = Duration::from_secs(3);
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum AgentMode {
Conversation,
FreeTime,
Wakeup,
Ingress { record_id: Option<String> },
}
#[derive(Clone, Debug)]
pub struct SessionOptions {
pub session_type: String,
pub root_node_ids: Vec<String>,
pub reference_root_node_ids: Vec<String>,
pub channel: Value,
pub free_time: Value,
pub orchestration: Value,
pub provenance_id: Option<String>,
pub mode: AgentMode,
pub source_session_type: Option<String>,
pub group_context: Value,
pub rust_lib_session_id: Option<String>,
}
impl SessionOptions {
pub fn conversation(session_type: impl Into<String>, roots: Vec<String>) -> Self {
Self {
session_type: session_type.into(),
root_node_ids: roots,
reference_root_node_ids: Vec::new(),
channel: Value::Null,
free_time: Value::Null,
orchestration: json!({"owner":"backend","status":"idle"}),
provenance_id: None,
mode: AgentMode::Conversation,
source_session_type: None,
group_context: Value::Null,
rust_lib_session_id: None,
}
}
}
fn restore_session_type(options: &mut SessionOptions, state: &Value) {
if !matches!(&options.mode, AgentMode::Ingress { .. }) {
options.session_type = state
.get("sessionType")
.and_then(Value::as_str)
.unwrap_or(&options.session_type)
.to_owned();
}
}
fn restore_commit_receipt(restored: Option<&Value>) -> anyhow::Result<Option<CommitReceipt>> {
restored
.and_then(|state| state.get("commitReceipt"))
.filter(|receipt| !receipt.is_null())
.cloned()
.map(serde_json::from_value)
.transpose()
.context("decoding the stored session commit receipt")
}
#[derive(Clone, Debug, Default, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct KwebPlan {
creates: Vec<StagedNodeCreate>,
updates: BTreeMap<String, PlannedNode>,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct StagedNodeCreate {
pending_id: String,
data: PlannedNode,
}
impl KwebPlan {
fn restore(restored: Option<&Value>, journal: &HistorySession) -> anyhow::Result<Self> {
if let Some(plan) = restored.and_then(|state| state.get("kwebPlan")) {
return serde_json::from_value(plan.clone()).context("decoding the staged Kweb plan");
}
// Transitional compatibility for journals written before Kweb plans
// moved into KennedyServer lifecycle state.
let latest = journal.state().events.iter().rev().find_map(|event| {
let EventKind::KwebPlanChanged { operation } = &event.kind else {
return None;
};
operation.get("plan")
});
latest
.cloned()
.map(serde_json::from_value)
.transpose()
.context("decoding the staged Kweb plan")
.map(Option::unwrap_or_default)
}
fn created(&self, id: &str) -> Option<&PlannedNode> {
self.creates
.iter()
.find(|create| create.pending_id == id)
.map(|create| &create.data)
}
fn created_mut(&mut self, id: &str) -> Option<&mut PlannedNode> {
self.creates
.iter_mut()
.find(|create| create.pending_id == id)
.map(|create| &mut create.data)
}
}
pub struct Session {
api: Service,
runtime: RuntimeModel,
journal: HistorySession,
plan: KwebPlan,
pub session_type: String,
pub channel: Value,
pub free_time: Value,
pub orchestration: Value,
pub provenance_id: Option<String>,
pub rust_lib_session_id: String,
pub root_node_ids: Vec<String>,
pub reference_root_node_ids: Vec<String>,
pub started_at: String,
pub transcript: Vec<Value>,
pub pending_turn: bool,
pub pending_external_event_id: Option<String>,
pub completed: bool,
pub rounds_used: u64,
commit_receipt: Option<CommitReceipt>,
commit_author: String,
mode: AgentMode,
source_session_type: Option<String>,
group_context: Value,
context: KwebContext,
free_time_end_reason: Option<String>,
fatal_persistence_error: Option<String>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ResolvedObject {
pub object_id: String,
pub bytes: Vec<u8>,
pub file_name: String,
pub media_type: String,
pub transport_kind: Option<String>,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum InputStage {
Accepted,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ContextRecovery {
NotNeeded,
Recovered,
Irreducible,
}
fn render_load_nodes_result(
journal: &HistorySession,
changed_box_ids: &[BoxId],
) -> anyhow::Result<String> {
if changed_box_ids.is_empty() {
return Ok("LoadNodes completed. The shared Kweb boxes were already current.".into());
}
let rendered = journal
.state()
.projection()
.items
.into_iter()
.filter(|item| !item.marker)
.map(|item| (item.box_id, item.text))
.collect::<BTreeMap<_, _>>();
changed_box_ids
.iter()
.map(|box_id| {
rendered
.get(box_id)
.cloned()
.with_context(|| format!("updated Kweb box {box_id} is absent from the projection"))
})
.collect::<anyhow::Result<Vec<_>>>()
.map(|boxes| boxes.join("\n\n"))
}
fn provider_tool_result_with_context_footer(journal: &HistorySession, result: &str) -> String {
// A provider turn may perform several inference/tool steps without being
// restarted, so put the newly projected status at the end of every tool
// continuation rather than leaving the model with the turn-opening value.
let footer = journal.state().projection().footer;
if result.is_empty() {
footer
} else {
format!("{result}\n\n{footer}")
}
}
fn append_slow_tool_duration(text: &mut String, elapsed: Duration) {
if elapsed <= SLOW_TOOL_THRESHOLD {
return;
}
if !text.is_empty() && !text.ends_with('\n') {
text.push('\n');
}
text.push_str(&format!("[tool duration: {:.3}s]", elapsed.as_secs_f64()));
}
fn render_web_search_result(result: &kcode_intelligence_router::SearchResponse) -> String {
let mut text = result.answer.clone();
if !result.sources.is_empty() {
text.push_str("\n\nSources:");
for source in &result.sources {
let title = if source.title.trim().is_empty() {
&source.url
} else {
&source.title
};
text.push_str("\n- ");
text.push_str(title);
if title != &source.url {
text.push_str(": ");
text.push_str(&source.url);
}
}
}
text
}
fn render_web_fetch_result(result: &kcode_intelligence_router::FetchResponse) -> String {
let mut text = format!("Source URL: {}", result.url);
if let Some(title) = result
.title
.as_deref()
.filter(|title| !title.trim().is_empty())
{
text.push_str("\nTitle: ");
text.push_str(title);
}
text.push_str("\nContent type: ");
text.push_str(&result.content_type);
if result.truncated {
text.push_str("\nThe returned page text was truncated.");
}
text.push_str("\n\n");
text.push_str(&result.content);
text
}
fn render_media_annotation_result(
object_id: &str,
file_name: &str,
content_type: &str,
result: &kcode_intelligence_router::AnnotationResponse,
) -> anyhow::Result<String> {
anyhow::ensure!(
!result.text.trim().is_empty(),
"media annotation response has no text"
);
let status = if result.complete {
"complete"
} else {
"incomplete"
};
let mut rendered = format!(
"Annotation for {object_id}\nFile: {file_name}\nContent type: {content_type}\nModel: {}\nStatus: {status}",
result.model
);
if let Some(reason) = result
.incomplete_reason
.as_deref()
.filter(|value| !value.trim().is_empty())
{
rendered.push_str("\nIncomplete reason: ");
rendered.push_str(reason);
}
rendered.push_str("\n\n");
rendered.push_str(&result.text);
Ok(rendered)
}
fn render_audio_transcription_result(
object_id: &str,
file_name: &str,
content_type: &str,
result: &kcode_intelligence_router::TranscriptionResponse,
) -> anyhow::Result<String> {
anyhow::ensure!(
!result.text.trim().is_empty(),
"audio transcription response has no text"
);
Ok(format!(
"Transcription for {object_id}\nFile: {file_name}\nContent type: {content_type}\nModel: {}\nStatus: complete\n\n{}",
result.model, result.text
))
}
fn render_document_extraction_result(
object_id: &str,
file_name: &str,
result: &kcode_intelligence_router::DocumentExtraction,
) -> String {
format!(
"Extracted text for {object_id}\nFile: {file_name}\nFormat: {}\nCharacters: {}\nTruncated: {}\n\n{}",
result.format, result.characters, result.truncated, result.text
)
}
struct ToolCall {
name: String,
arguments: Value,
}
struct RecordedToolInvocation {
invocation_id: String,
tool_instance: String,
tool_name: String,
}
struct PendingFreeformWrite {
request: FreeformWrite,
call_box_id: BoxId,
}
fn tool_call_box_content(call: &ToolCall) -> anyhow::Result<BoxContent> {
if matches!(
call.name.as_str(),
WRITE_RUST_LIB_TOOL | WRITE_WEB_LIB_TOOL | WRITE_RUST_BIN_TOOL
) {
let name = call
.arguments
.get("name")
.and_then(Value::as_str)
.map(|name| name.chars().take(255).collect::<String>());
let file_count = call
.arguments
.get("files")
.and_then(Value::as_array)
.map(Vec::len);
return Ok(BoxContent {
text: serde_json::to_string_pretty(&json!({
"name":call.name,
"arguments":{
"name":name,
"fileCount":file_count,
"completeFileContents":"omitted from active context; retained in the durable tool invocation"
}
}))?,
objects: Vec::new(),
metadata: json!({
"compactedToolInvocation":true,
"toolName":call.name,
}),
});
}
Ok(BoxContent::text(serde_json::to_string_pretty(
&json!({"name":call.name,"arguments":call.arguments}),
)?))
}
struct ToolOutcome {
text: String,
store_result: bool,
ok: bool,
end_session: bool,
freeform_write: Option<FreeformWrite>,
managed_source_snapshot: Option<SourceSnapshot>,
}
#[derive(Clone)]
struct ChangedSubagentState {
box_id: BoxId,
name: String,
text: Option<String>,
hide_from_parent: bool,
}
#[derive(Clone, Copy)]
struct CanonicalBoxVersion {
event_id: EventId,
active: bool,
tool_owned: bool,
dehydrated: bool,
}
type CanonicalBoxVersions = BTreeMap<BoxId, CanonicalBoxVersion>;
fn canonical_box_versions(journal: &HistorySession) -> CanonicalBoxVersions {
journal
.state()
.boxes
.iter()
.map(|(box_id, state)| {
(
*box_id,
CanonicalBoxVersion {
event_id: state.canonical.event_id,
active: state.active,
tool_owned: matches!(state.owner, BoxOwner::Tool { .. }),
dehydrated: matches!(state.representation, Representation::Dehydrated { .. }),
},
)
})
.collect()
}
fn changed_subagent_tool_states(
journal: &HistorySession,
previous: &CanonicalBoxVersions,
) -> Vec<ChangedSubagentState> {
journal
.state()
.boxes
.iter()
.filter_map(|(box_id, state)| {
let current_tool_owned = matches!(state.owner, BoxOwner::Tool { .. });
let prior = previous.get(box_id);
if !current_tool_owned && !prior.is_some_and(|prior| prior.tool_owned) {
return None;
}
let changed = prior.is_none_or(|prior| {
prior.event_id != state.canonical.event_id
|| prior.active != state.active
|| prior.tool_owned != current_tool_owned
});
changed.then(|| ChangedSubagentState {
box_id: *box_id,
name: state.name.clone(),
text: (current_tool_owned && !state.canonical.content.text.is_empty())
.then(|| state.canonical.content.text.clone()),
hide_from_parent: prior.is_none_or(|prior| prior.dehydrated || !prior.active),
})
})
.collect()
}
fn subagent_managed_write_fits(
journal: &HistorySession,
call: &ToolCall,
budget: &kcode_agent_runtime::ContextBudget,
) -> bool {
let Some(snapshot) = proposed_write_snapshot(&call.name, &call.arguments) else {
return true;
};
let kind = snapshot.kind;
let key = source_box_id(journal, kind, &snapshot.name)
.map(|box_id| format!("tool-state:{box_id}"))
.unwrap_or_else(|| format!("prospective-managed-state:{:?}:{}", kind, snapshot.name));
budget.fits_state(
key,
format!(
"Current Managed {} {}:\n{}",
kind.label(),
snapshot.name,
snapshot.text
),
)
}
fn subagent_state_updates(
states: &[ChangedSubagentState],
) -> Vec<kcode_agent_runtime::StateUpdate> {
states
.iter()
.map(|state| kcode_agent_runtime::StateUpdate {
key: format!("tool-state:{}", state.box_id),
text: state
.text
.as_ref()
.map(|text| format!("Current {}:\n{text}", state.name)),
})
.collect()
}
struct KennedySubagentHost<'a> {
session: &'a mut Session,
captures: HashMap<String, FreeformWrite>,
}
struct KennedySessionHost<'a, C> {
session: &'a mut Session,
checkpoint: &'a mut C,
accounting: Option<kcode_intelligence_chatend::TopLevelCall>,
pending_freeform_write: Option<PendingFreeformWrite>,
deadline_after_response: bool,
}
impl Session {
pub async fn new(
api: Service,
system_prompt: String,
runtime: RuntimeModel,
mut options: SessionOptions,
restored: Option<&Value>,
) -> anyhow::Result<Self> {
if let Some(state) = restored {
restore_session_type(&mut options, state);
options.channel = state.get("channel").cloned().unwrap_or(options.channel);
options.free_time = state.get("freeTime").cloned().unwrap_or(options.free_time);
options.orchestration = state
.get("orchestration")
.cloned()
.unwrap_or(options.orchestration);
}
if options.group_context.is_null() {
options.group_context = options
.channel
.get("groupContext")
.cloned()
.unwrap_or(Value::Null);
}
options
.reference_root_node_ids
.retain(|id| !options.root_node_ids.contains(id));
options.reference_root_node_ids.sort();
options.reference_root_node_ids.dedup();
let started_at = restored
.and_then(|state| state.get("startedAt"))
.and_then(Value::as_str)
.map(str::to_owned)
.unwrap_or_else(|| Utc::now().to_rfc3339());
let rust_lib_session_id = restored
.and_then(|state| state.get("rustLibSessionId"))
.and_then(Value::as_str)
.map(str::to_owned)
.or(options.rust_lib_session_id.clone())
.unwrap_or_else(|| format!("kennedy:{}", Uuid::new_v4()));
let history_session_id = restored
.and_then(|state| state.get("sessionId"))
.and_then(Value::as_str)
.map(str::to_owned);
let source_session_type = options.source_session_type.clone().or_else(|| {
restored
.and_then(|state| state.get("sourceSessionType"))
.and_then(Value::as_str)
.map(str::to_owned)
});
let session_id = history_session_id
.clone()
.unwrap_or_else(|| Uuid::new_v4().to_string());
let metadata = SessionMetadata {
session_id: session_id.clone(),
kind: session_kind(&options.session_type, &options.mode),
created_at: started_at.clone(),
effective_context_tokens: runtime.context_window_tokens,
channel: options.channel.clone(),
};
let mut journal = if history_session_id.is_some() {
api.history_session(metadata, &runtime.model)
.with_context(|| {
format!(
"opening authoritative session {session_id} (legacy snapshots are intentionally unsupported)"
)
})?
} else {
api.create_history_session(NewSession {
kind: metadata.kind,
created_at: metadata.created_at,
effective_context_tokens: metadata.effective_context_tokens,
channel: metadata.channel,
})?
};
let mut context =
KwebContext::new(options.root_node_ids.clone()).map_err(anyhow::Error::new)?;
restore_kweb_context(&journal, &mut context)?;
let plan = KwebPlan::restore(restored, &journal)?;
let transcript = transcript_from_journal(&journal);
let (pending_turn, pending_external_event_id) = restore_pending_turn(restored, &transcript);
let needs_initialization = !journal
.state()
.boxes
.values()
.any(|state| matches!(state.owner, BoxOwner::System));
let commit_receipt = restore_commit_receipt(restored)?;
let commit_author = restored
.and_then(|state| state.get("commitAuthor"))
.and_then(Value::as_str)
.map(str::to_owned)
.unwrap_or_else(|| runtime.attribution());
if let Some(receipt) = &commit_receipt {
journal.mark_completed(receipt.session_object_id.to_string());
}
let completed =
journal.state().completed_session_object.is_some() || commit_receipt.is_some();
let mut session = Self {
api,
runtime,
journal,
plan,
session_type: options.session_type,
channel: options.channel,
free_time: options.free_time,
orchestration: options.orchestration,
provenance_id: options.provenance_id,
rust_lib_session_id,
root_node_ids: options.root_node_ids,
reference_root_node_ids: options.reference_root_node_ids,
started_at,
transcript,
pending_turn,
pending_external_event_id,
completed,
rounds_used: restored
.and_then(|state| state.get("roundsUsed"))
.and_then(Value::as_u64)
.unwrap_or_default(),
commit_receipt,
commit_author,
mode: options.mode,
source_session_type,
group_context: options.group_context,
context,
free_time_end_reason: None,
fatal_persistence_error: None,
};
if matches!(session.mode, AgentMode::Ingress { .. }) && !session.journal.is_sealed() {
session.journal.repair_unfinished_tools(now())?;
}
if session.journal.is_sealed() {
anyhow::ensure!(
!matches!(session.mode, AgentMode::Conversation),
"a read-only conversation has an unexpectedly sealed session log"
);
if session.commit_receipt.is_none() {
session.finalize_kweb_session()?;
}
session.completed = true;
return Ok(session);
}
if needs_initialization {
session.journal.create_box(
now(),
"Kennedy system prompt",
BoxOwner::System,
BoxContent::text(&system_prompt),
)?;
if session.session_type == "telegram-group" && !session.group_context.is_null() {
session.journal.create_box(
now(),
"Telegram group context",
BoxOwner::Controller,
BoxContent::text(kcode_telegram_session_coordinator::format_group_context(
&session.group_context,
)),
)?;
}
let roots = session.root_node_ids.clone();
let invocation =
session.record_tool_invocation("LoadNodes", json!({"identifiers":&roots}))?;
let result = load_durable_batch(&session.api, &mut session.context, &roots)?;
session.sync_kweb_boxes()?;
session.record_tool_completion(
Some(&invocation),
json!({"ok":true,"automatic":true,"identifiers":roots,"result":result}),
)?;
} else {
session.sync_kweb_boxes()?;
}
if matches!(session.mode, AgentMode::Ingress { .. })
&& !session.completed
&& !session.journal.state().history_ingress_started
{
session.prepare_history_ingress(&system_prompt).await?;
}
Ok(session)
}
async fn prepare_history_ingress(&mut self, prompt: &str) -> anyhow::Result<()> {
let cost_at_ingress = self.journal.state().projection().status;
if !self.journal.state().source_terminated {
self.journal.record(
now(),
EventKind::SourceTerminated {
reason: "history_ingress".into(),
},
)?;
}
let system_box = self
.journal
.state()
.boxes
.values()
.find(|state| matches!(state.owner, BoxOwner::System))
.map(|state| state.id)
.context("session has no system-prompt box")?;
self.journal
.update_box(now(), system_box, BoxContent::text(prompt))?;
let ingress_kind = session_kind(&self.session_type, &self.mode);
if self.journal.state().metadata.effective_context_tokens
!= self.runtime.context_window_tokens
|| self.journal.state().metadata.kind != ingress_kind
{
self.journal
.configure_context(ingress_kind, self.runtime.context_window_tokens);
}
self.journal.create_box(
now(),
"Session cost at ingress",
BoxOwner::Controller,
BoxContent::text(cost_summary(
"session cost before history ingress",
cost_at_ingress.estimated_cost_usd_nanos,
cost_at_ingress.unpriced_provider_calls,
)),
)?;
self.revalidate_loaded_nodes().await?;
match kcode_history_ingress_context::prepare(&mut self.journal, now())? {
HistoryIngressContextOutcome::Ready => {}
HistoryIngressContextOutcome::OverCapacity {
estimated_tokens,
target_tokens,
} => {
self.journal.record(
now(),
EventKind::Note {
label: INGRESS_FORCE_COMMIT_NOTE.into(),
value: json!({
"reason":"fully_dehydrated_context_above_initial_target",
"estimatedTokens":estimated_tokens,
"initialTargetTokens":target_tokens,
}),
},
)?;
self.pending_turn = false;
self.finalize_kweb_session()?;
self.completed = true;
return Ok(());
}
}
self.journal
.record(now(), EventKind::HistoryIngressStarted)?;
self.pending_turn = true;
Ok(())
}
async fn revalidate_loaded_nodes(&mut self) -> anyhow::Result<()> {
let direct = self.context.loaded_node_ids().to_vec();
load_durable_batch(&self.api, &mut self.context, &direct)?;
self.sync_kweb_boxes()?;
Ok(())
}
fn stage_user_input(&mut self, text: &str, metadata: &Value) -> Option<InputStage> {
let text = text.trim();
let attachments = metadata
.get("attachments")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
if text.is_empty() && attachments.is_empty() && metadata.get("media").is_none() {
return None;
}
let result = self.stage_user_input_inner(text, metadata, attachments);
match result {
Ok(stage) => Some(stage),
Err(error) => {
self.fatal_persistence_error = Some(error.to_string());
tracing::error!(error=%error, "Could not durably stage session input");
Some(InputStage::Accepted)
}
}
}
fn stage_user_input_inner(
&mut self,
text: &str,
metadata: &Value,
attachments: Vec<Value>,
) -> anyhow::Result<InputStage> {
let mut content = BoxContent::text(text);
content.metadata = message_metadata_without_attachment_payloads(metadata);
let mut attachment_boxes = Vec::new();
let mut attachment_names = Vec::new();
let mut canonical_attachments = Vec::with_capacity(attachments.len());
for attachment in attachments {
let mut descriptor = attachment_metadata_without_payload(&attachment);
let mut file_name = ingress_object_filename(
attachment.get("fileName").and_then(Value::as_str),
"document",
);
if let Some(pending_id) = attachment.get("pendingId").and_then(Value::as_str) {
let pending_id = PendingId::parse(pending_id.to_owned())?;
anyhow::ensure!(
self.journal.objects().contains_key(&pending_id),
"attached object {pending_id} is not staged in this session"
);
content.objects.push(pending_id.to_string());
file_name = canonicalize_staged_file_descriptor(
&self.journal,
&pending_id,
&mut descriptor,
)?;
} else if let Some(data_url) = attachment.get("dataUrl").and_then(Value::as_str) {
let (media_type, bytes) = decode_data_url(data_url)?;
let id = self.journal.stage_object(
now(),
media_type,
Some(file_name.clone()),
descriptor.clone(),
&bytes,
)?;
content.objects.push(id.to_string());
file_name =
canonicalize_staged_file_descriptor(&self.journal, &id, &mut descriptor)?;
}
attachment_names.push(file_name.clone());
if let Some(extracted) = attachment
.get("text")
.and_then(Value::as_str)
.filter(|text| !text.is_empty())
{
attachment_boxes.push((
format!("User attachment text: {file_name}"),
BoxContent {
text: extracted.into(),
objects: Vec::new(),
metadata: json!({
"boxKind":"attachmentText",
"attachment":descriptor.clone(),
}),
},
));
}
canonical_attachments.push(descriptor);
}
if !content.metadata.is_object() {
content.metadata = json!({});
}
content.metadata["attachments"] = json!(canonical_attachments);
if let Some(media) = metadata.get("media") {
let mut descriptor = attachment_metadata_without_payload(media);
if let Some(pending_id) = media.get("pendingId").and_then(Value::as_str) {
let pending_id = PendingId::parse(pending_id.to_owned())?;
anyhow::ensure!(
self.journal.objects().contains_key(&pending_id),
"voice object {pending_id} is not staged in this session"
);
content.objects.push(pending_id.to_string());
canonicalize_staged_file_descriptor(&self.journal, &pending_id, &mut descriptor)?;
} else if let Some(data_url) = media.get("dataUrl").and_then(Value::as_str) {
let (media_type, bytes) = decode_data_url(data_url)?;
let file_name =
ingress_object_filename(media.get("fileName").and_then(Value::as_str), "media");
let id = self.journal.stage_object(
now(),
media_type,
Some(file_name),
descriptor.clone(),
&bytes,
)?;
content.objects.push(id.to_string());
canonicalize_staged_file_descriptor(&self.journal, &id, &mut descriptor)?;
}
content.metadata["media"] = descriptor;
}
if content.text.trim().is_empty()
&& content.objects.is_empty()
&& !attachment_names.is_empty()
{
content.text = attachment_names
.iter()
.map(|name| format!("Attachment provided: {name}"))
.collect::<Vec<_>>()
.join("\n");
}
let visible = if content.text.trim().is_empty() {
content
.objects
.iter()
.map(|id| format!("Object provided: {id}"))
.collect::<Vec<_>>()
.join("\n")
} else {
content.text.clone()
};
if !content.objects.is_empty() {
if !content.metadata.is_object() {
content.metadata = json!({});
}
content.metadata["transcriptText"] = json!(visible);
}
append_user_file_metadata(&self.journal, &mut content)?;
let mut prospective_boxes =
vec![("User message".to_owned(), BoxOwner::User, content.clone())];
prospective_boxes.extend(
attachment_boxes
.iter()
.map(|(name, content)| (name.clone(), BoxOwner::User, content.clone())),
);
let recorded_at = now();
let transcript_objects = content.objects.clone();
let mut transcript_attachments = content
.metadata
.get("attachments")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
if let Some(media) = content
.metadata
.get("media")
.filter(|value| value.is_object())
{
transcript_attachments.push(media.clone());
}
for (name, owner, content) in prospective_boxes {
self.journal
.create_box(recorded_at.clone(), name, owner, content)?;
}
let mut transcript = json!({"role":"user","content":visible});
if !transcript_objects.is_empty() {
transcript["objects"] = json!(transcript_objects);
}
if !transcript_attachments.is_empty() {
transcript["attachments"] = json!(transcript_attachments);
}
if let Some(id) = metadata.get("externalEventId").and_then(Value::as_str) {
transcript["externalEventId"] = json!(id);
}
self.transcript.push(transcript);
self.recover_context_overflow(
metadata.get("externalEventId").and_then(Value::as_str),
&[],
)?;
Ok(InputStage::Accepted)
}
pub fn append_final_user_message(&mut self, text: &str, metadata: &Value) -> bool {
self.stage_user_input(text, metadata).is_some()
}
pub fn stage_source_message(
&mut self,
kennedy: bool,
text: &str,
metadata: Value,
) -> anyhow::Result<()> {
let external_event_id = metadata
.get("externalEventId")
.and_then(Value::as_str)
.map(str::to_owned);
let owner = if kennedy {
BoxOwner::Kennedy
} else {
BoxOwner::User
};
let name = if kennedy {
"Kennedy message"
} else {
"User message"
};
self.journal.create_box(
now(),
name,
owner,
BoxContent {
text: text.into(),
objects: Vec::new(),
metadata: metadata.clone(),
},
)?;
let mut transcript = json!({
"role":if kennedy {"kennedy"} else {"user"},
"content":text,
"metadata":metadata,
});
if let Some(id) = &external_event_id {
transcript["externalEventId"] = json!(id);
}
self.transcript.push(transcript);
self.recover_context_overflow(external_event_id.as_deref(), &[])?;
Ok(())
}
pub fn answer_for_external_event(&self, id: &str) -> Option<&Value> {
self.transcript.iter().rev().find(|entry| {
is_terminal_external_response(entry)
&& entry.get("externalEventId").and_then(Value::as_str) == Some(id)
})
}
pub fn responses_for_external_event(&self, id: &str) -> Vec<&Value> {
self.transcript
.iter()
.filter(|entry| {
matches!(
entry.get("role").and_then(Value::as_str),
Some("kennedy" | "system")
) && entry.get("externalEventId").and_then(Value::as_str) == Some(id)
})
.collect()
}
pub fn resolve_object(&mut self, object_id: &str) -> anyhow::Result<ResolvedObject> {
let api = self.api.clone();
resolve_object_using(&mut self.journal, object_id, move |canonical_id| {
api.kmap_file(canonical_id).map_err(Into::into)
})
}
fn resolve_media_object(&mut self, object_id: &str) -> anyhow::Result<ResolvedObject> {
let mut resolved = self.resolve_object(object_id)?;
resolved.media_type = normalized_media_type(&resolved.media_type);
anyhow::ensure!(
!resolved.bytes.is_empty(),
"media object {} is empty",
resolved.object_id
);
anyhow::ensure!(
resolved.bytes.len() as u64 <= MAX_MEDIA_ENRICHMENT_BYTES,
"media object {} is {} bytes, over the {}-byte enrichment limit",
resolved.object_id,
resolved.bytes.len(),
MAX_MEDIA_ENRICHMENT_BYTES
);
Ok(resolved)
}
fn resolve_image_object(
&mut self,
object_id: &str,
) -> anyhow::Result<(Vec<u8>, String, String)> {
let resolved = self.resolve_media_object(object_id)?;
anyhow::ensure!(
resolved.media_type.starts_with("image/"),
"GenerateImage reference {object_id} is not an image"
);
Ok((resolved.bytes, resolved.file_name, resolved.media_type))
}
fn recover_context_overflow(
&mut self,
external_event_id: Option<&str>,
pinned_box_ids: &[BoxId],
) -> anyhow::Result<ContextRecovery> {
let projection = self.journal.state().projection();
let target_tokens = self.journal.state().active_context_limit();
if projection.estimated_tokens <= target_tokens {
return Ok(ContextRecovery::NotNeeded);
}
let projection_hash = hex::encode(Sha256::digest(projection.render().as_bytes()));
let already_irreducible = self
.journal
.state()
.events
.iter()
.rev()
.find_map(|event| match &event.kind {
EventKind::Note { label, value } if label == "context_overflow_recovery" => {
Some(value)
}
_ => None,
})
.is_some_and(|value| {
value.get("irreducible").and_then(Value::as_bool) == Some(true)
&& value.get("limitTokens").and_then(Value::as_u64) == Some(target_tokens)
&& value.get("projectionHash").and_then(Value::as_str)
== Some(projection_hash.as_str())
});
if already_irreducible {
return Ok(ContextRecovery::Irreducible);
}
let before_tokens = projection.estimated_tokens;
let mut metadata = json!({
"transcriptRole":"system",
"contextOverflowWarning":true,
"projectedTokens":before_tokens,
"limitTokens":target_tokens,
});
if let Some(id) = external_event_id {
metadata["externalEventId"] = json!(id);
}
let warning_box_id = self.journal.create_box(
now(),
CONTEXT_OVERFLOW_WARNING_BOX_NAME,
BoxOwner::Controller,
BoxContent {
text: CONTEXT_OVERFLOW_WARNING.into(),
objects: Vec::new(),
metadata,
},
)?;
let mut transcript = json!({
"role":"system",
"content":CONTEXT_OVERFLOW_WARNING,
"contextOverflowWarning":true,
});
if let Some(id) = external_event_id {
transcript["externalEventId"] = json!(id);
}
self.transcript.push(transcript);
let mut pins = pinned_box_ids.to_vec();
if !pins.contains(&warning_box_id) {
pins.push(warning_box_id);
}
let outcome = kcode_history_ingress_context::recover(&mut self.journal, now(), &pins)?;
let (dehydrated_box_ids, estimated_tokens, target_tokens, irreducible) = match outcome {
ContextRecoveryOutcome::Recovered {
dehydrated_box_ids,
estimated_tokens,
target_tokens,
} => (dehydrated_box_ids, estimated_tokens, target_tokens, false),
ContextRecoveryOutcome::OverCapacity {
dehydrated_box_ids,
estimated_tokens,
target_tokens,
} => (dehydrated_box_ids, estimated_tokens, target_tokens, true),
};
let final_projection_hash = hex::encode(Sha256::digest(
self.journal.state().projection().render().as_bytes(),
));
self.journal.record(
now(),
EventKind::Note {
label: "context_overflow_recovery".into(),
value: json!({
"beforeTokens":before_tokens,
"estimatedTokens":estimated_tokens,
"limitTokens":target_tokens,
"dehydratedBoxIds":dehydrated_box_ids,
"irreducible":irreducible,
"projectionHash":final_projection_hash,
}),
},
)?;
if irreducible {
if matches!(self.mode, AgentMode::Ingress { .. }) {
self.request_ingress_force_commit(
"irreducible_context_overflow",
estimated_tokens,
)?;
} else if !self.journal.state().source_terminated {
self.journal.record(
now(),
EventKind::SourceTerminated {
reason: "irreducible_context_overflow".into(),
},
)?;
}
Ok(ContextRecovery::Irreducible)
} else {
Ok(ContextRecovery::Recovered)
}
}
fn request_ingress_force_commit(
&mut self,
reason: &str,
projected_tokens: u64,
) -> anyhow::Result<()> {
if self.ingress_force_commit_requested() {
return Ok(());
}
self.journal.record(
now(),
EventKind::Note {
label: INGRESS_FORCE_COMMIT_NOTE.into(),
value: json!({
"reason":reason,
"projectedTokens":projected_tokens,
"limitTokens":self.journal.state().ingress_context_limit(),
}),
},
)?;
Ok(())
}
fn ingress_force_commit_requested(&self) -> bool {
self.journal.state().events.iter().rev().any(|event| {
matches!(
&event.kind,
EventKind::Note { label, .. } if label == INGRESS_FORCE_COMMIT_NOTE
)
})
}
pub fn requires_history_ingress(&self) -> bool {
matches!(self.mode, AgentMode::Conversation) && self.journal.state().source_terminated
}
pub fn stage_free_time_opening(&mut self) -> bool {
if self.pending_turn {
return false;
}
let mut blocks = vec![free_time_opening(&self.free_time)];
if let Some(message) = self
.free_time
.get("handoffMessage")
.and_then(Value::as_str)
.filter(|message| !message.trim().is_empty())
{
blocks.push(format!(
"Message from the previous self-time session:\n\n{message}"
));
}
let Some(stage) = self.stage_user_input(&blocks.join("\n\n"), &json!({"kind":"self-time"}))
else {
return false;
};
self.pending_turn = matches!(stage, InputStage::Accepted);
true
}
pub fn stage_wakeup_opening(&mut self) -> anyhow::Result<bool> {
if self.pending_turn {
return Ok(false);
}
let marker = self
.channel
.get("wakeupMarker")
.and_then(Value::as_str)
.context("wakeup session is missing its acquired time marker")?;
let marker = DateTime::parse_from_rfc3339(marker)
.context("wakeup session has an invalid acquired time marker")?
.with_timezone(&Utc);
let text = wakeup_opening(marker);
let Some(stage) = self.stage_user_input(
&text,
&json!({"kind":"wakeup","wakeupMarker":marker.to_rfc3339()}),
) else {
return Ok(false);
};
self.pending_turn = matches!(stage, InputStage::Accepted);
Ok(true)
}
pub fn begin_user_turn(&mut self, text: &str, metadata: &Value) -> bool {
if self.pending_turn {
return false;
}
let Some(stage) = self.stage_user_input(text, metadata) else {
return false;
};
debug_assert_eq!(stage, InputStage::Accepted);
self.rounds_used = 0;
self.pending_turn = true;
self.pending_external_event_id = metadata
.get("externalEventId")
.and_then(Value::as_str)
.map(str::to_owned);
true
}
pub fn reset_exhausted_turn_rounds_for_retry(&mut self) {
if matches!(self.mode, AgentMode::Conversation)
&& self.rounds_used >= AGENT_LOOP_ROUND_LIMIT
{
self.rounds_used = 0;
}
}
pub fn interrupt_current_turn(&mut self) -> anyhow::Result<()> {
self.journal.repair_unfinished_tools(now())?;
let notice = "The user stopped this agent turn.";
let mut metadata = json!({"transcriptRole":"system","userStopped":true});
let mut transcript_entry = json!({
"role":"system",
"content":notice,
"userStopped":true,
});
if let Some(external_event_id) = &self.pending_external_event_id {
metadata["externalEventId"] = json!(external_event_id);
transcript_entry["externalEventId"] = json!(external_event_id);
}
self.journal.create_box(
now(),
"Turn stopped",
BoxOwner::Controller,
BoxContent {
text: notice.into(),
objects: Vec::new(),
metadata,
},
)?;
self.transcript.push(transcript_entry);
self.pending_turn = false;
self.pending_external_event_id = None;
self.orchestration =
json!({"owner":"backend","status":"idle","lastOutcome":"user-stopped"});
Ok(())
}
pub async fn run_pending_turn<C, F>(
&mut self,
operation_id: Uuid,
mut checkpoint: C,
) -> anyhow::Result<Option<String>>
where
C: FnMut(Value) -> F + Send,
F: Future<Output = anyhow::Result<()>> + Send,
{
if let Some(error) = self.fatal_persistence_error.take() {
anyhow::bail!("session journal write failed: {error}");
}
if !self.pending_turn {
return Ok(None);
}
let runtime = self.api.agent_runtime();
let user_id = self
.root_node_ids
.first()
.context("session has no user root for intelligence accounting")?
.clone();
let request = kcode_agent_runtime::SessionRunRequest {
user_id,
operation_id,
rounds_used: self.rounds_used,
round_limit: AGENT_LOOP_ROUND_LIMIT,
};
let mut host = KennedySessionHost {
session: self,
checkpoint: &mut checkpoint,
accounting: None,
pending_freeform_write: None,
deadline_after_response: false,
};
let result = runtime.run_session(request, &mut host).await;
let round_limit = result
.as_ref()
.is_err_and(kcode_agent_runtime::is_session_round_limit);
let result = if round_limit && matches!(host.session.mode, AgentMode::Ingress { .. }) {
host.session.request_ingress_force_commit(
"agent_loop_round_limit",
host.session.journal.state().projection().estimated_tokens,
)?;
(host.checkpoint)(host.session.snapshot()?).await?;
None
} else {
result?
};
drop(host);
match self.mode {
AgentMode::Conversation => {
if self.journal.state().source_terminated {
self.pending_turn = false;
self.pending_external_event_id = None;
checkpoint(self.snapshot()?).await?;
return Ok(None);
}
let Some(answer) = result else {
if self
.pending_external_event_id
.as_deref()
.and_then(|id| self.answer_for_external_event(id))
.is_some()
{
self.pending_turn = false;
self.pending_external_event_id = None;
checkpoint(self.snapshot()?).await?;
return Ok(None);
}
anyhow::bail!(
"Kennedy ended a conversational turn without an assistant response"
);
};
self.pending_turn = false;
self.pending_external_event_id = None;
checkpoint(self.snapshot()?).await?;
Ok(Some(answer))
}
AgentMode::FreeTime | AgentMode::Wakeup | AgentMode::Ingress { .. } => {
self.pending_turn = false;
self.pending_external_event_id = None;
if matches!(
self.mode,
AgentMode::FreeTime | AgentMode::Wakeup | AgentMode::Ingress { .. }
) {
self.finalize_kweb_session()?;
self.completed = true;
}
checkpoint(self.snapshot()?).await?;
Ok(None)
}
}
}
fn project_descendant<T>(
&mut self,
outcome: Result<kcode_intelligence_router::Accounted<T>, services::ApiError>,
) -> anyhow::Result<T> {
match outcome {
Ok(accounted) => {
kcode_intelligence_chatend::record_descendant_receipt(
&mut self.journal,
&accounted.receipt,
)?;
Ok(accounted.value)
}
Err(error) => {
if let Some(receipt) = &error.receipt {
kcode_intelligence_chatend::record_descendant_receipt(
&mut self.journal,
receipt,
)?;
}
Err(error.into())
}
}
}
async fn run_subagent(
&mut self,
arguments: &Value,
parent_operation_id: Uuid,
) -> anyhow::Result<String> {
validate_arguments(
arguments,
&["model", "contextNodeIds", "task"],
&["reasoningEffort"],
)?;
let model = nonempty_string(arguments, "model", 128)?;
let reasoning_effort = arguments
.get("reasoningEffort")
.map(|_| nonempty_string(arguments, "reasoningEffort", 32))
.transpose()?
.unwrap_or_else(|| self.runtime.reasoning_effort.clone());
let task = bounded_nonempty_string(arguments, "task", 100_000)?;
let context_node_ids =
canonical_node_id_array(arguments, "contextNodeIds", SUBAGENT_CONTEXT_NODE_LIMIT)?;
let mut context = Vec::with_capacity(context_node_ids.len());
for node_id in &context_node_ids {
context.push(self.api.kmap_node(node_id)?.data.long_description);
}
let user_id = self
.root_node_ids
.first()
.context("session has no user root for subagent intelligence accounting")?
.clone();
let timeout = self.agent_request_timeout();
let runtime = self.api.agent_runtime();
let first_event = self.journal.state().events.len();
let cost_before = self.journal.state().projection().status;
let result = {
let mut host = KennedySubagentHost {
session: self,
captures: HashMap::new(),
};
runtime
.run(
kcode_agent_runtime::RunRequest {
user_id,
parent_operation_id,
model,
reasoning_effort,
context,
task,
timeout,
start_metadata: json!({"contextNodeIds":context_node_ids}),
},
&mut host,
)
.await
};
match result {
Ok(result) => {
let cost_after = self.journal.state().projection().status;
Ok(format!(
"{}\n\n[{}]",
result.answer,
cost_summary(
"subagent cost",
cost_after
.estimated_cost_usd_nanos
.saturating_sub(cost_before.estimated_cost_usd_nanos),
cost_after
.unpriced_provider_calls
.saturating_sub(cost_before.unpriced_provider_calls),
)
))
}
Err(error) => {
let may_have_effects =
self.journal.state().events[first_event..]
.iter()
.any(|event| {
matches!(
&event.kind,
EventKind::Note { label, .. } if label == "subagent_tool_call"
)
});
if may_have_effects {
Err(error.context(
"the subagent failed after making Ktool calls; some tool effects may already have occurred",
))
} else {
Err(error)
}
}
}
}
async fn complete_subagent_freeform_write(
&mut self,
request: FreeformWrite,
contents: String,
budget: &kcode_agent_runtime::ContextBudget,
) -> anyhow::Result<(ToolOutcome, Vec<ChangedSubagentState>)> {
let kind = request.kind();
let freeform_tool = request.write_tool();
let backend_arguments = request.capture_subagent(&mut self.journal, &now(), contents)?;
let preview = self
.api
.managed_source_execute(
&self.rust_lib_session_id,
request.preview_tool(),
backend_arguments.clone(),
Vec::new(),
)
.await?;
let preview = preview
.snapshot
.context("subagent freeform write preview omitted its source snapshot")?;
let source_box_id = request.source_box_id(&self.journal)?;
anyhow::ensure!(
budget.fits_state(
format!("tool-state:{source_box_id}"),
format!(
"Current Managed {} {}:\n{}",
kind.label(),
preview.name,
preview.text
),
),
"{freeform_tool} was not run because its resulting source state would exceed the subagent context limit"
);
let previous = canonical_box_versions(&self.journal);
let execution = self
.api
.managed_source_execute(
&self.rust_lib_session_id,
freeform_tool,
backend_arguments,
Vec::new(),
)
.await?;
let snapshot = execution
.snapshot
.context("subagent freeform write omitted its resulting source snapshot")?;
apply_snapshot(&mut self.journal, &now(), snapshot)?;
let states = changed_subagent_tool_states(&self.journal, &previous);
Ok((
ToolOutcome {
text: execution.text,
store_result: false,
ok: true,
end_session: false,
freeform_write: None,
managed_source_snapshot: None,
},
states,
))
}
async fn complete_freeform_write(
&mut self,
pending: PendingFreeformWrite,
contents: String,
) -> anyhow::Result<ToolOutcome> {
let request = pending.request;
let freeform_tool = request.write_tool();
let backend_arguments =
request.capture(&mut self.journal, &now(), pending.call_box_id, contents)?;
let preview_result = self
.api
.managed_source_execute(
&self.rust_lib_session_id,
request.preview_tool(),
backend_arguments.clone(),
Vec::new(),
)
.await;
let preview = match preview_result {
Ok(preview) => preview,
Err(error) => {
return Ok(ToolOutcome {
text: format!("{freeform_tool} failed: {error}"),
store_result: true,
ok: false,
end_session: false,
freeform_write: None,
managed_source_snapshot: None,
});
}
};
let _preview = preview
.snapshot
.context("freeform write preview omitted the resulting source snapshot")?;
request.source_box_id(&self.journal)?;
let execution_result = self
.api
.managed_source_execute(
&self.rust_lib_session_id,
freeform_tool,
backend_arguments,
Vec::new(),
)
.await;
let execution = match execution_result {
Ok(execution) => execution,
Err(error) => {
return Ok(ToolOutcome {
text: format!("{freeform_tool} failed: {error}"),
store_result: true,
ok: false,
end_session: false,
freeform_write: None,
managed_source_snapshot: None,
});
}
};
let snapshot = execution
.snapshot
.context("freeform write omitted the resulting source snapshot")?;
apply_snapshot(&mut self.journal, &now(), snapshot)?;
Ok(ToolOutcome {
text: execution.text,
store_result: false,
ok: true,
end_session: false,
freeform_write: None,
managed_source_snapshot: None,
})
}
async fn send_telegram_dm(&mut self, arguments: &Value) -> anyhow::Result<String> {
let request = kcode_telegram_session_coordinator::parse_private_request(arguments)?;
let attachments = self.telegram_delivery_attachments(request.attachments)?;
let caller_holds_user_lock = self.session_type == "telegram"
&& self.channel.get("telegramUserId").and_then(Value::as_i64)
== Some(request.telegram_user_id);
self.api
.telegram()
.send_private(kcode_telegram_session_coordinator::PrivateDelivery {
telegram_user_id: request.telegram_user_id,
message: request.message,
attachments,
caller_holds_user_lock,
})
.await
}
async fn send_telegram_group_message(&mut self, arguments: &Value) -> anyhow::Result<String> {
let request = kcode_telegram_session_coordinator::parse_group_request(arguments)?;
let attachments = self.telegram_delivery_attachments(request.attachments)?;
self.api
.telegram()
.send_group(kcode_telegram_session_coordinator::GroupDelivery {
root_node_id: request.root_node_id,
message: request.message,
attachments,
})
.await
}
fn telegram_delivery_attachments(
&mut self,
requests: Vec<kcode_telegram_session_coordinator::AttachmentRequest>,
) -> anyhow::Result<Vec<kcode_telegram_session_coordinator::Attachment>> {
requests
.into_iter()
.map(|request| {
let object = self.resolve_object(&request.object_id)?;
let file_name = request
.file_name
.unwrap_or_else(|| object.file_name.clone());
Ok(kcode_telegram_session_coordinator::Attachment {
object_id: object.object_id,
bytes: object.bytes,
file_name,
media_type: object.media_type,
transport_kind: object.transport_kind,
})
})
.collect()
}
async fn execute_tool(
&mut self,
call: &ToolCall,
operation_id: Uuid,
) -> anyhow::Result<ToolOutcome> {
self.assert_tool_allowed(&call.name)?;
let mut end_session = false;
let mut store_result = true;
let mut freeform_write = None;
let mut managed_source_snapshot = None;
let text = match call.name.as_str() {
"SendTelegramDM" => self.send_telegram_dm(&call.arguments).await?,
"SendTelegramGroupMessage" => self.send_telegram_group_message(&call.arguments).await?,
"RunSubagent" => {
store_result = true;
let first_event = self.journal.state().events.len();
match self.run_subagent(&call.arguments, operation_id).await {
Ok(response) => response,
Err(error) => {
let may_have_effects = self.journal.state().events[first_event..]
.iter()
.any(|event| {
matches!(
&event.kind,
EventKind::Note { label, .. }
if label == "subagent_tool_call"
)
});
if may_have_effects {
return Err(error.context(
"the subagent failed after making Ktool calls; some tool effects may already have occurred",
));
}
return Err(error);
}
}
}
"EndSession" => {
validate_arguments(&call.arguments, &[], &["message"])?;
anyhow::ensure!(
!matches!(self.mode, AgentMode::Conversation),
"EndSession is only available during an autonomous or history-ingress session"
);
end_session = true;
if matches!(self.mode, AgentMode::FreeTime)
&& let Some(message) = call
.arguments
.get("message")
.and_then(Value::as_str)
.filter(|message| !message.trim().is_empty())
{
self.free_time["nextSessionMessage"] = json!(message);
}
"Session ending.".into()
}
"DehydrateBoxes" => {
validate_arguments(&call.arguments, &["boxIds"], &[])?;
let ids = box_id_array(&call.arguments, "boxIds")?;
self.journal.dehydrate_boxes(now(), &ids)?;
format!(
"Dehydrated boxes {}.",
ids.iter()
.map(ToString::to_string)
.collect::<Vec<_>>()
.join(", ")
)
}
"SummarizeBox" => {
validate_arguments(&call.arguments, &["boxId", "summary"], &[])?;
let id = box_id(&call.arguments, "boxId")?;
let summary = nonempty_string(&call.arguments, "summary", 1_000_000)?;
self.journal.summarize_box(now(), id, summary)?;
format!("Summarized box {id}.")
}
"HydrateBox" => {
validate_arguments(&call.arguments, &["boxId"], &[])?;
let id = box_id(&call.arguments, "boxId")?;
self.journal.rehydrate_box(now(), id)?;
let external_event_id = self.pending_external_event_id.clone();
match self.recover_context_overflow(external_event_id.as_deref(), &[id])? {
ContextRecovery::NotNeeded => format!("Hydrated box {id}."),
ContextRecovery::Recovered => {
format!("Hydrated box {id}.\n\n{CONTEXT_OVERFLOW_WARNING}")
}
ContextRecovery::Irreducible => anyhow::bail!(CONTEXT_OVERFLOW_WARNING),
}
}
"BoxesIntoObjects" => {
validate_arguments(&call.arguments, &["boxIds"], &[])?;
let ids = box_id_array(&call.arguments, "boxIds")?;
let objects = stage_box_text_objects(&mut self.journal, &ids, &now())?;
render_box_text_objects(&objects)
}
"LoadNodes" => {
validate_arguments(&call.arguments, &["identifiers"], &[])?;
let identifiers = canonical_node_id_list(&call.arguments, "identifiers")?;
load_durable_batch(&self.api, &mut self.context, &identifiers)?;
let changed = self.sync_kweb_boxes()?;
store_result = false;
render_load_nodes_result(&self.journal, &changed)?
}
"EmitObject" => {
validate_arguments(&call.arguments, &["objectId"], &["fileName"])?;
anyhow::ensure!(
matches!(self.mode, AgentMode::Conversation),
"EmitObject is only available in a conversation"
);
let object_id = nonempty_string(&call.arguments, "objectId", 64)?;
let object = self.resolve_object(&object_id)?;
let file_name = optional_delivery_file_name(&call.arguments, "fileName")?
.unwrap_or_else(|| object.file_name.clone());
if let Some(maximum) = self.channel.get("maxObjectBytes").and_then(Value::as_u64) {
anyhow::ensure!(
!object.bytes.is_empty(),
"object {object_id} is empty and cannot be sent through this channel"
);
anyhow::ensure!(
object.bytes.len() as u64 <= maximum,
"object {object_id} is {} bytes, over this channel's {maximum}-byte limit",
object.bytes.len()
);
}
let descriptor = json!({
"objectId":object_id,
"fileName":file_name,
"mediaType":object.media_type,
"byteLength":object.bytes.len(),
});
let mut metadata = json!({
"outputKind":"object",
"attachments":[descriptor.clone()],
});
if let Some(external_event_id) = &self.pending_external_event_id {
metadata["externalEventId"] = json!(external_event_id);
}
let content = BoxContent {
text: String::new(),
objects: vec![object_id.clone()],
metadata,
};
self.journal
.create_box(now(), "Kennedy message", BoxOwner::Kennedy, content)?;
let mut transcript = json!({
"role":"kennedy",
"content":"",
"objects":[object_id],
"attachments":[descriptor],
});
if let Some(external_event_id) = &self.pending_external_event_id {
transcript["externalEventId"] = json!(external_event_id);
}
self.transcript.push(transcript);
store_result = false;
"Object emitted to the user.".into()
}
"WebSearch" => {
validate_arguments(&call.arguments, &["question", "model"], &[])?;
let model = nonempty_string(&call.arguments, "model", 128)?;
let user_id = self
.root_node_ids
.first()
.context("session has no user root for intelligence accounting")?
.clone();
let outcome = self
.api
.search(
&user_id,
kcode_intelligence_router::SearchRequest {
question: nonempty_string(&call.arguments, "question", 4_000)?,
model,
operation_id: Uuid::new_v4(),
parent_operation_id: Some(operation_id),
},
)
.await;
let result = self.project_descendant(outcome)?;
render_web_search_result(&result)
}
"WebFetch" => {
validate_arguments(&call.arguments, &["url"], &[])?;
let user_id = self
.root_node_ids
.first()
.context("session has no user root for intelligence accounting")?;
let result = self
.api
.fetch(
user_id,
kcode_intelligence_router::FetchRequest {
url: nonempty_string(&call.arguments, "url", 4_096)?,
operation_id: Uuid::new_v4(),
parent_operation_id: Some(operation_id),
},
)
.await?;
render_web_fetch_result(&result)
}
"StageTelegramGroupMedia" => {
validate_arguments(&call.arguments, &["messageId"], &[])?;
let message_id = positive_integer(&call.arguments, "messageId")?;
let message_id = i64::try_from(message_id)
.context("messageId exceeds Telegram's supported integer range")?;
let media_ref = kcode_telegram_session_coordinator::group_media_reference(
&self.group_context,
message_id,
)?;
let chat_id = media_ref.chat_id;
if let Some((pending_id, metadata, size_bytes)) =
staged_telegram_group_media(&self.journal, chat_id, message_id)
{
render_staged_telegram_group_media(
&pending_id,
&metadata,
size_bytes,
message_id,
true,
)?
} else {
let (bytes, downloaded_media_type) = self
.api
.telegram()
.group_message_media(chat_id, message_id)?;
anyhow::ensure!(
!bytes.is_empty(),
"Telegram group media message {message_id} is empty"
);
anyhow::ensure!(
bytes.len() as u64 <= MAX_MEDIA_ENRICHMENT_BYTES,
"Telegram group media message {message_id} is {} bytes, over the {}-byte enrichment limit",
bytes.len(),
MAX_MEDIA_ENRICHMENT_BYTES
);
let media_type = normalized_media_type(&downloaded_media_type);
let file_name = kcode_telegram_session_coordinator::group_media_file_name(
&media_ref,
&media_type,
);
let pending_id = self.journal.stage_object(
now(),
media_type.clone(),
Some(file_name),
media_ref.transport_metadata(),
&bytes,
)?;
let metadata = self
.journal
.objects()
.get(&pending_id)
.context("newly staged Telegram group media is missing")?
.metadata
.clone();
render_staged_telegram_group_media(
&pending_id,
&metadata,
bytes.len() as u64,
message_id,
false,
)?
}
}
"TranscribeAudio" => {
validate_arguments(&call.arguments, &["objectId", "model", "prompt"], &[])?;
let model = nonempty_string(&call.arguments, "model", 128)?;
let prompt = bounded_nonempty_string(&call.arguments, "prompt", 4_000)?;
let object_id = nonempty_string(&call.arguments, "objectId", 64)?;
let object = self.resolve_media_object(&object_id)?;
validate_transcribable_audio(&object.media_type)?;
validate_transcription_model(&model)?;
let user_id = self
.root_node_ids
.first()
.context("session has no user root for intelligence accounting")?
.clone();
let outcome = self
.api
.transcribe_audio(
&user_id,
&model,
&prompt,
object.bytes,
object.file_name.clone(),
&object.media_type,
operation_id,
)
.await;
let result = self.project_descendant(outcome)?;
render_audio_transcription_result(
&object.object_id,
&object.file_name,
&object.media_type,
&result,
)?
}
"AnnotateMedia" => {
validate_arguments(&call.arguments, &["objectId", "model", "prompt"], &[])?;
let model = nonempty_string(&call.arguments, "model", 128)?;
let prompt = bounded_nonempty_string(&call.arguments, "prompt", 4_000)?;
let object_id = nonempty_string(&call.arguments, "objectId", 64)?;
let media = self.resolve_media_object(&object_id)?;
validate_annotation_media(&model, &media.media_type)?;
let user_id = self
.root_node_ids
.first()
.context("session has no user root for intelligence accounting")?
.clone();
let outcome = self
.api
.annotate_media(
&user_id,
&model,
&prompt,
media.bytes,
media.file_name.clone(),
&media.media_type,
operation_id,
)
.await;
let result = self.project_descendant(outcome)?;
render_media_annotation_result(
&media.object_id,
&media.file_name,
&media.media_type,
&result,
)?
}
"GenerateImage" => {
validate_arguments(
&call.arguments,
&["model", "prompt"],
&["referenceObjectIds"],
)?;
let model = nonempty_string(&call.arguments, "model", 128)?;
validate_image_model(&model)?;
let prompt = bounded_nonempty_string(&call.arguments, "prompt", 100_000)?;
let reference_ids =
optional_object_id_array(&call.arguments, "referenceObjectIds", 14)?;
let mut references = Vec::with_capacity(reference_ids.len());
for object_id in &reference_ids {
references.push(self.resolve_image_object(object_id)?);
}
let user_id = self
.root_node_ids
.first()
.context("session has no user root for intelligence accounting")?
.clone();
let outcome = self
.api
.generate_image(&user_id, &model, &prompt, references, operation_id)
.await;
let result = self.project_descendant(outcome)?;
let size = result.bytes.len();
let file_name =
format!("generated-image.{}", image_extension(&result.content_type));
let object_id = self.api.save_generated_image(
result.bytes,
&file_name,
&result.content_type,
&result.model,
)?;
format!(
"Generated image.\nObject: {object_id}\nFile: {file_name}\nContent type: {}\nSize: {size} bytes\nModel: {}\nUse EmitObject with {object_id} to deliver it.",
result.content_type, result.model
)
}
"ExtractDocumentText" => {
validate_arguments(&call.arguments, &["objectId"], &[])?;
let object_id = nonempty_string(&call.arguments, "objectId", 64)?;
let object = self.resolve_media_object(&object_id)?;
validate_extractable_document(&object.media_type, &object.file_name)?;
let result = self
.api
.extract_document(object.bytes, object.file_name.clone(), &object.media_type)
.await?;
render_document_extraction_result(&object.object_id, &object.file_name, &result)
}
name if SPEECH_CLASSIFICATION_TOOLS.contains(&name) => {
self.api
.execute_speech_classification_tool(name, call.arguments.clone())
.await?
}
"ConnectNodes" => self.connect_nodes(&call.arguments)?,
"ConsolidateFanout" => self.consolidate_fanout(&call.arguments)?,
"SetFixedConnection" => self.set_fixed_connection(&call.arguments)?,
"CreateNode" => self.create_node(&call.arguments)?,
"UpdateNode" => self.update_node(&call.arguments)?,
name if RUST_LIB_TOOLS.contains(&name)
|| WEB_LIB_TOOLS.contains(&name)
|| RUST_BIN_TOOLS.contains(&name) =>
{
if let Some(request) = prepare_freeform_write(&self.journal, name, &call.arguments)?
{
store_result = false;
let acknowledgement = request.acknowledgement();
freeform_write = Some(request);
acknowledgement
} else {
let mut objects = Vec::new();
if name == CALL_RUST_BIN_TOOL {
let object_ids = rust_binary_object_ids(&call.arguments)?;
objects.reserve(object_ids.len());
for object_id in object_ids {
objects.push(self.resolve_object(&object_id)?.bytes);
}
} else if name == ATTACH_OBJECT_WEB_LIB_TOOL {
let object_id = web_library_object_id(&call.arguments)?;
objects.push(self.resolve_object(&object_id)?.bytes);
}
let execution = self
.api
.managed_source_execute(
&self.rust_lib_session_id,
name,
call.arguments.clone(),
objects,
)
.await?;
if let Some(snapshot) = execution.snapshot {
managed_source_snapshot = Some(snapshot);
store_result = false;
}
execution.text
}
}
_ => anyhow::bail!("Tool {} is not available", call.name),
};
Ok(ToolOutcome {
text,
store_result,
ok: true,
end_session,
freeform_write,
managed_source_snapshot,
})
}
fn assert_tool_allowed(&self, name: &str) -> anyhow::Result<()> {
let write = matches!(
name,
"ConnectNodes"
| "ConsolidateFanout"
| "SetFixedConnection"
| "CreateNode"
| "UpdateNode"
);
anyhow::ensure!(
!write || !matches!(self.mode, AgentMode::Conversation),
"{name} requires the global Kweb write lane and is unavailable in a read-only conversation"
);
if name == "EndSession" {
anyhow::ensure!(
!matches!(self.mode, AgentMode::Conversation),
"EndSession is unavailable in a conversation"
);
}
Ok(())
}
fn sync_kweb_boxes(&mut self) -> anyhow::Result<Vec<BoxId>> {
let updates = self
.plan
.updates
.iter()
.map(|(id, node)| (id.clone(), kweb_node_draft(node)))
.collect::<BTreeMap<_, _>>();
let creates = self
.plan
.creates
.iter()
.map(|create| KwebStagedCreate {
pending_id: create.pending_id.clone(),
data: kweb_node_draft(&create.data),
})
.collect::<Vec<_>>();
self.context
.sync_chatend(&mut self.journal, now(), &updates, &creates)
.map_err(anyhow::Error::new)
}
fn record_tool_invocation(
&mut self,
name: &str,
arguments: Value,
) -> anyhow::Result<RecordedToolInvocation> {
let invocation = RecordedToolInvocation {
invocation_id: Uuid::new_v4().to_string(),
tool_instance: tool_instance(name),
tool_name: name.into(),
};
self.journal.record(
now(),
EventKind::ToolInvoked {
tool_instance: invocation.tool_instance.clone(),
tool_name: invocation.tool_name.clone(),
arguments,
invocation_id: Some(invocation.invocation_id.clone()),
},
)?;
Ok(invocation)
}
fn record_tool_completion(
&mut self,
invocation: Option<&RecordedToolInvocation>,
outcome: Value,
) -> anyhow::Result<EventId> {
let (tool_instance, tool_name, invocation_id) = invocation
.map(|invocation| {
(
invocation.tool_instance.clone(),
invocation.tool_name.clone(),
Some(invocation.invocation_id.clone()),
)
})
.unwrap_or_else(|| ("call_ktool".into(), "call_ktool".into(), None));
self.journal.record(
now(),
EventKind::ToolCompleted {
tool_instance,
tool_name,
outcome,
invocation_id,
},
)
}
fn stage_plan(&mut self) -> anyhow::Result<()> {
self.sync_kweb_boxes()?;
Ok(())
}
fn node_data(&self, id: &str) -> anyhow::Result<PlannedNode> {
if let Some(data) = self.plan.created(id) {
return Ok(data.clone());
}
if let Some(data) = self.plan.updates.get(id) {
return Ok(data.clone());
}
let node = self
.context
.node(id)
.with_context(|| format!("Kweb context does not contain node {id}"))?;
Ok(planned_node(node))
}
fn put_node_data(&mut self, id: &str, data: PlannedNode) -> anyhow::Result<()> {
if let Some(created) = self.plan.created_mut(id) {
*created = data;
} else {
canonical_id(id)?;
self.plan.updates.insert(id.to_owned(), data);
}
Ok(())
}
fn connect_nodes(&mut self, args: &Value) -> anyhow::Result<String> {
validate_arguments(args, &["identifiers"], &[])?;
let ids = resource_id_array(args, "identifiers", 2)?;
for id in &ids {
self.ensure_known_node(id)?;
}
for id in &ids {
let mut data = self.node_data(id)?;
let mut recent = ids
.iter()
.filter(|other| *other != id)
.cloned()
.collect::<Vec<_>>();
for other in data.recent_connections {
if &other != id && !recent.contains(&other) {
recent.push(other);
}
}
data.recent_connections = recent;
self.put_node_data(id, data)?;
}
self.stage_plan()?;
Ok(format!(
"Staged connections among nodes {}.",
ids.join(", ")
))
}
fn consolidate_fanout(&mut self, args: &Value) -> anyhow::Result<String> {
validate_arguments(
args,
&[
"parentIdentifier",
"fanoutIdentifiers",
"aggregatorIdentifier",
],
&[],
)?;
let parent = resource_id(args, "parentIdentifier")?;
let aggregator = resource_id(args, "aggregatorIdentifier")?;
let fanout = resource_id_array(args, "fanoutIdentifiers", 1)?;
for id in std::iter::once(&parent)
.chain(std::iter::once(&aggregator))
.chain(fanout.iter())
{
self.ensure_known_node(id)?;
}
let mut parent_data = self.node_data(&parent)?;
parent_data
.recent_connections
.retain(|id| !fanout.contains(id));
if !parent_data.recent_connections.contains(&aggregator) {
parent_data.recent_connections.push(aggregator.clone());
}
let mut aggregator_data = self.node_data(&aggregator)?;
for id in fanout {
if !aggregator_data.recent_connections.contains(&id) {
aggregator_data.recent_connections.push(id);
}
}
self.put_node_data(&parent, parent_data)?;
self.put_node_data(&aggregator, aggregator_data)?;
self.stage_plan()?;
Ok(format!(
"Staged fanout consolidation from node {parent} into node {aggregator}."
))
}
fn set_fixed_connection(&mut self, args: &Value) -> anyhow::Result<String> {
validate_arguments(args, &["parentIdentifier", "childIdentifier", "slot"], &[])?;
let parent = resource_id(args, "parentIdentifier")?;
self.ensure_known_node(&parent)?;
let child = args
.get("childIdentifier")
.and_then(Value::as_str)
.filter(|value| *value != "blank")
.map(parse_resource_id)
.transpose()?;
if let Some(child) = &child {
self.ensure_known_node(child)?;
anyhow::ensure!(child != &parent, "a node cannot connect to itself");
}
let slot = positive_integer(args, "slot")? as usize;
let mut data = self.node_data(&parent)?;
if let Some(child) = child.clone() {
anyhow::ensure!(
slot <= data.fixed_connections.len() + 1,
"fixed connection positions must remain contiguous"
);
data.fixed_connections.retain(|id| id != &child);
if slot - 1 < data.fixed_connections.len() {
data.fixed_connections[slot - 1] = child;
} else {
data.fixed_connections.push(child);
}
} else if slot > 0 && slot - 1 < data.fixed_connections.len() {
data.fixed_connections.remove(slot - 1);
}
self.put_node_data(&parent, data)?;
self.stage_plan()?;
Ok(match child {
Some(child) => {
format!("Staged node {child} in fixed slot {slot} of node {parent}.")
}
None => format!("Cleared fixed slot {slot} of node {parent} in the staged plan."),
})
}
fn create_node(&mut self, args: &Value) -> anyhow::Result<String> {
validate_arguments(
args,
&[
"parentIdentifiers",
"ownerIdentifier",
"shortName",
"shortDescription",
"longDescription",
],
&[],
)?;
let (short_name, short_description, long_description) =
node_text_arguments(args, "shortName", "shortDescription", "longDescription")?;
let parents = resource_id_array(args, "parentIdentifiers", 1)?;
let owner = resource_id(args, "ownerIdentifier")?;
for id in parents.iter().chain(std::iter::once(&owner)) {
if id != "self" && id != "unowned" {
self.ensure_known_node(id)?;
}
}
let pending = self.journal.allocate_pending_node(now())?.to_string();
self.plan.creates.push(StagedNodeCreate {
pending_id: pending.clone(),
data: PlannedNode {
short_name,
short_description,
long_description,
owner,
fixed_connections: Vec::new(),
recent_connections: parents.clone(),
objects: Vec::new(),
attach_session_archive: true,
},
});
for parent in parents {
let mut data = self.node_data(&parent)?;
data.recent_connections.retain(|id| id != &pending);
data.recent_connections.insert(0, pending.clone());
self.put_node_data(&parent, data)?;
}
self.stage_plan()?;
Ok(format!("Created staged node {pending}."))
}
fn update_node(&mut self, args: &Value) -> anyhow::Result<String> {
validate_arguments(
args,
&[
"identifier",
"ownerIdentifier",
"newShortName",
"newShortDescription",
"newLongDescription",
],
&[],
)?;
let (short_name, short_description, long_description) = node_text_arguments(
args,
"newShortName",
"newShortDescription",
"newLongDescription",
)?;
let id = resource_id(args, "identifier")?;
let owner = resource_id(args, "ownerIdentifier")?;
self.ensure_known_node(&id)?;
if owner != "self" && owner != "unowned" {
self.ensure_known_node(&owner)?;
}
let mut data = self.node_data(&id)?;
data.owner = owner;
data.short_name = short_name;
data.short_description = short_description;
data.long_description = long_description;
data.attach_session_archive = true;
self.put_node_data(&id, data)?;
self.stage_plan()?;
Ok(format!("Staged the update to node {id}."))
}
fn ensure_known_node(&self, id: &str) -> anyhow::Result<()> {
if id.starts_with("pending:") {
anyhow::ensure!(
self.plan.created(id).is_some(),
"pending node {id} is not part of this session"
);
} else {
canonical_id(id)?;
anyhow::ensure!(
self.context.contains_full_node(id) || self.plan.updates.contains_key(id),
"node {id} is not loaded; call LoadNodes first"
);
}
Ok(())
}
fn finalize_kweb_session(&mut self) -> anyhow::Result<()> {
if self.commit_receipt.is_some() {
return Ok(());
}
self.journal.repair_unfinished_tools(now())?;
self.journal.seal()?;
let archive = self.journal.archive_bytes()?;
let object_locations = self
.journal
.objects()
.iter()
.map(|(id, location)| (id.clone(), location.clone()))
.collect::<Vec<_>>();
let mut objects = BTreeMap::new();
for (id, location) in object_locations {
let pending_id = id.to_string();
let bytes = encode_file(
&pending_id,
location.metadata.file_name.as_deref(),
&location.metadata.media_type,
staged_object_transport_kind(&self.journal, &id).as_deref(),
self.journal.read_object(&id)?,
)
.with_context(|| format!("encoding staged object {pending_id}"))?;
anyhow::ensure!(
objects.insert(pending_id.clone(), bytes).is_none(),
"duplicate staged object {pending_id}"
);
}
let mut creates = BTreeMap::new();
for create in &self.plan.creates {
anyhow::ensure!(
creates
.insert(create.pending_id.clone(), create.data.clone())
.is_none(),
"duplicate staged node {}",
create.pending_id
);
}
let updates = self
.plan
.updates
.iter()
.map(|(node_id, data)| {
node_id
.parse::<NodeId>()
.with_context(|| format!("{node_id:?} is not a canonical node ID"))
.map(|node_id| (node_id, data.clone()))
})
.collect::<anyhow::Result<BTreeMap<_, _>>>()?;
let result = self.api.commit_kweb_session(CommitRequest {
idempotency_key: self.journal.state().metadata.session_id.clone(),
author: self.commit_author.clone(),
source_created_at: DateTime::parse_from_rfc3339(&self.started_at)
.context("session start timestamp is invalid")?
.with_timezone(&Utc),
archive,
objects,
creates,
updates,
})?;
self.journal
.mark_completed(result.session_object_id.to_string());
self.commit_receipt = Some(result);
Ok(())
}
fn prepare_free_time_round(&mut self) -> anyhow::Result<bool> {
if !matches!(self.mode, AgentMode::FreeTime) {
return Ok(false);
}
let Some(deadline) = deadline(&self.free_time) else {
return Ok(false);
};
if Utc::now() >= deadline {
self.free_time_end_reason = Some("deadline".into());
self.journal.create_box(
now(),
"Self-time timer",
BoxOwner::Controller,
BoxContent::text(
"The self-time deadline has arrived. Finish without starting more tool work.",
),
)?;
return Ok(true);
}
Ok(false)
}
fn refresh_runtime_prompt(&mut self) -> anyhow::Result<()> {
let Some(system_box) = self
.journal
.state()
.boxes
.values()
.find(|state| matches!(state.owner, BoxOwner::System))
.map(|state| state.id)
else {
return Ok(());
};
let current = self
.journal
.state()
.boxes
.get(&system_box)
.context("system-prompt box disappeared")?
.canonical
.content
.text
.clone();
let marker = "\n\nCurrent runtime\n\n";
let Some((prefix, _)) = current.rsplit_once(marker) else {
return Ok(());
};
let refreshed = format!(
"{prefix}{marker}{}",
runtime_description(&self.runtime, Utc::now())
);
if refreshed != current {
self.journal
.update_box(now(), system_box, BoxContent::text(refreshed))?;
}
Ok(())
}
fn agent_request_timeout(&self) -> Option<Duration> {
if matches!(self.mode, AgentMode::Conversation) && self.session_type == "conversation" {
return Some(BROWSER_CONVERSATION_REQUEST_TIMEOUT);
}
if matches!(self.mode, AgentMode::Ingress { .. }) {
return Some(HISTORY_INGRESS_REQUEST_TIMEOUT);
}
if matches!(self.mode, AgentMode::Wakeup) {
return Some(WAKEUP_REQUEST_TIMEOUT);
}
if matches!(self.mode, AgentMode::FreeTime) {
let deadline = deadline(&self.free_time)?;
return Some(Duration::from_secs(
(deadline - Utc::now()).num_seconds().max(1) as u64 + 360,
));
}
None
}
pub fn refresh_telegram_group_context(
&mut self,
group_context: &Value,
current_message_id: Option<&str>,
) -> anyhow::Result<()> {
if self.session_type != "telegram-group" {
return Ok(());
}
self.channel["groupContext"] = group_context.clone();
self.group_context = group_context.clone();
self.journal.create_box(
now(),
"Telegram group update",
BoxOwner::Controller,
BoxContent::text(kcode_telegram_session_coordinator::format_group_context(
group_context,
)),
)?;
self.recover_context_overflow(current_message_id, &[])?;
Ok(())
}
pub fn finalize_free_time(&mut self, reason: &str) -> anyhow::Result<()> {
anyhow::ensure!(
matches!(reason, "tool" | "deadline" | "hard-stop" | "user-stop"),
"invalid self-time completion reason"
);
self.free_time["sliceEndedReason"] = json!(reason);
self.free_time["sliceEndedAt"] = json!(now());
self.pending_turn = false;
self.pending_external_event_id = None;
Ok(())
}
pub fn commit_current_write_session(&mut self) -> anyhow::Result<()> {
anyhow::ensure!(
matches!(
self.mode,
AgentMode::FreeTime | AgentMode::Wakeup | AgentMode::Ingress { .. }
),
"a read-only conversation cannot be committed as a Kweb write session"
);
self.finalize_kweb_session()?;
self.completed = true;
Ok(())
}
pub fn snapshot(&self) -> anyhow::Result<Value> {
let projection = self.journal.state().projection();
let chatend_text = projection.render();
let session_status = projection.status.clone();
Ok(json!({
"format":"kennedy-chatend",
"version":1,
"stateVersion":3,
"sessionId":self.journal.state().metadata.session_id,
"chatendMetadata":self.journal.state().metadata,
"sessionType":self.session_type,
"sourceSessionType":self.source_session_type,
"channel":self.channel,
"freeTime":self.free_time,
"orchestration":self.orchestration,
"provenanceId":self.provenance_id,
"rustLibSessionId":self.rust_lib_session_id,
"rootNodeIds":self.root_node_ids,
"referenceRootNodeIds":self.reference_root_node_ids,
"startedAt":self.started_at,
"transcript":self.transcript,
"pendingTurn":self.pending_turn,
"pendingExternalEventId":self.pending_external_event_id,
"roundsUsed":self.rounds_used,
"completed":self.completed,
"sessionObjectId":self.journal.state().completed_session_object,
"commitReceipt":self.commit_receipt,
"commitAuthor":self.commit_author,
"providerModel":self.runtime.model,
"kwebPlan":self.plan,
"boxCount":self.journal.state().boxes.len(),
"eventCount":self.journal.state().events.len(),
"boxes":self.journal.state().boxes,
"events":self.journal.state().events,
"context":projection,
"sessionStatus":session_status,
"chatendText":chatend_text,
}))
}
pub async fn release_managed_sources(&self) {
self.api
.release_managed_sources(&self.rust_lib_session_id)
.await;
}
}
impl<C, F> kcode_agent_runtime::SessionHost for KennedySessionHost<'_, C>
where
C: FnMut(Value) -> F + Send,
F: Future<Output = anyhow::Result<()>> + Send,
{
fn prepare_round<'a>(
&'a mut self,
round: u64,
) -> kcode_agent_runtime::HostFuture<'a, kcode_agent_runtime::RoundPreparation> {
Box::pin(async move {
self.session.rounds_used = round;
self.session.refresh_runtime_prompt()?;
self.deadline_after_response = self.session.prepare_free_time_round()?;
let external_event_id = self.session.pending_external_event_id.clone();
if self
.session
.recover_context_overflow(external_event_id.as_deref(), &[])?
== ContextRecovery::Irreducible
|| (matches!(self.session.mode, AgentMode::Ingress { .. })
&& self.session.ingress_force_commit_requested())
{
return Ok(kcode_agent_runtime::RoundPreparation::Complete(None));
}
let input = self.session.journal.state().projection().render();
Ok(kcode_agent_runtime::RoundPreparation::Run(
kcode_agent_runtime::PreparedRound {
input,
model: self.session.runtime.model.clone(),
reasoning_effort: self.session.runtime.reasoning_effort.clone(),
tool_description: call_ktool_description().into(),
timeout: self.session.agent_request_timeout(),
},
))
})
}
fn record<'a>(
&'a mut self,
event: kcode_agent_runtime::SessionEvent,
) -> kcode_agent_runtime::HostFuture<'a, ()> {
Box::pin(async move {
match event {
kcode_agent_runtime::SessionEvent::InferenceSubmitted {
manifest_hash,
model,
..
} => {
let projection = self.session.journal.state().projection();
self.accounting = Some(kcode_intelligence_chatend::TopLevelCall::new(
manifest_hash.clone(),
model,
));
self.session.journal.record(
now(),
EventKind::InferenceSubmitted {
manifest_hash,
estimated_input_tokens: projection.estimated_tokens,
raw_estimated_input_tokens: Some(projection.raw_estimated_tokens),
},
)?;
}
kcode_agent_runtime::SessionEvent::ProviderInput { input, .. } => {
self.session.journal.record(
now(),
EventKind::Note {
label: "provider_input".into(),
value: Value::String(input),
},
)?;
}
kcode_agent_runtime::SessionEvent::UsageUpdated { usage, .. } => {
self.accounting
.as_mut()
.context("provider usage arrived before inference submission")?
.usage_updated(&mut self.session.journal, &now(), &usage)?;
}
kcode_agent_runtime::SessionEvent::ProviderReceipt { usage, .. } => {
self.accounting
.take()
.context("provider receipt arrived before inference submission")?
.completed(&mut self.session.journal, &now(), usage.as_ref())?;
}
}
let snapshot = self.session.snapshot()?;
(self.checkpoint)(snapshot).await
})
}
fn execute_tool<'a>(
&'a mut self,
call: anyhow::Result<kcode_agent_runtime::ToolCall>,
operation_id: Uuid,
) -> kcode_agent_runtime::HostFuture<'a, kcode_agent_runtime::SessionToolOutcome> {
Box::pin(async move {
if let Some(pending) = &self.pending_freeform_write {
let text = format!(
"{} is awaiting the complete file contents; no other Ktool can run before that output.",
pending.request.write_tool()
);
self.session
.record_tool_completion(None, json!({"ok":false,"result":text}))?;
let text = provider_tool_result_with_context_footer(&self.session.journal, &text);
(self.checkpoint)(self.session.snapshot()?).await?;
return Ok(kcode_agent_runtime::SessionToolOutcome {
text,
ok: false,
capture: Some(json!(true)),
stop: false,
finish_after_round: false,
emitted_response: false,
});
}
let tool_started_at = std::time::Instant::now();
let mut created_call_box_id = None;
let mut recorded_invocation = None;
let transcript_start = self.session.transcript.len();
let mut emitted_response = false;
let mut outcome = match call {
Ok(call) => {
let call = ToolCall {
name: call.name,
arguments: call.arguments,
};
let call_name = format!("Kennedy tool call: {}", call.name);
let call_content = tool_call_box_content(&call)?;
recorded_invocation = Some(
self.session
.record_tool_invocation(&call.name, call.arguments.clone())?,
);
created_call_box_id = Some(self.session.journal.create_box(
now(),
call_name,
BoxOwner::Kennedy,
call_content,
)?);
let external_event_id = self.session.pending_external_event_id.clone();
if self
.session
.recover_context_overflow(external_event_id.as_deref(), &[])?
== ContextRecovery::Irreducible
{
ToolOutcome {
text: CONTEXT_OVERFLOW_WARNING.into(),
store_result: false,
ok: false,
end_session: false,
freeform_write: None,
managed_source_snapshot: None,
}
} else {
match self.session.execute_tool(&call, operation_id).await {
Ok(outcome) => {
emitted_response = call.name == "EmitObject" && outcome.ok;
outcome
}
Err(error) => ToolOutcome {
text: format!("{} failed: {error}", call.name),
store_result: call.name != "LoadNodes",
ok: false,
end_session: false,
freeform_write: None,
managed_source_snapshot: None,
},
}
}
}
Err(error) => ToolOutcome {
text: error.to_string(),
store_result: true,
ok: false,
end_session: false,
freeform_write: None,
managed_source_snapshot: None,
},
};
if let Some(snapshot) = outcome.managed_source_snapshot.take() {
apply_snapshot(&mut self.session.journal, &now(), snapshot)?;
outcome.store_result = false;
}
append_slow_tool_duration(&mut outcome.text, tool_started_at.elapsed());
let capture = if let Some(request) = outcome.freeform_write.take() {
self.pending_freeform_write = Some(PendingFreeformWrite {
request,
call_box_id: created_call_box_id
.context("freeform write call box was not created")?,
});
Some(json!(true))
} else {
None
};
if outcome.store_result {
self.session.journal.create_box(
now(),
"Kennedy tool result",
BoxOwner::Controller,
BoxContent::text(&outcome.text),
)?;
}
let external_event_id = self.session.pending_external_event_id.clone();
let recovery = self
.session
.recover_context_overflow(external_event_id.as_deref(), &[])?;
let context_warning_added =
self.session.transcript[transcript_start..]
.iter()
.any(|entry| {
entry.get("contextOverflowWarning").and_then(Value::as_bool) == Some(true)
});
let mut provider_text = outcome.text.clone();
if context_warning_added && !provider_text.contains(CONTEXT_OVERFLOW_WARNING) {
if !provider_text.is_empty() {
provider_text.push_str("\n\n");
}
provider_text.push_str(CONTEXT_OVERFLOW_WARNING);
}
let provider_text =
provider_tool_result_with_context_footer(&self.session.journal, &provider_text);
self.session.record_tool_completion(
recorded_invocation.as_ref(),
json!({"ok":outcome.ok,"result":outcome.text}),
)?;
let stop = recovery == ContextRecovery::Irreducible
|| (matches!(self.session.mode, AgentMode::Ingress { .. })
&& self.session.ingress_force_commit_requested())
|| (!matches!(self.session.mode, AgentMode::Ingress { .. })
&& self.session.journal.state().source_terminated);
let snapshot = self.session.snapshot()?;
(self.checkpoint)(snapshot).await?;
Ok(kcode_agent_runtime::SessionToolOutcome {
text: provider_text,
ok: outcome.ok,
capture,
stop,
finish_after_round: outcome.end_session,
emitted_response,
})
})
}
fn complete_capture<'a>(
&'a mut self,
_capture: Value,
contents: String,
) -> kcode_agent_runtime::HostFuture<'a, kcode_agent_runtime::SessionControl> {
Box::pin(async move {
let pending = self
.pending_freeform_write
.take()
.context("provider completed without a pending freeform write")?;
let result_metadata = pending.request.clone();
let outcome = self
.session
.complete_freeform_write(pending, contents)
.await?;
if outcome.store_result {
self.session.journal.create_box(
now(),
"Kennedy tool result",
BoxOwner::Controller,
BoxContent::text(&outcome.text),
)?;
}
self.session.journal.record(
now(),
EventKind::Note {
label: "write_file_freeform_result".into(),
value: result_metadata.result_record(outcome.ok, &outcome.text),
},
)?;
let external_event_id = self.session.pending_external_event_id.clone();
let recovery = self
.session
.recover_context_overflow(external_event_id.as_deref(), &[])?;
let snapshot = self.session.snapshot()?;
(self.checkpoint)(snapshot).await?;
if recovery == ContextRecovery::Irreducible
|| (matches!(self.session.mode, AgentMode::Ingress { .. })
&& self.session.ingress_force_commit_requested())
|| (!matches!(self.session.mode, AgentMode::Ingress { .. })
&& self.session.journal.state().source_terminated)
|| self.deadline_after_response
{
return Ok(kcode_agent_runtime::SessionControl::Complete(None));
}
self.session.journal.create_box(
now(),
controller_box_name(&self.session.mode),
BoxOwner::Controller,
BoxContent::text(controller_message(
&self.session.mode,
&self.session.free_time,
)),
)?;
Ok(kcode_agent_runtime::SessionControl::Continue)
})
}
fn complete_round<'a>(
&'a mut self,
completion: kcode_agent_runtime::RoundCompletion,
) -> kcode_agent_runtime::HostFuture<'a, kcode_agent_runtime::SessionControl> {
Box::pin(async move {
let answer = completion.answer.trim().to_owned();
let mut completion_recovery = ContextRecovery::NotNeeded;
if !answer.is_empty() {
let mut content = BoxContent::text(answer.clone());
if let Some(id) = &self.session.pending_external_event_id {
content.metadata["externalEventId"] = json!(id);
}
self.session.journal.create_box(
now(),
"Kennedy message",
BoxOwner::Kennedy,
content,
)?;
let mut transcript = json!({"role":"kennedy","content":answer});
if let Some(id) = &self.session.pending_external_event_id {
transcript["externalEventId"] = json!(id);
}
self.session.transcript.push(transcript);
let external_event_id = self.session.pending_external_event_id.clone();
completion_recovery = self
.session
.recover_context_overflow(external_event_id.as_deref(), &[])?;
}
let snapshot = self.session.snapshot()?;
(self.checkpoint)(snapshot).await?;
if completion_recovery == ContextRecovery::Irreducible
|| (matches!(self.session.mode, AgentMode::Ingress { .. })
&& self.session.ingress_force_commit_requested())
|| (!matches!(self.session.mode, AgentMode::Ingress { .. })
&& self.session.journal.state().source_terminated)
{
return Ok(kcode_agent_runtime::SessionControl::Complete(None));
}
if completion.finish_requested || self.deadline_after_response {
return Ok(kcode_agent_runtime::SessionControl::Complete(
(!answer.is_empty()).then_some(answer),
));
}
if matches!(self.session.mode, AgentMode::Conversation) && !answer.is_empty() {
return Ok(kcode_agent_runtime::SessionControl::Complete(Some(answer)));
}
if matches!(self.session.mode, AgentMode::Conversation) && completion.emitted_response {
return Ok(kcode_agent_runtime::SessionControl::Complete(None));
}
let solo_ingress_response =
matches!(self.session.mode, AgentMode::Ingress { .. }) && !answer.is_empty();
anyhow::ensure!(
completion.used_tool || solo_ingress_response,
"provider completed without a response or tool call"
);
self.session.journal.create_box(
now(),
controller_box_name(&self.session.mode),
BoxOwner::Controller,
BoxContent::text(controller_message(
&self.session.mode,
&self.session.free_time,
)),
)?;
Ok(kcode_agent_runtime::SessionControl::Continue)
})
}
}
impl kcode_agent_runtime::Host for KennedySubagentHost<'_> {
fn render_tool_call(&mut self, call: &kcode_agent_runtime::ToolCall) -> anyhow::Result<String> {
Ok(tool_call_box_content(&ToolCall {
name: call.name.clone(),
arguments: call.arguments.clone(),
})?
.text)
}
fn execute_tool<'a>(
&'a mut self,
call: kcode_agent_runtime::ToolCall,
operation_id: Uuid,
budget: kcode_agent_runtime::ContextBudget,
) -> kcode_agent_runtime::HostFuture<'a, kcode_agent_runtime::ToolOutcome> {
Box::pin(async move {
let call = ToolCall {
name: call.name,
arguments: call.arguments,
};
if call.name == "RunSubagent" {
return Ok(kcode_agent_runtime::ToolOutcome::failure(
"RunSubagent is unavailable inside a subagent. Only Kennedy may launch subagents.",
));
}
if matches!(
call.name.as_str(),
"SendTelegramDM" | "SendTelegramGroupMessage"
) {
return Ok(kcode_agent_runtime::ToolOutcome::failure(
"Telegram cold delivery is unavailable inside a subagent. Only Kennedy may initiate a Telegram message.",
));
}
if budget.estimated_tokens() > budget.max_input_tokens() {
return Ok(kcode_agent_runtime::ToolOutcome::failure(
"The Ktool call was not run because its retained invocation would exceed the subagent context limit.",
));
}
if !subagent_managed_write_fits(&self.session.journal, &call, &budget) {
return Ok(kcode_agent_runtime::ToolOutcome::failure(
"The managed-source write was not run because its resulting current state would exceed the subagent context limit.",
));
}
let tool_started_at = std::time::Instant::now();
let previous = canonical_box_versions(&self.session.journal);
let mut outcome = match self.session.execute_tool(&call, operation_id).await {
Ok(outcome) => outcome,
Err(error) => {
let mut text = format!("{} failed: {error}", call.name);
append_slow_tool_duration(&mut text, tool_started_at.elapsed());
return Ok(kcode_agent_runtime::ToolOutcome::failure(text));
}
};
append_slow_tool_duration(&mut outcome.text, tool_started_at.elapsed());
if let Some(snapshot) = outcome.managed_source_snapshot.take() {
apply_snapshot(&mut self.session.journal, &now(), snapshot)?;
outcome.store_result = false;
}
let states = if outcome.ok {
changed_subagent_tool_states(&self.session.journal, &previous)
} else {
Vec::new()
};
let hidden = states
.iter()
.filter(|state| state.hide_from_parent && state.text.is_some())
.map(|state| state.box_id)
.collect::<Vec<_>>();
if !hidden.is_empty() {
self.session.journal.dehydrate_boxes(now(), &hidden)?;
}
let capture = outcome.freeform_write.take().map(|request| {
let id = Uuid::new_v4().to_string();
self.captures.insert(id.clone(), request);
Value::String(id)
});
Ok(kcode_agent_runtime::ToolOutcome {
text: outcome.text,
ok: outcome.ok,
state_updates: subagent_state_updates(&states),
capture,
})
})
}
fn complete_capture<'a>(
&'a mut self,
capture: Value,
contents: String,
budget: kcode_agent_runtime::ContextBudget,
) -> kcode_agent_runtime::HostFuture<'a, kcode_agent_runtime::ToolOutcome> {
Box::pin(async move {
let id = capture
.as_str()
.context("subagent freeform capture token is invalid")?;
let request = self
.captures
.remove(id)
.context("subagent freeform capture token is unknown")?;
let (outcome, states) = self
.session
.complete_subagent_freeform_write(request, contents, &budget)
.await?;
let hidden = states
.iter()
.filter(|state| state.hide_from_parent && state.text.is_some())
.map(|state| state.box_id)
.collect::<Vec<_>>();
if !hidden.is_empty() {
self.session.journal.dehydrate_boxes(now(), &hidden)?;
}
Ok(kcode_agent_runtime::ToolOutcome {
text: outcome.text,
ok: outcome.ok,
state_updates: subagent_state_updates(&states),
capture: None,
})
})
}
fn record(&mut self, event: kcode_agent_runtime::AuditEvent) -> anyhow::Result<()> {
kcode_intelligence_chatend::record_subagent_event(&mut self.session.journal, &now(), &event)
}
}
fn cost_summary(label: &str, estimated_cost_usd_nanos: u64, unpriced_calls: u64) -> String {
let rounded_milli_pennies = estimated_cost_usd_nanos.saturating_add(5_000) / 10_000;
let pennies = format!(
"{}.{:03}",
rounded_milli_pennies / 1_000,
rounded_milli_pennies % 1_000
);
if unpriced_calls == 0 {
format!("Estimated {label}: {pennies} pennies at standard API rates.")
} else {
format!(
"Estimated {label}: {pennies} pennies at standard API rates; {unpriced_calls} provider {} could not be priced.",
if unpriced_calls == 1 { "call" } else { "calls" }
)
}
}
fn restore_kweb_context(journal: &HistorySession, context: &mut KwebContext) -> anyhow::Result<()> {
let Some(tool) = journal.state().tools.get(KWEB_TOOL_INSTANCE) else {
return Ok(());
};
let mut nodes = BTreeMap::new();
for slot in &tool.slots {
let state = journal
.state()
.box_state(slot.box_id)
.context("Kweb slot references a missing box")?;
if let Some(node) = state.canonical.content.metadata.get("storedNode") {
let node = match serde_json::from_value::<KwebNode>(node.clone()) {
Ok(node) => node,
Err(_) => node_from_value(node).context("decoding a stored Kweb context node")?,
};
nodes.insert(node.id.clone(), node);
}
}
let mut direct = journal
.state()
.events
.iter()
.flat_map(|event| {
let EventKind::ToolInvoked {
tool_name,
arguments,
..
} = &event.kind
else {
return Vec::new();
};
match tool_name.as_str() {
"LoadNodes" => arguments
.get("identifiers")
.and_then(Value::as_array)
.into_iter()
.flatten()
.filter_map(Value::as_str)
.map(str::to_owned)
.collect(),
// This is replay-only compatibility for persisted pre-batch sessions.
"LoadNode" => arguments
.get("identifier")
.and_then(Value::as_str)
.map(str::to_owned)
.into_iter()
.collect(),
_ => Vec::new(),
}
})
.collect::<Vec<_>>();
if direct.is_empty() {
direct = context.root_node_ids().to_vec();
}
context
.restore(nodes.into_values(), direct)
.map_err(anyhow::Error::new)
}
fn transcript_from_journal(journal: &HistorySession) -> Vec<Value> {
journal
.state()
.boxes
.values()
.filter_map(|state| {
let role = match state.owner {
BoxOwner::User if state.name == "User message" => "user",
BoxOwner::Kennedy if state.name == "Kennedy message" => "kennedy",
BoxOwner::Controller
if state
.canonical
.content
.metadata
.get("transcriptRole")
.and_then(Value::as_str)
== Some("system") =>
{
"system"
}
_ => return None,
};
let transcript_text = state
.canonical
.content
.metadata
.get("transcriptText")
.and_then(Value::as_str)
.unwrap_or(&state.canonical.content.text);
let mut entry = json!({
"role":role,
"content":transcript_text,
});
if !state.canonical.content.objects.is_empty() {
entry["objects"] = json!(state.canonical.content.objects);
}
if let Some(attachments) = state
.canonical
.content
.metadata
.get("attachments")
.filter(|value| value.is_array())
{
entry["attachments"] = attachments.clone();
} else if let Some(media) = state
.canonical
.content
.metadata
.get("media")
.filter(|value| value.is_object())
{
entry["attachments"] = json!([media]);
}
if let Some(id) = state.canonical.content.metadata.get("externalEventId") {
entry["externalEventId"] = id.clone();
}
if state
.canonical
.content
.metadata
.get("contextOverflowWarning")
.and_then(Value::as_bool)
== Some(true)
{
entry["contextOverflowWarning"] = json!(true);
}
if state
.canonical
.content
.metadata
.get("userStopped")
.and_then(Value::as_bool)
== Some(true)
{
entry["userStopped"] = json!(true);
}
Some(entry)
})
.collect()
}
fn is_terminal_external_response(entry: &Value) -> bool {
match entry.get("role").and_then(Value::as_str) {
Some("kennedy") => true,
Some("system") => {
entry.get("contextOverflowWarning").and_then(Value::as_bool) != Some(true)
}
_ => false,
}
}
fn restore_pending_turn(restored: Option<&Value>, transcript: &[Value]) -> (bool, Option<String>) {
let mut unanswered_external = Vec::<String>::new();
let mut answered_external = HashSet::<String>::new();
for entry in transcript {
if entry.get("role").and_then(Value::as_str) == Some("user") {
if let Some(id) = entry.get("externalEventId").and_then(Value::as_str) {
answered_external.remove(id);
unanswered_external.retain(|candidate| candidate != id);
unanswered_external.push(id.to_owned());
}
continue;
}
if !is_terminal_external_response(entry) {
continue;
}
if let Some(id) = entry.get("externalEventId").and_then(Value::as_str) {
answered_external.insert(id.to_owned());
unanswered_external.retain(|candidate| candidate != id);
} else if entry.get("userStopped").and_then(Value::as_bool) == Some(true)
&& let Some(id) = unanswered_external.pop()
{
// Older stop boxes did not identify their user event. A stop closes
// the most recent unanswered turn that precedes it in the journal.
answered_external.insert(id);
}
}
let journal_pending_external = unanswered_external.last().cloned();
let restored_pending = restored
.and_then(|state| state.get("pendingTurn"))
.and_then(Value::as_bool)
.unwrap_or(false);
let restored_external = restored
.and_then(|state| state.get("pendingExternalEventId"))
.and_then(Value::as_str);
// The response box is journaled before the lifecycle checkpoint that clears
// the turn, so recovery must tolerate a crash between those two writes.
let restored_external_answered =
restored_external.is_some_and(|id| answered_external.contains(id));
let pending_external_event_id = journal_pending_external.or_else(|| {
restored_external
.filter(|id| !answered_external.contains(*id))
.map(str::to_owned)
});
let pending_turn =
pending_external_event_id.is_some() || (restored_pending && !restored_external_answered);
(pending_turn, pending_external_event_id)
}
fn staged_object_transport_kind(
journal: &HistorySession,
pending_id: &PendingId,
) -> Option<String> {
let pending_id_text = pending_id.to_string();
for state in journal.state().boxes.values() {
let Some(index) = state
.canonical
.content
.objects
.iter()
.position(|object_id| object_id == &pending_id_text)
else {
continue;
};
let metadata = &state.canonical.content.metadata;
let descriptor = metadata
.get("attachments")
.and_then(Value::as_array)
.and_then(|attachments| {
attachments
.iter()
.find(|attachment| {
attachment.get("pendingId").and_then(Value::as_str)
== Some(pending_id_text.as_str())
})
.or_else(|| attachments.get(index))
})
.or_else(|| metadata.get("media").filter(|value| value.is_object()));
if let Some(kind) = descriptor
.and_then(|descriptor| descriptor.get("kind"))
.and_then(Value::as_str)
.filter(|kind| !kind.trim().is_empty())
{
return Some(kind.to_owned());
}
}
journal
.objects()
.get(pending_id)
.and_then(|location| location.metadata.transport.get("kind"))
.and_then(Value::as_str)
.filter(|kind| !kind.trim().is_empty())
.map(str::to_owned)
}
fn kweb_node_draft(node: &PlannedNode) -> NodeDraft {
NodeDraft {
short_name: node.short_name.clone(),
short_description: node.short_description.clone(),
long_description: node.long_description.clone(),
owner: node.owner.clone(),
fixed_connections: node.fixed_connections.clone(),
recent_connections: node.recent_connections.clone(),
objects: node.objects.clone(),
}
}
fn planned_node(node: &KwebNode) -> PlannedNode {
PlannedNode {
short_name: node.short_name.clone(),
short_description: node.short_description.clone(),
long_description: node.long_description.clone(),
owner: node.owner.clone(),
fixed_connections: node
.fixed_connections
.iter()
.map(|connection| connection.id.clone())
.collect(),
recent_connections: node
.recent_connections
.iter()
.map(|connection| connection.id.clone())
.collect(),
objects: node.objects.clone(),
attach_session_archive: true,
}
}
fn session_kind(session_type: &str, mode: &AgentMode) -> SessionKind {
if matches!(mode, AgentMode::Ingress { .. }) {
return SessionKind::HistoryIngress;
}
match session_type {
"conversation" => SessionKind::Conversation,
"telegram" => SessionKind::Telegram,
"telegram-group" => SessionKind::TelegramGroup,
"free-time" => SessionKind::SelfTime,
"wakeup" => SessionKind::Other("wakeup".into()),
"audio" => SessionKind::AudioIngress,
other => SessionKind::Other(other.into()),
}
}
fn tool_instance(name: &str) -> String {
if name == "LoadNodes" {
return KWEB_TOOL_INSTANCE.into();
}
format!("{name}:{}", Uuid::new_v4())
}
fn resolve_object_using(
journal: &mut HistorySession,
object_id: &str,
read_canonical: impl FnOnce(&str) -> anyhow::Result<StoredFile>,
) -> anyhow::Result<ResolvedObject> {
if object_id.starts_with("pending:") {
let pending_id = PendingId::parse(object_id.to_owned())?;
let location = journal
.objects()
.get(&pending_id)
.cloned()
.with_context(|| {
format!("staged object {pending_id} does not exist in this session")
})?;
let transport_kind = staged_object_transport_kind(journal, &pending_id);
let bytes = journal.read_object(&pending_id)?;
anyhow::ensure!(
bytes.len() as u64 == location.payload_len,
"staged object {pending_id} declared {} bytes but resolved to {}",
location.payload_len,
bytes.len()
);
let fallback = format!("object-{}.bin", pending_id.number());
Ok(ResolvedObject {
object_id: pending_id.to_string(),
bytes,
file_name: sanitize_file_name(
location.metadata.file_name.as_deref().unwrap_or_default(),
&fallback,
),
media_type: location.metadata.media_type,
transport_kind,
})
} else {
let canonical_id = object_id
.parse::<ObjectId>()
.with_context(|| format!("{object_id:?} is not an object ID"))?;
let file = read_canonical(object_id)?;
anyhow::ensure!(
file.object_id == canonical_id,
"object store returned {} while resolving {canonical_id}",
file.object_id
);
Ok(ResolvedObject {
object_id: canonical_id.to_string(),
bytes: file.bytes,
file_name: file.file_name,
media_type: file.media_type,
transport_kind: file.transport_kind,
})
}
}
fn rust_binary_object_ids(arguments: &Value) -> anyhow::Result<Vec<String>> {
let Some(object_ids) = arguments.get("objectIds") else {
return Ok(Vec::new());
};
object_ids
.as_array()
.context("Rust-binary objectIds must be an array")?
.iter()
.map(|object_id| {
object_id
.as_str()
.map(str::to_owned)
.context("Rust-binary objectIds must contain only strings")
})
.collect()
}
fn web_library_object_id(arguments: &Value) -> anyhow::Result<String> {
arguments
.get("objectId")
.and_then(Value::as_str)
.filter(|object_id| !object_id.trim().is_empty())
.map(str::to_owned)
.context("Web-library attachment objectId must be a nonempty string")
}
fn ingress_object_filename(value: Option<&str>, fallback: &str) -> String {
value
.filter(|value| !value.trim().is_empty())
.unwrap_or(fallback)
.to_owned()
}
fn file_name_extension(file_name: &str) -> String {
file_name
.rsplit_once('.')
.and_then(|(stem, extension)| {
(!stem.is_empty() && !extension.is_empty()).then_some(extension)
})
.map(|extension| format!(".{extension}"))
.unwrap_or_else(|| "(none)".into())
}
fn render_user_file_metadata(
ordinal: usize,
object_id: &str,
file_name: &str,
media_type: &str,
size_bytes: u64,
) -> String {
format!(
"User-provided file {ordinal}\nObject reference: {object_id}\nOriginal filename: {file_name}\nExtension: {}\nMIME type: {}\nSize: {size_bytes} bytes",
file_name_extension(file_name),
normalized_media_type(media_type),
)
}
fn authoritative_staged_file_metadata(
journal: &HistorySession,
pending_id: &PendingId,
) -> anyhow::Result<(String, String, u64)> {
let location = journal
.objects()
.get(pending_id)
.with_context(|| format!("user-provided object {pending_id} is not staged"))?;
let fallback = format!("object-{}.bin", pending_id.number());
let file_name = sanitize_file_name(
location.metadata.file_name.as_deref().unwrap_or_default(),
&fallback,
);
Ok((
file_name,
normalized_media_type(&location.metadata.media_type),
location.payload_len,
))
}
fn canonicalize_staged_file_descriptor(
journal: &HistorySession,
pending_id: &PendingId,
descriptor: &mut Value,
) -> anyhow::Result<String> {
let (file_name, media_type, size_bytes) =
authoritative_staged_file_metadata(journal, pending_id)?;
if !descriptor.is_object() {
*descriptor = json!({});
}
descriptor["pendingId"] = json!(pending_id.to_string());
descriptor["fileName"] = json!(file_name);
descriptor["extension"] = json!(file_name_extension(&file_name));
descriptor["mimeType"] = json!(media_type);
descriptor["sizeBytes"] = json!(size_bytes);
Ok(file_name)
}
fn append_user_file_metadata(
journal: &HistorySession,
content: &mut BoxContent,
) -> anyhow::Result<()> {
let mut blocks = Vec::with_capacity(content.objects.len());
for (index, object_id) in content.objects.iter().enumerate() {
let pending_id = PendingId::parse(object_id.clone())?;
let (file_name, media_type, size_bytes) =
authoritative_staged_file_metadata(journal, &pending_id)?;
blocks.push(render_user_file_metadata(
index + 1,
object_id,
&file_name,
&media_type,
size_bytes,
));
}
if blocks.is_empty() {
return Ok(());
}
if !content.text.is_empty() && !content.text.ends_with('\n') {
content.text.push_str("\n\n");
}
content.text.push_str(&blocks.join("\n\n"));
Ok(())
}
fn authoritative_object_filename(metadata: &ObjectMetadata) -> anyhow::Result<&str> {
metadata
.file_name
.as_deref()
.filter(|value| !value.trim().is_empty())
.with_context(|| {
format!(
"staged object {} has no authoritative filename",
metadata.pending_id
)
})
}
fn staged_telegram_group_media(
journal: &HistorySession,
chat_id: i64,
message_id: i64,
) -> Option<(PendingId, ObjectMetadata, u64)> {
journal.objects().iter().find_map(|(pending_id, location)| {
let transport = &location.metadata.transport;
(transport.get("source").and_then(Value::as_str) == Some("telegram-group")
&& transport.get("chatId").and_then(Value::as_i64) == Some(chat_id)
&& transport.get("messageId").and_then(Value::as_i64) == Some(message_id))
.then(|| {
(
pending_id.clone(),
location.metadata.clone(),
location.payload_len,
)
})
})
}
fn render_staged_telegram_group_media(
pending_id: &PendingId,
metadata: &ObjectMetadata,
size_bytes: u64,
message_id: i64,
reused: bool,
) -> anyhow::Result<String> {
Ok(format!(
"{} Telegram group media\nMessage ID: {message_id}\nObject: {pending_id}\nKind: {}\nOriginal filename: {}\nExtension: {}\nMIME type: {}\nSize: {size_bytes} bytes\n\nUse Object {pending_id} with AnnotateMedia, GenerateImage (for images), TranscribeAudio, or ExtractDocumentText as appropriate.",
if reused {
"Reused already-staged"
} else {
"Staged"
},
metadata
.transport
.get("kind")
.and_then(Value::as_str)
.unwrap_or("media"),
authoritative_object_filename(metadata)?,
file_name_extension(authoritative_object_filename(metadata)?),
normalized_media_type(&metadata.media_type),
))
}
fn normalized_media_type(value: &str) -> String {
value
.split(';')
.next()
.unwrap_or(value)
.trim()
.to_ascii_lowercase()
}
fn validate_annotation_media(model: &str, media_type: &str) -> anyhow::Result<()> {
match model {
"gpt-5.6" | "gpt-5.6-sol" | "gpt-5.6-terra" | "gpt-5.6-luna" => anyhow::ensure!(
media_type.starts_with("image/"),
"{model} annotations accept images only"
),
"gemini-2.5-flash" | "gemini-3.1-flash-lite" | "gemini-3.1-pro-preview" => anyhow::ensure!(
media_type.starts_with("image/")
|| media_type.starts_with("audio/")
|| media_type.starts_with("video/")
|| media_type == "application/ogg",
"{model} annotations accept images, audio, or video only"
),
_ => anyhow::bail!("unsupported exact annotation model {model}"),
}
Ok(())
}
fn validate_image_model(model: &str) -> anyhow::Result<()> {
anyhow::ensure!(
matches!(model, "gpt-image-2" | "gemini-3-pro-image"),
"unsupported exact image model {model}; use gpt-image-2 or gemini-3-pro-image"
);
Ok(())
}
fn image_extension(media_type: &str) -> &'static str {
match normalized_media_type(media_type).as_str() {
"image/jpeg" => "jpg",
"image/webp" => "webp",
_ => "png",
}
}
fn optional_object_id_array(
value: &Value,
key: &str,
maximum: usize,
) -> anyhow::Result<Vec<String>> {
let Some(values) = value.get(key) else {
return Ok(Vec::new());
};
let values = values
.as_array()
.with_context(|| format!("{key} must be an array"))?;
anyhow::ensure!(
values.len() <= maximum,
"{key} must contain at most {maximum} object IDs"
);
let ids = values
.iter()
.map(|value| {
let id = value
.as_str()
.with_context(|| format!("{key} entries must be strings"))?;
anyhow::ensure!(
!id.trim().is_empty() && id.chars().count() <= 64,
"{key} entries must contain between 1 and 64 characters"
);
Ok(id.to_owned())
})
.collect::<anyhow::Result<Vec<_>>>()?;
anyhow::ensure!(
ids.iter().collect::<HashSet<_>>().len() == ids.len(),
"{key} must not contain duplicate object IDs"
);
Ok(ids)
}
fn optional_delivery_file_name(value: &Value, key: &str) -> anyhow::Result<Option<String>> {
let Some(file_name) = value.get(key) else {
return Ok(None);
};
let file_name = file_name
.as_str()
.with_context(|| format!("{key} must be a string"))?;
kcode_telegram_session_coordinator::validate_file_name(file_name)?;
Ok(Some(file_name.to_owned()))
}
fn validate_transcription_model(model: &str) -> anyhow::Result<()> {
anyhow::ensure!(
matches!(
model,
"gpt-4o-transcribe"
| "gemini-2.5-flash"
| "gemini-3.1-flash-lite"
| "gemini-3.1-pro-preview"
),
"unsupported exact transcription model {model}"
);
Ok(())
}
fn validate_transcribable_audio(media_type: &str) -> anyhow::Result<()> {
anyhow::ensure!(
matches!(
media_type,
"audio/flac"
| "audio/x-flac"
| "audio/m4a"
| "audio/mp3"
| "audio/mp4"
| "audio/mpeg"
| "audio/mpga"
| "audio/ogg"
| "audio/opus"
| "audio/wav"
| "audio/x-wav"
| "audio/webm"
| "application/ogg"
),
"TranscribeAudio accepts a supported FLAC, MP3, MP4, M4A, OGG, WAV, or WebM audio object only"
);
Ok(())
}
fn validate_extractable_document(media_type: &str, file_name: &str) -> anyhow::Result<()> {
let media_type = normalized_media_type(media_type);
let extension = file_name
.rsplit_once('.')
.map(|(_, extension)| extension)
.map(str::to_ascii_lowercase);
let supported_media_type = media_type.starts_with("text/")
|| matches!(
media_type.as_str(),
"application/pdf"
| "application/msword"
| "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
| "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
| "application/vnd.ms-excel"
| "application/vnd.ms-excel.sheet.binary.macroenabled.12"
| "application/vnd.oasis.opendocument.spreadsheet"
| "application/json"
| "application/xml"
| "application/yaml"
| "application/x-yaml"
);
anyhow::ensure!(
supported_media_type
|| matches!(
extension.as_deref(),
Some(
"pdf"
| "doc"
| "docx"
| "xlsx"
| "xls"
| "xlsb"
| "ods"
| "csv"
| "tsv"
| "txt"
| "md"
| "json"
| "yaml"
| "yml"
| "xml"
)
),
"ExtractDocumentText accepts supported PDF, Word, spreadsheet, and text-family objects only"
);
Ok(())
}
fn decode_data_url(value: &str) -> anyhow::Result<(String, Vec<u8>)> {
let value = value
.strip_prefix("data:")
.context("object data URL must begin with data:")?;
let (header, data) = value.split_once(',').context("invalid object data URL")?;
let media_type = header
.strip_suffix(";base64")
.context("object data URL must use Base64")?;
Ok((media_type.into(), BASE64.decode(data)?))
}
fn attachment_metadata_without_payload(value: &Value) -> Value {
let mut value = value.clone();
if let Some(object) = value.as_object_mut() {
object.remove("dataUrl");
object.remove("text");
}
value
}
fn message_metadata_without_attachment_payloads(value: &Value) -> Value {
let mut value = value.clone();
let Some(object) = value.as_object_mut() else {
return value;
};
if let Some(attachments) = object.get_mut("attachments").and_then(Value::as_array_mut) {
for attachment in attachments {
*attachment = attachment_metadata_without_payload(attachment);
}
}
if let Some(media) = object.get_mut("media") {
*media = attachment_metadata_without_payload(media);
}
value
}
#[derive(Debug, Eq, PartialEq)]
struct BoxTextObject {
box_id: BoxId,
pending_id: PendingId,
reused: bool,
}
fn stage_box_text_objects(
journal: &mut HistorySession,
box_ids: &[BoxId],
recorded_at: &str,
) -> anyhow::Result<Vec<BoxTextObject>> {
let selections = box_ids
.iter()
.map(|box_id| {
let state = journal
.state()
.box_state(*box_id)
.with_context(|| format!("box {box_id} does not exist"))?;
anyhow::ensure!(state.active, "box {box_id} is not active");
anyhow::ensure!(
!state.canonical.content.text.is_empty(),
"box {box_id} has no text content"
);
Ok((
*box_id,
state.name.clone(),
state.canonical.event_id,
state.canonical.content.text.clone(),
))
})
.collect::<anyhow::Result<Vec<_>>>()?;
let mut objects = Vec::with_capacity(selections.len());
for (box_id, box_name, canonical_event_id, text) in selections {
if let Some(pending_id) = existing_box_text_object(journal, box_id, canonical_event_id) {
objects.push(BoxTextObject {
box_id,
pending_id,
reused: true,
});
continue;
}
let pending_id = journal.stage_object(
recorded_at,
BOX_TEXT_MEDIA_TYPE,
Some(format!("box-{box_id}.txt")),
json!({
"source":BOX_TEXT_OBJECT_SOURCE,
"boxId":box_id.0,
"boxName":box_name,
"canonicalEventId":canonical_event_id.0,
}),
text.as_bytes(),
)?;
objects.push(BoxTextObject {
box_id,
pending_id,
reused: false,
});
}
Ok(objects)
}
fn existing_box_text_object(
journal: &HistorySession,
box_id: BoxId,
canonical_event_id: EventId,
) -> Option<PendingId> {
journal
.objects()
.iter()
.find(|(_, location)| {
let transport = &location.metadata.transport;
transport.get("source").and_then(Value::as_str) == Some(BOX_TEXT_OBJECT_SOURCE)
&& transport.get("boxId").and_then(Value::as_u64) == Some(box_id.0)
&& transport.get("canonicalEventId").and_then(Value::as_u64)
== Some(canonical_event_id.0)
})
.map(|(pending_id, _)| pending_id.clone())
}
fn render_box_text_objects(objects: &[BoxTextObject]) -> String {
let mut text = String::from("Box text objects:");
for object in objects {
text.push_str(&format!("\nBox {}: {}", object.box_id, object.pending_id));
if object.reused {
text.push_str(" (already staged)");
}
}
text.push_str(
"\nThese pending object references resolve to canonical object IDs when the logical session commits.",
);
text
}
fn call_ktool_description() -> &'static str {
"Call one Kennedy Ktool. The provider function remains registered even if its explaining system-prompt box is dehydrated. Kennedy may display an object with an optional recipient-visible filename using {\"name\":\"EmitObject\",\"arguments\":{\"objectId\":\"AAECAwQF\",\"fileName\":\"report.pdf\"}}. She may make an out-of-band cold delivery to an authorized user's private Telegram chat, optionally with Kweb object attachments and per-attachment delivery filenames, from any session with {\"name\":\"SendTelegramDM\",\"arguments\":{\"user\":{\"telegramUserId\":42},\"message\":\"Exact message text.\",\"attachments\":[\"pending:1\",{\"objectId\":\"AAECAwQF\",\"fileName\":\"report.pdf\"}]}}. Kennedy may likewise send text and attachments to any known Telegram group, addressed by its canonical Kweb root, with {\"name\":\"SendTelegramGroupMessage\",\"arguments\":{\"group\":{\"rootNodeId\":\"AAAAAAAE\"},\"message\":\"Exact message text.\",\"attachments\":[\"pending:1\"]}}."
}
fn validate_arguments(value: &Value, required: &[&str], optional: &[&str]) -> anyhow::Result<()> {
let map = value
.as_object()
.context("arguments must be a JSON object")?;
let allowed = required
.iter()
.chain(optional)
.copied()
.collect::<HashSet<_>>();
anyhow::ensure!(
required.iter().all(|key| map.contains_key(*key))
&& map.keys().all(|key| allowed.contains(key.as_str())),
"expected exactly: {}{}",
required.join(", "),
if optional.is_empty() {
String::new()
} else {
format!(" (optional: {})", optional.join(", "))
}
);
Ok(())
}
fn positive_integer(value: &Value, key: &str) -> anyhow::Result<u64> {
value
.get(key)
.and_then(Value::as_u64)
.filter(|value| *value > 0)
.with_context(|| format!("{key} must be a positive integer"))
}
fn box_id(value: &Value, key: &str) -> anyhow::Result<BoxId> {
positive_integer(value, key).map(BoxId)
}
fn box_id_array(value: &Value, key: &str) -> anyhow::Result<Vec<BoxId>> {
let ids = value
.get(key)
.and_then(Value::as_array)
.with_context(|| format!("{key} must be an array"))?
.iter()
.map(|value| {
value
.as_u64()
.filter(|value| *value > 0)
.map(BoxId)
.with_context(|| format!("{key} must contain only positive integers"))
})
.collect::<anyhow::Result<Vec<_>>>()?;
anyhow::ensure!(!ids.is_empty(), "{key} must contain at least one box ID");
let unique = ids.iter().copied().collect::<HashSet<_>>();
anyhow::ensure!(
unique.len() == ids.len(),
"{key} must not contain duplicate box IDs"
);
Ok(ids)
}
fn canonical_id(value: &str) -> anyhow::Result<String> {
value
.parse::<NodeId>()
.with_context(|| format!("{value:?} is not a canonical node ID"))?;
Ok(value.into())
}
fn canonical_node_id_array(
value: &Value,
key: &str,
maximum: usize,
) -> anyhow::Result<Vec<String>> {
let ids = canonical_node_ids(value, key)?;
anyhow::ensure!(
ids.len() <= maximum,
"{key} must contain at most {maximum} identifiers"
);
Ok(ids)
}
fn canonical_node_id_list(value: &Value, key: &str) -> anyhow::Result<Vec<String>> {
let ids = canonical_node_ids(value, key)?;
anyhow::ensure!(
!ids.is_empty(),
"{key} must contain at least one identifier"
);
Ok(ids)
}
fn canonical_node_ids(value: &Value, key: &str) -> anyhow::Result<Vec<String>> {
let ids = value
.get(key)
.and_then(Value::as_array)
.with_context(|| format!("{key} must be an array"))?
.iter()
.map(|value| {
canonical_id(
value
.as_str()
.with_context(|| format!("{key} entries must be canonical node IDs"))?,
)
})
.collect::<anyhow::Result<Vec<_>>>()?;
anyhow::ensure!(
ids.iter().collect::<HashSet<_>>().len() == ids.len(),
"{key} must not contain duplicate identifiers"
);
Ok(ids)
}
fn parse_resource_id(value: &str) -> anyhow::Result<String> {
if value.starts_with("pending:") {
PendingId::parse(value.to_owned())?;
Ok(value.into())
} else if matches!(value, "self" | "unowned") {
Ok(value.into())
} else {
canonical_id(value)
}
}
fn resource_id(value: &Value, key: &str) -> anyhow::Result<String> {
parse_resource_id(
value
.get(key)
.and_then(Value::as_str)
.with_context(|| format!("{key} must be a node identifier"))?,
)
}
fn resource_id_array(value: &Value, key: &str, minimum: usize) -> anyhow::Result<Vec<String>> {
let ids = value
.get(key)
.and_then(Value::as_array)
.with_context(|| format!("{key} must be an array"))?
.iter()
.map(|value| parse_resource_id(value.as_str().context("node identifier must be a string")?))
.collect::<anyhow::Result<Vec<_>>>()?;
anyhow::ensure!(
ids.len() >= minimum && ids.iter().collect::<HashSet<_>>().len() == ids.len(),
"{key} has invalid length or duplicate identifiers"
);
Ok(ids)
}
fn string_value(value: &Value, key: &str) -> anyhow::Result<String> {
value
.get(key)
.and_then(Value::as_str)
.map(str::to_owned)
.with_context(|| format!("{key} must be a string"))
}
fn node_text_arguments(
value: &Value,
short_name_key: &str,
short_description_key: &str,
long_description_key: &str,
) -> anyhow::Result<(String, String, String)> {
let short_name = string_value(value, short_name_key)?;
let short_description = string_value(value, short_description_key)?;
let long_description = string_value(value, long_description_key)?;
let short_name_characters = short_name.chars().count();
let short_description_characters = short_description.chars().count();
let long_description_characters = long_description.chars().count();
anyhow::ensure!(
(MIN_NODE_SHORT_NAME_CHARACTERS..=MAX_NODE_SHORT_NAME_CHARACTERS)
.contains(&short_name_characters),
"{short_name_key} must contain between {MIN_NODE_SHORT_NAME_CHARACTERS} and \
{MAX_NODE_SHORT_NAME_CHARACTERS} characters; received {short_name_characters}. \
Correct it and retry."
);
anyhow::ensure!(
short_description_characters <= MAX_NODE_SHORT_DESCRIPTION_CHARACTERS,
"{short_description_key} must be at most {MAX_NODE_SHORT_DESCRIPTION_CHARACTERS} \
characters; received {short_description_characters}. Shorten it and retry."
);
anyhow::ensure!(
long_description_characters <= MAX_NODE_LONG_DESCRIPTION_CHARACTERS,
"{long_description_key} must be at most {MAX_NODE_LONG_DESCRIPTION_CHARACTERS} \
characters; received {long_description_characters}. Shorten it and retry."
);
Ok((short_name, short_description, long_description))
}
fn nonempty_string(value: &Value, key: &str, max: usize) -> anyhow::Result<String> {
let value = string_value(value, key)?;
let trimmed = value.trim();
anyhow::ensure!(
!trimmed.is_empty() && trimmed.chars().count() <= max,
"{key} must contain between 1 and {max} characters"
);
Ok(trimmed.into())
}
fn bounded_nonempty_string(value: &Value, key: &str, max: usize) -> anyhow::Result<String> {
let value = string_value(value, key)?;
anyhow::ensure!(
!value.trim().is_empty() && value.chars().count() <= max,
"{key} must contain between 1 and {max} characters"
);
Ok(value)
}
fn now() -> String {
Utc::now().to_rfc3339()
}
fn runtime_description(runtime: &RuntimeModel, current_time: DateTime<Utc>) -> String {
format!(
"You are currently running on {} with {} thinking mode. The current date and time is {}.",
runtime.model,
runtime.reasoning_effort,
human_utc_datetime(current_time)
)
}
fn human_utc_datetime(value: DateTime<Utc>) -> String {
let day = value.day();
let suffix = match day % 100 {
11..=13 => "th",
_ => match day % 10 {
1 => "st",
2 => "nd",
3 => "rd",
_ => "th",
},
};
let hour = match value.hour() % 12 {
0 => 12,
hour => hour,
};
let period = if value.hour() < 12 { "am" } else { "pm" };
format!(
"{} {day}{suffix}, {}, {hour}:{:02}{period} UTC",
value.format("%B"),
value.year(),
value.minute()
)
}
fn deadline(value: &Value) -> Option<DateTime<Utc>> {
value
.get("deadlineAt")
.and_then(Value::as_str)
.and_then(|value| DateTime::parse_from_rfc3339(value).ok())
.map(|value| value.with_timezone(&Utc))
}
fn free_time_schedule(value: &Value) -> String {
deadline(value)
.map(|deadline| {
format!(
"The self-time deadline is {}.",
human_utc_datetime(deadline)
)
})
.unwrap_or_else(|| "The self-time deadline was not supplied.".into())
}
fn free_time_opening(value: &Value) -> String {
let custom = value
.get("customPrompt")
.and_then(Value::as_str)
.unwrap_or_default();
if custom.trim().is_empty() {
"Begin this self-time session.".into()
} else {
format!("Begin this self-time session.\n\nRequested focus:\n{custom}")
}
}
fn wakeup_opening(marker: DateTime<Utc>) -> String {
format!(
"The time is {} UTC on {}. Determine whether you have any messages you would like to send the user",
marker.format("%H:%M"),
marker.format("%Y-%m-%d"),
)
}
fn controller_box_name(mode: &AgentMode) -> &'static str {
match mode {
AgentMode::Conversation => "Turn continuation",
AgentMode::FreeTime => "Self-time continuation",
AgentMode::Wakeup => "Wakeup continuation",
AgentMode::Ingress { .. } => "History-ingress continuation",
}
}
fn controller_message(mode: &AgentMode, free_time: &Value) -> String {
match mode {
AgentMode::Conversation => {
"Continue the turn. Use tools if needed, then answer the user.".into()
}
AgentMode::FreeTime => format!("Continue self time. {}", free_time_schedule(free_time)),
AgentMode::Wakeup => {
"Continue this autonomous wakeup session. Sending no message is a valid outcome; call EndSession when you have finished.".into()
}
AgentMode::Ingress { .. } => {
"You are in a solo history-ingress session; there is no user to receive a conversational response. If you have completed all useful memory work, call EndSession now through the native call_ktool function with no arguments. A normal response does not end this session. If work remains, continue it with tools, then call EndSession when finished.".into()
}
}
}