use std::collections::{BTreeSet, HashMap};
use base64::Engine as _;
use chrono::{DateTime, Utc};
use serde_json::{Value, json};
use crate::core::knowledge::{ImportMerge, KnowledgeFact, ProjectKnowledge};
use super::request::{BridgeRequest, Op, is_fact_ref, is_profile_name, is_safe_webhook};
use super::state::ProState;
use super::wire::{AlertRecord, AppliedItem, Baseline, FileOut, Plan, Profile, ProjectColumn};
use super::{
GUARD_KINDS, ProError, ProResult, ProjectScope, collect, export, ledger, plain_text,
security_fields, snapshots,
};
const MAX_PLANS: usize = 8;
const SNAPSHOT_HINT: &str = "A snapshot was saved first; restore it under Memory Care → Snapshots.";
pub trait Notifier: Send + Sync {
fn desktop(&self, text: &str) -> bool;
fn webhook(&self, url: &str, body: &[u8]) -> bool;
}
pub struct SystemNotifier;
impl Notifier for SystemNotifier {
fn desktop(&self, text: &str) -> bool {
crate::core::value::notify::send("LeanCTX", text)
}
fn webhook(&self, url: &str, body: &[u8]) -> bool {
if !is_safe_webhook(url) {
return false;
}
let timeout = Some(std::time::Duration::from_secs(6));
let agent =
crate::core::http_client::ureq_agent_without_redirects(timeout, timeout, timeout);
agent
.post(url)
.header("Content-Type", "application/json")
.send(body)
.is_ok_and(|r| r.status().is_success())
}
}
pub(crate) enum Outcome {
Applied(Value),
File(FileOut),
}
pub(crate) struct Ctx<'a> {
pub(crate) scope: &'a ProjectScope,
pub(crate) request: &'a BridgeRequest,
pub(crate) now: DateTime<Utc>,
pub(crate) notifier: &'a dyn Notifier,
}
pub(crate) fn parse_plans(op: Op, raw: Vec<Value>) -> ProResult<Vec<Plan>> {
if raw.is_empty() || raw.len() > MAX_PLANS {
return Err(ProError::conflict("Unexpected number of planned changes"));
}
raw.into_iter()
.map(|value| {
let plan: Plan = serde_json::from_value(value)
.map_err(|_| ProError::conflict("Unknown planned change"))?;
if !op.allowed_effects().contains(&plan.effect()) {
return Err(ProError::conflict(
"Planned change does not fit the request",
));
}
Ok(plan)
})
.collect()
}
fn ensure(condition: bool, message: &str) -> ProResult<()> {
if condition {
Ok(())
} else {
Err(ProError::conflict(message))
}
}
fn is_time(value: &str) -> bool {
DateTime::parse_from_rfc3339(value).is_ok()
}
fn fact_ref(fact: &KnowledgeFact) -> String {
format!("{}/{}", fact.category, fact.key)
}
fn current_map(facts: &[KnowledgeFact]) -> HashMap<String, KnowledgeFact> {
facts
.iter()
.filter(|f| f.is_current())
.map(|f| (fact_ref(f), f.clone()))
.collect()
}
pub(crate) struct RestorePlan {
snapshot: HashMap<String, KnowledgeFact>,
now: HashMap<String, KnowledgeFact>,
pub(crate) missing: Vec<String>,
pub(crate) changed: Vec<String>,
}
pub(crate) fn restore_plan(scope: &ProjectScope, id: &str) -> ProResult<RestorePlan> {
let snapshot = current_map(&snapshots::read_facts(scope, id)?);
let records = collect::knowledge_records(scope);
let now = current_map(&records);
let mut missing: Vec<String> = snapshot
.keys()
.filter(|r| !now.contains_key(*r))
.cloned()
.collect();
let mut changed: Vec<String> = snapshot
.iter()
.filter(|(r, f)| now.get(*r).is_some_and(|n| n.value != f.value))
.map(|(r, _)| r.clone())
.collect();
missing.sort();
changed.sort();
Ok(RestorePlan {
snapshot,
now,
missing,
changed,
})
}
fn same_set(a: &[String], b: &[String]) -> bool {
a.iter().collect::<BTreeSet<_>>() == b.iter().collect::<BTreeSet<_>>()
}
fn requested_profile(ctx: &Ctx<'_>) -> Option<Profile> {
let name = ctx.request.b_str("name");
ProState::new(ctx.scope)
.profiles()
.into_iter()
.find(|p| p.name == name)
}
pub(crate) fn validate(ctx: &Ctx<'_>, plan: &Plan) -> ProResult<()> {
let request = ctx.request;
let state = ProState::new(ctx.scope);
match plan {
Plan::AlertsSettings { settings } => {
let stored = state.alert_settings();
let wanted = |key: &str| request.body.get(key);
let kinds: Vec<Value> = settings.kinds.iter().map(|k| json!(k)).collect();
ensure(
settings
.kinds
.iter()
.all(|k| GUARD_KINDS.contains(&k.as_str()))
&& settings.kinds.iter().collect::<BTreeSet<_>>().len() == settings.kinds.len(),
"Alert kinds outside the guard list",
)?;
ensure(
settings.webhook.is_empty() || is_safe_webhook(&settings.webhook),
"Unsafe webhook",
)?;
let matches = |key: &str, planned: Value, stored: Value| {
wanted(key).map_or(planned == stored, |w| *w == planned)
};
ensure(
matches("enabled", json!(settings.enabled), json!(stored.enabled))
&& matches("desktop", json!(settings.desktop), json!(stored.desktop))
&& matches("webhook", json!(settings.webhook), json!(stored.webhook))
&& matches("kinds", Value::Array(kinds), json!(stored.kinds)),
"Alert settings differ from the request",
)
}
Plan::AlertsNotify {
at,
text,
counts,
test,
desktop,
webhook,
} => {
let stored = state.alert_settings();
ensure(*test == (request.op == Op::AlertsTest), "Unexpected alert")?;
ensure(*test || stored.enabled, "Alerts are off")?;
ensure(is_time(at) && plain_text(text, 240), "Invalid alert text")?;
ensure(
counts.iter().all(|(k, n)| {
GUARD_KINDS.contains(&k.as_str()) && (0..=1_000_000_000).contains(n)
}),
"Alert counts outside the guard list",
)?;
ensure(
*desktop == stored.desktop && *webhook == stored.webhook,
"Alert channels differ from the settings",
)
}
Plan::ProfilesPut {
name,
values,
saved_at,
} => {
ensure(
request.b_str("action") == "save" && name == request.b_str("name"),
"Profile differs from the request",
)?;
ensure(
is_profile_name(name) && is_time(saved_at),
"Invalid profile",
)?;
for (key, value) in values {
let field = security_fields::lookup(key)
.ok_or_else(|| ProError::conflict("Profile holds an unmanaged setting"))?;
security_fields::wire_of(field, value)?;
}
Ok(())
}
Plan::ProfilesDelete { name } => ensure(
request.b_str("action") == "delete" && name == request.b_str("name"),
"Profile differs from the request",
),
Plan::ConfigSet {
key,
wire,
expected,
weakening,
} => {
ensure(
request.b_str("action") == "apply",
"Unexpected setting change",
)?;
let field = security_fields::lookup(key)
.ok_or_else(|| ProError::conflict("This setting is not managed here"))?;
ensure(
security_fields::wire_of(field, expected)? == *wire
&& security_fields::expected_of(field, wire) == *expected,
"Setting value does not match its wire form",
)?;
let profile =
requested_profile(ctx).ok_or_else(|| ProError::not_found("Unknown profile"))?;
let stored = profile
.values
.get(key)
.map(|v| security_fields::wire_of(field, v))
.transpose()?;
ensure(
stored.as_deref() == Some(wire.as_str()),
"Setting is not part of the profile",
)?;
let before = security_fields::effective(key).unwrap_or_default();
let lowers = security_fields::weakening(field, &before, wire);
if lowers && !request.b_true("confirm_weaker") {
return Err(ProError::conflict(
"This lowers protection. Confirm the change to continue.",
));
}
ensure(lowers == *weakening, "Protection change misreported")
}
Plan::MemorySnapshot { reason } => ensure(
reason == request.b_str("reason"),
"Snapshot differs from the request",
),
Plan::MemoryRestore {
id,
missing,
changed,
} => {
ensure(
id == request.b_str("id"),
"Snapshot differs from the request",
)?;
let plan = restore_plan(ctx.scope, id)?;
ensure(
same_set(missing, &plan.missing) && same_set(changed, &plan.changed),
"Memory changed; compare the snapshot again",
)
}
Plan::MemoryForget { refs, reason } => {
let asked: Vec<String> = request
.body
.get("refs")
.and_then(Value::as_array)
.map(|a| {
a.iter()
.filter_map(Value::as_str)
.map(str::to_string)
.collect()
})
.unwrap_or_default();
ensure(
same_set(refs, &asked) && reason == request.b_str("reason"),
"Cleanup differs from the request",
)?;
ensure(refs.iter().all(|r| is_fact_ref(r)), "Invalid insight key")?;
let current = current_map(&collect::knowledge_records(ctx.scope));
ensure(
refs.iter().all(|r| current.contains_key(r)),
"Memory changed; refresh before cleaning up",
)
}
Plan::MemoryConsolidate { run } => ensure(
*run && request.b_true("run"),
"Consolidation differs from the request",
),
Plan::ContextOverlay { action, path } => {
ensure(
action == request.b_str("action") && path == request.b_str("path"),
"Rule differs from the request",
)?;
ensure(
ctx.scope.jailed(path)?.is_file(),
"The path is not a file of this project",
)?;
ensure(
collect::heatmap(ctx.scope).contains_key(path),
"The path is not a current recommendation",
)
}
Plan::AdaptiveRecord { item } => ensure(
item.path == request.b_str("path") && item.action == request.b_str("action"),
"Record differs from the request",
),
Plan::AdaptiveDismiss { path } => {
ensure(
path == request.b_str("path"),
"Dismissal differs from the request",
)?;
ctx.scope.jailed(path).map(|_| ())
}
Plan::LedgerExport {
format,
file_name,
mime,
filters,
project_column,
} => {
ensure(format == request.q("format"), "Export format differs")?;
ensure(
*file_name == export::file_name(ctx.now.date_naive(), format)
&& mime == export::mime_of(format),
"Export file differs",
)?;
let mine = ledger::filters(ctx.scope, &ctx.request.query)?;
let normalize = |f: &super::wire::LedgerFilters| {
let mut f = f.clone();
if let Some(scope) = f.scope.as_mut() {
scope.sort();
scope.dedup();
}
f
};
ensure(
normalize(filters) == normalize(&mine),
"Export filters differ",
)?;
let labels = ctx.scope.labels();
ensure(
match project_column {
ProjectColumn::Fixed { name } => *name == ctx.scope.label,
ProjectColumn::Labels { labels: planned } => {
planned.iter().all(|l| labels.contains(l))
}
},
"Export project column outside this project",
)
}
}
}
fn snapshot_now(ctx: &Ctx<'_>, reason: &str) -> ProResult<super::wire::SnapshotMeta> {
let knowledge = ProjectKnowledge::load_or_create(&ctx.scope.root);
snapshots::create(ctx.scope, &knowledge, reason, ctx.now)
}
fn policy() -> ProResult<crate::core::memory_policy::MemoryPolicy> {
crate::core::config::Config::load()
.memory_policy_effective()
.map_err(|_| ProError::internal("Invalid memory policy"))
}
fn import(ctx: &Ctx<'_>, facts: Vec<KnowledgeFact>, merge: ImportMerge) -> ProResult<()> {
if facts.is_empty() {
return Ok(());
}
ProjectKnowledge::mutate_locked(&ctx.scope.root, |k| {
k.import_facts(facts, merge, "dashboard-pro");
})
.map(|_| ())
.map_err(|_| ProError::internal(format!("The restore was not saved. {SNAPSHOT_HINT}")))
}
pub(crate) fn execute(ctx: &Ctx<'_>, plan: Plan) -> ProResult<Outcome> {
let state = ProState::new(ctx.scope);
let effect = plan.effect();
let applied = |readback: Value| {
Ok(Outcome::Applied(
json!({ "effect": effect, "readback": readback }),
))
};
match plan {
Plan::AlertsSettings { settings } => {
state.put_alert_settings(&settings)?;
let back = state.alert_settings();
ensure(back == settings, "Alert settings did not read back")?;
applied(json!({ "settings": back }))
}
Plan::AlertsNotify {
at,
text,
counts,
test,
desktop,
webhook,
} => {
let mut channels = Vec::new();
if desktop {
channels.push(
if ctx.notifier.desktop(&text) {
"desktop"
} else {
"desktop failed"
}
.to_string(),
);
}
if !webhook.is_empty() {
let body = json!({ "source": "leanctx", "test": test, "counts": counts, "at": at });
let sent = ctx.notifier.webhook(&webhook, body.to_string().as_bytes());
channels.push(if sent { "webhook" } else { "webhook failed" }.to_string());
}
let record = AlertRecord {
at,
text,
channels,
test,
};
state.append_alert(record.clone())?;
ensure(
state.alerts_file().history.last() == Some(&record),
"Alert history did not read back",
)?;
applied(json!({ "alert": record }))
}
Plan::ProfilesPut {
name,
values,
saved_at,
} => {
let profile = Profile {
name: name.clone(),
values,
saved_at,
};
state.put_profile(profile.clone())?;
ensure(
state.profiles().contains(&profile),
"Profile did not read back",
)?;
applied(json!({ "profiles": state.profiles() }))
}
Plan::ProfilesDelete { name } => {
state.delete_profile(&name)?;
ensure(
!state.profiles().iter().any(|p| p.name == name),
"Profile was not deleted",
)?;
applied(json!({ "profiles": state.profiles() }))
}
Plan::ConfigSet {
key,
wire,
expected,
weakening,
} => {
crate::core::config::setter::set_by_key(&key, &wire)
.map_err(|e| ProError::conflict(format!("The setting was refused: {e}")))?;
let field = security_fields::lookup(&key)
.ok_or_else(|| ProError::conflict("This setting is not managed here"))?;
let after =
security_fields::effective(&key).map(|w| security_fields::expected_of(field, &w));
ensure(
after.as_ref() == Some(&expected),
"The instance did not report the new value. A setting in this project's own configuration may take precedence over the device setting.",
)?;
applied(json!({ "key": key, "value": expected, "weakened": weakening }))
}
Plan::MemorySnapshot { reason } => {
let meta = snapshot_now(ctx, &reason)?;
ensure(
snapshots::list(ctx.scope, snapshots::LIST_LIMIT)
.iter()
.any(|m| m.id == meta.id),
"Snapshot did not read back",
)?;
applied(json!({ "snapshot": meta }))
}
Plan::MemoryRestore { id, .. } => {
let plan = restore_plan(ctx.scope, &id)?;
let safety = snapshot_now(ctx, "Before restoring a snapshot")?;
let pick = |refs: &[String]| -> Vec<KnowledgeFact> {
refs.iter()
.filter_map(|r| plan.snapshot.get(r).cloned())
.collect()
};
import(ctx, pick(&plan.missing), ImportMerge::SkipExisting)?;
import(ctx, pick(&plan.changed), ImportMerge::Replace)?;
let after = current_map(&collect::knowledge_records(ctx.scope));
let lost = plan.now.keys().any(|r| !after.contains_key(r));
let differs = plan
.snapshot
.iter()
.any(|(r, f)| after.get(r).is_none_or(|a| a.value != f.value));
if lost || differs {
return Err(ProError::conflict(format!(
"The restore did not read back completely. {SNAPSHOT_HINT}"
)));
}
applied(json!({
"restored": plan.missing.len(),
"updated": plan.changed.len(),
"insights": after.len(),
"safety_snapshot": safety.id,
}))
}
Plan::MemoryForget { refs, reason } => {
let safety = snapshot_now(ctx, &reason)?;
let policy = policy()?;
let pairs: Vec<(String, String)> = refs
.iter()
.filter_map(|r| r.split_once('/'))
.map(|(c, k)| (c.to_string(), k.to_string()))
.collect();
ProjectKnowledge::mutate_locked(&ctx.scope.root, |k| {
let mut removed = false;
for (category, key) in &pairs {
removed |= k.remove_fact(category, key);
}
if removed {
let _ = k.run_memory_lifecycle(&policy);
}
})
.map_err(|_| {
ProError::internal(format!("The removal was not saved. {SNAPSHOT_HINT}"))
})?;
let left = current_map(&collect::knowledge_records(ctx.scope));
if refs.iter().any(|r| left.contains_key(r)) {
return Err(ProError::conflict(format!(
"Some insights were not removed. {SNAPSHOT_HINT}"
)));
}
applied(json!({ "removed": refs.len(), "safety_snapshot": safety.id }))
}
Plan::MemoryConsolidate { .. } => {
let safety = snapshot_now(ctx, "Before consolidation")?;
let report = crate::core::knowledge_consolidate::consolidate_project_knowledge_with(
&ctx.scope.root,
&crate::core::consolidation_engine::ConsolidateOptions::manual(),
)
.map_err(|_| ProError::internal(format!("Consolidation failed. {SNAPSHOT_HINT}")))?;
let text = crate::core::knowledge_consolidate::format_consolidation_report(&report);
applied(json!({
"summary": consolidation_summary(&text),
"safety_snapshot": safety.id,
}))
}
Plan::ContextOverlay { action, path } => {
use crate::core::context_overlay::{
ContextOverlay, OverlayAuthor, OverlayOp, OverlayScope, OverlayStore,
};
let root = std::path::PathBuf::from(&ctx.scope.root);
let target = crate::core::context_field::ContextItemId::from_file(&path);
let operation = if action == "compress" {
OverlayOp::SetView(crate::core::context_field::ViewKind::Map)
} else {
OverlayOp::Exclude {
reason: "adaptive recommendation".into(),
}
};
let mut store = OverlayStore::load_project(&root);
store.add(ContextOverlay::new(
target.clone(),
operation.clone(),
OverlayScope::Project,
String::new(),
OverlayAuthor::User,
));
store
.save_project(&root)
.map_err(|_| ProError::internal("The rule was not saved"))?;
let back = OverlayStore::load_project(&root);
ensure(
back.all()
.iter()
.any(|o| o.target == target && o.operation == operation),
"The rule did not read back",
)?;
applied(json!({ "action": action, "path": path }))
}
Plan::AdaptiveRecord { item } => {
let heat: super::wire::HeatmapEntry = collect::heatmap(ctx.scope)
.remove(&item.path)
.and_then(|v| serde_json::from_value(v).ok())
.unwrap_or_default();
let original = heat.total_original_tokens.unwrap_or(0);
let saved = heat.total_tokens_saved.unwrap_or(0);
let record = AppliedItem {
path: item.path,
action: item.action,
applied_at: ctx.now.to_rfc3339(),
baseline: Baseline {
reads: heat.access_count.unwrap_or(0),
original,
saved,
ratio: (original > 0)
.then(|| ((saved as f64 / original as f64) * 1000.0).round() / 1000.0),
},
};
let mut adaptive = state.adaptive();
adaptive.applied.push(record.clone());
state.put_adaptive(adaptive)?;
ensure(
state.adaptive().applied.last() == Some(&record),
"The record did not read back",
)?;
applied(json!({ "item": record }))
}
Plan::AdaptiveDismiss { path } => {
let mut adaptive = state.adaptive();
if !adaptive.dismissed.contains(&path) {
adaptive.dismissed.push(path.clone());
}
adaptive
.applied
.retain(|item| !(item.path == path && item.action == "compress"));
state.put_adaptive(adaptive)?;
let back = state.adaptive();
ensure(
back.dismissed.contains(&path)
&& !back
.applied
.iter()
.any(|item| item.path == path && item.action == "compress"),
"The dismissal did not read back",
)?;
applied(json!({ "path": path }))
}
Plan::LedgerExport {
format,
file_name,
mime,
filters,
project_column,
} => {
let bytes = export::build(ctx.scope, &filters, &format, &project_column)?;
Ok(Outcome::File(FileOut {
name: file_name,
mime,
bytes_b64: base64::engine::general_purpose::STANDARD.encode(bytes),
}))
}
}
}
fn plural(count: i64, word: &str) -> String {
format!("{count} {word}{}", if count == 1 { "" } else { "s" })
}
pub(crate) fn consolidation_summary(output: &str) -> Value {
let number = |pattern: &str| -> i64 {
regex::Regex::new(pattern)
.ok()
.and_then(|re| re.captures(output))
.and_then(|c| c.get(1))
.and_then(|m| m.as_str().parse().ok())
.unwrap_or(0)
};
let merged = number(r"consolidated (\d+)");
let archived = number(r"archived-to-target (\d+)")
.max(number(r"Lifecycle:[^\n]*?archived (\d+)"))
+ number(r"Patterns:[^\n]*?compacted (\d+)")
+ number(r"History:[^\n]*?compacted (\d+)");
let kept = match number(r"remaining (\d+)") {
0 => number(r"Facts: (\d+) active"),
n => n,
};
let session = number(r"Session import: \S+ \((\d+) item");
let mut lines = Vec::new();
if merged > 0 {
lines.push(format!(
"{} were merged into one.",
plural(merged, "similar insight")
));
}
if archived > 0 {
lines.push(format!(
"{} were archived; archived items can be restored.",
plural(archived, "item")
));
}
if session > 0 {
lines.push(format!(
"{} from the current session were added.",
plural(session, "item")
));
}
if lines.is_empty() {
lines.push("Nothing needed consolidating.".into());
}
lines.push(format!("{} stayed in memory.", plural(kept, "insight")));
json!({ "merged": merged, "archived": archived, "kept": kept, "preview": false, "lines": lines })
}
pub(crate) fn apply(ctx: &Ctx<'_>, plans: Vec<Plan>) -> ProResult<Vec<Outcome>> {
for plan in &plans {
validate(ctx, plan)?;
}
plans.into_iter().map(|plan| execute(ctx, plan)).collect()
}