#![forbid(unsafe_code)]
mod services;
pub use kcode_kennedy_session_objects::ResolvedObject;
pub use kcode_telegram_session_coordinator::validate_file_name as validate_delivery_file_name;
pub use services::{Api as Service, LocalServices as Capabilities};
use std::{
collections::{BTreeMap, HashMap},
future::Future,
time::{Duration, Instant},
};
use anyhow::Context as _;
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,
proposed_write_snapshot,
};
use kcode_dev_tools_chatend::{
FreeformWrite, SourceSnapshot, apply_snapshot, decode_freeform_write, prepare_freeform_write,
};
use kcode_history_ingress_context::{
Outcome as HistoryIngressContextOutcome, RecoveryOutcome as ContextRecoveryOutcome,
};
use kcode_kennedy_kweb_loader::{load_durable_batch, node_from_value};
use kcode_kennedy_session_ingress::{is_terminal_external_response, restore_pending_turn};
use kcode_kennedy_session_presentation::{RenderRequest, render};
use kcode_kennedy_session_tool_contracts::{
DecodedTool, ManagedObjectArguments, ValidationRequest, decode, decode_managed_objects,
decode_note_to_self, validate,
};
use kcode_kennedy_session_tool_presentation::invocation_box_content;
use kcode_kennedy_subagent_context::Context as SubagentContext;
use kcode_kweb_context::{
Context as KwebContext, Node as KwebNode, NodeDraft, StagedCreate as KwebStagedCreate,
};
use kcode_kweb_db::NodeId;
use kcode_server_object_envelopes::encode_file;
use kcode_session_history::{
NewSession, Session as HistorySession,
chatend::{
BoxContent, BoxId, BoxOwner, CacheExpectation, ContextProjection, EventId, EventKind,
ProviderContext, ProviderToolDefinition, SessionKind, SessionMetadata,
},
};
use kcode_session_runtime_budget::{RoundBudget, RuntimeBudget, TimeBudget, TimeBudgetKind};
use kcode_speaker_system::KTOOLS as SPEECH_CLASSIFICATION_TOOLS;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use uuid::Uuid;
const BROWSER_CONVERSATION_REQUEST_TIMEOUT: Duration = Duration::from_secs(225 * 60);
const HISTORY_INGRESS_REQUEST_TIMEOUT: Duration = Duration::from_secs(225 * 60);
const HISTORY_INGRESS_ATTEMPT_DURATION: Duration = Duration::from_secs(45 * 60);
const WAKEUP_REQUEST_TIMEOUT: Duration = Duration::from_secs(225 * 60);
const SELF_TIME_HARD_STOP_ALLOWANCE: Duration = Duration::from_secs(15 * 60);
const MAX_MEDIA_ENRICHMENT_BYTES: u64 = 20 * 1024 * 1024;
const KWEB_TOOL_INSTANCE: &str = "kweb";
const TASK_BOARD_TOOLS: [&str; 8] = [
"CreateTaskCategory",
"GetTaskCategory",
"RemoveTaskCategory",
"CreateTask",
"GetTask",
"UpdateTask",
"RemoveTask",
"GetTopTaskOrphan",
];
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";
#[derive(Debug)]
struct IngressTimeExpired;
impl std::fmt::Display for IngressTimeExpired {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str("history ingress time expired before EndSession")
}
}
impl std::error::Error for IngressTimeExpired {}
pub fn is_ingress_time_expired(error: &anyhow::Error) -> bool {
error.is::<IngressTimeExpired>()
}
fn ingress_time_remaining_at(deadline: &mut Option<Instant>, now: Instant) -> anyhow::Result<u64> {
let Some(current) = *deadline else {
*deadline = Some(
now.checked_add(HISTORY_INGRESS_ATTEMPT_DURATION)
.context("history ingress deadline overflow")?,
);
return Ok(HISTORY_INGRESS_ATTEMPT_DURATION.as_secs());
};
if now >= current {
return Err(anyhow::Error::new(IngressTimeExpired));
}
Ok(current.duration_since(now).as_secs())
}
#[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)
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum AgentMode {
Conversation,
FreeTime,
Wakeup,
Ingress { record_id: Option<String> },
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TurnDeadlineKind {
Telegram,
SelfTimeHardStop,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct TurnDeadline {
pub kind: TurnDeadlineKind,
pub at: DateTime<Utc>,
}
#[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,
}
struct CreateNodeArguments {
parents: Vec<String>,
owner: String,
short_name: String,
short_description: String,
long_description: String,
}
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");
}
let latest = journal
.state()
.current_ingress_attempt_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,
subagent_codex_prompt: String,
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>,
active_provider_deadline: Option<DateTime<Utc>>,
active_turn_deadline: Option<TurnDeadline>,
provider_affinity: Option<ProviderAffinityState>,
next_thread_reset_reason: Option<String>,
ingress_deadline: Option<Instant>,
previous_ingress_attempt_timed_out: bool,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct ProviderAffinityState {
continuation: kcode_intelligence_router::AgentContinuation,
synchronized_event_id: EventId,
material_fingerprint: 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],
footer_lines: &[String],
) -> anyhow::Result<String> {
let changed_box_ids = changed_box_ids
.iter()
.map(ToString::to_string)
.collect::<Vec<_>>();
let projected_boxes = journal
.state()
.projection_with_footer_lines(footer_lines)
.items
.into_iter()
.filter(|item| !item.marker)
.map(|item| (item.box_id.to_string(), item.text))
.collect::<Vec<_>>();
render(RenderRequest::LoadNodes {
changed_box_ids: &changed_box_ids,
projected_boxes: &projected_boxes,
})
}
fn provider_tool_result_with_context_footer(footer: &str, result: &str) -> String {
render(RenderRequest::ProviderFooter { result, footer })
.expect("provider-footer rendering is infallible")
}
fn append_slow_tool_duration(text: &mut String, elapsed: Duration) {
*text = render(RenderRequest::SlowTool { text, elapsed })
.expect("slow-tool rendering is infallible");
}
fn log_primary_thread_observation(
operation_id: Uuid,
round: u64,
requested_model: &str,
prepared: &PreparedCacheObservation,
provider_thread_id: Option<&str>,
input_tokens: u64,
cached_input_tokens: u64,
) {
tracing::info!(
affinity_scope = "primary",
%operation_id,
round,
provider = prepared.provider,
requested_model,
model = prepared.model,
thread_action = prepared.thread_action,
provider_thread_id = provider_thread_id.unwrap_or(""),
thread_reset_reason = prepared.thread_reset_reason.as_deref().unwrap_or(""),
projection_hash = prepared.projection_hash,
provider_input_hash = prepared.provider_input_hash,
provider_input_bytes = prepared.provider_input_bytes,
input_tokens,
cached_input_tokens,
"Provider thread-affinity observation"
);
}
fn render_web_search_result(
result: &kcode_intelligence_router::SearchResponse,
) -> anyhow::Result<String> {
let sources = result
.sources
.iter()
.map(|source| (source.title.clone(), source.url.clone()))
.collect::<Vec<_>>();
render(RenderRequest::WebSearch {
answer: &result.answer,
sources: &sources,
})
}
fn render_web_fetch_result(
result: &kcode_intelligence_router::FetchResponse,
) -> anyhow::Result<String> {
render(RenderRequest::WebFetch {
url: &result.url,
title: result.title.as_deref(),
content_type: &result.content_type,
truncated: result.truncated,
content: &result.content,
})
}
fn render_media_annotation_result(
object_id: &str,
file_name: &str,
content_type: &str,
result: &kcode_intelligence_router::AnnotationResponse,
) -> anyhow::Result<String> {
render(RenderRequest::MediaAnnotation {
object_id,
file_name,
content_type,
model: &result.model,
complete: result.complete,
incomplete_reason: result.incomplete_reason.as_deref(),
text: &result.text,
})
}
fn render_audio_transcription_result(
object_id: &str,
file_name: &str,
content_type: &str,
result: &kcode_intelligence_router::TranscriptionResponse,
) -> anyhow::Result<String> {
render(RenderRequest::AudioTranscription {
object_id,
file_name,
content_type,
model: &result.model,
text: &result.text,
})
}
fn render_document_extraction_result(
object_id: &str,
file_name: &str,
result: &kcode_intelligence_router::DocumentExtraction,
) -> anyhow::Result<String> {
render(RenderRequest::DocumentExtraction {
object_id,
file_name,
format: &result.format,
characters: result.characters,
truncated: result.truncated,
text: &result.text,
})
}
struct ToolCall {
name: String,
arguments: Value,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct TaskId {
task_id: String,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct CategoryId {
category_id: String,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct CategoryCall {
category_id: String,
#[serde(default)]
offset: u64,
#[serde(default = "task_page_limit")]
limit: u32,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct EmptyCall {}
fn task_page_limit() -> u32 {
50
}
struct RecordedToolInvocation {
invocation_id: String,
tool_instance: String,
tool_name: String,
}
struct PendingFreeformWrite {
request: FreeformWrite,
call_box_id: BoxId,
}
struct ToolOutcome {
text: String,
store_result: bool,
ok: bool,
end_session: bool,
freeform_write: Option<FreeformWrite>,
managed_source_snapshot: Option<SourceSnapshot>,
}
fn result_displays_snapshot(result: &str, snapshot: &SourceSnapshot) -> bool {
result == snapshot.text
}
fn subagent_managed_write_fits(
context: &SubagentContext,
call: &ToolCall,
budget: &kcode_agent_runtime::ContextBudget,
) -> bool {
let Some(snapshot) = proposed_write_snapshot(&call.name, &call.arguments) else {
return true;
};
let state = context.source_state(&snapshot);
budget.fits_state(state.key, state.text)
}
struct KennedySubagentHost<'a> {
session: &'a mut Session,
context: SubagentContext,
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,
operation_id: Uuid,
prepared_cache: Option<PreparedCacheObservation>,
}
struct PreparedCacheObservation {
cacheable_prefix_bytes: u64,
expectation: CacheExpectation,
material_fingerprint: String,
projection_hash: String,
logical_input: String,
provider_input_hash: String,
provider_input_bytes: u64,
thread_action: String,
thread_reset_reason: Option<String>,
estimated_input_tokens: u64,
raw_estimated_input_tokens: u64,
provider: String,
model: String,
}
fn is_kweb_mutation(name: &str) -> bool {
matches!(
name,
"ConnectNodes" | "ConsolidateFanout" | "SetFixedConnection" | "CreateNode" | "UpdateNode"
)
}
fn referenced_pending_nodes(decoded: &DecodedTool) -> Vec<String> {
let mut ids = Vec::new();
match decoded {
DecodedTool::ConnectNodes(nodes) => ids.extend(nodes.iter()),
DecodedTool::ConsolidateFanout {
parent,
fanout,
aggregator,
} => {
ids.push(parent);
ids.extend(fanout.iter());
ids.push(aggregator);
}
DecodedTool::SetFixedConnection {
parent,
child,
slot: _,
} => {
ids.push(parent);
ids.extend(child.iter());
}
DecodedTool::CreateNode { parents, owner, .. } => {
ids.extend(parents.iter());
ids.push(owner);
}
DecodedTool::UpdateNode { id, owner, .. } => {
ids.push(id);
ids.push(owner);
}
_ => {}
}
ids.into_iter()
.filter(|id| id.starts_with("pending:"))
.cloned()
.collect()
}
fn subagent_unavailable_reason(name: &str) -> Option<&'static str> {
match name {
"RunSubagent" => {
Some("RunSubagent is unavailable inside a subagent. Only Kennedy may launch subagents.")
}
"EndSession" => Some(
"EndSession is unavailable inside a subagent. A child cannot control the parent session lifecycle.",
),
"DehydrateBoxes" | "SummarizeBox" | "HydrateBox" | "BoxesIntoObjects" => {
Some("Parent box controls are unavailable inside a box-free subagent context.")
}
_ => None,
}
}
fn execute_kweb_mutation(
name: &str,
decoded: DecodedTool,
context: &KwebContext,
plan: &mut KwebPlan,
journal: &mut HistorySession,
) -> anyhow::Result<String> {
match (name, decoded) {
("ConnectNodes", DecodedTool::ConnectNodes(ids)) => {
Session::connect_nodes(plan, context, ids)
}
(
"ConsolidateFanout",
DecodedTool::ConsolidateFanout {
parent,
fanout,
aggregator,
},
) => Session::consolidate_fanout(plan, context, parent, fanout, aggregator),
(
"SetFixedConnection",
DecodedTool::SetFixedConnection {
parent,
child,
slot,
},
) => Session::set_fixed_connection(plan, context, parent, child, slot),
(
"CreateNode",
DecodedTool::CreateNode {
parents,
owner,
short_name,
short_description,
long_description,
},
) => Session::create_node(
plan,
context,
journal,
CreateNodeArguments {
parents,
owner,
short_name,
short_description,
long_description,
},
),
(
"UpdateNode",
DecodedTool::UpdateNode {
id,
owner,
short_name,
short_description,
long_description,
},
) => Session::update_node(
plan,
context,
id,
owner,
short_name,
short_description,
long_description,
),
_ => anyhow::bail!("decoded contract for {name} did not match its Kweb mutation"),
}
}
fn kweb_plan_projection(plan: &KwebPlan) -> (BTreeMap<String, NodeDraft>, Vec<KwebStagedCreate>) {
let updates = plan
.updates
.iter()
.map(|(id, node)| (id.clone(), kweb_node_draft(node)))
.collect();
let creates = plan
.creates
.iter()
.map(|create| KwebStagedCreate {
pending_id: create.pending_id.clone(),
data: kweb_node_draft(&create.data),
})
.collect();
(updates, creates)
}
impl Session {
pub fn mark_previous_ingress_attempt_timed_out(&mut self) {
if matches!(self.mode, AgentMode::Ingress { .. }) {
self.previous_ingress_attempt_timed_out = true;
}
}
fn ingress_time_remaining(&mut self) -> anyhow::Result<Option<u64>> {
if !matches!(self.mode, AgentMode::Ingress { .. }) {
return Ok(None);
}
ingress_time_remaining_at(&mut self.ingress_deadline, Instant::now()).map(Some)
}
fn runtime_budget(&self) -> RuntimeBudget {
let Some(provider_deadline) = self.active_provider_deadline else {
return RuntimeBudget::default();
};
let mut time_limits = vec![TimeBudget {
kind: TimeBudgetKind::ProviderCall,
remaining: remaining_until(provider_deadline),
}];
if matches!(self.mode, AgentMode::FreeTime)
&& let Some(work_deadline) = deadline(&self.free_time)
{
time_limits.push(TimeBudget {
kind: TimeBudgetKind::SelfTimeWork,
remaining: remaining_until(work_deadline),
});
}
if let Some(outer) = self.active_turn_deadline {
time_limits.push(TimeBudget {
kind: match outer.kind {
TurnDeadlineKind::Telegram => TimeBudgetKind::TelegramTurn,
TurnDeadlineKind::SelfTimeHardStop => TimeBudgetKind::SelfTimeHardStop,
},
remaining: remaining_until(outer.at),
});
}
RuntimeBudget {
rounds: Some(RoundBudget {
used: self.rounds_used,
limit: kcode_agent_runtime::DEFAULT_ROUND_LIMIT,
}),
time_limits,
}
}
fn projection(&self) -> ContextProjection {
self.journal
.state()
.projection_with_footer_lines(&self.runtime_budget().footer_lines())
}
fn provider_material_fingerprint(&self, tool_description: &str) -> String {
let material = json!({
"model":self.runtime.model,
"reasoningEffort":self.runtime.reasoning_effort,
"tool":"call_ktool",
"toolDescription":tool_description,
});
hex::encode(Sha256::digest(
serde_json::to_vec(&material).expect("provider material always serializes"),
))
}
fn begin_provider_call_budget(&mut self, timeout: Option<Duration>) {
self.active_provider_deadline = timeout.and_then(|timeout| {
chrono::Duration::from_std(timeout)
.ok()
.map(|timeout| Utc::now() + timeout)
});
}
fn synchronize_provider_known_events(&mut self) {
if let Some(affinity) = self.provider_affinity.as_mut()
&& let Some(event) = self.journal.state().events.last()
{
affinity.synchronized_event_id = event.id;
}
}
pub async fn new(
api: Service,
system_prompt: String,
subagent_codex_prompt: String,
runtime: RuntimeModel,
started_at: String,
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();
DateTime::parse_from_rfc3339(&started_at).context("session start timestamp is invalid")?;
if let Some(restored_started_at) = restored
.and_then(|state| state.get("startedAt"))
.and_then(Value::as_str)
{
anyhow::ensure!(
restored_started_at == started_at,
"restored session start timestamp changed"
);
}
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 fresh_ingress_attempt = matches!(options.mode, AgentMode::Ingress { .. })
&& !journal.is_sealed()
&& journal.state().history_ingress_started;
if fresh_ingress_attempt {
journal.reset_history_ingress_attempt(now())?;
}
let mut context = KwebContext::with_fixed_connections(
options.root_node_ids.clone(),
api.loads_fixed_connections(),
)
.map_err(anyhow::Error::new)?;
restore_kweb_context(&journal, &mut context)?;
let plan = if fresh_ingress_attempt {
KwebPlan::default()
} else {
KwebPlan::restore(restored, &journal)?
};
let transcript = kcode_kennedy_session_ingress::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());
let provider_affinity = (!fresh_ingress_attempt)
.then(|| restored.and_then(|state| state.get("providerAffinity")))
.flatten()
.filter(|value| !value.is_null())
.cloned()
.map(serde_json::from_value)
.transpose()
.context("restored provider affinity is invalid")?;
let next_thread_reset_reason = (!fresh_ingress_attempt)
.then(|| {
restored
.and_then(|state| state.get("nextThreadResetReason"))
.and_then(Value::as_str)
.map(str::to_owned)
})
.flatten();
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,
subagent_codex_prompt,
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: (!fresh_ingress_attempt)
.then(|| {
restored
.and_then(|state| state.get("roundsUsed"))
.and_then(Value::as_u64)
})
.flatten()
.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,
active_provider_deadline: None,
active_turn_deadline: None,
provider_affinity,
next_thread_reset_reason,
ingress_deadline: None,
previous_ingress_attempt_timed_out: false,
};
if matches!(session.mode, AgentMode::Ingress { .. }) && !session.journal.is_sealed() {
session.journal.repair_unfinished_tools(now())?;
}
if session.journal.is_sealed() {
session.provider_affinity = None;
session.next_thread_reset_reason = None;
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.kmap(), &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 fresh_ingress_attempt {
session.revalidate_loaded_nodes().await?;
session.pending_turn = true;
}
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.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.kmap(), &mut self.context, &direct)?;
self.sync_kweb_boxes()?;
Ok(())
}
fn stage_user_input(&mut self, text: &str, metadata: &Value) -> Option<InputStage> {
let recorded_at = now();
let result = (|| -> anyhow::Result<Option<InputStage>> {
let Some(staged) = kcode_kennedy_session_ingress::stage_user_input(
&mut self.journal,
text,
metadata,
&recorded_at,
)?
else {
return Ok(None);
};
self.transcript.push(staged.transcript);
self.recover_context_overflow(staged.external_event_id.as_deref(), &[])?;
Ok(Some(InputStage::Accepted))
})();
match result {
Ok(stage) => stage,
Err(error) => {
self.fatal_persistence_error = Some(error.to_string());
tracing::error!(error=%error, "Could not durably stage session input");
Some(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 staged = kcode_kennedy_session_ingress::stage_source_input(
&mut self.journal,
kennedy,
text,
metadata,
&now(),
)?;
self.transcript.push(staged.transcript);
self.recover_context_overflow(staged.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();
kcode_kennedy_session_objects::resolve_object(
&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 api = self.api.clone();
kcode_kennedy_session_objects::resolve_media_object(
&mut self.journal,
object_id,
MAX_MEDIA_ENRICHMENT_BYTES,
move |canonical_id| api.kmap_file(canonical_id).map_err(Into::into),
)
}
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.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.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()
.current_ingress_attempt_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![
render(RenderRequest::FreeTimeOpening {
free_time: &self.free_time,
})
.expect("free-time opening rendering is infallible"),
];
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 = render(RenderRequest::WakeupOpening { 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 >= kcode_agent_runtime::DEFAULT_ROUND_LIMIT
{
self.rounds_used = 0;
}
}
pub fn interrupt_current_turn(&mut self) -> anyhow::Result<()> {
self.provider_affinity = None;
self.next_thread_reset_reason = Some("prior_provider_turn_interrupted".into());
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,
turn_deadline: Option<TurnDeadline>,
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);
}
self.active_turn_deadline = turn_deadline;
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: kcode_agent_runtime::DEFAULT_ROUND_LIMIT,
};
let mut host = KennedySessionHost {
session: self,
checkpoint: &mut checkpoint,
accounting: None,
pending_freeform_write: None,
deadline_after_response: false,
operation_id,
prepared_cache: None,
};
let result = runtime.run_session(request, &mut host).await;
drop(host);
self.active_provider_deadline = None;
self.active_turn_deadline = None;
let result = result?;
match self.mode {
AgentMode::Conversation => {
if self.journal.state().source_terminated {
self.provider_affinity = None;
self.next_thread_reset_reason = None;
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;
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,
model: String,
reasoning_effort: Option<String>,
context_node_ids: Vec<String>,
task: String,
parent_operation_id: Uuid,
) -> anyhow::Result<String> {
let reasoning_effort =
reasoning_effort.unwrap_or_else(|| self.runtime.reasoning_effort.clone());
let mut selected_node_descriptions = Vec::with_capacity(context_node_ids.len());
for node_id in &context_node_ids {
selected_node_descriptions.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 provider = runtime.resolve_model(&model).await?.provider;
let first_event = self.journal.state().events.len();
let cost_before = self.projection().status;
let subagent_context = SubagentContext::new(
self.root_node_ids.clone(),
self.api.loads_fixed_connections(),
provider,
self.subagent_codex_prompt.clone(),
selected_node_descriptions,
)?;
let initial_sections = subagent_context.initial_sections().to_vec();
let result = {
let mut host = KennedySubagentHost {
session: self,
context: subagent_context,
captures: HashMap::new(),
};
runtime
.run(
kcode_agent_runtime::RunRequest {
user_id,
parent_operation_id,
model,
reasoning_effort,
context: initial_sections,
task,
timeout,
start_metadata: json!({"contextNodeIds":context_node_ids}),
},
&mut host,
)
.await
};
match result {
Ok(result) => {
let cost_after = self.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,
context: &mut SubagentContext,
request: FreeformWrite,
contents: String,
budget: &kcode_agent_runtime::ContextBudget,
) -> anyhow::Result<kcode_agent_runtime::ToolOutcome> {
let kind = request.kind();
let freeform_tool = request.write_tool();
anyhow::ensure!(
context.source_is_open(kind, request.name()),
"{} {:?} is not open in this subagent context. Call {} first.",
kind.label(),
request.name(),
kind.open_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 preview_state = context.source_state(&preview);
anyhow::ensure!(
budget.fits_state(preview_state.key, preview_state.text),
"{freeform_tool} was not run because its resulting source state would exceed the subagent context limit"
);
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")?;
let state = context.apply_source_snapshot(snapshot);
Ok(kcode_agent_runtime::ToolOutcome {
text: execution.text,
ok: true,
state_updates: state.update.into_iter().collect(),
displayed_state_keys: Vec::new(),
capture: None,
})
}
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>> {
let api = self.api.clone();
kcode_kennedy_session_objects::delivery_attachments(
&mut self.journal,
requests,
move |canonical_id| api.kmap_file(canonical_id).map_err(Into::into),
)
}
async fn execute_tool(
&mut self,
call: &ToolCall,
operation_id: Uuid,
) -> anyhow::Result<ToolOutcome> {
self.assert_tool_allowed(&call.name)?;
let decoded = decode(&call.name, &call.arguments)?;
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(), decoded) {
("NoteToSelf", None) => {
decode_note_to_self(&call.arguments)?;
store_result = false;
"Note saved.".into()
}
("SendTelegramDM", _) => self.send_telegram_dm(&call.arguments).await?,
("SendTelegramGroupMessage", _) => {
self.send_telegram_group_message(&call.arguments).await?
}
(
"RunSubagent",
Some(DecodedTool::RunSubagent {
model,
reasoning_effort,
context_node_ids,
task,
}),
) => {
let first_event = self.journal.state().events.len();
match self
.run_subagent(
model,
reasoning_effort,
context_node_ids,
task,
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", Some(DecodedTool::EndSession { 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) = message.filter(|message| !message.trim().is_empty())
{
self.free_time["nextSessionMessage"] = json!(message);
}
"Session ending.".into()
}
("DehydrateBoxes", Some(DecodedTool::BoxIds(ids))) => {
self.journal.dehydrate_boxes(now(), &ids)?;
format!(
"Dehydrated boxes {}.",
ids.iter()
.map(ToString::to_string)
.collect::<Vec<_>>()
.join(", ")
)
}
("SummarizeBox", Some(DecodedTool::SummarizeBox { box_id, summary })) => {
self.journal.summarize_box(now(), box_id, summary)?;
format!("Summarized box {box_id}.")
}
("HydrateBox", Some(DecodedTool::BoxId(id))) => {
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", Some(DecodedTool::BoxIds(ids))) => {
kcode_kennedy_box_text_objects::stage_box_text_objects(
&mut self.journal,
&ids,
&now(),
)?
}
("LoadNodes", Some(DecodedTool::LoadNodes(identifiers))) => {
load_durable_batch(self.api.kmap(), &mut self.context, &identifiers)?;
let changed = self.sync_kweb_boxes()?;
store_result = false;
render_load_nodes_result(
&self.journal,
&changed,
&self.runtime_budget().footer_lines(),
)?
}
(
"EmitObject",
Some(DecodedTool::EmitObject {
object_id,
file_name,
}),
) => {
anyhow::ensure!(
matches!(self.mode, AgentMode::Conversation),
"EmitObject is only available in a conversation"
);
let object = self.resolve_object(&object_id)?;
let file_name = file_name.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", Some(DecodedTool::WebSearch { question, model })) => {
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,
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", Some(DecodedTool::WebFetch(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,
operation_id: Uuid::new_v4(),
parent_operation_id: Some(operation_id),
},
)
.await?;
render_web_fetch_result(&result)?
}
("StageTelegramGroupMedia", Some(DecodedTool::StageTelegramGroupMedia(message_id))) => {
let media_ref = kcode_telegram_session_coordinator::group_media_reference(
&self.group_context,
message_id,
)?;
let chat_id = media_ref.chat_id;
let api = self.api.clone();
let staged = kcode_kennedy_session_objects::stage_telegram_group_media(
&mut self.journal,
kcode_kennedy_session_objects::TelegramStageRequest {
chat_id,
message_id,
maximum_bytes: MAX_MEDIA_ENRICHMENT_BYTES,
transport_metadata: media_ref.transport_metadata(),
recorded_at: now(),
},
|| api.telegram().group_message_media(chat_id, message_id),
|media_type| {
kcode_telegram_session_coordinator::group_media_file_name(
&media_ref, media_type,
)
},
)?;
render(RenderRequest::StagedTelegramMedia {
pending_id: &staged.descriptor.pending_id,
kind: &staged.kind,
file_name: &staged.descriptor.file_name,
media_type: &staged.descriptor.media_type,
size_bytes: staged.descriptor.size_bytes,
message_id,
reused: staged.reused,
})?
}
(
"TranscribeAudio",
Some(DecodedTool::MediaEnrichment {
object_id,
model,
prompt,
}),
) => {
let object = self.resolve_media_object(&object_id)?;
validate(ValidationRequest::TranscribableAudio(&object.media_type))?;
validate(ValidationRequest::TranscriptionModel(&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,
None,
operation_id,
)
.await;
let result = self.project_descendant(outcome)?;
render_audio_transcription_result(
&object.object_id,
&object.file_name,
&object.media_type,
&result,
)?
}
(
"AnnotateMedia",
Some(DecodedTool::MediaEnrichment {
object_id,
model,
prompt,
}),
) => {
let media = self.resolve_media_object(&object_id)?;
validate(ValidationRequest::Annotation {
model: &model,
media_type: &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",
Some(DecodedTool::GenerateImage {
model,
prompt,
reference_object_ids,
}),
) => {
let mut references = Vec::with_capacity(reference_object_ids.len());
for object_id in &reference_object_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", Some(DecodedTool::ObjectId(object_id))) => {
let object = self.resolve_media_object(&object_id)?;
validate(ValidationRequest::ExtractableDocument {
media_type: &object.media_type,
file_name: &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, None) if SPEECH_CLASSIFICATION_TOOLS.contains(&name) => {
self.api
.execute_speech_classification_tool(name, call.arguments.clone())
.await?
}
(name, None) if TASK_BOARD_TOOLS.contains(&name) => {
self.execute_task_board_tool(name, &call.arguments).await?
}
(name, Some(decoded)) if is_kweb_mutation(name) => {
let text = execute_kweb_mutation(
name,
decoded,
&self.context,
&mut self.plan,
&mut self.journal,
)?;
self.sync_kweb_boxes()?;
text
}
(name, None)
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 object_ids = if name == CALL_RUST_BIN_TOOL {
decode_managed_objects(ManagedObjectArguments::RustBinary(&call.arguments))?
} else if name == ATTACH_OBJECT_WEB_LIB_TOOL {
decode_managed_objects(ManagedObjectArguments::WebLibraryAttachment(
&call.arguments,
))?
} else {
Vec::new()
};
let mut objects = Vec::with_capacity(object_ids.len());
for object_id in object_ids {
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
}
}
(name, Some(_)) => {
anyhow::bail!("decoded contract for {name} did not match its dispatch lane")
}
(name, None) => anyhow::bail!("Tool {name} is not available"),
};
Ok(ToolOutcome {
text,
store_result,
ok: true,
end_session,
freeform_write,
managed_source_snapshot,
})
}
async fn execute_task_board_tool(
&self,
name: &str,
arguments: &Value,
) -> anyhow::Result<String> {
let board = self
.api
.task_board()
.context("task board is not configured")?
.clone();
let name = name.to_owned();
let arguments = arguments.clone();
let user_id = self
.root_node_ids
.first()
.context("session has no user root for task-category lookup")?
.clone();
tokio::task::spawn_blocking(move || -> anyhow::Result<String> {
let output = match name.as_str() {
"CreateTaskCategory" => serde_json::to_string_pretty(
&board.create_category(serde_json::from_value(arguments)?)?,
)?,
"GetTaskCategory" => {
let call: CategoryCall = serde_json::from_value(arguments)?;
serde_json::to_string_pretty(&board.category(
&call.category_id,
kcode_task_board::BrowsePage {
user_id,
offset: call.offset,
limit: call.limit,
},
)?)?
}
"RemoveTaskCategory" => {
let call: CategoryId = serde_json::from_value(arguments)?;
board.remove_category(&call.category_id)?;
format!("Removed category {}.", call.category_id)
}
"CreateTask" => serde_json::to_string_pretty(
&board.create_task(serde_json::from_value(arguments)?)?,
)?,
"GetTask" => {
let call: TaskId = serde_json::from_value(arguments)?;
serde_json::to_string_pretty(&board.task(&call.task_id)?)?
}
"UpdateTask" => serde_json::to_string_pretty(
&board.update_task(serde_json::from_value(arguments)?)?,
)?,
"RemoveTask" => {
let call: TaskId = serde_json::from_value(arguments)?;
board.remove_task(&call.task_id)?;
format!("Removed task {}.", call.task_id)
}
"GetTopTaskOrphan" => {
let _: EmptyCall = serde_json::from_value(arguments)?;
serde_json::to_string_pretty(&board.top_orphan()?)?
}
_ => anyhow::bail!("Tool {name} is not a task-board operation"),
};
Ok(output)
})
.await
.context("task-board worker stopped")?
}
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, creates) = kweb_plan_projection(&self.plan);
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 node_data(plan: &KwebPlan, context: &KwebContext, id: &str) -> anyhow::Result<PlannedNode> {
if let Some(data) = plan.created(id) {
return Ok(data.clone());
}
if let Some(data) = plan.updates.get(id) {
return Ok(data.clone());
}
let node = context
.node(id)
.with_context(|| format!("Kweb context does not contain node {id}"))?;
Ok(planned_node(node))
}
fn put_node_data(plan: &mut KwebPlan, id: &str, data: PlannedNode) -> anyhow::Result<()> {
if let Some(created) = plan.created_mut(id) {
*created = data;
} else {
canonical_id(id)?;
plan.updates.insert(id.to_owned(), data);
}
Ok(())
}
fn connect_nodes(
plan: &mut KwebPlan,
context: &KwebContext,
ids: Vec<String>,
) -> anyhow::Result<String> {
for id in &ids {
Self::ensure_known_node(plan, context, id)?;
}
for id in &ids {
let mut data = Self::node_data(plan, context, 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(plan, id, data)?;
}
Ok(format!(
"Staged connections among nodes {}.",
ids.join(", ")
))
}
fn consolidate_fanout(
plan: &mut KwebPlan,
context: &KwebContext,
parent: String,
fanout: Vec<String>,
aggregator: String,
) -> anyhow::Result<String> {
for id in std::iter::once(&parent)
.chain(std::iter::once(&aggregator))
.chain(fanout.iter())
{
Self::ensure_known_node(plan, context, id)?;
}
let mut parent_data = Self::node_data(plan, context, &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(plan, context, &aggregator)?;
for id in fanout {
if !aggregator_data.recent_connections.contains(&id) {
aggregator_data.recent_connections.push(id);
}
}
Self::put_node_data(plan, &parent, parent_data)?;
Self::put_node_data(plan, &aggregator, aggregator_data)?;
Ok(format!(
"Staged fanout consolidation from node {parent} into node {aggregator}."
))
}
fn set_fixed_connection(
plan: &mut KwebPlan,
context: &KwebContext,
parent: String,
child: Option<String>,
slot: usize,
) -> anyhow::Result<String> {
Self::ensure_known_node(plan, context, &parent)?;
if let Some(child) = &child {
Self::ensure_known_node(plan, context, child)?;
anyhow::ensure!(child != &parent, "a node cannot connect to itself");
}
let mut data = Self::node_data(plan, context, &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(plan, &parent, data)?;
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(
plan: &mut KwebPlan,
context: &KwebContext,
journal: &mut HistorySession,
arguments: CreateNodeArguments,
) -> anyhow::Result<String> {
let CreateNodeArguments {
parents,
owner,
short_name,
short_description,
long_description,
} = arguments;
for id in parents.iter().chain(std::iter::once(&owner)) {
if id != "self" && id != "unowned" {
Self::ensure_known_node(plan, context, id)?;
}
}
let pending = journal.allocate_pending_node(now())?.to_string();
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(plan, context, &parent)?;
data.recent_connections.retain(|id| id != &pending);
data.recent_connections.insert(0, pending.clone());
Self::put_node_data(plan, &parent, data)?;
}
Ok(format!("Created staged node {pending}."))
}
fn update_node(
plan: &mut KwebPlan,
context: &KwebContext,
id: String,
owner: String,
short_name: String,
short_description: String,
long_description: String,
) -> anyhow::Result<String> {
Self::ensure_known_node(plan, context, &id)?;
if owner != "self" && owner != "unowned" {
Self::ensure_known_node(plan, context, &owner)?;
}
let mut data = Self::node_data(plan, context, &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(plan, &id, data)?;
Ok(format!("Staged the update to node {id}."))
}
fn ensure_known_node(plan: &KwebPlan, context: &KwebContext, id: &str) -> anyhow::Result<()> {
if id.starts_with("pending:") {
anyhow::ensure!(
plan.created(id).is_some(),
"pending node {id} is not part of this session"
);
} else {
canonical_id(id)?;
anyhow::ensure!(
context.contains_full_node(id) || plan.updates.contains_key(id),
"node {id} is not loaded; call LoadNodes first"
);
}
Ok(())
}
fn finalize_kweb_session(&mut self) -> anyhow::Result<()> {
self.provider_affinity = None;
self.next_thread_reset_reason = None;
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 transport_kind =
kcode_kennedy_session_objects::staged_descriptor(&self.journal, &id)?
.transport_kind;
let bytes = encode_file(
&pending_id,
location.metadata.file_name.as_deref(),
&location.metadata.media_type,
transport_kind.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 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
+ SELF_TIME_HARD_STOP_ALLOWANCE.as_secs(),
));
}
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.projection();
let submitted = self
.journal
.state()
.current_ingress_attempt_events()
.iter()
.rev()
.find_map(|event| {
let EventKind::ProviderInputSubmitted { round, context, .. } = &event.kind else {
return None;
};
Some((event.recorded_at.as_str(), *round, context))
});
let (chatend_text, chatend_text_source, structured_material) = match submitted {
Some((submitted_at, round, submitted)) => (
submitted.input.clone(),
"submitted",
json!({
"provider":submitted.provider,
"model":submitted.model,
"reasoningEffort":submitted.reasoning_effort,
"baseInstructions":submitted.base_instructions,
"developerInstructions":submitted.developer_instructions,
"tools":submitted.tools,
"round":round,
"submittedAt":submitted_at,
}),
),
None => (projection.render(), "reconstructed", Value::Null),
};
let session_status = projection.status.clone();
Ok(json!({
"format":"kennedy-chatend",
"version":1,
"stateVersion":4,
"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,
"providerAffinity":self.provider_affinity,
"nextThreadResetReason":self.next_thread_reset_reason,
"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,
"chatendTextSource":chatend_text_source,
"structuredMaterial":structured_material,
}))
}
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.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 ingress_time_remaining = self.session.ingress_time_remaining()?;
let timeout = self.session.agent_request_timeout();
self.session.begin_provider_call_budget(timeout);
let tool_description = call_ktool_description();
let material_fingerprint = self
.session
.provider_material_fingerprint(&tool_description);
let mut thread_reset_reason = self.session.next_thread_reset_reason.take();
let mut continuation = None;
let mut resume_after = None;
if let Some(affinity) = &self.session.provider_affinity {
if affinity.material_fingerprint == material_fingerprint {
continuation = Some(affinity.continuation.clone());
resume_after = Some(affinity.synchronized_event_id);
} else {
self.session.provider_affinity = None;
thread_reset_reason = Some("provider_material_changed".into());
}
}
if continuation.is_some() {
self.session.provider_affinity = None;
self.session.next_thread_reset_reason =
Some("prior_provider_turn_ambiguous".into());
}
let footer_lines = self.session.runtime_budget().footer_lines();
let prepared = if let Some(remaining_seconds) = ingress_time_remaining {
self.session
.journal
.prepare_provider_projection_with_ingress_time(
now(),
&footer_lines,
&material_fingerprint,
resume_after,
remaining_seconds,
self.session.previous_ingress_attempt_timed_out,
)?
} else {
self.session.journal.prepare_provider_projection(
now(),
&footer_lines,
&material_fingerprint,
resume_after,
)?
};
if let Some(reason) = prepared.thread_reset_reason.clone() {
self.session.provider_affinity = None;
continuation = None;
thread_reset_reason = Some(reason);
}
if continuation.is_none() && thread_reset_reason.is_some() {
self.session.next_thread_reset_reason = thread_reset_reason.clone();
}
let input = prepared.projection.render();
let projection_hash = hex::encode(Sha256::digest(input.as_bytes()));
let provider_input_hash =
hex::encode(Sha256::digest(prepared.provider_input.as_bytes()));
let provider_input_bytes = prepared.provider_input.len() as u64;
let thread_action = if continuation.is_some() {
"resume"
} else {
"start"
}
.to_owned();
self.prepared_cache = Some(PreparedCacheObservation {
cacheable_prefix_bytes: prepared.cacheable_prefix_bytes,
expectation: prepared.expectation,
material_fingerprint,
projection_hash,
logical_input: input.clone(),
provider_input_hash,
provider_input_bytes,
thread_action,
thread_reset_reason,
estimated_input_tokens: prepared.projection.estimated_tokens,
raw_estimated_input_tokens: prepared.projection.raw_estimated_tokens,
provider: String::new(),
model: self.session.runtime.model.clone(),
});
Ok(kcode_agent_runtime::RoundPreparation::Run(
kcode_agent_runtime::PreparedRound {
input,
provider_input: prepared.provider_input,
continuation,
model: self.session.runtime.model.clone(),
reasoning_effort: self.session.runtime.reasoning_effort.clone(),
tool_description,
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 prepared = self
.prepared_cache
.as_ref()
.context("inference was submitted before provider context preparation")?;
anyhow::ensure!(
prepared.projection_hash == manifest_hash,
"provider input hash changed after context preparation"
);
self.accounting = Some(kcode_intelligence_chatend::TopLevelCall::new(
manifest_hash.clone(),
model,
));
self.session.journal.record(
now(),
EventKind::InferenceSubmitted {
manifest_hash,
estimated_input_tokens: prepared.estimated_input_tokens,
raw_estimated_input_tokens: Some(prepared.raw_estimated_input_tokens),
},
)?;
}
kcode_agent_runtime::SessionEvent::ProviderInput { round, context } => {
let prepared = self
.prepared_cache
.as_ref()
.context("provider context arrived before context preparation")?;
anyhow::ensure!(
hex::encode(Sha256::digest(context.input.as_bytes()))
== prepared.provider_input_hash,
"provider submitted transport input different from the prepared continuation delta"
);
let provider = context.provider.clone();
let model = context.model.clone();
self.session.journal.record(
now(),
EventKind::ProviderInputSubmitted {
round,
context: ProviderContext {
input: prepared.logical_input.clone(),
provider: context.provider,
model: context.model,
reasoning_effort: context.reasoning_effort,
base_instructions: context.base_instructions,
developer_instructions: context.developer_instructions,
tools: context
.tools
.into_iter()
.map(|tool| ProviderToolDefinition {
name: tool.name,
description: tool.description,
input_schema: tool.input_schema,
})
.collect(),
},
transport_input_hash: Some(prepared.provider_input_hash.clone()),
transport_input_bytes: Some(prepared.provider_input_bytes),
thread_action: Some(prepared.thread_action.clone()),
thread_reset_reason: prepared.thread_reset_reason.clone(),
cacheable_prefix_bytes: prepared.cacheable_prefix_bytes,
material_fingerprint: prepared.material_fingerprint.clone(),
cache_expectation: prepared.expectation.label().into(),
planned_invalidation_reason: prepared
.expectation
.planned_reason()
.map(str::to_owned),
},
)?;
if let Some(prepared) = self.prepared_cache.as_mut() {
prepared.provider = provider;
prepared.model = model;
}
}
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,
receipt,
continuation,
..
} => {
self.accounting
.take()
.context("provider receipt arrived before inference submission")?
.completed(&mut self.session.journal, &now(), usage.as_ref())?;
let prepared = self
.prepared_cache
.take()
.context("provider receipt arrived before context preparation")?;
if let Some(continuation) = continuation {
anyhow::ensure!(
receipt.provider_thread_id.as_deref()
== Some(continuation.thread_id.as_str()),
"provider receipt thread differs from continuation state"
);
let synchronized_event_id = self
.session
.journal
.state()
.events
.last()
.context("provider completion did not create a journal event")?
.id;
self.session.provider_affinity = Some(ProviderAffinityState {
continuation,
synchronized_event_id,
material_fingerprint: prepared.material_fingerprint.clone(),
});
self.session.next_thread_reset_reason = None;
} else {
self.session.provider_affinity = None;
self.session.next_thread_reset_reason = Some(
if prepared.thread_action == "resume" {
"provider_thread_resume_unavailable"
} else {
"provider_continuation_unavailable"
}
.into(),
);
}
log_primary_thread_observation(
self.operation_id,
self.session.rounds_used,
&self.session.runtime.model,
&prepared,
receipt.provider_thread_id.as_deref(),
usage.as_ref().map_or(0, |usage| usage.input_tokens),
usage.as_ref().map_or(0, |usage| usage.cached_input_tokens),
);
}
}
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 footer = self.session.projection().footer;
let text = provider_tool_result_with_context_footer(&footer, &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 = invocation_box_content(&call.name, &call.arguments)?;
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 footer = self.session.projection().footer;
let provider_text = provider_tool_result_with_context_footer(&footer, &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);
self.session.synchronize_provider_known_events();
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(invocation_box_content(&call.name, &call.arguments)?.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 let Some(reason) = subagent_unavailable_reason(&call.name) {
return Ok(kcode_agent_runtime::ToolOutcome::failure(reason));
}
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.context, &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();
if call.name == "LoadNodes" {
let Some(DecodedTool::LoadNodes(identifiers)) =
decode(&call.name, &call.arguments)?
else {
return Ok(kcode_agent_runtime::ToolOutcome::failure(
"LoadNodes did not match its tool contract.",
));
};
load_durable_batch(
self.session.api.kmap(),
self.context.kweb_mut(),
&identifiers,
)?;
let (updates, creates) = kweb_plan_projection(&self.session.plan);
let changes = self.context.reconcile_kweb(&updates, &creates)?;
let displayed_state_keys = changes.displayed_state_keys();
let mut text = if changes.is_empty() {
"LoadNodes completed. The subagent Kweb projection was already current.".into()
} else {
changes.display_text()
};
append_slow_tool_duration(&mut text, tool_started_at.elapsed());
return Ok(kcode_agent_runtime::ToolOutcome {
text,
ok: true,
state_updates: changes.updates,
displayed_state_keys,
capture: None,
});
}
if is_kweb_mutation(&call.name) {
self.session.assert_tool_allowed(&call.name)?;
let decoded = decode(&call.name, &call.arguments)?
.with_context(|| format!("{} did not match its tool contract", call.name))?;
let referenced_pending = referenced_pending_nodes(&decoded);
let prior_create_count = self.session.plan.creates.len();
let mut text = execute_kweb_mutation(
&call.name,
decoded,
self.context.kweb(),
&mut self.session.plan,
&mut self.session.journal,
)?;
self.context.include_staged_nodes(
referenced_pending.into_iter().chain(
self.session.plan.creates[prior_create_count..]
.iter()
.map(|create| create.pending_id.clone()),
),
);
let (updates, creates) = kweb_plan_projection(&self.session.plan);
let changes = self.context.reconcile_kweb(&updates, &creates)?;
append_slow_tool_duration(&mut text, tool_started_at.elapsed());
return Ok(kcode_agent_runtime::ToolOutcome {
text,
ok: true,
state_updates: changes.updates,
displayed_state_keys: Vec::new(),
capture: None,
});
}
if let Some(request) = decode_freeform_write(&call.name, &call.arguments)? {
if !self.context.source_is_open(request.kind(), request.name()) {
return Ok(kcode_agent_runtime::ToolOutcome::failure(format!(
"{} {:?} is not open in this subagent context. Call {} first.",
request.kind().label(),
request.name(),
request.kind().open_tool()
)));
}
let acknowledgement = request.acknowledgement();
let id = Uuid::new_v4().to_string();
self.captures.insert(id.clone(), request);
return Ok(kcode_agent_runtime::ToolOutcome {
text: acknowledgement,
ok: true,
state_updates: Vec::new(),
displayed_state_keys: Vec::new(),
capture: Some(Value::String(id)),
});
}
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));
}
};
let displays_managed_snapshot = outcome
.managed_source_snapshot
.as_ref()
.is_some_and(|snapshot| result_displays_snapshot(&outcome.text, snapshot));
append_slow_tool_duration(&mut outcome.text, tool_started_at.elapsed());
let (state_updates, displayed_state_keys) =
if let Some(snapshot) = outcome.managed_source_snapshot.take() {
let state = self.context.apply_source_snapshot(snapshot);
let displayed = displays_managed_snapshot.then_some(state.key);
(
state.update.into_iter().collect(),
displayed.into_iter().collect(),
)
} else {
(Vec::new(), Vec::new())
};
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,
displayed_state_keys,
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")?;
self.session
.complete_subagent_freeform_write(&mut self.context, request, contents, &budget)
.await
})
}
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 {
render(RenderRequest::CostSummary {
label,
estimated_cost_usd_nanos,
unpriced_calls,
})
.expect("cost-summary rendering is infallible")
}
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()
.current_ingress_attempt_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(),
"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 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 canonical_id(value: &str) -> anyhow::Result<String> {
value
.parse::<NodeId>()
.with_context(|| format!("{value:?} is not a canonical node ID"))?;
Ok(value.into())
}
fn image_extension(media_type: &str) -> &'static str {
match media_type
.split(';')
.next()
.unwrap_or(media_type)
.trim()
.to_ascii_lowercase()
.as_str()
{
"image/jpeg" => "jpg",
"image/webp" => "webp",
_ => "png",
}
}
fn call_ktool_description() -> String {
render(RenderRequest::CallKtoolDescription).expect("Ktool-description rendering is infallible")
}
fn now() -> String {
Utc::now().to_rfc3339()
}
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 remaining_until(deadline: DateTime<Utc>) -> Duration {
(deadline - Utc::now()).to_std().unwrap_or(Duration::ZERO)
}
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 {
let mode = match mode {
AgentMode::Conversation => "conversation",
AgentMode::FreeTime => "free-time",
AgentMode::Wakeup => "wakeup",
AgentMode::Ingress { .. } => "ingress",
};
render(RenderRequest::ControllerMessage { mode, free_time })
.expect("known controller modes render successfully")
}
#[cfg(test)]
mod tests {
use std::time::{SystemTime, UNIX_EPOCH};
use kcode_session_history::{Config as HistoryConfig, NewSession, SessionHistory};
use super::*;
fn test_journal(label: &str) -> (std::path::PathBuf, HistorySession) {
let root = std::env::temp_dir().join(format!(
"kcode-kennedy-sessions-{label}-{}-{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos()
));
let history = SessionHistory::open(HistoryConfig {
directory: root.join("sessions"),
completed_list: root.join("completed.jsonl"),
provider_cost_compatibility: None,
})
.unwrap();
let journal = history
.create_session(NewSession {
kind: SessionKind::SelfTime,
created_at: "2026-08-05T00:00:00Z".into(),
effective_context_tokens: 10_000,
channel: Value::Null,
})
.unwrap();
(root, journal)
}
#[test]
fn subagents_reject_parent_controls_but_allow_delivery_effects() {
for unavailable in [
"RunSubagent",
"EndSession",
"DehydrateBoxes",
"SummarizeBox",
"HydrateBox",
"BoxesIntoObjects",
] {
assert!(subagent_unavailable_reason(unavailable).is_some());
}
for delegated in [
"NoteToSelf",
"EmitObject",
"SendTelegramDM",
"SendTelegramGroupMessage",
"LoadNodes",
"ExtractDocumentText",
] {
assert_eq!(subagent_unavailable_reason(delegated), None);
}
}
#[test]
fn only_complete_snapshot_results_claim_to_display_managed_state() {
let snapshot = SourceSnapshot {
kind: kcode_dev_tools::ManagedSourceKind::RustLibrary,
name: "example".into(),
text: "complete source".into(),
};
assert!(result_displays_snapshot("complete source", &snapshot));
assert!(!result_displays_snapshot(
"Wrote file src/lib.rs in Rust library example.",
&snapshot
));
}
#[test]
fn provider_affinity_round_trips_through_snapshot_json() {
let affinity = ProviderAffinityState {
continuation: kcode_intelligence_router::AgentContinuation {
thread_id: "thread-1".into(),
provider_model: "gpt-5.6-sol".into(),
cumulative_input_tokens: 120,
cumulative_output_tokens: 30,
cumulative_cached_input_tokens: 80,
cumulative_reasoning_output_tokens: 10,
},
synchronized_event_id: EventId(42),
material_fingerprint: "material".into(),
};
let restored: ProviderAffinityState =
serde_json::from_value(serde_json::to_value(&affinity).unwrap()).unwrap();
assert_eq!(restored, affinity);
}
#[test]
fn ingress_deadline_starts_at_2700_seconds_rounds_down_and_expires_safely() {
let now = Instant::now();
let mut deadline = None;
assert_eq!(
ingress_time_remaining_at(&mut deadline, now).unwrap(),
2_700
);
let mut near_deadline = Some(now + Duration::from_millis(1_500));
assert_eq!(
ingress_time_remaining_at(&mut near_deadline, now).unwrap(),
1
);
let expired =
ingress_time_remaining_at(&mut near_deadline, now + Duration::from_millis(1_500))
.unwrap_err();
assert!(is_ingress_time_expired(&expired));
}
#[test]
fn child_kweb_mutation_changes_plan_without_creating_parent_boxes() {
let (root, mut journal) = test_journal("subagent-kweb");
let node_id = "AAAAAAAE".to_owned();
let mut context = KwebContext::new(vec![node_id.clone()]).unwrap();
context
.apply_load(
KwebNode {
id: node_id.clone(),
short_name: "Old".into(),
short_description: "Old summary".into(),
long_description: "Old details".into(),
owner: "self".into(),
fixed_connections: Vec::new(),
recent_connections: Vec::new(),
objects: Vec::new(),
last_modified_by: "test".into(),
last_modified_at: None,
},
Vec::new(),
)
.unwrap();
let mut plan = KwebPlan::default();
let result = execute_kweb_mutation(
"UpdateNode",
DecodedTool::UpdateNode {
id: node_id.clone(),
owner: "self".into(),
short_name: "New".into(),
short_description: "New summary".into(),
long_description: "New details".into(),
},
&context,
&mut plan,
&mut journal,
)
.unwrap();
assert_eq!(result, format!("Staged the update to node {node_id}."));
assert_eq!(plan.updates[&node_id].long_description, "New details");
assert!(journal.state().boxes.is_empty());
std::fs::remove_dir_all(root).unwrap();
}
}