use crate::out::Out;
#[derive(Clone, Debug)]
pub struct Triage {
storage: crate::storage::Storage,
}
#[derive(Clone, Copy, Debug)]
pub struct InspectOptions {
pub recent: usize,
pub loop_min: usize,
pub stale_min: i64,
}
impl Default for InspectOptions {
fn default() -> Self {
Self {
recent: 40,
loop_min: 3,
stale_min: 15,
}
}
}
impl Triage {
pub fn new(pile: PathBuf, key: Option<PathBuf>) -> Self {
Self::with_storage(crate::storage::Storage::new(pile, key))
}
pub fn with_storage(storage: crate::storage::Storage) -> Self {
Self { storage }
}
fn with_snapshot(&self, operation: impl FnOnce(&TriageSnapshot) -> Result<()>) -> Result<()> {
self.storage.with_store(|store, signer, runtime| {
let snapshot = TriageSnapshot::load(store, signer, runtime)?;
operation(&snapshot)
})
}
pub fn scan(&self, options: &InspectOptions, out: &mut Out<'_>) -> Result<()> {
self.with_snapshot(|snapshot| {
scan(
snapshot,
self.storage.path(),
options.recent,
options.loop_min,
options.stale_min,
out,
)
})
}
pub fn loops(&self, recent: usize, min_repeat: usize, out: &mut Out<'_>) -> Result<()> {
self.with_snapshot(|snapshot| loops(snapshot, recent, min_repeat, out))
}
pub fn timeline(&self, recent: usize, out: &mut Out<'_>) -> Result<()> {
self.with_snapshot(|snapshot| timeline(snapshot, recent, out))
}
pub fn cover(&self, full: bool, out: &mut Out<'_>) -> Result<()> {
self.with_snapshot(|snapshot| cover(snapshot, full, out))
}
pub fn chunk(&self, id: &str, out: &mut Out<'_>) -> Result<()> {
self.with_snapshot(|snapshot| chunk(snapshot, id, out))
}
pub fn turn(&self, turn: usize, full: bool, out: &mut Out<'_>) -> Result<()> {
self.with_snapshot(|snapshot| self::turn(snapshot, turn, full, out))
}
pub fn context(&self, turn: usize, full: bool, raw: bool, out: &mut Out<'_>) -> Result<()> {
self.with_snapshot(|snapshot| context(snapshot, turn, full, raw, out))
}
}
use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use crate::memory::{self as memory_model};
use crate::memory_cover::{
all_chunk_ids, chunk_about_archive_message, chunk_about_exec_result, chunk_aliases,
chunk_end_at, chunk_image_handle, chunk_lens_handle, chunk_observed_at, chunk_references,
chunk_start_at, chunk_summary_handle,
};
use crate::schemas::cognition::DEFAULT_SCOPE_ID as COGNITION_SCOPE_ID;
use crate::schemas::headspace::DEFAULT_SCOPE_ID as HEADSPACE_SCOPE_ID;
use crate::schemas::memory::DEFAULT_SCOPE_ID as MEMORY_SCOPE_ID;
use crate::schemas::message::DEFAULT_SCOPE_ID as MESSAGE_SCOPE_ID;
use crate::schemas::relations::DEFAULT_SCOPE_ID as RELATIONS_SCOPE_ID;
use crate::schemas::triage::cog;
use crate::secrets::{storage as secret_storage, SecretsSnapshot};
#[cfg(test)]
use crate::storage::{load_signer, open_pile_strict};
use crate::storage::{open_secrets_collection_acquiring, AcquiringReader, FacultySnapshot, FacultyStore, FactArchive};
use crate::triage::{
self as triage_model, build_loop_report, collect_exec_state, collect_model_chat_state,
collect_reason_state, ExecRequestRow, ExecState, ModelChatState, ModelResultRow,
ReasonEventRow, ScanOptions, ScanSources, SourceView, TriageHeadspace, UnreadMessages,
UnreadUnavailable,
};
use anybytes::View;
use anyhow::{anyhow, bail, Context, Result};
use hifitime::Epoch;
use serde::{Deserialize, Serialize};
use triblespace::core::blob::encodings::succinctarchive::{
Rank9AcceleratedSuccinctArchiveBlob, SuccinctArchiveBlob,
};
use triblespace::core::collection::CollectionStoreExt;
use triblespace::core::query::TriblePattern;
use triblespace::core::repo::BlobStoreGet;
use triblespace::macros::{find, pattern};
use triblespace::prelude::*;
type TextHandle = Inline<inlineencodings::Handle<blobencodings::UTF8String>>;
type Interval = Inline<inlineencodings::NsTAIInterval>;
type TriageReader = AcquiringReader<FacultySnapshot>;
struct CollectionView {
facts: FactArchive,
reader: TriageReader,
}
impl CollectionView {
fn source(&self) -> SourceView<'_, FactArchive, TriageReader> {
SourceView {
facts: &self.facts,
reader: &self.reader,
}
}
}
struct TriageSnapshot {
store_snapshot: TriageReader,
collections: BTreeMap<Id, FactArchive>,
secrets: SecretsSnapshot<TriageReader>,
}
impl TriageSnapshot {
fn load(
pile: &mut FacultyStore,
signer: &ed25519_dalek::SigningKey,
runtime: &std::sync::Arc<tokio::runtime::Runtime>,
) -> Result<Self> {
let mut registered = Vec::new();
let mut succinct = Vec::new();
let mut rank9 = Vec::new();
for (scope, label) in [
(COGNITION_SCOPE_ID, "Cognition"),
(HEADSPACE_SCOPE_ID, "Headspace"),
(MEMORY_SCOPE_ID, "Memory"),
(RELATIONS_SCOPE_ID, "Relations"),
(MESSAGE_SCOPE_ID, "Message"),
] {
let source =
crate::collection_names::open_configured_acquiring(pile, scope, signer.verifying_key(), runtime)
.with_context(|| format!("register {label} collection"))?;
let succinct_collection = pile
.attach::<SuccinctArchiveBlob>(source, ())
.with_context(|| format!("register succinct {label} collection"))?;
let rank9_collection = pile
.attach::<Rank9AcceleratedSuccinctArchiveBlob>(source, succinct_collection)
.with_context(|| format!("register Rank9 {label} collection"))?;
registered.push((scope, label));
succinct.push(succinct_collection);
rank9.push(rank9_collection);
}
let secrets_collection = open_secrets_collection_acquiring(pile, signer.verifying_key(), runtime)?;
let snapshot = runtime.block_on(async {
for (index, (_, label)) in registered.iter().enumerate() {
crate::storage::tolerate_own_lag(
pile.maintain_attached(succinct[index], signer).await,
)
.with_context(|| format!("maintain {label} succinct fact archive"))?;
crate::storage::tolerate_own_lag(
pile.maintain_attached(rank9[index], signer).await,
)
.with_context(|| format!("maintain {label} fact archive"))?;
}
let snapshot = secrets_collection.ensure(pile, signer)
.await
.context("observe configured Secrets collection")?;
Ok::<_, anyhow::Error>(snapshot)
})?;
let store_snapshot = AcquiringReader::new(snapshot, std::sync::Arc::clone(runtime));
let secrets = secret_storage::snapshot_acquiring(store_snapshot.clone(), secrets_collection)?;
let mut collections = BTreeMap::new();
for ((scope, label), collection) in registered.iter().zip(&rank9) {
let archive = crate::storage::acquire_facts(&store_snapshot, *collection)
.with_context(|| format!("read maintained {label} collection"))?;
collections.insert(*scope, archive);
}
Ok(Self {
store_snapshot,
collections,
secrets,
})
}
#[cfg(test)]
fn open(pile_path: &Path, key: Option<&Path>) -> Result<Self> {
crate::storage::Storage::new(pile_path.to_owned(), key.map(Path::to_owned))
.with_store(Self::load)
}
fn view(&self, scope: Id, label: &str) -> Result<CollectionView> {
let facts = self
.collections
.get(&scope)
.cloned()
.with_context(|| format!("{label} collection was not attached in snapshot"))?;
Ok(CollectionView {
facts,
reader: self.store_snapshot.clone(),
})
}
fn cognition(&self) -> Result<CollectionView> {
self.view(COGNITION_SCOPE_ID, "Cognition")
}
fn headspace(&self) -> Result<(CollectionView, TriageHeadspace)> {
let secrets = self.secrets();
let view = self.view(HEADSPACE_SCOPE_ID, "Headspace")?;
let projected = triage_model::project_headspace(view.source(), secrets)?;
Ok((view, projected))
}
fn secrets(&self) -> &SecretsSnapshot<TriageReader> {
&self.secrets
}
fn memory(&self) -> Result<CollectionView> {
self.view(MEMORY_SCOPE_ID, "Memory")
}
fn relations(&self) -> Result<CollectionView> {
self.view(RELATIONS_SCOPE_ID, "Relations")
}
fn messages(&self) -> Result<CollectionView> {
self.view(MESSAGE_SCOPE_ID, "Message")
}
}
#[derive(Debug, Clone)]
struct TimelineRow {
at: i128,
source: &'static str,
detail: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
enum ChatRole {
System,
User,
Assistant,
}
impl std::fmt::Display for ChatRole {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::System => formatter.write_str("system"),
Self::User => formatter.write_str("user"),
Self::Assistant => formatter.write_str("assistant"),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct ChatMessage {
role: ChatRole,
content: String,
}
#[derive(Debug, Clone)]
struct ContextCandidate {
result: Id,
thought: Id,
messages: Vec<ChatMessage>,
}
fn fmt_id(id: Id) -> String {
format!("{id:x}")
}
fn now_key() -> Result<i128> {
triage_model::now_tai_ns()
}
fn interval_key(interval: Interval) -> i128 {
triage_model::interval_key(interval)
}
fn format_tai_ns(ns: i128) -> String {
let ns = ns.clamp(i64::MIN as i128, i64::MAX as i128) as i64;
let epoch = Epoch::from_tai_duration(hifitime::Duration::from_truncated_nanoseconds(ns));
let (year, month, day, hour, minute, second, _) = epoch.to_gregorian_utc();
format!("{year:04}-{month:02}-{day:02}T{hour:02}:{minute:02}:{second:02}")
}
fn format_age(now: i128, past: i128) -> String {
let seconds = (now.saturating_sub(past) / 1_000_000_000).max(0) as i64;
if seconds < 60 {
format!("{seconds}s")
} else if seconds < 3_600 {
format!("{}m", seconds / 60)
} else if seconds < 86_400 {
format!("{}h", seconds / 3_600)
} else {
format!("{}d", seconds / 86_400)
}
}
fn format_duration_ns(ns: i128) -> String {
let ns = ns.max(0);
let milliseconds = ns / 1_000_000;
if milliseconds < 1_000 {
format!("{milliseconds}ms")
} else if milliseconds < 60_000 {
format!("{:.2}s", milliseconds as f64 / 1_000.0)
} else {
format!("{:.1}m", milliseconds as f64 / 60_000.0)
}
}
fn format_exit_code(code: Option<u64>) -> String {
code.map(|code| code.to_string())
.unwrap_or_else(|| "-".to_owned())
}
fn truncate_single_line(text: &str, max: usize) -> String {
let mut out = String::with_capacity(max + 3);
for ch in text.chars() {
if out.chars().count() >= max {
out.push_str("...");
break;
}
out.push(if ch == '\n' || ch == '\r' { ' ' } else { ch });
}
out
}
fn first_line(text: &str) -> String {
text.lines()
.find(|line| !line.trim().is_empty())
.unwrap_or(text)
.trim()
.to_owned()
}
fn read_text(reader: &impl BlobStoreGet, handle: TextHandle) -> Result<String> {
let view: View<str> = reader
.get(handle)
.with_context(|| format!("read UTF8String {}", hex::encode(handle.raw)))?;
Ok(view.to_string())
}
fn scan(
snapshot: &TriageSnapshot,
pile_path: &Path,
recent: usize,
loop_min: usize,
stale_min: i64,
out: &mut Out<'_>,
) -> Result<()> {
let cognition = snapshot.cognition()?;
let headspace_view = snapshot.view(HEADSPACE_SCOPE_ID, "Headspace")?;
let secrets = snapshot.secrets();
let relations = snapshot.relations()?;
let messages = snapshot.messages()?;
let now = now_key()?;
let stale_ns = stale_min.max(0) as i128 * 60 * 1_000_000_000;
let report = triage_model::project_scan(
ScanSources {
cognition: cognition.source(),
headspace: headspace_view.source(),
secrets,
relations: relations.source(),
messages: messages.source(),
},
ScanOptions {
now: Some(now),
stale_after_ns: stale_ns,
recent_attempts: recent,
loop_min,
},
)?;
out.line(format!("Triage scan"))?;
out.line(format!("- pile: {}", pile_path.display()))?;
let config_heads = report.headspace.config_heads();
let active_profile_heads = report.headspace.active_profile_heads();
if let Some(error) = report.headspace.unsettled_reason() {
out.line(format!("- Headspace active state: unresolved ({error})"))?;
} else {
let state = if config_heads.len() > 1 || active_profile_heads.len() > 1 {
"agreed"
} else {
"unique"
};
out.line(format!(
"- Headspace active state: {state} (config heads={}, profile heads={})",
config_heads.len(),
active_profile_heads.len()
))?;
}
if let Some(persona) = report.headspace.persona_id {
out.line(format!("- persona id: {persona:x}"))?;
}
if !report.relations.forked_profiles.is_empty() {
out.line(format!(
"- Relations profile forks: {}",
report.relations.forked_profiles.len()
))?;
}
out.line("")?;
out.line(format!("Queues"))?;
out.line(format!(
"- exec: requests={} pending={} running={} age_unknown={} stale={} forked={} invalid={} done={}",
report.exec_queue.requests,
report.exec_queue.pending,
report.exec_queue.running,
report.exec_queue.age_unknown,
report.exec_queue.stale,
report.exec_queue.forked,
report.exec_queue.invalid,
report.exec_queue.done
))?;
out.line(format!(
"- model: requests={} pending={} running={} age_unknown={} stale={} forked={} invalid={} done={}",
report.model_queue.requests,
report.model_queue.pending,
report.model_queue.running,
report.model_queue.age_unknown,
report.model_queue.stale,
report.model_queue.forked,
report.model_queue.invalid,
report.model_queue.done
))?;
match report.unread_messages {
UnreadMessages::Available { count, .. } => {
out.line(format!("- unread canonical inbox messages: {count}"))?
}
UnreadMessages::Unavailable(reason) => {
let reason = match reason {
UnreadUnavailable::HeadspaceUnsettled => "Headspace is unsettled",
UnreadUnavailable::PersonaNotConfigured => "no persona is configured",
};
out.line(format!(
"- unread canonical inbox messages: unavailable ({reason})"
))?;
}
}
out.line("")?;
out.line(format!("Loop heuristics"))?;
if let Some(pattern) = report.probable_loop.as_ref() {
out.line(format!(
"- probable loop: {} repeated {}x (exit={}): {}",
truncate_single_line(&pattern.command, 80),
pattern.count,
pattern
.exit_code
.map(|code| code.to_string())
.unwrap_or_else(|| "-".to_owned()),
truncate_single_line(&pattern.fingerprint, 120)
))?;
} else {
out.line(format!(
"- no repeated failure loop >= {loop_min} in recent exec results"
))?;
}
let mut model_failures: Vec<_> = report
.model_state
.results
.iter()
.filter(|row| row.error.is_some())
.collect();
model_failures.sort_by_key(|row| (row.finished_at, row.id));
model_failures.reverse();
out.line("")?;
out.line(format!("Recent model failures"))?;
if model_failures.is_empty() {
out.line(format!("- none"))?;
}
for row in model_failures.into_iter().take(recent.min(5)) {
out.line(format!(
"- {} | {}",
format_age(now, row.finished_at),
truncate_single_line(row.error.as_deref().unwrap_or("<missing error>"), 140)
))?;
}
out.line("")?;
out.line(format!("Suggested next checks"))?;
for suggestion in &report.suggestions {
out.line(format!("- {suggestion}"))?;
}
Ok(())
}
fn loops(
snapshot: &TriageSnapshot,
recent: usize,
min_repeat: usize,
out: &mut Out<'_>,
) -> Result<()> {
let cognition = snapshot.cognition()?;
let state = collect_exec_state(&cognition.reader, &cognition.facts)?;
let report = build_loop_report(&state, recent, min_repeat);
let now = now_key()?;
out.line(format!("Triage loops"))?;
out.line(format!("- recent attempts: {}", report.recent.len()))?;
if let Some(head) = &report.contiguous_head {
out.line(format!(
"- contiguous head loop: {}x, exit={}, command={}",
head.count,
head.exit_code
.map(|code| code.to_string())
.unwrap_or_else(|| "-".to_owned()),
truncate_single_line(&head.command, 90)
))?;
} else {
out.line(format!(
"- contiguous head loop: none (threshold {min_repeat})"
))?;
}
out.line("")?;
out.line(format!("Top patterns"))?;
for pattern in report.top_patterns.iter().take(5) {
out.line(format!(
"- {}x | exit={} | {} | {}",
pattern.count,
pattern
.exit_code
.map(|code| code.to_string())
.unwrap_or_else(|| "-".to_owned()),
truncate_single_line(&pattern.command, 70),
truncate_single_line(&pattern.fingerprint, 80)
))?;
}
out.line("")?;
out.line(format!("Recent attempts"))?;
for row in report.recent {
out.line(format!(
"- [{}:{}] {} | exit={} | {} | {}",
fmt_id(row.request_id),
fmt_id(row.result_id),
format_age(now, row.finished_at),
row.exit_code
.map(|code| code.to_string())
.unwrap_or_else(|| "-".to_owned()),
truncate_single_line(&row.command, 70),
truncate_single_line(&row.fingerprint, 90)
))?;
}
Ok(())
}
fn build_timeline_rows(
exec_state: &ExecState,
model_state: &ModelChatState,
reason_rows: &[ReasonEventRow],
recent: usize,
) -> Vec<TimelineRow> {
let mut rows = Vec::new();
for request in exec_state.requests.values() {
rows.push(TimelineRow {
at: request.requested_at,
source: "exec",
detail: format!(
"[{}] {}",
fmt_id(request.id),
truncate_single_line(&request.command, 120)
),
});
}
for result in &exec_state.results {
let command = exec_state
.requests
.get(&result.about_request)
.map(|request| request.command.as_str())
.unwrap_or("<missing request>");
let status = result
.error
.as_deref()
.map(|error| format!("error {}", truncate_single_line(error, 72)))
.or_else(|| {
result.stderr_text.as_deref().map(|stderr| {
format!(
"exit {} stderr {}",
format_exit_code(result.exit_code),
truncate_single_line(&first_line(stderr), 72)
)
})
})
.unwrap_or_else(|| format!("exit {}", format_exit_code(result.exit_code)));
rows.push(TimelineRow {
at: result.finished_at,
source: "exec-result",
detail: format!(
"[{}:{}] {} | {status}",
fmt_id(result.about_request),
fmt_id(result.id),
truncate_single_line(command, 100)
),
});
}
for request in model_state.requests.values() {
rows.push(TimelineRow {
at: request.requested_at,
source: "model",
detail: format!("[{}] request", fmt_id(request.id)),
});
}
for result in &model_state.results {
if let Some(error) = &result.error {
rows.push(TimelineRow {
at: result.finished_at,
source: "model-error",
detail: format!(
"[{}] {}",
fmt_id(result.id),
truncate_single_line(error, 130)
),
});
}
}
for row in reason_rows {
let mut detail = format!("[{}] ", fmt_id(row.id));
if let Some(turn) = row.about_turn {
detail.push_str(&format!("[turn {}] ", fmt_id(turn)));
}
detail.push_str(&truncate_single_line(
row.text.as_deref().unwrap_or("<missing>"),
120,
));
if let Some(command) = &row.command_text {
detail.push_str(" | ");
detail.push_str(&truncate_single_line(command, 96));
}
rows.push(TimelineRow {
at: row.created_at.unwrap_or(i128::MIN),
source: "reason",
detail,
});
}
rows.sort_by_key(|row| row.at);
rows.reverse();
rows.truncate(recent);
rows
}
fn timeline(snapshot: &TriageSnapshot, recent: usize, out: &mut Out<'_>) -> Result<()> {
let cognition = snapshot.cognition()?;
let exec_state = collect_exec_state(&cognition.reader, &cognition.facts)?;
let model_state = collect_model_chat_state(&cognition.reader, &cognition.facts)?;
let reason_state = collect_reason_state(&cognition.reader, &cognition.facts)?;
let rows = build_timeline_rows(&exec_state, &model_state, &reason_state, recent);
let now = now_key()?;
out.line(format!("Triage timeline"))?;
out.line(format!("- rows: {}", rows.len()))?;
out.line("")?;
for row in rows {
out.line(format!(
"- {:>5} {:>11} | {}",
format_age(now, row.at),
row.source,
row.detail
))?;
}
Ok(())
}
fn chunk_text<P: TriblePattern>(reader: &impl BlobStoreGet, space: &P, id: Id) -> Result<String> {
if let Some(handle) = chunk_summary_handle(space, id) {
return memory_model::read_text(reader, handle);
}
if let Some(handle) = chunk_image_handle(space, id) {
return Ok(format!(
"<image: {} bytes>",
memory_model::read_image(reader, handle)?.len()
));
}
Ok(String::new())
}
fn format_span<P: TriblePattern>(space: &P, id: Id) -> String {
let (Some(s), Some(e)) = (chunk_start_at(space, id), chunk_end_at(space, id)) else {
return "?".to_string();
};
let start = interval_key(s);
let end = interval_key(e);
format!(
"{}..{} ({})",
format_tai_ns(start),
format_tai_ns(end),
format_duration_ns(end.saturating_sub(start))
)
}
fn cover(snapshot: &TriageSnapshot, full: bool, out: &mut Out<'_>) -> Result<()> {
let memory = snapshot.memory()?;
let (_, headspace) = snapshot.headspace()?;
let space = &memory.facts;
let mut chunk_ids = all_chunk_ids(space);
chunk_ids.sort();
let mut all_chunk_chars = 0usize;
for id in &chunk_ids {
all_chunk_chars += chunk_text(&memory.reader, space, *id)?.len();
}
let budget = headspace.budget()?;
let fill = if budget.body_budget_chars > 0 {
all_chunk_chars as f64 / budget.body_budget_chars as f64 * 100.0
} else {
0.0
};
out.line(format!("Memory cover"))?;
out.line(format!("- chunks: {}", chunk_ids.len()))?;
out.line("")?;
out.line(format!("Budget"))?;
out.line(format!(
"- context={} output={} safety={} chars/token={}",
budget.context_window_tokens,
budget.max_output_tokens,
budget.safety_margin_tokens,
budget.chars_per_token
))?;
out.line(format!(
"- system={} chars body={} chars all-chunks={} chars ratio={fill:.1}%",
budget.system_prompt_chars, budget.body_budget_chars, all_chunk_chars
))?;
out.line("")?;
out.line(format!(
"Chunks (canonical ID order; every episode coexists)"
))?;
if chunk_ids.is_empty() {
out.line(format!("- empty"))?;
}
for id in chunk_ids {
let text = chunk_text(&memory.reader, space, id)?;
out.line(format!(
"- chunk {} | {} | {}",
fmt_id(id),
format_span(space, id),
if full {
text
} else {
truncate_single_line(&text, 100)
}
))?;
}
Ok(())
}
fn matching_memory_nodes<P: TriblePattern>(space: &P, prefix: &str) -> BTreeSet<Id> {
let prefix = prefix.trim().to_ascii_uppercase();
let mut matches: BTreeSet<Id> = BTreeSet::new();
for id in all_chunk_ids(space) {
if format!("{id:X}").starts_with(&prefix)
|| chunk_aliases(space, id)
.iter()
.any(|alias| format!("{alias:X}").starts_with(&prefix))
{
matches.insert(id);
}
}
matches
}
fn print_ids(label: &str, ids: impl IntoIterator<Item = Id>, out: &mut Out<'_>) -> Result<()> {
let ids: Vec<_> = ids.into_iter().collect();
if !ids.is_empty() {
out.line(format!(
" {label}: {}",
ids.iter()
.map(|id| fmt_id(*id))
.collect::<Vec<_>>()
.join(", ")
))?;
}
Ok(())
}
fn print_observations(observations: &[Interval], out: &mut Out<'_>) -> Result<()> {
if !observations.is_empty() {
out.line(format!(" observations:"))?;
for observation in observations {
out.line(format!(
" - {}",
format_tai_ns(interval_key(*observation))
))?;
}
}
Ok(())
}
fn chunk(snapshot: &TriageSnapshot, prefix: &str, out: &mut Out<'_>) -> Result<()> {
let memory = snapshot.memory()?;
let matches = matching_memory_nodes(&memory.facts, prefix);
if matches.is_empty() {
bail!("no canonical Memory chunk or alias matches prefix '{prefix}'");
}
out.line(format!(
"Memory match set for '{prefix}' ({} chunk(s))",
matches.len()
))?;
for (index, id) in matches.into_iter().enumerate() {
if index > 0 {
out.line("")?;
}
let space = &memory.facts;
out.line(format!("Chunk {}", fmt_id(id)))?;
out.line(format!(" span: {}", format_span(space, id)))?;
print_ids("references", chunk_references(space, id).into_iter(), out)?;
print_ids("aliases", chunk_aliases(space, id).into_iter(), out)?;
if let Some(exec) = chunk_about_exec_result(space, id) {
out.line(format!(" about exec result: {}", fmt_id(exec)))?;
}
if let Some(message) = chunk_about_archive_message(space, id) {
out.line(format!(" about archive message: {}", fmt_id(message)))?;
}
if let Some(lens) = chunk_lens_handle(space, id) {
out.line(format!(
" lens: {}",
memory_model::read_text(&memory.reader, lens)?
))?;
}
print_observations(&chunk_observed_at(space, id), out)?;
out.line(format!(" content:"))?;
for line in chunk_text(&memory.reader, space, id)?.lines() {
out.line(format!(" {line}"))?;
}
}
Ok(())
}
fn select_request(state: &ExecState, turn: usize) -> Result<&ExecRequestRow> {
if turn == 0 {
bail!("turn is one-based");
}
let mut requests: Vec<_> = state.requests.values().collect();
requests.sort_by_key(|request| (request.requested_at, request.id));
requests.reverse();
requests.get(turn - 1).copied().ok_or_else(|| {
anyhow!(
"turn #{turn} not found; only {} request(s) exist",
requests.len()
)
})
}
fn contexts_for_turn<P: TriblePattern>(
reader: &impl BlobStoreGet,
space: &P,
exec_state: &ExecState,
request: Id,
) -> Result<Vec<ContextCandidate>> {
let mut pairs = BTreeSet::new();
for result in exec_state
.results
.iter()
.filter(|result| result.about_request == request)
{
if let Some(thought) = result.about_thought {
pairs.insert((result.id, thought));
}
}
let mut contexts = Vec::new();
for (result, thought) in pairs {
for handle in
find!(value: TextHandle, pattern!(space, [{ thought @ cog::context: ?value }]))
{
let json = read_text(reader, handle)?;
let messages = serde_json::from_str(&json)
.with_context(|| format!("parse context JSON for thought {thought:x}"))?;
contexts.push(ContextCandidate {
result,
thought,
messages,
});
}
}
Ok(contexts)
}
fn print_model_result(row: &ModelResultRow, full: bool, out: &mut Out<'_>) -> Result<()> {
out.line(format!(" Model result {}", fmt_id(row.id)))?;
out.line(format!(" finished: {}", format_tai_ns(row.finished_at)))?;
if let Some(error) = &row.error {
out.line(format!(
" error: {}",
if full {
error.clone()
} else {
truncate_single_line(error, 120)
}
))?;
}
if row.input_tokens.is_some() || row.output_tokens.is_some() {
let token = |value: Option<u64>| value.map_or_else(|| "-".to_owned(), |n| n.to_string());
out.line(format!(
" tokens: in={} out={} cache_create={} cache_read={}",
token(row.input_tokens),
token(row.output_tokens),
token(row.cache_creation_input_tokens),
token(row.cache_read_input_tokens)
))?;
}
for (label, text) in [
("reasoning", row.reasoning_text.as_ref()),
("output", row.output_text.as_ref()),
] {
if let Some(text) = text {
if full {
out.line(format!(" {label} ({} chars):", text.len()))?;
for line in text.lines() {
out.line(format!(" {line}"))?;
}
} else {
out.line(format!(
" {label}: {} chars \"{}\"",
text.len(),
truncate_single_line(text, 80)
))?;
}
}
}
Ok(())
}
fn turn(snapshot: &TriageSnapshot, turn: usize, full: bool, out: &mut Out<'_>) -> Result<()> {
let cognition = snapshot.cognition()?;
let exec_state = collect_exec_state(&cognition.reader, &cognition.facts)?;
let model_state = collect_model_chat_state(&cognition.reader, &cognition.facts)?;
let request = select_request(&exec_state, turn)?;
let now = now_key()?;
out.line(format!("Turn #{turn}"))?;
out.line(format!("- request: {}", fmt_id(request.id)))?;
out.line(format!(
"- requested: {} ({})",
format_tai_ns(request.requested_at),
format_age(now, request.requested_at)
))?;
out.line(format!(
"- command: {}",
if full {
request.command.clone()
} else {
truncate_single_line(&request.command, 100)
}
))?;
let mut results: Vec<_> = exec_state
.results
.iter()
.filter(|result| result.about_request == request.id)
.collect();
results.sort_by_key(|result| (result.finished_at, result.id));
if results.is_empty() {
out.line("")?;
out.line(format!(
"Exec results: none (turn may still be in progress)"
))?;
return Ok(());
}
out.line("")?;
out.line(format!("Exec results ({})", results.len()))?;
for result in results {
out.line(format!("- result {}", fmt_id(result.id)))?;
out.line(format!(
" exit: {}",
result
.exit_code
.map(|code| code.to_string())
.unwrap_or_else(|| "-".to_owned())
))?;
out.line(format!(
" finished: {} (latency {})",
format_tai_ns(result.finished_at),
format_duration_ns(result.finished_at.saturating_sub(request.requested_at))
))?;
for (label, text) in [
("error", result.error.as_ref()),
("stderr", result.stderr_text.as_ref()),
("stdout", result.stdout_text.as_ref()),
] {
if let Some(text) = text {
if full {
out.line(format!(" {label} ({} chars):", text.len()))?;
for line in text.lines() {
out.line(format!(" {line}"))?;
}
} else {
out.line(format!(" {label}: {}", truncate_single_line(text, 120)))?;
}
}
}
if let Some(thought) = result.about_thought {
out.line(format!(" thought: {}", fmt_id(thought)))?;
let mut model_requests: Vec<_> = model_state
.requests
.values()
.filter(|candidate| candidate.about_thought == Some(thought))
.collect();
model_requests.sort_by_key(|candidate| (candidate.requested_at, candidate.id));
for model_request in model_requests {
out.line(format!(" Model request {}", fmt_id(model_request.id)))?;
let mut model_results: Vec<_> = model_state
.results
.iter()
.filter(|candidate| candidate.about_request == model_request.id)
.collect();
model_results.sort_by_key(|candidate| (candidate.finished_at, candidate.id));
for model_result in model_results {
print_model_result(model_result, full, out)?;
}
}
}
}
let contexts = contexts_for_turn(&cognition.reader, &cognition.facts, &exec_state, request.id)?;
out.line("")?;
out.line(format!("Context candidates ({})", contexts.len()))?;
for candidate in contexts {
let chars: usize = candidate
.messages
.iter()
.map(|message| message.content.len())
.sum();
out.line(format!(
"- result {} thought {}: {} messages, {} chars",
fmt_id(candidate.result),
fmt_id(candidate.thought),
candidate.messages.len(),
chars
))?;
for (index, message) in candidate.messages.iter().enumerate() {
if full {
out.line(format!(
" #{index} [{}] ({} chars)",
message.role,
message.content.len()
))?;
for line in message.content.lines() {
out.line(format!(" {line}"))?;
}
} else {
out.line(format!(
" #{index} [{:<9}] ({:>5} chars) \"{}\"",
message.role.to_string(),
message.content.len(),
truncate_single_line(&message.content, 60)
))?;
}
}
}
Ok(())
}
fn context(
snapshot: &TriageSnapshot,
turn: usize,
full: bool,
raw: bool,
out: &mut Out<'_>,
) -> Result<()> {
let cognition = snapshot.cognition()?;
let (_, headspace) = snapshot.headspace()?;
let exec_state = collect_exec_state(&cognition.reader, &cognition.facts)?;
let request = select_request(&exec_state, turn)?;
let contexts = contexts_for_turn(&cognition.reader, &cognition.facts, &exec_state, request.id)?;
if contexts.is_empty() {
bail!("turn #{turn} has no recorded context candidate");
}
if raw {
let values: Vec<_> = contexts
.iter()
.map(|candidate| {
serde_json::json!({
"result": fmt_id(candidate.result),
"thought": fmt_id(candidate.thought),
"messages": candidate.messages,
})
})
.collect();
out.line(format!("{}", serde_json::to_string_pretty(&values)?))?;
return Ok(());
}
out.line(format!(
"Contexts for turn #{turn} [{}]",
fmt_id(request.id)
))?;
out.line(format!(
"- command: {}",
truncate_single_line(&request.command, 60)
))?;
out.line(format!("- candidates: {}", contexts.len()))?;
for candidate in contexts {
let chars: usize = candidate
.messages
.iter()
.map(|message| message.content.len())
.sum();
let budget = headspace.budget()?;
let fill = if budget.body_budget_chars > 0 {
chars as f64 / budget.body_budget_chars as f64 * 100.0
} else {
0.0
};
out.line("")?;
out.line(format!(
"Result {} / thought {}: {} messages, {} chars, fill={fill:.1}%",
fmt_id(candidate.result),
fmt_id(candidate.thought),
candidate.messages.len(),
chars
))?;
for (index, message) in candidate.messages.iter().enumerate() {
if full {
out.line(format!(
" #{index} [{}] ({} chars)",
message.role,
message.content.len()
))?;
for line in message.content.lines() {
out.line(format!(" {line}"))?;
}
} else {
out.line(format!(
" #{index} [{:<9}] ({:>5} chars) \"{}\"",
message.role.to_string(),
message.content.len(),
truncate_single_line(&message.content, 70)
))?;
}
}
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::super::cli::Cli;
use super::*;
use clap::Parser;
use std::fs::File;
use std::path::PathBuf;
use crate::headspace::{self, Resolution};
use crate::memory::{ChunkDraft, ChunkDraftContent};
use crate::schemas::headspace::{
playground_config, KIND_CONFIG_ID, KIND_LIVE_RECORD, KIND_MODEL_PROFILE_ID,
};
use crate::schemas::triage::{exec, KIND_EXEC_REQUEST_ID};
use crate::storage::initialize_signer;
use triblespace::core::metadata;
use triblespace::core::repo::StoreSnapshot;
use triblespace::macros::entity;
fn test_id(byte: u8) -> Id {
Id::new([byte; 16]).unwrap()
}
fn point(seconds: f64) -> Interval {
let at = Epoch::from_tai_seconds(seconds);
(at, at).try_to_inline().unwrap()
}
fn exec_request(id: Id, command: &str, at: f64) -> Fragment {
entity! { ExclusiveId::force_ref(&id) @
metadata::tag: &KIND_EXEC_REQUEST_ID,
exec::command_text: command.to_owned(),
metadata::created_at: point(at),
}
}
struct Fixture {
_directory: tempfile::TempDir,
pile: PathBuf,
key: PathBuf,
}
impl Fixture {
fn new() -> Self {
let directory = tempfile::tempdir().unwrap();
let pile = directory.path().join("triage.pile");
let key = directory.path().join("triage.key");
File::create(&pile).unwrap();
initialize_signer(&pile, Some(&key)).unwrap();
Self {
_directory: directory,
pile,
key,
}
}
fn publish(&self, scope: Id, fragment: Fragment) {
let signer = load_signer(&self.pile, Some(&self.key)).unwrap();
let mut pile = open_pile_strict(&self.pile).unwrap();
let collection =
crate::collection_names::open(&mut pile, scope, signer.verifying_key()).unwrap();
pile.commit(collection, &signer, fragment).unwrap();
pile.close().unwrap();
}
fn snapshot(&self) -> TriageSnapshot {
TriageSnapshot::open(&self.pile, Some(&self.key)).unwrap()
}
}
fn chunk(text: &str, start: f64) -> (Fragment, Id) {
memory_model::chunk_fragment(ChunkDraft {
content: ChunkDraftContent::Text(text.to_owned()),
start_at: point(start),
end_at: point(start + 1.0),
lens: None,
references: BTreeSet::new(),
about_exec_result: None,
about_archive_message: None,
observed_at: BTreeSet::from([point(start + 2.0)]),
aliases: BTreeSet::new(),
})
.unwrap()
}
fn extrinsic_headspace_fragment(anchor: Id, profile_id: Id, config_id: Id) -> Fragment {
let profile = headspace::default_profile(anchor, "extrinsic");
let config = headspace::default_config(anchor);
let mut fragment = headspace::profile_anchor_fragment(anchor);
let name = fragment.put(profile.name);
let left_model = fragment.put("left".to_owned());
let right_model = fragment.put("right".to_owned());
let base_url = fragment.put(profile.base_url);
fragment += entity! { ExclusiveId::force_ref(&profile_id) @
metadata::tag: &KIND_LIVE_RECORD,
metadata::tag: &KIND_MODEL_PROFILE_ID,
playground_config::model_profile_id: &anchor,
metadata::name: name,
playground_config::model_name: left_model,
playground_config::model_name: right_model,
playground_config::model_base_url: base_url,
playground_config::model_stream: 0_u64.to_inline(),
playground_config::model_context_window_tokens: profile.context_window_tokens.to_inline(),
playground_config::model_max_output_tokens: profile.max_output_tokens.to_inline(),
playground_config::model_context_safety_margin_tokens: profile.context_safety_margin_tokens.to_inline(),
playground_config::model_chars_per_token: profile.chars_per_token.to_inline(),
};
let system_prompt = fragment.put(config.system_prompt);
let author = fragment.put(config.author);
let author_role = fragment.put(config.author_role);
fragment += entity! { ExclusiveId::force_ref(&config_id) @
metadata::tag: &KIND_LIVE_RECORD,
metadata::tag: &KIND_CONFIG_ID,
playground_config::active_model_profile_id: &anchor,
playground_config::system_prompt: system_prompt,
playground_config::cognition_scope: &config.cognition_scope,
playground_config::author: author,
playground_config::author_role: author_role,
playground_config::poll_ms: config.poll_ms.to_inline(),
metadata::description: "an additive fact the reader does not model",
};
fragment
}
#[test]
fn memory_chunks_coexist() {
let fixture = Fixture::new();
let (left, left_id) = chunk("one telling", 10.0);
let (right, right_id) = chunk("another telling", 10.0);
fixture.publish(MEMORY_SCOPE_ID, left);
fixture.publish(MEMORY_SCOPE_ID, right);
let view = fixture.snapshot().memory().unwrap();
let ids: BTreeSet<Id> = all_chunk_ids(&view.facts).into_iter().collect();
assert_eq!(ids, BTreeSet::from([left_id, right_id]));
}
#[test]
fn headspace_profile_fork_is_reported_instead_of_arbitrated() {
let fixture = Fixture::new();
let anchor = test_id(0x51);
let profile = headspace::default_profile(anchor, "triage");
let config = headspace::default_config(anchor);
let (genesis, profile_head, _) =
headspace::add_profile_fragment(&profile, &config, &[]).unwrap();
fixture.publish(HEADSPACE_SCOPE_ID, genesis);
let mut left = profile.clone();
left.model = "left".to_owned();
let mut right = profile;
right.model = "right".to_owned();
fixture.publish(
HEADSPACE_SCOPE_ID,
headspace::profile_snapshot_fragment(&left, &[profile_head])
.unwrap()
.0,
);
fixture.publish(
HEADSPACE_SCOPE_ID,
headspace::profile_snapshot_fragment(&right, &[profile_head])
.unwrap()
.0,
);
let (_, projected) = fixture.snapshot().headspace().unwrap();
assert!(projected.budget.is_none());
assert!(matches!(
projected.active_profile,
Some(Resolution::Forked(_))
));
assert!(projected.unsettled_reason().unwrap().contains("forked"));
}
#[test]
fn headspace_projection_keeps_ids_opaque_and_scalar_multiplicity_visible() {
let fixture = Fixture::new();
let anchor = test_id(0x56);
let profile_id = test_id(0x57);
let config_id = test_id(0x58);
fixture.publish(
HEADSPACE_SCOPE_ID,
extrinsic_headspace_fragment(anchor, profile_id, config_id),
);
let (_, projected) = fixture.snapshot().headspace().unwrap();
assert!(matches!(
projected.config,
Resolution::Unique(ref snapshot) if snapshot.id == config_id
));
assert!(matches!(
projected.active_profile,
Some(Resolution::Forked(ref variants))
if variants.len() == 2 && variants.iter().all(|snapshot| snapshot.id == profile_id)
));
}
#[test]
fn missing_exact_headspace_secret_is_a_visible_current_state_error() {
let fixture = Fixture::new();
let anchor = test_id(0x61);
let mut profile = headspace::default_profile(anchor, "private");
profile.model_secret_version = Some(test_id(0x62));
let config = headspace::default_config(anchor);
fixture.publish(
HEADSPACE_SCOPE_ID,
headspace::add_profile_fragment(&profile, &config, &[])
.unwrap()
.0,
);
let error = match fixture.snapshot().headspace() {
Ok(_) => panic!("missing exact secret unexpectedly validated"),
Err(error) => error,
};
assert!(format!("{error:#}").contains("missing exact model Secrets version"));
}
#[test]
fn retired_headspace_secret_reference_does_not_poison_current_state() {
let fixture = Fixture::new();
let anchor = test_id(0x63);
let mut historical = headspace::default_profile(anchor, "private");
historical.model_secret_version = Some(test_id(0x64));
let config = headspace::default_config(anchor);
let (genesis, historical_head, _) =
headspace::add_profile_fragment(&historical, &config, &[]).unwrap();
fixture.publish(HEADSPACE_SCOPE_ID, genesis);
let current = headspace::default_profile(anchor, "public");
fixture.publish(
HEADSPACE_SCOPE_ID,
headspace::profile_snapshot_fragment(¤t, &[historical_head])
.unwrap()
.0,
);
let (_, projected) = fixture.snapshot().headspace().unwrap();
assert!(projected.is_settled());
assert!(matches!(
projected.active_profile,
Some(Resolution::Unique(_))
));
}
#[test]
fn maintained_snapshot_is_idempotent_after_first_open() {
let fixture = Fixture::new();
fixture.publish(
COGNITION_SCOPE_ID,
exec_request(test_id(0x71), "read only", 10.0),
);
let snapshot = fixture.snapshot();
snapshot.cognition().unwrap();
let secrets_reader = snapshot.secrets.store_snapshot();
assert!(snapshot
.store_snapshot
.changes_since(secrets_reader)
.is_empty());
assert!(secrets_reader
.changes_since(&snapshot.store_snapshot)
.is_empty());
drop(snapshot);
let maintained = std::fs::metadata(&fixture.pile).unwrap().len();
let snapshot = fixture.snapshot();
snapshot.cognition().unwrap();
drop(snapshot);
let reopened = std::fs::metadata(&fixture.pile).unwrap().len();
assert_eq!(reopened, maintained);
}
#[test]
fn retired_branch_commands_and_selector_are_not_cli_surface() {
assert!(
Cli::try_parse_from(["triage", "--pile", "x", "--branch", "cognition", "scan"])
.is_err()
);
assert!(Cli::try_parse_from(["triage", "--pile", "x", "chain"]).is_err());
assert!(Cli::try_parse_from(["triage", "--pile", "x", "repair"]).is_err());
assert!(Cli::try_parse_from(["triage", "--pile", "x", "migrate-legacy"]).is_err());
}
}