# Kennedy orchestration
This library owns shared typed session control and non-transport scheduling. Session History owns durable records and commands, transport runtimes own their queues, and Kweb Manager owns graph state. Telegram event, group, delivery, rollover, wakeup scheduling, and mailbox adoption remain outside this library.
## Public API
### Re-exports
pub use kcode_kennedy_sessions::{
AgentMode, RuntimeModel, Service as SessionService, Session, TurnDeadline,
TurnDeadlineKind,
};
pub use kcode_kennedy_stepped_turn_runtime::{
ControlSignal, DriveFailure, Mailbox, MailboxSender, PendingTurnAdmission,
ProcessedAdmission, QueuedAdmission, TurnExit,
};
pub use prompts::Manuals;
pub use services::{Api, ApiError, LocalServices, data_url, telegram_caption_for};
pub use worker::{Orchestrator, SessionRuntime, TurnCompletion, persist_record};
The re-exported session and stepped-runtime types retain their defining libraries’ public fields, variants, methods, trait implementations, and error contracts.
### Configuration
#[derive(Clone, Debug)]
pub struct Config {
pub user_root_node_id: String,
pub kennedy_root_node_id: String,
pub telegram_max_media_bytes: usize,
pub runtime_model: RuntimeModel,
}
The root IDs must be validated canonical system roots encoded for session state. `telegram_max_media_bytes` is copied into restored Telegram channel state so session tools preserve the transport object-size limit.
### Manuals
#[derive(Clone, Debug)]
pub struct Manuals(/* private fields */);
impl Manuals {
pub fn open() -> Self;
pub fn compose_conversation(
&self,
runtime: &RuntimeModel,
session_type: &str,
session_context: &str,
opened_at: chrono::DateTime<chrono::Utc>,
) -> String;
pub fn compose_ingress(
&self,
runtime: &RuntimeModel,
source_session_type: &str,
opened_at: chrono::DateTime<chrono::Utc>,
) -> String;
pub fn subagent_codex_prompt(&self) -> &'static str;
}
`Manuals::open` opens the bundled prompt layers without filesystem I/O. Its static read-tools layer excludes the `DehydrateBoxes`, `SummarizeBox`, and `HydrateBox` manuals; native tool registration and the Root-loadable Kmap context-management instruction are owned outside this library. The writable manual uses model-facing map-marker terminology without changing storage, wire, or Ktool names.
`compose_conversation` selects conversation, self-time, or wakeup instructions and includes writable tools for self-time and wakeup. Telegram layers are included only for `telegram` and `telegram-group`, with the group layer limited to `telegram-group`. Nonempty trimmed `session_context` is included as the self-time schedule.
`compose_ingress` selects audio ingress only for `audio` and history ingress otherwise. It includes writable tools and retains Telegram group context when the source type is `telegram-group`.
Both compositions preserve a stable layer order and append runtime facts containing the supplied UTC `opened_at`, model, reasoning effort, and active failure-avoidance limit. Conversation prompts use 70% of the model context window; ingress prompts use the full window. Session opening derives `opened_at` once from durable restored state or session creation. Later rounds retain the original prompt rather than refreshing its timestamp or static prefix.
`subagent_codex_prompt` returns exactly the bundled Codex harness layer. A session applies that layer to a child only when the child’s resolved provider is Codex and does not combine it with another child static prompt.
### Service façade
#[derive(Clone)]
pub struct LocalServices {
pub kmap: kcode_kweb_manager::KwebManager,
pub intelligence: kcode_intelligence_router::Intelligence,
pub history: kcode_session_history::SessionHistory,
pub audio: kcode_audio_session_ingress::Coordinator,
pub directory: std::sync::Arc<kcode_telegram_identity::Directory>,
pub dev_tools: kcode_dev_tools::Service,
pub telegram: kcode_tg_kennedy_bot::Service,
}
All handles must belong to the same process and persistence graph.
#[derive(Debug, Clone)]
pub struct ApiError {
pub code: String,
pub message: String,
}
impl std::fmt::Display for ApiError;
impl std::error::Error for ApiError;
Expected history or identity conflicts use `state_conflict`, missing data uses `not_found`, bad input uses `invalid_request`, and unexpected owner failures use `internal_error`. Kmap conflicts retain `conflict`. Provider cancellation retains `operation_cancelled`. Internal owner diagnostics are not exposed when the owner classifies them as internal.
#[derive(Clone)]
pub struct Api(/* private fields */);
impl Api {
pub fn new(config: &Config, services: LocalServices) -> Self;
pub async fn telegram_user_lock(
&self,
telegram_user_id: i64,
) -> std::sync::Arc<tokio::sync::Mutex<()>>;
pub fn kmap_node(
&self,
node_id: &str,
) -> Result<kcode_kweb_db::Node, ApiError>;
pub fn user_root_node_id(&self) -> &str;
pub fn kennedy_root_node_id(&self) -> &str;
pub fn cancel_intelligence(
&self,
operation_id: uuid::Uuid,
) -> Result<bool, ApiError>;
pub fn history_health(&self) -> Result<(), ApiError>;
pub async fn history_list(
&self,
) -> Result<Vec<kcode_session_history::SessionRecord>, ApiError>;
pub async fn history_get_session(
&self,
id: &str,
) -> Result<kcode_session_history::SessionRecord, ApiError>;
pub async fn history_register(
&self,
input: kcode_session_history::RegisterSession,
) -> Result<kcode_session_history::SessionRecord, ApiError>;
pub async fn history_command_heads(
&self,
) -> Result<Vec<kcode_session_history::SessionCommand>, ApiError>;
pub async fn history_claim_command(
&self,
id: &str,
) -> Result<kcode_session_history::SessionCommand, ApiError>;
pub async fn history_complete_command(
&self,
id: &str,
outcome: serde_json::Value,
) -> Result<kcode_session_history::SessionCommand, ApiError>;
pub fn history_listen_for_stop(
&self,
id: &str,
) -> Result<kcode_session_history::StopListener, ApiError>;
pub async fn history_stop_heads(
&self,
) -> Result<Vec<kcode_session_history::SessionStopRequest>, ApiError>;
pub async fn history_complete_stop(
&self,
id: &str,
outcome: serde_json::Value,
) -> Result<kcode_session_history::SessionStopRequest, ApiError>;
pub async fn history_checkpoint(
&self,
id: &str,
input: kcode_session_history::Checkpoint,
) -> Result<kcode_session_history::SessionRecord, ApiError>;
pub async fn history_request_ingress(
&self,
id: &str,
input: kcode_session_history::Checkpoint,
) -> Result<kcode_session_history::SessionRecord, ApiError>;
pub async fn history_start_ingress(
&self,
id: &str,
input: kcode_session_history::StartIngress,
) -> Result<kcode_session_history::SessionRecord, ApiError>;
pub async fn history_complete_ingress(
&self,
id: &str,
expected_version: i64,
) -> Result<kcode_session_history::SessionRecord, ApiError>;
pub async fn history_fail_ingress(
&self,
id: &str,
input: kcode_session_history::IngressFailure,
) -> Result<kcode_session_history::SessionRecord, ApiError>;
pub async fn history_complete(
&self,
id: &str,
input: kcode_session_history::Checkpoint,
) -> Result<kcode_session_history::SessionRecord, ApiError>;
pub async fn history_release_interrupted_ingress(
&self,
) -> Result<Vec<String>, ApiError>;
pub fn directory_user(
&self,
telegram_user_id: i64,
) -> Result<kcode_telegram_identity::User, ApiError>;
pub fn directory_group(
&self,
group_id: &str,
) -> Result<kcode_telegram_identity::Group, ApiError>;
pub async fn release_managed_sources(&self, session_id: &str);
pub fn telegram_health(&self);
pub async fn telegram_private_sessions(
&self,
) -> Result<Vec<kcode_tg_kennedy_bot::PrivateSession>, ApiError>;
pub async fn telegram_events(
&self,
) -> Result<serde_json::Value, ApiError>;
pub async fn telegram_group_ingress(
&self,
) -> Result<serde_json::Value, ApiError>;
pub async fn telegram_complete_group_ingress(
&self,
batch_id: &str,
) -> Result<serde_json::Value, ApiError>;
pub async fn telegram_group_session_updates(
&self,
) -> Result<serde_json::Value, ApiError>;
pub async fn telegram_complete_silent_group_reset(
&self,
conversation_id: &str,
) -> Result<serde_json::Value, ApiError>;
pub async fn telegram_acknowledge_group_context(
&self,
conversation_id: &str,
through_message_id: i64,
) -> Result<serde_json::Value, ApiError>;
pub async fn telegram_detach_group_session(
&self,
conversation_id: &str,
group_id: &str,
telegram_user_id: i64,
) -> Result<serde_json::Value, ApiError>;
pub async fn telegram_save_group_message_preparation(
&self,
chat_id: i64,
message_id: i64,
text: &str,
model: Option<&str>,
format: Option<&str>,
truncated: bool,
) -> Result<serde_json::Value, ApiError>;
pub async fn telegram_bind_event(
&self,
event_id: &str,
conversation_id: &str,
expected_conversation_id: Option<&str>,
) -> Result<serde_json::Value, ApiError>;
pub async fn telegram_reply_event(
&self,
event_id: &str,
conversation_id: &str,
text: &str,
context_warning: Option<&str>,
) -> Result<serde_json::Value, ApiError>;
pub async fn telegram_abort_event(
&self,
event_id: &str,
conversation_id: Option<&str>,
message: &str,
) -> Result<serde_json::Value, ApiError>;
pub async fn telegram_interrupt_event(
&self,
event_id: &str,
conversation_id: &str,
) -> Result<serde_json::Value, ApiError>;
pub async fn telegram_complete_reset(
&self,
event_id: &str,
message: Option<&str>,
) -> Result<serde_json::Value, ApiError>;
pub fn telegram_event_media(
&self,
event_id: &str,
) -> Result<(Vec<u8>, String), ApiError>;
pub fn telegram_group_message_media(
&self,
chat_id: i64,
message_id: i64,
) -> Result<(Vec<u8>, String), ApiError>;
pub fn telegram_group_message_media_metadata(
&self,
chat_id: i64,
message_id: i64,
) -> Result<(u64, String), ApiError>;
pub async fn telegram_send_object(
&self,
event_id: &str,
conversation_id: &str,
file: &kcode_kennedy_sessions::ResolvedObject,
caption: Option<&str>,
complete: bool,
) -> Result<serde_json::Value, ApiError>;
pub async fn extract_document(
&self,
bytes: Vec<u8>,
filename: String,
mime: &str,
) -> Result<kcode_intelligence_router::DocumentExtraction, ApiError>;
pub async fn synchronize_audio_ingress(
&self,
) -> Result<(), ApiError>;
}
Mutating façade operations preserve the owning subsystem’s durable idempotency and state-transition contracts. `release_managed_sources` logs release failure instead of returning it, preventing cleanup failure from replacing the owning session outcome. A Telegram user lock is shared by all clones of the same `Api`; callers must hold it for the complete conflicting per-user operation.
pub fn data_url(mime: &str, bytes: &[u8]) -> String;
pub fn telegram_caption_for<'a>(
file: &kcode_kennedy_sessions::ResolvedObject,
text: &'a str,
) -> Option<&'a str>;
`telegram_caption_for` returns the exact supplied text only when it is nonempty, no longer than 1,024 UTF-16 code units, and the resolved native Telegram kind is not `video_note` or `sticker`.
### Runtime and orchestration
#[derive(Clone)]
pub struct SessionRuntime {
pub model: RuntimeModel,
pub user_root_node_id: String,
pub kennedy_root_node_id: String,
// manuals are private
}
pub enum TurnCompletion {
Finished,
Stopped,
}
`Stopped` means the Session History stop listener won and the provider operation was cancelled. It does not persist interruption; the caller must call `Session::interrupt_current_turn`, checkpoint the resulting state, and complete any owning external work.
pub struct Orchestrator(/* private fields */);
impl Orchestrator {
pub fn new(
config: Config,
api: Api,
sessions: SessionService,
) -> Self;
pub fn api(&self) -> &Api;
pub fn writer(
&self,
) -> &std::sync::Arc<tokio::sync::Mutex<()>>;
pub async fn run(
self: std::sync::Arc<Self>,
) -> anyhow::Result<()>;
pub async fn initialize_until_ready(&self);
pub fn runtime(&self) -> anyhow::Result<&SessionRuntime>;
pub async fn open_session(
&self,
runtime: SessionRuntime,
options: kcode_kennedy_sessions::SessionOptions,
restored: Option<&serde_json::Value>,
) -> anyhow::Result<Session>;
pub async fn pending_stop(
&self,
session_id: &str,
) -> anyhow::Result<Option<kcode_session_history::SessionStopRequest>>;
pub async fn complete_pending_stop(
&self,
session_id: &str,
outcome: serde_json::Value,
) -> anyhow::Result<()>;
pub async fn operation_is_active(
&self,
session_id: &str,
) -> bool;
pub async fn register_operation(
&self,
session_id: &str,
operation_id: uuid::Uuid,
);
pub async fn remove_operation(
&self,
session_id: &str,
operation_id: uuid::Uuid,
);
pub async fn drive_session_turn<D, C, F>(
&self,
session_id: &str,
session: &mut Session,
operation_id: uuid::Uuid,
turn_deadline: Option<TurnDeadline>,
mailbox: &mut Mailbox<D>,
checkpoint: C,
) -> Result<TurnExit<D>, DriveFailure<D>>
where
D: Send,
C: FnMut(serde_json::Value) -> F + Send,
F: std::future::Future<Output = anyhow::Result<()>> + Send;
pub async fn run_session_turn<C, F>(
&self,
session_id: &str,
session: &mut Session,
operation_id: uuid::Uuid,
turn_deadline: Option<TurnDeadline>,
checkpoint: C,
) -> anyhow::Result<TurnCompletion>
where
C: FnMut(serde_json::Value) -> F + Send,
F: std::future::Future<Output = anyhow::Result<()>> + Send;
pub async fn list_history(
&self,
) -> anyhow::Result<Vec<kcode_session_history::SessionRecord>>;
pub async fn conversation_lock(
&self,
id: &str,
) -> std::sync::Arc<tokio::sync::Mutex<()>>;
pub async fn session_for_record(
&self,
record: &kcode_session_history::SessionRecord,
) -> anyhow::Result<Session>;
pub async fn close_conversation(
&self,
record: &std::sync::Arc<
tokio::sync::Mutex<kcode_session_history::SessionRecord>,
>,
session: &Session,
) -> anyhow::Result<()>;
pub async fn request_conversation_ingress(
&self,
record: &std::sync::Arc<
tokio::sync::Mutex<kcode_session_history::SessionRecord>,
>,
state: Option<serde_json::Value>,
) -> anyhow::Result<()>;
pub async fn get_conversation(
&self,
id: &str,
) -> anyhow::Result<kcode_session_history::SessionRecord>;
pub async fn get_listed_conversation(
&self,
id: &str,
) -> anyhow::Result<Option<kcode_session_history::SessionRecord>>;
}
`initialize_until_ready` serializes concurrent initialization callers and retries root, history, and interrupted-ingress readiness until successful. `runtime` fails before readiness. `run` initializes and then schedules browser commands, history ingress, self-time, and audio synchronization indefinitely. Polling failures are logged distinctly and retried.
`open_session` creates or restores the authoritative session implementation. Restored `startedAt` is the stable session opening time; a missing value is created once. Invalid restored timestamps fail opening. `session_for_record` repairs compatibility defaults in the in-memory restoration input without mutating durable state and restores configured Telegram media limits for Telegram session types.
`get_listed_conversation` returns `None` when a record obtained from a prior listing completed and disappeared. `close_conversation` releases the session’s managed sources before requesting the versioned transition to history ingress.
All history-ingress work entering the common path uses the session library’s 45-minute safe-boundary timer. Provider or persistence errors, provider timeout, round exhaustion, user stop, and timer expiry transition the record to `ingress_failed`; orchestration does not automatically retry terminal failures. Startup repair releases interrupted writer work for explicit retry. A retry creates a fresh attempt, and only a prior timer failure adds the prior-timeout marker. A user stop checkpoints interruption without committing a partial Kweb write plan.
### Stepped turns
`drive_session_turn` registers `operation_id` before creating the durable stop listener. Listener creation failure removes the registration and returns `DriveFailure` with an empty processed list. On every later outcome, registration remains visible through provider cancellation and stepped-driver settlement and is removed before return.
The stop listener and optional absolute deadline remain active for the complete owned provider wait. Arbitration is stop-biased: if stop and deadline are simultaneously ready, `ControlSignal::Stop` wins. The deadline is also forwarded to the session for transient provider projection and is not persisted by orchestration.
`Mailbox` and `MailboxSender` carry admissions into a pending turn but are not durable queues. Admissions are accepted only at a session `Yield`, at the session’s safe boundary. The checkpoint is completed before those admissions are reported as processed. Admissions unavailable at that checkpoint remain pending for a later `Yield`.
Every ordinary `TurnExit` and `DriveFailure` after listener creation includes processed-admission metadata, including completion, interruption, and operation failure. Callers may durably acknowledge or remove only entries covered by a completed checkpoint. The caller owns the transport listener, sender lifetime, durable queue, checkpoint storage, external delivery, post-interruption `Session::interrupt_current_turn` persistence, and reconstruction of the session and unprocessed admissions after restart.
When stop or deadline wins, provider cancellation is requested before the stepped operation is aborted and joined. Cancellation failure does not replace the controlling signal. `TurnExit::Interrupted` reports the winning signal and retains processed metadata.
`run_session_turn` creates an empty `Mailbox<()>`. Completion maps to `TurnCompletion::Finished`, stop interruption to `TurnCompletion::Stopped`, deadline interruption to an error, and `DriveFailure` to its contained error. Processed admissions must remain empty. This adapter and the stepped-runtime re-exports do not establish Telegram mailbox adoption.
Self-time passes its hard stop as an internal `TurnDeadline` rather than wrapping the turn in an outer timeout. Consequently, hard-stop processing preserves `Yield` checkpointing, stop bias, provider cancellation, driver abort and join, and operation removal in that order.
### Persistence and lifecycle adapters
pub async fn persist_record(
api: &Api,
record: &std::sync::Arc<
tokio::sync::Mutex<kcode_session_history::SessionRecord>,
>,
state: serde_json::Value,
user_activity: bool,
) -> anyhow::Result<()>;
pub fn build(
config: Config,
api: Api,
sessions: SessionService,
) -> std::sync::Arc<Orchestrator>;
pub async fn run(
worker: std::sync::Arc<Orchestrator>,
) -> anyhow::Result<()>;
`persist_record` holds the supplied record mutex through the versioned checkpoint and updates the record with the returned value. After `state_conflict`, an identical state already stored is idempotent success; a different state remains an error.
The free `run` function delegates to `Orchestrator::run`. Callers own task cancellation and the lifetime of that indefinitely running operation.
## Concurrency, boundaries, and performance
The library performs no environment or filesystem discovery. Capabilities are supplied through `LocalServices`, and manuals are bundled. Generated, vendor, cache, log, and output material are not package state.
Conflicting work is bounded by one mutex per adopted conversation, one global Kweb writer mutex, one initialization mutex, one mutex per Telegram user requested through an `Api`, and short-held registries. Independent conversations and independent façade operations may proceed concurrently subject to their owning services. At most one self-time or ingress writer job is launched by an orchestrator at once. Callers that bypass the returned conversation or user locks do not receive their serialization guarantee.
Accessors and registry updates are constant work apart from mutex contention. Manual composition and `data_url` are linear in output size. Listings, caption validation, and mailbox processing are linear in returned records, UTF-16 text units, or admissions respectively. Persistence, provider, media, listener, identity, and transport methods perform one corresponding owner operation unless their stated conflict handling or lifecycle requires the additional read, cancellation, settlement, or retry. They may wait for the owner and do not add an unbounded per-operation retry. `initialize_until_ready` and the long-running scheduler are the explicit unbounded retry loops.