use clap::{Args as ClapArgs, Subcommand, ValueEnum};
use serde_json::json;
use memstead_base::binding::{
BINDING_VERSION, Binding, BuildMode, BuildOperation, CapabilityError, DEFAULT_ADJUDICATION_CAP,
DEFAULT_FULL_RESYNC_EVERY, Operations, PruneConfig, SyncOperation, VerifyOperation,
prune_guarantee_for_medium, validate_binding,
};
use memstead_base::binding_migrate::{
BindingMigrateError, check_all_consumed, fold_v1_binding, migrate_gen2_bindings,
};
use memstead_base::ingest::advance::{
AdvanceError, DispositionInput, ExcludeError, advance_baseline, record_exclusions,
};
use memstead_base::ingest::findings::{
FindingsError, FullResyncDecision, record_anchor_hash_backfill, record_verified_baseline,
verify_binding, verify_binding_full,
};
use memstead_base::ingest::report::{
DEFAULT_REPORT_BUDGET, compute_fidelity_report, render_fidelity_report,
};
use memstead_base::ingest::resolve::{ResolveError, ResolvedSource, resolve_binding_run};
use memstead_base::ingest::{
OperationFilter, OperationKind, RenderBriefError, render_ingest_brief, render_sync_brief_for,
render_verify_brief_for, select_next_due_operation,
};
use memstead_base::pipeline::{IngestTrigger, MediumType, PatternEntry, PatternMode, Source};
use memstead_base::pipeline_store::{
ProjectionGeneration, delete_ingest, load_legacy_pipeline_configs, load_pipeline_configs,
load_projection_generations, read_binding, remove_mediums_and_facets_trees, write_binding,
};
use memstead_base::workspace_store::StoreError;
use memstead_base::{migrate_legacy_pipeline, read_legacy_pipeline_configs};
use crate::CliError;
use crate::output::{ExitKind, print_json, print_markdown};
use crate::setup::{CliContext, workspace_not_initialised_error};
#[derive(ClapArgs, Debug)]
pub struct Args {
#[command(subcommand)]
pub command: ProjectionCommand,
}
#[derive(Subcommand, Debug)]
pub enum ProjectionCommand {
Brief(BriefArgs),
Init(InitArgs),
Migrate(MigrateArgs),
Enable(EnableArgs),
Advance(AdvanceArgs),
Exclude(ExcludeArgs),
Verify(VerifyArgs),
}
#[derive(Clone, Copy, Debug, ValueEnum)]
pub enum MediumTypeArg {
Codebase,
Filesystem,
Git,
Graph,
Web,
}
impl MediumTypeArg {
fn to_medium_type(self) -> MediumType {
match self {
MediumTypeArg::Codebase => MediumType::Codebase,
MediumTypeArg::Filesystem => MediumType::Filesystem,
MediumTypeArg::Git => MediumType::Git,
MediumTypeArg::Graph => MediumType::Graph,
MediumTypeArg::Web => MediumType::Web,
}
}
}
#[derive(ClapArgs, Debug)]
pub struct BriefArgs {
pub binding: Option<String>,
#[arg(long)]
pub all: bool,
#[arg(long, value_enum, default_value_t = BriefOperationArg::Build, requires = "all", conflicts_with_all = ["verify", "sync"])]
pub operation: BriefOperationArg,
#[arg(long, conflicts_with = "sync")]
pub verify: bool,
#[arg(long, conflicts_with = "verify")]
pub sync: bool,
}
#[derive(Clone, Copy, Debug, ValueEnum, PartialEq, Eq)]
pub enum BriefOperationArg {
Build,
Sync,
Verify,
Any,
}
impl BriefOperationArg {
fn to_filter(self) -> OperationFilter {
match self {
BriefOperationArg::Build => OperationFilter::Only(OperationKind::Build),
BriefOperationArg::Sync => OperationFilter::Only(OperationKind::Sync),
BriefOperationArg::Verify => OperationFilter::Only(OperationKind::Verify),
BriefOperationArg::Any => OperationFilter::Any,
}
}
}
#[derive(ClapArgs, Debug)]
pub struct InitArgs {
#[arg(long)]
pub mem: String,
#[arg(long)]
pub source: String,
#[arg(long = "medium-type", value_enum)]
pub medium_type: MediumTypeArg,
#[arg(long)]
pub intent: Option<String>,
#[arg(long)]
pub name: Option<String>,
}
#[derive(ClapArgs, Debug)]
pub struct MigrateArgs {
#[arg(long)]
pub dry_run: bool,
}
#[derive(Clone, Copy, Debug, ValueEnum, PartialEq, Eq)]
pub enum EnableOperationArg {
Build,
Sync,
Verify,
}
impl EnableOperationArg {
fn name(self) -> &'static str {
match self {
EnableOperationArg::Build => "build",
EnableOperationArg::Sync => "sync",
EnableOperationArg::Verify => "verify",
}
}
}
#[derive(ClapArgs, Debug)]
pub struct EnableArgs {
#[arg(value_enum)]
pub operation: EnableOperationArg,
pub binding: String,
}
#[derive(ClapArgs, Debug)]
pub struct AdvanceArgs {
pub binding: String,
#[arg(long)]
pub dispositions: String,
}
#[derive(ClapArgs, Debug)]
pub struct ExcludeArgs {
pub binding: String,
#[arg(long)]
pub exclusions: String,
}
#[derive(ClapArgs, Debug)]
pub struct VerifyArgs {
pub binding: String,
#[arg(long)]
pub budget: Option<usize>,
#[arg(long = "include")]
pub include: Vec<String>,
#[arg(long)]
pub full: bool,
}
pub fn run(ctx: &CliContext, args: Args) -> anyhow::Result<()> {
match args.command {
ProjectionCommand::Brief(a) => brief(ctx, a),
ProjectionCommand::Init(a) => init(ctx, a),
ProjectionCommand::Migrate(a) => migrate(ctx, a),
ProjectionCommand::Enable(a) => enable(ctx, a),
ProjectionCommand::Advance(a) => advance(ctx, a),
ProjectionCommand::Exclude(a) => exclude(ctx, a),
ProjectionCommand::Verify(a) => verify(ctx, a),
}
}
fn map_brief_err(binding_id: &str, err: RenderBriefError) -> CliError {
let message = err.to_string();
let mapped = match &err {
RenderBriefError::ConfigLoad(_) => {
CliError::new(ExitKind::Generic, "PROJECTION_LOAD_FAILED", message)
}
RenderBriefError::BuildOperationAbsent { .. } => CliError::new(
ExitKind::Validation,
"PROJECTION_BUILD_NOT_ENABLED",
message,
),
RenderBriefError::FindingsRead { .. } => CliError::new(
ExitKind::Generic,
"PROJECTION_FINDINGS_READ_FAILED",
message,
),
RenderBriefError::Resolve(inner) => match inner {
ResolveError::BindingNotFound { .. } => {
CliError::new(ExitKind::NotFound, "PROJECTION_NOT_FOUND", message)
}
ResolveError::MalformedProjectionRef { .. } => {
CliError::new(ExitKind::Validation, "PROJECTION_INVALID_NAME", message)
}
},
};
mapped.with_details(json!({ "binding": binding_id }))
}
fn brief(ctx: &CliContext, args: BriefArgs) -> anyhow::Result<()> {
let (_shape, root) = ctx.workspace_shape().ok_or_else(|| {
workspace_not_initialised_error(
"not inside a Memstead workspace (no `.memstead/workspace.toml` in any ancestor)",
)
})?;
let cli_engine = ctx.cli_engine_at(&root)?;
let engine = cli_engine.base();
if let Some(binding_id) = args.binding.as_deref()
&& let Ok(configs) = load_pipeline_configs(&root)
&& configs
.quarantined
.iter()
.any(|q| format!("{}/{}", q.mem, q.name) == binding_id)
{
return Err(binding_miss_error(&configs, binding_id).into());
}
if args.verify || args.sync {
let binding_id = args.binding.ok_or_else(|| {
CliError::new(
ExitKind::Validation,
"PROJECTION_BRIEF_BINDING_REQUIRED",
format!(
"`projection brief --{}` needs a binding id `<mem>/<stem>` — it renders one \
binding's brief, not an `--all` rotation",
if args.verify { "verify" } else { "sync" }
),
)
})?;
let (rendered, operation) = if args.verify {
(
render_verify_brief_for(engine, &root, &binding_id),
OperationKind::Verify,
)
} else {
(
render_sync_brief_for(engine, &root, &binding_id),
OperationKind::Sync,
)
};
let rendered = rendered.map_err(|e| map_brief_err(&binding_id, e))?;
if ctx.json {
print_json(&json!({ "brief": rendered, "operation": operation.as_wire() }))?;
} else {
print!("{rendered}");
}
return Ok(());
}
let selected = match args.binding {
Some(binding) if !args.all => Some((binding, OperationKind::Build)),
_ => {
let configs = load_pipeline_configs(&root).map_err(|e| {
CliError::new(
ExitKind::Generic,
"PROJECTION_LOAD_FAILED",
format!("could not load binding store: {e}"),
)
.with_details(json!({ "error": e.to_string() }))
})?;
if configs.bindings.is_empty() {
if ctx.json {
print_json(&json!({ "no_bindings": true }))?;
} else {
println!("> **[projection] No bindings configured in this workspace yet.**");
}
return Ok(());
}
select_next_due_operation(engine, &root, &configs, args.operation.to_filter())
}
};
let Some((binding_id, operation)) = selected else {
if ctx.json {
print_json(&json!({ "skipped": true }))?;
} else {
println!(
"> **[projection] Skipped — every eligible binding is backing off this pass.**"
);
}
return Ok(());
};
let rendered = match operation {
OperationKind::Build => render_ingest_brief(engine, &root, &binding_id),
OperationKind::Sync => render_sync_brief_for(engine, &root, &binding_id),
OperationKind::Verify => render_verify_brief_for(engine, &root, &binding_id),
}
.map_err(|e| map_brief_err(&binding_id, e))?;
if ctx.json {
print_json(&json!({ "brief": rendered, "operation": operation.as_wire() }))?;
} else {
print!("{rendered}");
}
Ok(())
}
fn is_single_component(value: &str) -> bool {
!value.is_empty()
&& value != "."
&& value != ".."
&& !value.contains('/')
&& !value.contains('\\')
&& !value.contains(':')
&& !value.contains('\0')
}
fn derive_stem(source: &str) -> String {
source
.trim_end_matches('/')
.rsplit('/')
.next()
.unwrap_or(source)
.to_string()
}
fn init_write_error(binding_id: &str, err: StoreError) -> CliError {
CliError::new(
ExitKind::Generic,
"PROJECTION_INIT_FAILED",
format!("could not scaffold binding `{binding_id}`: {err}"),
)
.with_details(json!({ "binding": binding_id, "error": err.to_string() }))
}
fn init(ctx: &CliContext, args: InitArgs) -> anyhow::Result<()> {
let (_shape, root) = ctx.workspace_shape().ok_or_else(|| {
workspace_not_initialised_error(
"not inside a Memstead workspace (no `.memstead/workspace.toml` in any ancestor)",
)
})?;
let mem = args.mem;
let stem = args
.name
.clone()
.unwrap_or_else(|| derive_stem(&args.source));
for (kind, value) in [("mem", mem.as_str()), ("name", stem.as_str())] {
if !is_single_component(value) {
return Err(CliError::new(
ExitKind::Validation,
"PROJECTION_INVALID_NAME",
format!(
"invalid {kind} '{}': must be a single path component (no separators, \
traversal segments, ':' or NUL) — pass an explicit --name",
value.escape_default()
),
)
.with_details(json!({ "kind": kind, "value": value }))
.into());
}
}
let binding_id = format!("{mem}/{stem}");
let medium_type = args.medium_type.to_medium_type();
let binding_path = root
.join(".memstead")
.join("projections")
.join(&mem)
.join(format!("{stem}.json"));
if binding_path.exists() {
return Err(CliError::new(
ExitKind::Validation,
"PROJECTION_EXISTS",
format!(
"a binding `{binding_id}` already exists at \
.memstead/projections/{mem}/{stem}.json — `projection init` never overwrites; \
choose a different --name or edit the existing binding"
),
)
.with_details(json!({ "binding": binding_id }))
.into());
}
let source = Source {
name: stem.clone(),
medium_type,
pointer: args.source.clone(),
change_detection: None,
scope: vec![PatternEntry {
path: "**/*".to_string(),
mode: PatternMode::Allow,
}],
engagement: None,
preparation: None,
};
let deny_paths: Vec<String> = if matches!(
medium_type,
memstead_base::MediumType::Codebase | memstead_base::MediumType::Filesystem
) {
memstead_base::binding::DEFAULT_SCAFFOLD_DENY_PATHS
.iter()
.map(|s| s.to_string())
.collect()
} else {
Vec::new()
};
let mut binding = Binding {
version: BINDING_VERSION,
intent: args.intent.clone(),
sources: vec![source],
reference_mems: Vec::new(),
destination_mem: mem.clone(),
deny_paths,
coverage_semantics: None,
rules: None,
prune: None,
operations: Operations {
build: Some(BuildOperation {
mode: BuildMode::Discovery,
trigger: IngestTrigger::Loop,
batch_size: 20,
post_actions: None,
}),
sync: Some(SyncOperation {
trigger: IngestTrigger::Manual,
batch_size: 20,
}),
verify: Some(VerifyOperation {
trigger: IngestTrigger::Manual,
batch_size: 20,
adjudication_cap: DEFAULT_ADJUDICATION_CAP,
full_resync_every: DEFAULT_FULL_RESYNC_EVERY,
}),
},
};
let mut warnings: Vec<String> = Vec::new();
if matches!(
medium_type,
memstead_base::MediumType::Codebase | memstead_base::MediumType::Filesystem
) {
let base = memstead_base::ingest::cursor::medium_base(&args.source, &root);
let canon_base = std::fs::canonicalize(&base).unwrap_or(base);
let canon_root = std::fs::canonicalize(&root).unwrap_or_else(|_| root.clone());
if !canon_base.starts_with(&canon_root) {
warnings.push(format!(
"medium base '{}' resolves outside the workspace root '{}': artifact ids will be \
workspace-relative ('../…' chains), and anchors written against source-relative \
paths will fail to resolve (orphaned). Consider rooting the workspace at the \
source tree.",
canon_base.display(),
canon_root.display()
));
}
}
if let Err(refusals) = validate_binding(&binding) {
for r in &refusals {
if let CapabilityError::OperationOutOfScope { operation, .. } = r {
match *operation {
"sync" => binding.operations.sync = None,
"verify" => binding.operations.verify = None,
_ => {}
}
}
warnings.push(r.to_string());
}
}
if binding.operations.sync.is_some() {
binding.prune = Some(PruneConfig {
guarantee: prune_guarantee_for_medium(medium_type),
});
}
let mut operations: Vec<&str> = vec!["build"];
if binding.operations.sync.is_some() {
operations.push("sync");
}
if binding.operations.verify.is_some() {
operations.push("verify");
}
write_binding(&root, &mem, &stem, &binding).map_err(|e| init_write_error(&binding_id, e))?;
let created = vec![format!(".memstead/projections/{mem}/{stem}.json")];
if ctx.json {
print_json(&json!({
"binding": binding_id,
"created": created,
"operations": operations,
"warnings": warnings,
}))?;
} else {
let mut out = format!("# Projection init\n\nScaffolded binding `{binding_id}`:\n");
for c in &created {
out.push_str(&format!("- `{c}`\n"));
}
out.push_str(&format!("\nOperations: {}\n", operations.join(", ")));
if !warnings.is_empty() {
out.push_str("\n## Warnings\n\n");
for w in &warnings {
out.push_str(&format!("- {w}\n"));
}
}
print_markdown(&out);
}
Ok(())
}
fn map_migrate_err(err: BindingMigrateError) -> CliError {
let message = err.to_string();
match &err {
BindingMigrateError::RefinementModeDeleted { .. } => CliError::new(
ExitKind::Validation,
"PROJECTION_MIGRATE_REFINEMENT",
message,
),
BindingMigrateError::MalformedProjectionRef { .. } => CliError::new(
ExitKind::Validation,
"PROJECTION_MIGRATE_MALFORMED_REF",
message,
),
BindingMigrateError::DanglingProjectionRef { .. }
| BindingMigrateError::DanglingFacetRef { .. }
| BindingMigrateError::DanglingMediumRef { .. } => CliError::new(
ExitKind::Validation,
"PROJECTION_MIGRATE_DANGLING_REF",
message,
),
BindingMigrateError::OrphanRecords { .. } => CliError::new(
ExitKind::Validation,
"PROJECTION_MIGRATE_ORPHAN_RECORDS",
message,
),
}
}
fn has_legacy_root_layout(root: &std::path::Path) -> bool {
["scopes", "projections", "ingests"]
.iter()
.any(|d| root.join(d).is_dir())
}
fn migrate_load_err(err: StoreError) -> CliError {
CliError::new(
ExitKind::Generic,
"PROJECTION_MIGRATE_FAILED",
format!("could not load pipeline config: {err}"),
)
.with_details(json!({ "error": err.to_string() }))
}
fn pointer_resolves_to(root: &std::path::Path, medium_pointer: &str, abs_path: &str) -> bool {
let resolved = if medium_pointer.is_empty() {
root.to_path_buf()
} else {
root.join(medium_pointer)
};
match (
std::fs::canonicalize(&resolved),
std::fs::canonicalize(abs_path),
) {
(Ok(a), Ok(b)) => a == b,
_ => resolved == std::path::Path::new(abs_path),
}
}
fn propose_workspace_toml(root: &std::path::Path) -> Option<String> {
let path = root.join(".memstead").join("workspace.toml");
let content = std::fs::read_to_string(path).ok()?;
let hits: Vec<(usize, &str)> = content
.lines()
.enumerate()
.filter(|(_, l)| {
let low = l.to_lowercase();
low.contains("reconcile-cursors") || low.contains("ingests/") || low.contains("ingest ")
})
.collect();
if hits.is_empty() {
return None;
}
let mut block = String::from(
"## Proposal: workspace.toml (NOT applied)\n\n`projection migrate` never edits \
`workspace.toml`. It found references to retired pipeline vocabulary — review and \
update these lines by hand, then commit:\n\n",
);
for (i, line) in hits {
block.push_str(&format!("- L{}: `{}`\n", i + 1, line.trim()));
}
Some(block)
}
fn binding_miss_error(configs: &memstead_base::BindingConfigs, binding_id: &str) -> CliError {
if let Some(q) = configs
.quarantined
.iter()
.find(|q| format!("{}/{}", q.mem, q.name) == binding_id)
{
return CliError::new(
ExitKind::Validation,
"PROJECTION_QUARANTINED",
format!(
"binding `{binding_id}` is quarantined — its stored file failed the load and \
it serves no operations until repaired: [{}] {}",
q.reason_code, q.reason_message
),
)
.with_details(json!({
"binding": binding_id,
"reason_code": q.reason_code,
"reason_message": q.reason_message,
"path": q.path,
}));
}
CliError::new(
ExitKind::NotFound,
"PROJECTION_NOT_FOUND",
format!(
"no binding `{binding_id}` in this workspace — scaffold one with \
`projection init` or migrate a legacy workspace with `projection migrate`"
),
)
.with_details(json!({ "binding": binding_id }))
}
fn consume_reconcile_cursors(
ctx: &CliContext,
root: &std::path::Path,
) -> anyhow::Result<(Vec<String>, Option<String>)> {
let cursor_path = root.join(".memstead").join("reconcile-cursors.json");
if !cursor_path.exists() {
return Ok((Vec::new(), None));
}
let cursors: std::collections::BTreeMap<String, String> = std::fs::read(&cursor_path)
.ok()
.and_then(|b| serde_json::from_slice(&b).ok())
.unwrap_or_default();
let mut seeded: Vec<String> = Vec::new();
if !cursors.is_empty() {
let configs = load_pipeline_configs(root).map_err(migrate_load_err)?;
let mut cli_engine = match ctx.cli_engine_at(root) {
Ok(e) => e,
Err(boot_err) => {
return Ok((
Vec::new(),
Some(format!(
"RECONCILE_CURSORS_DEFERRED: the workspace does not boot yet \
({boot_err:#}); reconcile-cursors.json was kept — repair the boot, \
then re-run `memstead projection migrate` to seed the sync baselines"
)),
));
}
};
let engine = cli_engine.base_mut();
for (cursor_key, sha) in &cursors {
let Some((_cursor_mem, abs_path)) = cursor_key.split_once(':') else {
continue;
};
for record in &configs.bindings {
let binding_id = format!("{}/{}", record.mem, record.name);
let Ok(resolved) = resolve_binding_run(&binding_id, &record.config) else {
continue;
};
for source in &resolved.sources {
if let ResolvedSource::Primary(p) = source
&& pointer_resolves_to(root, &p.pointer, abs_path)
{
let key = format!("{binding_id}/{}#synced", p.name);
if engine
.set_mem_sync_state(
&resolved.destination_mem,
&key,
sha,
Some("projection migrate: seeded from reconcile-cursors.json"),
)
.is_ok()
{
seeded.push(key);
}
}
}
}
}
}
let _ = std::fs::remove_file(&cursor_path);
Ok((seeded, None))
}
fn migrate(ctx: &CliContext, args: MigrateArgs) -> anyhow::Result<()> {
let (_shape, root) = ctx.workspace_shape().ok_or_else(|| {
workspace_not_initialised_error(
"not inside a Memstead workspace (no `.memstead/workspace.toml` in any ancestor)",
)
})?;
let gen1 = has_legacy_root_layout(&root);
if gen1 && !args.dry_run {
migrate_legacy_pipeline(&root).map_err(|e| {
CliError::new(
ExitKind::Generic,
"PROJECTION_MIGRATE_FAILED",
format!("could not convert root-folder (gen-1) pipeline layout: {e}"),
)
.with_details(json!({ "error": e.to_string() }))
})?;
}
let configs = if gen1 && args.dry_run {
read_legacy_pipeline_configs(&root).map_err(migrate_load_err)?
} else {
load_legacy_pipeline_configs(&root).map_err(migrate_load_err)?
};
let mut migrated = migrate_gen2_bindings(&configs).map_err(map_migrate_err)?;
let mut already_v2 = 0usize;
if !(gen1 && args.dry_run) {
let generations = load_projection_generations(&root).map_err(migrate_load_err)?;
for (mem, name, generation) in generations {
let binding_id = format!("{mem}/{name}");
match generation {
ProjectionGeneration::V2 => already_v2 += 1,
ProjectionGeneration::V1(v1) => {
let consumed = v1.source_facets.clone();
let binding = fold_v1_binding(&binding_id, &mem, v1.as_ref(), &configs)
.map_err(map_migrate_err)?;
migrated.push(memstead_base::binding_migrate::MigratedBinding {
id: binding_id,
mem,
name,
ingest_name: String::new(),
consumed_facets: consumed,
binding,
notes: Vec::new(),
});
}
ProjectionGeneration::VersionLess => {
if !migrated.iter().any(|m| m.mem == mem && m.name == name) {
return Err(CliError::new(
ExitKind::Validation,
"PROJECTION_MIGRATE_INERT_PROJECTION",
format!(
"projection `{binding_id}` is a version-less gen-2 file no \
ingest schedules — inert leftovers the loader refuses; delete \
.memstead/projections/{mem}/{name}.json (or add an ingest) and \
re-run `projection migrate`"
),
)
.with_details(json!({ "binding": binding_id }))
.into());
}
}
}
}
migrated.sort_by(|a, b| a.id.cmp(&b.id));
let consumed: Vec<(String, String)> = migrated
.iter()
.flat_map(|m| m.consumed_facets.iter().map(|f| (m.mem.clone(), f.clone())))
.collect();
check_all_consumed(&configs, &consumed).map_err(map_migrate_err)?;
}
let mut warnings: Vec<serde_json::Value> = Vec::new();
for m in &migrated {
if let Err(refusals) = validate_binding(&m.binding) {
for r in refusals {
warnings.push(json!({
"binding": m.id,
"kind": "capability",
"message": r.to_string(),
}));
}
}
for note in &m.notes {
warnings.push(json!({
"binding": m.id,
"kind": "note",
"message": note,
}));
}
}
if !args.dry_run {
for m in &migrated {
write_binding(&root, &m.mem, &m.name, &m.binding).map_err(|e| {
CliError::new(
ExitKind::Generic,
"PROJECTION_MIGRATE_FAILED",
format!("could not write binding `{}`: {e}", m.id),
)
.with_details(json!({ "binding": m.id, "error": e.to_string() }))
})?;
if !m.ingest_name.is_empty() {
delete_ingest(&root, &m.ingest_name).map_err(|e| {
CliError::new(
ExitKind::Generic,
"PROJECTION_MIGRATE_FAILED",
format!("could not remove merged ingest `{}`: {e}", m.ingest_name),
)
.with_details(json!({ "ingest": m.ingest_name, "error": e.to_string() }))
})?;
}
}
remove_mediums_and_facets_trees(&root).map_err(|e| {
CliError::new(
ExitKind::Generic,
"PROJECTION_MIGRATE_FAILED",
format!("could not remove the emptied mediums/facets trees: {e}"),
)
.with_details(json!({ "error": e.to_string() }))
})?;
}
let ((seeded, cursors_deferred), proposal) = if args.dry_run {
((Vec::new(), None), None)
} else {
(
consume_reconcile_cursors(ctx, &root)?,
propose_workspace_toml(&root),
)
};
let bindings: Vec<&str> = migrated.iter().map(|m| m.id.as_str()).collect();
if ctx.json {
print_json(&json!({
"ok": true,
"dry_run": args.dry_run,
"migrated": migrated.len(),
"already_v2": already_v2,
"bindings": bindings,
"warnings": warnings,
"cursors_seeded": seeded,
"cursors_deferred": cursors_deferred,
"workspace_toml_proposal": proposal,
}))?;
} else {
let verb = if args.dry_run {
"Would migrate"
} else {
"Migrated"
};
let mut out = format!(
"# Projection migration\n\n{verb} {} binding(s) to v2 ({already_v2} already v2):\n",
migrated.len()
);
for id in &bindings {
out.push_str(&format!("- `{id}`\n"));
}
if !warnings.is_empty() {
out.push_str("\n## Warnings\n\n");
for w in &warnings {
out.push_str(&format!(
"- [{}] `{}`: {}\n",
w["kind"].as_str().unwrap_or(""),
w["binding"].as_str().unwrap_or(""),
w["message"].as_str().unwrap_or(""),
));
}
}
if !seeded.is_empty() {
out.push_str("\n## Baselines seeded from reconcile-cursors.json\n\n");
for key in &seeded {
out.push_str(&format!("- `{key}`\n"));
}
}
if let Some(notice) = &cursors_deferred {
out.push_str(&format!("\n## Reconcile cursors deferred\n\n{notice}\n"));
}
if let Some(block) = &proposal {
out.push('\n');
out.push_str(block);
}
if !args.dry_run {
out.push_str(
"\nEach projection file was converted to a v2 single-record binding in place \
(medium + facet content folded inline, source names preserved verbatim); \
merged ingests and the emptied mediums/ and facets/ trees were removed.\n",
);
}
print_markdown(&out);
}
Ok(())
}
fn invalid_binding_id(binding_id: &str) -> CliError {
CliError::new(
ExitKind::Validation,
"PROJECTION_INVALID_NAME",
format!(
"invalid binding id '{}': expected `<mem>/<stem>` with each half a single path \
component (no extra separators, traversal segments, ':' or NUL)",
binding_id.escape_default()
),
)
.with_details(json!({ "binding": binding_id }))
}
fn enable_failed(binding_id: &str, err: StoreError) -> CliError {
CliError::new(
ExitKind::Generic,
"PROJECTION_ENABLE_FAILED",
format!("could not enable operation on binding `{binding_id}`: {err}"),
)
.with_details(json!({ "binding": binding_id, "error": err.to_string() }))
}
fn enable(ctx: &CliContext, args: EnableArgs) -> anyhow::Result<()> {
let (_shape, root) = ctx.workspace_shape().ok_or_else(|| {
workspace_not_initialised_error(
"not inside a Memstead workspace (no `.memstead/workspace.toml` in any ancestor)",
)
})?;
let binding_id = args.binding;
let op = args.operation;
let (mem, stem) = binding_id
.split_once('/')
.filter(|(m, n)| !m.is_empty() && !n.is_empty())
.filter(|(m, n)| is_single_component(m) && is_single_component(n))
.ok_or_else(|| invalid_binding_id(&binding_id))?;
let mem = mem.to_string();
let stem = stem.to_string();
let binding_path = root
.join(".memstead")
.join("projections")
.join(&mem)
.join(format!("{stem}.json"));
if !binding_path.exists() {
return Err(CliError::new(
ExitKind::NotFound,
"PROJECTION_NOT_FOUND",
format!(
"no binding `{binding_id}` at .memstead/projections/{mem}/{stem}.json — \
scaffold one with `projection init` or migrate a legacy workspace with \
`projection migrate`"
),
)
.with_details(json!({ "binding": binding_id }))
.into());
}
if let Ok(configs) = load_pipeline_configs(&root)
&& configs
.quarantined
.iter()
.any(|q| format!("{}/{}", q.mem, q.name) == binding_id)
{
return Err(binding_miss_error(&configs, &binding_id).into());
}
let mut binding =
read_binding(&root, &mem, &stem).map_err(|e| enable_failed(&binding_id, e))?;
let already = match op {
EnableOperationArg::Build => binding.operations.build.is_some(),
EnableOperationArg::Sync => binding.operations.sync.is_some(),
EnableOperationArg::Verify => binding.operations.verify.is_some(),
};
if already {
return Err(CliError::new(
ExitKind::Validation,
"PROJECTION_OP_ALREADY_ENABLED",
format!(
"operation `{}` is already enabled on binding `{binding_id}` — nothing to do",
op.name()
),
)
.with_details(json!({ "binding": binding_id, "operation": op.name() }))
.into());
}
let batch_size = binding
.operations
.build
.as_ref()
.map_or(20, |b| b.batch_size);
match op {
EnableOperationArg::Build => {
binding.operations.build = Some(BuildOperation {
mode: BuildMode::Discovery,
trigger: IngestTrigger::Loop,
batch_size,
post_actions: None,
});
}
EnableOperationArg::Sync => {
binding.operations.sync = Some(SyncOperation {
trigger: IngestTrigger::Manual,
batch_size,
});
}
EnableOperationArg::Verify => {
binding.operations.verify = Some(VerifyOperation {
trigger: IngestTrigger::Manual,
batch_size,
adjudication_cap: DEFAULT_ADJUDICATION_CAP,
full_resync_every: DEFAULT_FULL_RESYNC_EVERY,
});
}
}
if let Err(refusals) = validate_binding(&binding)
&& let Some(err) = refusals.iter().find(|r| {
matches!(
r,
CapabilityError::OperationOutOfScope { operation, .. } if *operation == op.name()
)
})
{
return Err(CliError::new(
ExitKind::Validation,
"PROJECTION_CAPABILITY_UNSUPPORTED",
err.to_string(),
)
.with_details(json!({ "binding": binding_id, "operation": op.name() }))
.into());
}
write_binding(&root, &mem, &stem, &binding).map_err(|e| enable_failed(&binding_id, e))?;
let mut operations: Vec<&str> = Vec::new();
if binding.operations.build.is_some() {
operations.push("build");
}
if binding.operations.sync.is_some() {
operations.push("sync");
}
if binding.operations.verify.is_some() {
operations.push("verify");
}
if ctx.json {
print_json(&json!({
"binding": binding_id,
"enabled": op.name(),
"operations": operations,
}))?;
} else {
print_markdown(&format!(
"# Projection enable\n\nEnabled `{}` on binding `{binding_id}`.\n\nOperations: {}\n",
op.name(),
operations.join(", ")
));
}
Ok(())
}
fn map_resolve_err(binding_id: &str, err: ResolveError) -> CliError {
let message = err.to_string();
let mapped = match err {
ResolveError::MalformedProjectionRef { .. } => {
CliError::new(ExitKind::Validation, "PROJECTION_INVALID_NAME", message)
}
_ => CliError::new(ExitKind::Generic, "PROJECTION_ADVANCE_FAILED", message),
};
mapped.with_details(json!({ "binding": binding_id }))
}
fn map_advance_err(binding_id: &str, err: AdvanceError) -> CliError {
let message = err.to_string();
match &err {
AdvanceError::MalformedId(_) => {
CliError::new(ExitKind::Validation, "PROJECTION_INVALID_NAME", message)
.with_details(json!({ "binding": binding_id }))
}
AdvanceError::UnknownArtifact {
artifacts,
suggestions,
..
} => {
let corrected: serde_json::Map<String, serde_json::Value> = suggestions
.iter()
.map(|(supplied, corrected)| {
(
supplied.clone(),
serde_json::Value::String(corrected.clone()),
)
})
.collect();
CliError::new(
ExitKind::Validation,
"PROJECTION_ADVANCE_UNKNOWN_ARTIFACT",
message,
)
.with_details(json!({
"binding": binding_id,
"unknown_artifacts": artifacts,
"corrected_artifacts": corrected,
}))
}
AdvanceError::Store(_) | AdvanceError::Engine(_) => {
CliError::new(ExitKind::Generic, "PROJECTION_ADVANCE_FAILED", message)
.with_details(json!({ "binding": binding_id }))
}
}
}
fn advance(ctx: &CliContext, args: AdvanceArgs) -> anyhow::Result<()> {
let (_shape, root) = ctx.workspace_shape().ok_or_else(|| {
workspace_not_initialised_error(
"not inside a Memstead workspace (no `.memstead/workspace.toml` in any ancestor)",
)
})?;
let binding_id = args.binding;
let dispositions: std::collections::BTreeMap<String, DispositionInput> =
serde_json::from_str(&args.dispositions).map_err(|e| {
CliError::new(
ExitKind::Validation,
"PROJECTION_INVALID_DISPOSITIONS",
format!(
"--dispositions must be a JSON object mapping artifact id → either a \
disposition string (e.g. \"worked\") or an object \
{{\"disposition\": \"excluded\", \"rationale\": \"...\"}}: {e}"
),
)
.with_details(json!({ "error": e.to_string() }))
})?;
let configs = load_pipeline_configs(&root).map_err(|e| {
CliError::new(
ExitKind::Generic,
"PROJECTION_ADVANCE_FAILED",
format!("could not load pipeline config: {e}"),
)
.with_details(json!({ "error": e.to_string() }))
})?;
let record = configs
.bindings
.iter()
.find(|r| format!("{}/{}", r.mem, r.name) == binding_id)
.ok_or_else(|| binding_miss_error(&configs, &binding_id))?;
if record.config.operations.sync.is_none() {
return Err(CliError::new(
ExitKind::Validation,
"PROJECTION_SYNC_NOT_ENABLED",
format!(
"binding `{binding_id}` has no sync operation — enable it with \
`memstead projection enable sync {binding_id}`"
),
)
.with_details(json!({ "binding": binding_id }))
.into());
}
let resolved = resolve_binding_run(&binding_id, &record.config)
.map_err(|e| map_resolve_err(&binding_id, e))?;
let mut cli_engine = ctx.cli_engine_at(&root)?;
let engine = cli_engine.base_mut();
let outcome = advance_baseline(engine, &root, &resolved, &dispositions)
.map_err(|e| map_advance_err(&binding_id, e))?;
if ctx.json {
print_json(&json!({
"binding": outcome.binding,
"completed": outcome.completed,
"disposed": outcome.disposed,
"pending": outcome.pending,
"remainder": outcome.remainder,
"tokens_written": outcome.tokens_written,
"warnings": outcome.warnings,
}))?;
} else {
let mut out = format!(
"# Projection advance\n\nBinding `{}`: {} artifact(s) disposed, {} remaining.\n",
outcome.binding, outcome.disposed, outcome.pending
);
if outcome.completed {
out.push_str("\nEvery presented artifact is disposed — the sync baseline advanced.\n");
if !outcome.tokens_written.is_empty() {
out.push_str("\nBaseline tokens written:\n");
for key in &outcome.tokens_written {
out.push_str(&format!("- `{key}`\n"));
}
}
} else {
out.push_str(
"\nRemainder still pending — re-run `projection advance` after judging the rest \
(a brief re-render shows what is left).\n",
);
}
if !outcome.warnings.is_empty() {
out.push_str("\n## Warnings\n\n");
for w in &outcome.warnings {
out.push_str(&format!("- {w}\n"));
}
}
print_markdown(&out);
}
Ok(())
}
fn map_exclude_err(binding_id: &str, err: ExcludeError) -> CliError {
let message = err.to_string();
match &err {
ExcludeError::MalformedId(_) => {
CliError::new(ExitKind::Validation, "PROJECTION_INVALID_NAME", message)
.with_details(json!({ "binding": binding_id }))
}
ExcludeError::NotSourceMember { artifacts, .. } => CliError::new(
ExitKind::Validation,
"PROJECTION_EXCLUDE_NOT_SOURCE_MEMBER",
message,
)
.with_details(json!({ "binding": binding_id, "not_source_members": artifacts })),
ExcludeError::Store(_) => {
CliError::new(ExitKind::Generic, "PROJECTION_EXCLUDE_FAILED", message)
.with_details(json!({ "binding": binding_id }))
}
}
}
fn exclude(ctx: &CliContext, args: ExcludeArgs) -> anyhow::Result<()> {
let (_shape, root) = ctx.workspace_shape().ok_or_else(|| {
workspace_not_initialised_error(
"not inside a Memstead workspace (no `.memstead/workspace.toml` in any ancestor)",
)
})?;
let binding_id = args.binding;
let exclusions: std::collections::BTreeMap<String, String> =
serde_json::from_str(&args.exclusions).map_err(|e| {
CliError::new(
ExitKind::Validation,
"PROJECTION_INVALID_EXCLUSIONS",
format!(
"--exclusions must be a JSON object mapping in-scope artifact id → \
rationale string: {e}"
),
)
.with_details(json!({ "error": e.to_string() }))
})?;
let configs = load_pipeline_configs(&root).map_err(|e| {
CliError::new(
ExitKind::Generic,
"PROJECTION_EXCLUDE_FAILED",
format!("could not load pipeline config: {e}"),
)
.with_details(json!({ "error": e.to_string() }))
})?;
let record = configs
.bindings
.iter()
.find(|r| format!("{}/{}", r.mem, r.name) == binding_id)
.ok_or_else(|| binding_miss_error(&configs, &binding_id))?;
let resolved = resolve_binding_run(&binding_id, &record.config)
.map_err(|e| map_resolve_err(&binding_id, e))?;
let outcome = record_exclusions(&root, &resolved, &exclusions)
.map_err(|e| map_exclude_err(&binding_id, e))?;
if ctx.json {
print_json(&json!({
"binding": outcome.binding,
"excluded": outcome.excluded,
"added": outcome.added,
}))?;
} else {
print_markdown(&format!(
"# Projection exclude\n\nBinding `{}`: {} artifact(s) newly excluded, \
{} in the ledger.\n",
outcome.binding, outcome.added, outcome.excluded
));
}
Ok(())
}
fn render_full_resync_note(decision: &FullResyncDecision) -> String {
match decision {
FullResyncDecision::Disabled => String::new(),
FullResyncDecision::NotDue { .. } => String::new(),
FullResyncDecision::Forced { walked_facets } => {
let facets = if walked_facets.is_empty() {
"(no primary facets)".to_string()
} else {
walked_facets.join(", ")
};
format!(
"> **Full measurement (`--full`)** — full-enumeration walk over: {facets}. \
Sampling scheduler bypassed; adjudication cap unlimited. Coverage and \
accuracy figures below are computed over the whole source, not sampled.\n\n"
)
}
FullResyncDecision::Due {
walked_facets,
refused,
..
} => {
let mut s = String::from("> **Scheduled full resync (D3)** — ");
if walked_facets.is_empty() {
s.push_str("no enumerable facet to walk this run.");
} else {
s.push_str(&format!(
"full-enumeration coverage walk fired for: {}.",
walked_facets.join(", ")
));
}
for r in refused {
s.push_str(&format!(
"\n> **Refused (non-enumerable):** `{}` ({}) — {}",
r.facet, r.medium_type, r.reason
));
}
s.push_str("\n\n");
s
}
}
}
fn verify(ctx: &CliContext, args: VerifyArgs) -> anyhow::Result<()> {
let (_shape, root) = ctx.workspace_shape().ok_or_else(|| {
workspace_not_initialised_error(
"not inside a Memstead workspace (no `.memstead/workspace.toml` in any ancestor)",
)
})?;
let binding_id = args.binding;
let configs = load_pipeline_configs(&root).map_err(|e| {
CliError::new(
ExitKind::Generic,
"PROJECTION_VERIFY_FAILED",
format!("could not load pipeline config: {e}"),
)
.with_details(json!({ "error": e.to_string() }))
})?;
let record = configs
.bindings
.iter()
.find(|r| format!("{}/{}", r.mem, r.name) == binding_id)
.ok_or_else(|| binding_miss_error(&configs, &binding_id))?;
let resolved = resolve_binding_run(&binding_id, &record.config)
.map_err(|e| map_resolve_err(&binding_id, e))?;
let mut cli_engine = ctx.cli_engine_at(&root)?;
let engine = cli_engine.base_mut();
let run = if args.full {
verify_binding_full
} else {
verify_binding
};
let outcome = run(engine, &root, &record.config, &resolved).map_err(|e| match &e {
FindingsError::SourceUnreachable { source_name, path } => CliError::new(
ExitKind::Validation,
"SOURCE_UNREACHABLE",
format!(
"verify refused for `{binding_id}`: source '{source_name}' resolves to \
`{path}`, which does not exist — restore or remount the source (or \
repoint its pointer); the recorded `#verified` baseline was left \
untouched"
),
)
.with_details(json!({
"binding": binding_id,
"source": source_name,
"path": path,
})),
FindingsError::FullWalkNonEnumerable(refusal) => CliError::new(
ExitKind::Validation,
"PROJECTION_CAPABILITY_UNSUPPORTED",
format!("verify --full refused for `{binding_id}`: {e}"),
)
.with_details(json!({
"binding": binding_id,
"facet": refusal.facet,
"medium_type": refusal.medium_type,
"reason": refusal.reason,
})),
_ => CliError::new(
ExitKind::Generic,
"PROJECTION_VERIFY_FAILED",
format!("verify failed for `{binding_id}`: {e}"),
)
.with_details(json!({ "binding": binding_id, "error": e.to_string() })),
})?;
let hashes_backfilled = record_anchor_hash_backfill(
engine,
&resolved.destination_mem,
&outcome,
Some("projection verify: prepared-hash backfill onto hash-less anchors"),
)
.map_err(|e| {
CliError::new(
ExitKind::Generic,
"PROJECTION_VERIFY_BACKFILL_FAILED",
format!(
"verify completed and findings were recorded for `{binding_id}`, but \
recording the prepared-hash backfill onto the anchors sidecar failed: {e}"
),
)
.with_details(json!({ "binding": binding_id, "error": e.to_string() }))
})?;
let budget = args.budget.unwrap_or(DEFAULT_REPORT_BUDGET);
let report = compute_fidelity_report(engine, &root, &record.config, &resolved, &outcome.key);
let rendered = render_fidelity_report(&report, budget, &args.include);
let verified_baseline = record_verified_baseline(
engine,
&resolved.destination_mem,
&outcome,
Some("projection verify: completed-run #verified baseline"),
)
.map_err(|e| {
CliError::new(
ExitKind::Generic,
"PROJECTION_VERIFY_BASELINE_FAILED",
format!(
"verify completed and findings were recorded for `{binding_id}`, but writing \
the `#verified` baseline failed: {e}"
),
)
.with_details(json!({ "binding": binding_id, "error": e.to_string() }))
})?;
if ctx.json {
print_json(&json!({
"binding": outcome.binding,
"key": {
"binding_hash": outcome.key.binding_hash,
"source_head": outcome.key.source_head,
},
"recorded": outcome.recorded,
"superseded": outcome.superseded,
"backlog": outcome.backlog,
"full_resync": outcome.full_resync,
"verified_baseline": verified_baseline,
"hash_backfilled": hashes_backfilled,
"report": report,
"report_mode": rendered.mode,
"report_markdown": rendered.markdown,
}))?;
} else {
let baseline_note = if verified_baseline.is_empty() {
String::new()
} else {
format!(
"\n> **Verified baseline recorded** — {}\n",
verified_baseline
.iter()
.map(|k| format!("`{k}`"))
.collect::<Vec<_>>()
.join(", ")
)
};
let backfill_note = if hashes_backfilled == 0 {
String::new()
} else {
format!(
"\n> **Prepared-hash backfill recorded** — {hashes_backfilled} hash-less \
anchor(s) now carry their observed prepared-content hash; subsequent \
verifies adjudicate them deterministically.\n"
)
};
print_markdown(&format!(
"{}{}{}{}",
render_full_resync_note(&outcome.full_resync),
rendered.markdown,
backfill_note,
baseline_note
));
}
Ok(())
}