From 7362e54b66dce3fb89d33fe6ddc0140f0d1e5a31 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 17 Jul 2026 01:14:41 +0100 Subject: [PATCH] Checkpoint live workspace projection batches --- .../src/bin/epiphany-daemon-supervisor.rs | 4 + .../epiphany-workspace-coverage-projector.rs | 206 ++++-- ...ce_coverage_projection_batch_checkpoint.rs | 189 ++++++ .../src/workspace_coverage_projector.rs | 620 +++++++++++++++--- .../workspace_coverage_projector_service.rs | 59 +- 5 files changed, 928 insertions(+), 150 deletions(-) diff --git a/epiphany-core/src/bin/epiphany-daemon-supervisor.rs b/epiphany-core/src/bin/epiphany-daemon-supervisor.rs index 8ce6d94a..3c42c1a0 100644 --- a/epiphany-core/src/bin/epiphany-daemon-supervisor.rs +++ b/epiphany-core/src/bin/epiphany-daemon-supervisor.rs @@ -1790,6 +1790,8 @@ fn workspace_coverage_projector_service_args( runtime_id.to_string(), "--interval-seconds".to_string(), interval_seconds.to_string(), + "--heartbeat-interval-seconds".to_string(), + "10".to_string(), "--qdrant-url".to_string(), qdrant_url.to_string(), "--ollama-base-url".to_string(), @@ -3715,6 +3717,8 @@ mod semantic_projector_authority_tests { "local-runtime", "--interval-seconds", "60", + "--heartbeat-interval-seconds", + "10", "--qdrant-url", "http://127.0.0.1:16333", "--ollama-base-url", diff --git a/epiphany-core/src/bin/epiphany-workspace-coverage-projector.rs b/epiphany-core/src/bin/epiphany-workspace-coverage-projector.rs index 0e893bf6..d28ed7b4 100644 --- a/epiphany-core/src/bin/epiphany-workspace-coverage-projector.rs +++ b/epiphany-core/src/bin/epiphany-workspace-coverage-projector.rs @@ -15,6 +15,7 @@ use serde_json::json; use std::env; use std::io; use std::path::PathBuf; +use std::sync::{Arc, Mutex}; use std::thread; use std::time::{Duration, Instant}; use uuid::Uuid; @@ -43,53 +44,45 @@ fn main() -> Result<()> { &launch.provider_incarnation_id, &launch.launch_id, )?; + let trusted_host = open_default_host_identity()?; + projector.install_execution_authority( + &args.local_verse_store, + &args.runtime_id, + trusted_host.entry().clone(), + provider_key.clone(), + )?; - let mut sequence = 0_u64; + let first_heartbeat = publish_heartbeat(&args, &launch, &launch_digest, &provider_key, 1)?; + let latest_heartbeat = Arc::new(Mutex::new(HeartbeatProjection { + heartbeat_id: first_heartbeat.heartbeat_id, + provider_status: first_heartbeat.status, + })); + spawn_heartbeat_nerve( + &args, + &launch, + &launch_digest, + &provider_key, + Arc::clone(&latest_heartbeat), + )?; + + let mut pulse_sequence = 0_u64; loop { - sequence = sequence + pulse_sequence = pulse_sequence .checked_add(1) - .ok_or_else(|| anyhow!("workspace coverage heartbeat sequence exhausted"))?; + .ok_or_else(|| anyhow!("workspace coverage pulse sequence exhausted"))?; let pulse = projector.pulse(); - // Contention describes canonical claim ownership, not provider health. - // Publishing it as degraded would make a live successor unable to - // establish the ready heartbeat required to fence an abandoned owner. - let degraded = matches!(pulse.status, WorkspaceCoverageProjectorPulseStatus::Refused); - let mut heartbeat = WorkspaceCoverageProviderHeartbeatEntry { - schema_version: WORKSPACE_COVERAGE_PROVIDER_HEARTBEAT_SCHEMA_VERSION.to_string(), - heartbeat_id: Uuid::new_v4().to_string(), - launch_id: launch.launch_id.clone(), - launch_envelope_digest: launch_digest.clone(), - service_id: launch.service_id.clone(), - provider_daemon_id: launch.provider_daemon_id.clone(), - runtime_id: launch.runtime_id.clone(), - host_identity_id: launch.host_identity_id.clone(), - host_identity_record_digest: launch.host_identity_record_digest.clone(), - boot_identity: launch.boot_identity.clone(), - process_id: launch.process_id, - process_creation_token: launch.process_creation_token, - process_executable_path: launch.process_executable_path.clone(), - provider_incarnation_id: launch.provider_incarnation_id.clone(), - provider_public_key: launch.provider_public_key.clone(), - sequence, - status: if degraded { "degraded" } else { "ready" }.to_string(), - observed_at_utc: chrono::Utc::now().to_rfc3339(), - provider_signature: Vec::new(), - signature_algorithm: "ed25519".to_string(), - }; - sign_workspace_coverage_heartbeat(&mut heartbeat, &provider_key)?; - let heartbeat = write_workspace_coverage_provider_heartbeat( - &args.local_verse_store, - args.runtime_id.clone(), - heartbeat, - )?; + let heartbeat = latest_heartbeat + .lock() + .map_err(|_| anyhow!("workspace coverage heartbeat projection lock poisoned"))? + .clone(); println!( "{}", serde_json::to_string(&json!({ "schemaVersion": "epiphany.workspace_coverage_projector_pulse.v0", - "pulseSequence": sequence, + "pulseSequence": pulse_sequence, "providerIncarnation": projector.provider_incarnation(), "heartbeatId": heartbeat.heartbeat_id, - "providerStatus": heartbeat.status, + "providerStatus": heartbeat.provider_status, "pulseStatus": pulse_status(pulse.status), "bodyObservationId": pulse.body_observation_id, "bodyGeneration": pulse.body_generation, @@ -104,6 +97,134 @@ fn main() -> Result<()> { } } +#[derive(Clone)] +struct HeartbeatProjection { + heartbeat_id: String, + provider_status: String, +} + +fn spawn_heartbeat_nerve( + args: &Args, + launch: &WorkspaceCoverageManagedProcessLaunchEntry, + launch_digest: &str, + provider_key: &SigningKey, + latest: Arc>, +) -> Result<()> { + let local_verse_store = args.local_verse_store.clone(); + let runtime_id = args.runtime_id.clone(); + let interval = Duration::from_secs(args.heartbeat_interval_seconds); + let launch = launch.clone(); + let launch_digest = launch_digest.to_string(); + let provider_key = provider_key.clone(); + thread::Builder::new() + .name("workspace-coverage-heartbeat".into()) + .spawn(move || { + let mut sequence = 1_u64; + loop { + thread::sleep(interval); + sequence = match sequence.checked_add(1) { + Some(sequence) => sequence, + None => { + fatal_heartbeat_nerve("workspace coverage heartbeat sequence exhausted"); + } + }; + let heartbeat = match publish_heartbeat_to( + &local_verse_store, + &runtime_id, + &launch, + &launch_digest, + &provider_key, + sequence, + ) { + Ok(heartbeat) => heartbeat, + Err(error) => { + fatal_heartbeat_nerve(&format!( + "workspace coverage heartbeat publication failed: {error:#}" + )); + } + }; + match latest.lock() { + Ok(mut projection) => { + projection.heartbeat_id = heartbeat.heartbeat_id; + projection.provider_status = heartbeat.status; + } + Err(_) => { + fatal_heartbeat_nerve( + "workspace coverage heartbeat projection lock poisoned", + ); + } + } + } + }) + .context("failed to start workspace coverage heartbeat nerve")?; + Ok(()) +} + +fn fatal_heartbeat_nerve(message: &str) -> ! { + eprintln!("{message}"); + // This exact process owns both projection and its authenticated liveness + // publisher. Continuing after the nerve dies would leave an unfenced + // actuator relying on an external repair loop to notice the corpse. + std::process::exit(1) +} + +fn publish_heartbeat( + args: &Args, + launch: &WorkspaceCoverageManagedProcessLaunchEntry, + launch_digest: &str, + provider_key: &SigningKey, + sequence: u64, +) -> Result { + publish_heartbeat_to( + &args.local_verse_store, + &args.runtime_id, + launch, + launch_digest, + provider_key, + sequence, + ) +} + +fn publish_heartbeat_to( + local_verse_store: &PathBuf, + runtime_id: &str, + launch: &WorkspaceCoverageManagedProcessLaunchEntry, + launch_digest: &str, + provider_key: &SigningKey, + sequence: u64, +) -> Result { + let mut heartbeat = WorkspaceCoverageProviderHeartbeatEntry { + schema_version: WORKSPACE_COVERAGE_PROVIDER_HEARTBEAT_SCHEMA_VERSION.to_string(), + heartbeat_id: Uuid::new_v4().to_string(), + launch_id: launch.launch_id.clone(), + launch_envelope_digest: launch_digest.to_string(), + service_id: launch.service_id.clone(), + provider_daemon_id: launch.provider_daemon_id.clone(), + runtime_id: launch.runtime_id.clone(), + host_identity_id: launch.host_identity_id.clone(), + host_identity_record_digest: launch.host_identity_record_digest.clone(), + boot_identity: launch.boot_identity.clone(), + process_id: launch.process_id, + process_creation_token: launch.process_creation_token, + process_executable_path: launch.process_executable_path.clone(), + provider_incarnation_id: launch.provider_incarnation_id.clone(), + provider_public_key: launch.provider_public_key.clone(), + sequence, + // Heartbeat proves only that this exact provider process remains able + // to publish. Projection faults and progress belong to separate state. + status: "ready".to_string(), + observed_at_utc: chrono::Utc::now().to_rfc3339(), + provider_signature: Vec::new(), + signature_algorithm: "ed25519".to_string(), + }; + sign_workspace_coverage_heartbeat(&mut heartbeat, provider_key)?; + write_workspace_coverage_provider_heartbeat( + local_verse_store, + runtime_id.to_string(), + heartbeat, + ) +} + fn authenticate_managed_launch( args: &Args, provider_key: &SigningKey, @@ -168,6 +289,7 @@ struct Args { local_verse_store: PathBuf, runtime_id: String, interval_seconds: u64, + heartbeat_interval_seconds: u64, managed_process_launch_id: String, qdrant_url: String, ollama_base_url: String, @@ -185,6 +307,7 @@ impl Args { let mut local_verse_store = None; let mut runtime_id = None; let mut interval_seconds = None; + let mut heartbeat_interval_seconds = None; let mut qdrant_url = None; let mut ollama_base_url = None; let mut ollama_model = None; @@ -199,6 +322,9 @@ impl Args { "--local-verse-store" => local_verse_store = Some(PathBuf::from(value()?)), "--runtime-id" => runtime_id = Some(value()?), "--interval-seconds" => interval_seconds = Some(value()?.parse::()?), + "--heartbeat-interval-seconds" => { + heartbeat_interval_seconds = Some(value()?.parse::()?) + } "--qdrant-url" => qdrant_url = Some(value()?), "--ollama-base-url" => ollama_base_url = Some(value()?), "--ollama-model" => ollama_model = Some(value()?), @@ -209,6 +335,11 @@ impl Args { if interval_seconds == 0 { return Err(anyhow!("--interval-seconds must be positive")); } + let heartbeat_interval_seconds = + heartbeat_interval_seconds.context("missing --heartbeat-interval-seconds")?; + if heartbeat_interval_seconds == 0 { + return Err(anyhow!("--heartbeat-interval-seconds must be positive")); + } let managed_process_launch_id = env::var("EPIPHANY_WORKSPACE_COVERAGE_LAUNCH_ID").unwrap_or_default(); if managed_process_launch_id.trim().is_empty() { @@ -221,6 +352,7 @@ impl Args { local_verse_store: local_verse_store.context("missing --local-verse-store")?, runtime_id: runtime_id.context("missing --runtime-id")?, interval_seconds, + heartbeat_interval_seconds, managed_process_launch_id, qdrant_url: required_nonempty(qdrant_url, "--qdrant-url")?, ollama_base_url: required_nonempty(ollama_base_url, "--ollama-base-url")?, diff --git a/epiphany-core/src/workspace_coverage_projection_batch_checkpoint.rs b/epiphany-core/src/workspace_coverage_projection_batch_checkpoint.rs index c6ac4ad3..f781ed58 100644 --- a/epiphany-core/src/workspace_coverage_projection_batch_checkpoint.rs +++ b/epiphany-core/src/workspace_coverage_projection_batch_checkpoint.rs @@ -471,6 +471,151 @@ pub(crate) fn load_authenticated_current_checkpoint( })) } +pub(crate) fn load_authenticated_checkpoint_chain( + body_store: impl AsRef, + local_verse_store: impl AsRef, + trusted_host: &HostIncarnationIdentityEntry, + claim_id: &str, + claim_epoch: u64, +) -> Result> { + let Some(current) = load_authenticated_current_checkpoint( + body_store.as_ref(), + local_verse_store.as_ref(), + trusted_host, + claim_id, + claim_epoch, + )? else { + return Ok(Vec::new()); + }; + let opening = SingleFileMessagePackBackingStore::new(body_store.as_ref()).pull_all()?; + let head_env = find( + &opening, + HEAD_TYPE, + &checkpoint_head_key(claim_id, claim_epoch), + )?; + let head: WorkspaceCoverageProjectionBatchCheckpointHeadEntry = decode(head_env)?; + if head.checkpoint_id != current.checkpoint.checkpoint_id + || head.checkpoint_envelope_digest != current.checkpoint_envelope_digest + { + bail!("checkpoint head changed while reconstructing authenticated chain"); + } + let binding_env = find(&opening, BODY_BINDING_TYPE, BODY_BINDING_KEY)?; + let body_head_env = find(&opening, BODY_HEAD_TYPE, BODY_HEAD_KEY)?; + let observation_env = find( + &opening, + BODY_OBSERVATION_TYPE, + ¤t.checkpoint.body_observation_id, + )?; + let manifest_env = find( + &opening, + BODY_MANIFEST_TYPE, + ¤t.checkpoint.manifest_root_sha256, + )?; + let obligation_env = find(&opening, OBLIGATION_TYPE, ¤t.checkpoint.obligation_id)?; + let plan_env = find(&opening, PLAN_TYPE, ¤t.checkpoint.plan_id)?; + let claim_env = find(&opening, CLAIM_TYPE, CLAIM_KEY)?; + let attempt_env = find(&opening, ATTEMPT_TYPE, ¤t.checkpoint.attempt_id)?; + let binding: RepositoryBodyBinding = decode(binding_env)?; + let body_head: RepositoryBodyHead = decode(body_head_env)?; + let observation: RepositoryBodyObservation = decode(observation_env)?; + let manifest: RepositoryBodyManifest = decode(manifest_env)?; + let obligation: WorkspaceCoverageObligation = decode(obligation_env)?; + let plan: WorkspaceCoverageProjectionPlan = decode(plan_env)?; + let claim: WorkspaceCoverageProjectionClaim = decode(claim_env)?; + let attempt: WorkspaceCoverageProjectionAttempt = decode(attempt_env)?; + validate_projection_claim(&claim)?; + validate_projection_attempt(&attempt)?; + validate_claim_attempt_link(&claim, &attempt)?; + validate_current_projection_binding(¤t.checkpoint, CurrentProjectionBinding { + claim_status: &claim.status, claim_id: &claim.claim_id, claim_epoch: claim.claim_epoch, + attempt_status: &attempt.status, attempt_id: &attempt.attempt_id, + claim_plan_id: &claim.plan_id, attempt_plan_id: &attempt.plan_id, plan_id: &plan.plan_id, + claim_obligation_id: &claim.obligation_id, obligation_id: &obligation.obligation_id, + plan_obligation_id: &plan.obligation_id, + claim_body_observation_id: &claim.body_observation_id, + obligation_body_observation_id: &obligation.body_observation_id, + claim_body_generation: claim.body_generation, + obligation_body_generation: obligation.body_generation, + claim_manifest_root: &claim.manifest_root_sha256, + obligation_manifest_root: &obligation.manifest_root_sha256, + })?; + validate_workspace_coverage_projection_plan(&obligation, &plan)?; + exact_obligation_body_authority(&opening, &obligation)?; + let (launch, launch_digest) = + authenticate_workspace_coverage_managed_process_launch_with_envelope_digest( + local_verse_store, + current.checkpoint.runtime_id.clone(), + ¤t.checkpoint.managed_process_launch_id, + trusted_host, + )?; + + let mut reversed = Vec::new(); + let mut next_id = Some(head.checkpoint_id.clone()); + let mut next_digest = Some(head.checkpoint_envelope_digest.clone()); + let mut seen_checkpoints = std::collections::HashSet::new(); + while let (Some(id), Some(expected_digest)) = (next_id.take(), next_digest.take()) { + if !seen_checkpoints.insert(id.clone()) { + bail!("checkpoint predecessor chain contains a cycle or duplicate event"); + } + let event_env = find( + &opening, + WORKSPACE_COVERAGE_PROJECTION_BATCH_CHECKPOINT_TYPE, + &checkpoint_key(claim_id, &id), + )?; + if envelope_digest(event_env) != expected_digest { + bail!("checkpoint predecessor digest disagrees with exact event envelope"); + } + let event: WorkspaceCoverageProjectionBatchCheckpointEntry = decode(event_env)?; + validate_shape(&event, true)?; + authenticate_signature(&event)?; + validate_launch(&event, &launch, &launch_digest)?; + validate_authority( + &event, &binding, binding_env, &body_head, body_head_env, &observation, + observation_env, &manifest, manifest_env, &obligation, obligation_env, &plan, + plan_env, &claim, claim_env, &attempt, attempt_env, + )?; + validate_batch_against_plan(&event, &obligation, &plan)?; + next_id = event.predecessor_checkpoint_id.clone(); + next_digest = event.predecessor_checkpoint_envelope_digest.clone(); + reversed.push(WorkspaceCoverageProjectionBatchCheckpointAdmission { + checkpoint: event, + checkpoint_envelope_digest: expected_digest, + }); + } + reversed.reverse(); + validate_reconstructed_chain(&reversed, &head)?; + Ok(reversed) +} + +fn validate_reconstructed_chain( + chain: &[WorkspaceCoverageProjectionBatchCheckpointAdmission], + head: &WorkspaceCoverageProjectionBatchCheckpointHeadEntry, +) -> Result<()> { + let mut prior: Option<&WorkspaceCoverageProjectionBatchCheckpointEntry> = None; + let mut seen_points = std::collections::HashSet::new(); + for admission in chain { + let event = &admission.checkpoint; + let expected_sequence = match prior { + Some(p) => p.sequence.checked_add(1).ok_or_else(|| anyhow!("checkpoint sequence exhausted"))?, + None => 1, + }; + if event.sequence != expected_sequence + || event.first_plan_ordinal != prior.map_or(0, |p| p.cumulative_point_count) + || event.batch_ordinal != event.sequence - 1 + || event.point_ids.iter().any(|id| !seen_points.insert(id.clone())) + { + bail!("checkpoint chain is noncontiguous or overlaps prior sealed points"); + } + prior = Some(event); + } + if prior.map(|p| p.sequence) != Some(head.sequence) + || prior.map(|p| p.cumulative_point_count) != Some(head.cumulative_point_count) + { + bail!("checkpoint chain does not reconstruct the authenticated current head"); + } + Ok(()) +} + #[derive(Clone, Copy)] struct CurrentProjectionBinding<'a> { claim_status: &'a str, @@ -1313,6 +1458,50 @@ mod tests { } } #[test] + fn reconstructed_chain_rejects_corrupted_earlier_batch_and_overlap() { + let k = SigningKey::from_bytes(&[7; 32]); + let first = sample(&k); + let mut second = first.clone(); + second.checkpoint_id = uuid::Uuid::new_v4().to_string(); + second.sequence = 2; + second.batch_ordinal = 1; + second.first_plan_ordinal = 1; + second.cumulative_point_count = 2; + second.point_ids = vec!["point-2".into()]; + second.point_bindings[0].point_id = "point-2".into(); + second.vector_bindings[0].point_id = "point-2".into(); + second.predecessor_checkpoint_id = Some(first.checkpoint_id.clone()); + second.predecessor_checkpoint_envelope_digest = Some(format!("sha256-{}", "8".repeat(64))); + let chain = vec![ + WorkspaceCoverageProjectionBatchCheckpointAdmission { + checkpoint: first.clone(), + checkpoint_envelope_digest: format!("sha256-{}", "8".repeat(64)), + }, + WorkspaceCoverageProjectionBatchCheckpointAdmission { + checkpoint: second.clone(), + checkpoint_envelope_digest: format!("sha256-{}", "9".repeat(64)), + }, + ]; + let head = WorkspaceCoverageProjectionBatchCheckpointHeadEntry { + schema_version: HEAD_SCHEMA.into(), + claim_id: second.claim_id.clone(), + claim_epoch: second.claim_epoch, + checkpoint_id: second.checkpoint_id.clone(), + checkpoint_envelope_digest: chain[1].checkpoint_envelope_digest.clone(), + sequence: 2, + cumulative_point_count: 2, + total_point_count: 2, + collection_name: second.collection_name.clone(), + }; + validate_reconstructed_chain(&chain, &head).unwrap(); + let mut corrupt = chain.clone(); + corrupt[0].checkpoint.sequence = 2; + assert!(validate_reconstructed_chain(&corrupt, &head).is_err()); + let mut overlap = chain.clone(); + overlap[1].checkpoint.point_ids[0] = overlap[0].checkpoint.point_ids[0].clone(); + assert!(validate_reconstructed_chain(&overlap, &head).is_err()); + } + #[test] fn current_body_projection_rejects_historical_or_terminal_checkpoint_authority() { let k = SigningKey::from_bytes(&[7; 32]); let event = sample(&k); diff --git a/epiphany-core/src/workspace_coverage_projector.rs b/epiphany-core/src/workspace_coverage_projector.rs index 9f115548..ec9f9d10 100644 --- a/epiphany-core/src/workspace_coverage_projector.rs +++ b/epiphany-core/src/workspace_coverage_projector.rs @@ -1,6 +1,17 @@ use crate::repository_body_observer::RepositoryBodyReadSession; use crate::semantic_backend::{ - CollectionCompatibility, OllamaEmbedder, QdrantBackend, SemanticPoint, SemanticStoredPoint, + CollectionCompatibility, OllamaEmbedder, QDRANT_POINT_BATCH_MAX, QdrantBackend, SemanticPoint, + SemanticStoredPoint, +}; +use crate::workspace_coverage_projection_batch_checkpoint::{ + ObservedWorkspaceCoverageBatchInput, WorkspaceCoverageProjectionBatchCheckpointAdmission, + admit_observed_workspace_coverage_batch, load_authenticated_checkpoint_chain, +}; +use crate::workspace_coverage_projection_progress::{ + authenticate_workspace_coverage_projection_progress, + load_latest_workspace_coverage_projection_progress, + publish_workspace_coverage_progress_for_checkpoint, + publish_workspace_coverage_progress_genesis, reconcile_workspace_coverage_checkpoint_progress, }; use crate::{ BODY_BINDING_KEY, BODY_BINDING_TYPE, BODY_HEAD_KEY, BODY_HEAD_TYPE, BODY_MANIFEST_TYPE, @@ -25,6 +36,7 @@ use anyhow::{Result, anyhow, bail}; use cultcache_rs::{ CacheBackingStore, CultCacheEnvelope, DatabaseEntry, SingleFileMessagePackBackingStore, }; +use ed25519_dalek::SigningKey; use serde::Serialize; use sha2::{Digest, Sha256}; use std::path::{Path, PathBuf}; @@ -51,6 +63,7 @@ const HEAD_KEY: &str = "current"; pub(crate) const OBLIGATION_TYPE: &str = "gamecult.epiphany.workspace_coverage_obligation"; pub(crate) const PLAN_TYPE: &str = "gamecult.epiphany.workspace_coverage_projection_plan"; +#[cfg(test)] #[derive(Clone, Debug)] struct WorkspaceCoverageProjectionInput { pub point_id: String, @@ -58,6 +71,14 @@ struct WorkspaceCoverageProjectionInput { pub vector: Vec, } +pub(crate) struct WorkspaceCoverageProjectionExecutionAuthority<'a> { + pub local_verse_store: &'a Path, + pub runtime_id: &'a str, + pub trusted_host: &'a HostIncarnationIdentityEntry, + pub provider_signing_key: &'a SigningKey, + pub operation_timeout_ms: u64, +} + #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) struct WorkspaceCoverageObservedBinding { collection_name: String, @@ -91,11 +112,16 @@ pub(crate) trait WorkspaceCoverageProjectionPort { collection: &str, compatibility: &CollectionCompatibility, ) -> Result<()>; - fn upsert( + fn upsert_waited_batch( &mut self, collection: &str, points: &[SemanticPoint], ) -> Result<()>; + fn observe_exact_ids( + &mut self, + collection: &str, + point_ids: &[String], + ) -> Result>>; fn observe_all( &mut self, collection: &str, @@ -150,12 +176,19 @@ impl WorkspaceCoverageProjectionPort for QdrantBackend { ) -> Result<()> { QdrantBackend::ensure_exact_collection(self, name, expected) } - fn upsert( + fn upsert_waited_batch( &mut self, name: &str, points: &[SemanticPoint], ) -> Result<()> { - self.upsert_points(name, points) + self.upsert_point_batch_waited(name, points).map(|_| ()) + } + fn observe_exact_ids( + &mut self, + name: &str, + point_ids: &[String], + ) -> Result>> { + self.retrieve_points_by_ids(name, point_ids) } fn observe_all( &mut self, @@ -303,7 +336,9 @@ pub(crate) fn validate_projection_claim(claim: &WorkspaceCoverageProjectionClaim Ok(()) } -pub(crate) fn validate_projection_attempt(attempt: &WorkspaceCoverageProjectionAttempt) -> Result<()> { +pub(crate) fn validate_projection_attempt( + attempt: &WorkspaceCoverageProjectionAttempt, +) -> Result<()> { if !matches!( attempt.schema_version.as_str(), ATTEMPT_SCHEMA | ATTEMPT_SCHEMA_V2 | ATTEMPT_SCHEMA_V3 @@ -1605,6 +1640,7 @@ pub(crate) fn execute_workspace_coverage_projection( prepared: &PreparedWorkspaceCoverageProjection, embedder: &OllamaEmbedder, port: &mut impl WorkspaceCoverageProjectionPort, + authority: &WorkspaceCoverageProjectionExecutionAuthority<'_>, ) -> Result { let result = (|| { if prepared.body_store != acquisition.body_store @@ -1613,30 +1649,247 @@ pub(crate) fn execute_workspace_coverage_projection( { bail!("execution refuses a prepared projection other than the acquired projection"); } - let texts = prepared - .points + execute_batched_and_commit(acquisition, prepared, embedder, port, authority) + })(); + terminalize_execution_result(acquisition, result) +} + +fn execute_batched_and_commit( + acquisition: &WorkspaceCoverageAcquisition, + prepared: &PreparedWorkspaceCoverageProjection, + embedder: &OllamaEmbedder, + port: &mut impl WorkspaceCoverageProjectionPort, + authority: &WorkspaceCoverageProjectionExecutionAuthority<'_>, +) -> Result { + if authority.operation_timeout_ms == 0 { + bail!("workspace coverage execution authority requires a positive operation timeout"); + } + let planned = &acquisition.plan.planned_points; + if prepared.points.len() != planned.len() { + bail!("prepared projection does not equal the sealed plan point count"); + } + for (ordinal, (prepared_point, planned_point)) in + prepared.points.iter().zip(planned).enumerate() + { + if prepared_point.point_id != planned_point.point_id + || format!("{:x}", Sha256::digest(prepared_point.text.as_bytes())) + != planned_point.chunk_sha256 + { + bail!("prepared projection disagrees with sealed plan ordinal {ordinal}"); + } + } + let collection_name = workspace_coverage_execution_collection( + &acquisition.plan.plan_id, + &acquisition.claim.claim_id, + acquisition.claim.claim_epoch, + )?; + port.ensure_exact_collection( + &collection_name, + &collection_compatibility(&acquisition.plan), + )?; + + let checkpoint_chain = load_authenticated_checkpoint_chain( + &acquisition.body_store, + authority.local_verse_store, + authority.trusted_host, + &acquisition.claim.claim_id, + acquisition.claim.claim_epoch, + )?; + let mut next_ordinal = if let Some(current) = checkpoint_chain.last() { + reconcile_workspace_coverage_checkpoint_progress( + &acquisition.body_store, + authority.local_verse_store, + authority.runtime_id, + authority.trusted_host, + authority.provider_signing_key, + &acquisition.claim.claim_id, + acquisition.claim.claim_epoch, + )?; + for checkpoint in &checkpoint_chain { + authenticate_checkpoint_readback(&collection_name, &checkpoint.checkpoint, port)?; + } + current.checkpoint.cumulative_point_count as usize + } else { + if let Some(existing) = load_latest_workspace_coverage_projection_progress( + authority.local_verse_store, + authority.runtime_id, + &acquisition.claim.managed_process_launch_id, + &acquisition.claim.claim_id, + )? { + let existing = authenticate_workspace_coverage_projection_progress( + authority.local_verse_store, + authority.runtime_id, + &existing.progress_id, + authority.trusted_host, + )?; + if existing.sequence != 1 + || existing.completed_units != 0 + || existing.checkpoint_id.is_some() + || existing.plan_id != acquisition.plan.plan_id + || existing.attempt_id != acquisition.attempt.attempt_id + { + bail!("checkpoint-free restart found progress other than exact genesis"); + } + } else { + publish_workspace_coverage_progress_genesis( + &acquisition.body_store, + authority.local_verse_store, + authority.runtime_id, + authority.trusted_host, + authority.provider_signing_key, + authority.operation_timeout_ms, + )?; + } + 0 + }; + + while next_ordinal < planned.len() { + let end = (next_ordinal + QDRANT_POINT_BATCH_MAX).min(planned.len()); + let texts = prepared.points[next_ordinal..end] .iter() .map(|point| point.text.clone()) .collect::>(); - let vectors = if texts.is_empty() { - Vec::new() - } else { - embedder.embed_documents(&texts)? - }; - let inputs = prepared - .points + let vectors = embedder.embed_documents(&texts)?; + if vectors.len() != texts.len() { + bail!("embedding provider omitted sealed batch vectors"); + } + let mut points = Vec::with_capacity(end - next_ordinal); + let mut expected_point_bindings = Vec::with_capacity(end - next_ordinal); + let mut expected_vector_bindings = Vec::with_capacity(end - next_ordinal); + for ((prepared_point, expected), vector) in prepared.points[next_ordinal..end] .iter() - .cloned() + .zip(&planned[next_ordinal..end]) .zip(vectors) - .map(|(point, vector)| WorkspaceCoverageProjectionInput { - point_id: point.point_id, - text: point.text, + { + if vector.len() != acquisition.plan.vector_dimensions as usize + || vector.iter().any(|value| !value.is_finite()) + { + bail!("embedding provider returned an invalid sealed batch vector"); + } + let payload = payload_for(&acquisition.obligation, &acquisition.plan, expected); + expected_point_bindings.push(WorkspaceCoveragePointBinding { + point_id: expected.point_id.clone(), + payload_sha256: digest(&payload)?, + }); + expected_vector_bindings.push(WorkspaceCoverageVectorBinding { + point_id: expected.point_id.clone(), + vector_sha256: digest(&vector)?, + }); + points.push(SemanticPoint { + id: prepared_point.point_id.clone(), vector, - }) - .collect(); - execute_and_commit(acquisition, inputs, port) - })(); - terminalize_execution_result(acquisition, result) + payload, + }); + } + port.upsert_waited_batch(&collection_name, &points)?; + let ids = points + .iter() + .map(|point| point.id.clone()) + .collect::>(); + let observed = port.observe_exact_ids(&collection_name, &ids)?; + authenticate_batch_readback( + &observed, + &expected_point_bindings, + &expected_vector_bindings, + acquisition.plan.vector_dimensions, + )?; + let checkpoint = admit_observed_workspace_coverage_batch( + &acquisition.body_store, + authority.local_verse_store, + authority.runtime_id, + authority.trusted_host, + authority.provider_signing_key, + ObservedWorkspaceCoverageBatchInput { + claim_id: acquisition.claim.claim_id.clone(), + attempt_id: acquisition.attempt.attempt_id.clone(), + plan_id: acquisition.plan.plan_id.clone(), + first_plan_ordinal: next_ordinal as u64, + point_bindings: expected_point_bindings, + vector_bindings: expected_vector_bindings, + }, + )?; + publish_workspace_coverage_progress_for_checkpoint( + authority.local_verse_store, + authority.runtime_id, + authority.provider_signing_key, + &checkpoint, + )?; + next_ordinal = end; + } + + let final_chain = load_authenticated_checkpoint_chain( + &acquisition.body_store, + authority.local_verse_store, + authority.trusted_host, + &acquisition.claim.claim_id, + acquisition.claim.claim_epoch, + )?; + let observed = observe_final_projection_against_authenticated_checkpoint_chain( + &acquisition.obligation, + &acquisition.plan, + &collection_name, + port.observe_all(&collection_name)?, + &final_chain, + )?; + commit_workspace_coverage_success(acquisition, observed) +} + +fn authenticate_checkpoint_readback( + collection_name: &str, + checkpoint: &crate::workspace_coverage_projection_batch_checkpoint::WorkspaceCoverageProjectionBatchCheckpointEntry, + port: &mut impl WorkspaceCoverageProjectionPort, +) -> Result<()> { + if checkpoint.collection_name != collection_name { + bail!("checkpoint collection disagrees with the sealed execution collection"); + } + let observed = port.observe_exact_ids(collection_name, &checkpoint.point_ids)?; + authenticate_batch_readback( + &observed, + &checkpoint.point_bindings, + &checkpoint.vector_bindings, + checkpoint.vector_dimensions, + ) +} + +fn authenticate_batch_readback( + observed: &[SemanticStoredPoint], + expected_points: &[WorkspaceCoveragePointBinding], + expected_vectors: &[WorkspaceCoverageVectorBinding], + vector_dimensions: u32, +) -> Result<()> { + if observed.len() != expected_points.len() || observed.len() != expected_vectors.len() { + bail!("exact batch readback count disagrees with submitted bindings"); + } + for ((point, expected_point), expected_vector) in + observed.iter().zip(expected_points).zip(expected_vectors) + { + let payload = point + .payload + .as_ref() + .ok_or_else(|| anyhow!("exact batch readback omitted payload"))?; + let vector = point + .vector + .as_ref() + .ok_or_else(|| anyhow!("exact batch readback omitted vector"))?; + if point.id != expected_point.point_id + || point.id != expected_vector.point_id + || digest(payload)? != expected_point.payload_sha256 + || vector.len() != vector_dimensions as usize + || vector.iter().any(|value| !value.is_finite()) + || digest(vector)? != expected_vector.vector_sha256 + { + bail!("exact batch readback disagrees with submitted payload/vector bindings"); + } + } + Ok(()) +} + +#[cfg(test)] +fn sealed_batch_ranges(point_count: usize) -> Vec> { + (0..point_count) + .step_by(QDRANT_POINT_BATCH_MAX) + .map(|start| start..(start + QDRANT_POINT_BATCH_MAX).min(point_count)) + .collect() } #[cfg(test)] @@ -1667,6 +1920,7 @@ fn terminalize_execution_result( } } +#[cfg(test)] fn execute_and_commit( acquisition: &WorkspaceCoverageAcquisition, mut inputs: Vec, @@ -1706,7 +1960,7 @@ fn execute_and_commit( let compatibility = collection_compatibility(&acquisition.plan); port.ensure_exact_collection(&collection_name, &compatibility)?; if !points.is_empty() { - port.upsert(&collection_name, &points)?; + port.upsert_waited_batch(&collection_name, &points)?; } let observed = observe_exact_bindings( acquisition, @@ -1717,6 +1971,7 @@ fn execute_and_commit( commit_workspace_coverage_success(acquisition, observed) } +#[cfg(test)] fn observe_exact_bindings( acquisition: &WorkspaceCoverageAcquisition, collection_name: &str, @@ -1733,6 +1988,95 @@ fn observe_exact_bindings( Ok(binding) } +pub(crate) fn observe_final_projection_against_authenticated_checkpoint_chain( + obligation: &WorkspaceCoverageObligation, + plan: &WorkspaceCoverageProjectionPlan, + collection_name: &str, + observed: Vec>, + chain: &[WorkspaceCoverageProjectionBatchCheckpointAdmission], +) -> Result { + let mut expected_points = Vec::new(); + let mut expected_vectors = Vec::new(); + let mut expected_ordinal = 0_u64; + for admission in chain { + let checkpoint = &admission.checkpoint; + if checkpoint.obligation_id != obligation.obligation_id + || checkpoint.plan_id != plan.plan_id + || checkpoint.first_plan_ordinal != expected_ordinal + || checkpoint.point_bindings.len() != checkpoint.vector_bindings.len() + { + bail!("authenticated checkpoint chain disagrees with final sealed projection"); + } + expected_points.extend(checkpoint.point_bindings.iter().cloned()); + expected_vectors.extend(checkpoint.vector_bindings.iter().cloned()); + expected_ordinal = checkpoint.cumulative_point_count; + } + if expected_ordinal != plan.expected_point_count + || expected_points.len() as u64 != plan.expected_point_count + || expected_vectors.len() != expected_points.len() + { + bail!("authenticated checkpoint chain does not cover the complete sealed plan"); + } + for ((planned, point), vector) in plan + .planned_points + .iter() + .zip(&expected_points) + .zip(&expected_vectors) + { + if point.point_id != planned.point_id + || vector.point_id != planned.point_id + || point.payload_sha256 != digest(&payload_for(obligation, plan, planned))? + { + bail!("authenticated checkpoint chain order or payload binding is invalid"); + } + } + let ordered = authenticate_final_observation_against_checkpoint_bindings( + observed, + &expected_points, + &expected_vectors, + )?; + observe_sealed_bindings(obligation, plan, collection_name, ordered, Some(&expected_vectors)) +} + +fn authenticate_final_observation_against_checkpoint_bindings( + observed: Vec>, + expected_points: &[WorkspaceCoveragePointBinding], + expected_vectors: &[WorkspaceCoverageVectorBinding], +) -> Result>> { + let mut observed_by_id = std::collections::HashMap::with_capacity(observed.len()); + for point in observed { + let point_id = point.id.clone(); + if observed_by_id.insert(point_id.clone(), point).is_some() { + bail!("final Qdrant observation duplicated checkpointed point {point_id}"); + } + } + if observed_by_id.len() != expected_points.len() { + bail!("final Qdrant observation cardinality disagrees with authenticated checkpoint chain"); + } + let mut ordered = Vec::with_capacity(expected_points.len()); + for (point, vector) in expected_points.iter().zip(expected_vectors) { + let observed = observed_by_id + .remove(&point.point_id) + .ok_or_else(|| anyhow!("final Qdrant observation omitted a checkpointed point"))?; + let payload = observed + .payload + .as_ref() + .ok_or_else(|| anyhow!("final Qdrant observation omitted checkpointed payload"))?; + let observed_vector = observed + .vector + .as_ref() + .ok_or_else(|| anyhow!("final Qdrant observation omitted checkpointed vector"))?; + if digest(payload)? != point.payload_sha256 || digest(observed_vector)? != vector.vector_sha256 { + bail!("final Qdrant payload/vector disagrees with authenticated checkpoint evidence"); + } + ordered.push(observed); + } + if !observed_by_id.is_empty() { + bail!("final Qdrant observation contains uncheckpointed points"); + } + Ok(ordered) +} + fn observe_sealed_bindings( obligation: &WorkspaceCoverageObligation, plan: &WorkspaceCoverageProjectionPlan, @@ -2141,22 +2485,42 @@ mod tests { self.collection_compatibility = Some(compatibility.clone()); Ok(()) } - fn upsert( + fn upsert_waited_batch( &mut self, _: &str, points: &[SemanticPoint], ) -> Result<()> { self.upsert_calls += 1; - self.points = points - .iter() - .map(|point| SemanticStoredPoint { + for point in points { + let stored = SemanticStoredPoint { id: point.id.clone(), payload: Some(point.payload.clone()), vector: Some(point.vector.clone()), - }) - .collect(); + }; + if let Some(existing) = self.points.iter_mut().find(|item| item.id == point.id) { + *existing = stored; + } else { + self.points.push(stored); + } + } Ok(()) } + fn observe_exact_ids( + &mut self, + _: &str, + point_ids: &[String], + ) -> Result>> { + point_ids + .iter() + .map(|id| { + self.points + .iter() + .find(|point| &point.id == id) + .cloned() + .ok_or_else(|| anyhow!("requested exact point is absent")) + }) + .collect() + } fn observe_all( &mut self, _: &str, @@ -2208,6 +2572,124 @@ mod tests { } } + #[test] + fn sealed_batch_ranges_preserve_plan_order_and_never_exceed_qdrant_bound() { + let ranges = sealed_batch_ranges(QDRANT_POINT_BATCH_MAX * 2 + 3); + assert_eq!( + ranges, + vec![ + 0..QDRANT_POINT_BATCH_MAX, + QDRANT_POINT_BATCH_MAX..QDRANT_POINT_BATCH_MAX * 2, + QDRANT_POINT_BATCH_MAX * 2..QDRANT_POINT_BATCH_MAX * 2 + 3, + ] + ); + assert!( + ranges + .iter() + .all(|range| range.len() <= QDRANT_POINT_BATCH_MAX) + ); + assert_eq!( + ranges.into_iter().flatten().collect::>(), + (0..QDRANT_POINT_BATCH_MAX * 2 + 3).collect::>() + ); + } + + #[test] + fn exact_batch_readback_rejects_vector_substitution() -> Result<()> { + let (repo, _state, runtime, basis) = coverage_fixture()?; + let (_, acquisition) = acquire_test(&runtime, &basis, "provider", "model", 3)?; + let planned = acquisition + .plan + .planned_points + .first() + .ok_or_else(|| anyhow!("fixture produced no planned points"))?; + let payload = payload_for(&acquisition.obligation, &acquisition.plan, planned); + let expected_vector = vec![0.25_f32; 3]; + let point_bindings = vec![WorkspaceCoveragePointBinding { + point_id: planned.point_id.clone(), + payload_sha256: digest(&payload)?, + }]; + let vector_bindings = vec![WorkspaceCoverageVectorBinding { + point_id: planned.point_id.clone(), + vector_sha256: digest(&expected_vector)?, + }]; + let observed = vec![SemanticStoredPoint { + id: planned.point_id.clone(), + payload: Some(payload), + vector: Some(vec![0.5_f32; 3]), + }]; + assert!( + authenticate_batch_readback(&observed, &point_bindings, &vector_bindings, 3).is_err() + ); + drop(repo); + Ok(()) + } + + #[test] + fn final_checkpoint_proof_rejects_mutated_or_missing_earlier_point() -> Result<()> { + let (repo, _state, runtime, basis) = coverage_fixture()?; + let (_, acquisition) = acquire_test(&runtime, &basis, "provider", "model", 3)?; + let planned = acquisition + .plan + .planned_points + .first() + .ok_or_else(|| anyhow!("fixture produced no planned points"))?; + let payload = payload_for(&acquisition.obligation, &acquisition.plan, planned); + let vector = vec![0.25_f32; 3]; + let point_bindings = vec![WorkspaceCoveragePointBinding { + point_id: planned.point_id.clone(), + payload_sha256: digest(&payload)?, + }]; + let vector_bindings = vec![WorkspaceCoverageVectorBinding { + point_id: planned.point_id.clone(), + vector_sha256: digest(&vector)?, + }]; + let observed = vec![SemanticStoredPoint { + id: planned.point_id.clone(), + payload: Some(payload), + vector: Some(vector), + }]; + authenticate_final_observation_against_checkpoint_bindings( + observed.clone(), + &point_bindings, + &vector_bindings, + )?; + let mut mutated = observed.clone(); + mutated[0].vector.as_mut().unwrap()[0] += 1.0; + assert!(authenticate_final_observation_against_checkpoint_bindings( + mutated, + &point_bindings, + &vector_bindings, + ) + .is_err()); + assert!(authenticate_final_observation_against_checkpoint_bindings( + Vec::new(), + &point_bindings, + &vector_bindings, + ) + .is_err()); + let mut duplicated = observed.clone(); + duplicated.push(observed[0].clone()); + assert!(authenticate_final_observation_against_checkpoint_bindings( + duplicated, + &point_bindings, + &vector_bindings, + ) + .is_err()); + let mut substituted_duplicate = observed[0].clone(); + substituted_duplicate.vector.as_mut().unwrap()[0] += 1.0; + let mut duplicated = observed.clone(); + duplicated.push(substituted_duplicate); + assert!(authenticate_final_observation_against_checkpoint_bindings( + duplicated, + &point_bindings, + &vector_bindings, + ) + .is_err()); + drop(repo); + Ok(()) + } + struct AdvancingEvidencePort { inner: FakeProjectionPort, body_store: PathBuf, @@ -3065,82 +3547,4 @@ mod tests { ); Ok(()) } - - #[test] - #[ignore = "requires live local Qdrant and Ollama"] - fn live_qdrant_ollama_projection_proves_vectors_and_currentness() -> Result<()> { - let (_repo, _state, runtime, basis) = coverage_fixture()?; - let ollama_url = std::env::var("EPIPHANY_OLLAMA_BASE_URL") - .unwrap_or_else(|_| "http://127.0.0.1:11434".into()); - let qdrant_url = - std::env::var("EPIPHANY_QDRANT_URL").unwrap_or_else(|_| "http://127.0.0.1:6333".into()); - let model = std::env::var("EPIPHANY_OLLAMA_MODEL") - .unwrap_or_else(|_| "qwen3-embedding:0.6b".into()); - let embedder = OllamaEmbedder::new(crate::semantic_backend::OllamaConfig { - base_url: ollama_url, - model, - timeout_ms: 30_000, - query_instruction: String::new(), - })?; - let model_identity = embedder.model_artifact()?.canonical_identity(); - let dimensions = embedder.embedding_dimensions()?; - let body = RepositoryBodyReadSession::open(&runtime, &basis)?; - let prepared = prepare_workspace_coverage_projection( - &body, - "gamecult-ollama-embedding", - &model_identity, - dimensions, - )?; - let acquisition = match acquire_workspace_coverage_projection( - &prepared, - "epiphany-workspace-coverage-projector", - &uuid::Uuid::new_v4().to_string(), - &uuid::Uuid::new_v4().to_string(), - )? { - WorkspaceCoverageAcquireResult::Acquired(value) => value, - _ => bail!("isolated live smoke did not acquire its projection"), - }; - let compatibility = CollectionCompatibility { - managed_by: "epiphany-workspace-coverage-projector".into(), - corpus_kind: "repository_body_workspace_coverage".into(), - schema_version: 0, - projection_version: acquisition.plan.plan_id.clone(), - embedding_provider_id: acquisition.plan.embedding_provider_id.clone(), - embedding_model: acquisition.plan.embedding_model.clone(), - vector_size: acquisition.plan.vector_dimensions as usize, - }; - let mut qdrant = QdrantBackend::new(crate::semantic_backend::QdrantConfig { - url: qdrant_url, - api_key: None, - timeout_ms: 30_000, - })?; - let result = - execute_workspace_coverage_projection(&acquisition, &prepared, &embedder, &mut qdrant); - let collection = workspace_coverage_execution_collection( - &acquisition.plan.plan_id, - &acquisition.claim.claim_id, - acquisition.claim.claim_epoch, - )?; - let receipt = match result { - Ok(receipt) => receipt, - Err(error) => { - let _ = qdrant.retire_exact_collection(&collection, &compatibility); - return Err(error); - } - }; - assert_eq!(receipt.collection_name, collection); - assert!(matches!( - classify_current_workspace_coverage( - &runtime, - &basis, - "gamecult-ollama-embedding", - &model_identity, - dimensions, - )?, - WorkspaceCoverageCurrentState::Current(_) - )); - qdrant.retire_exact_collection(&collection, &compatibility)?; - assert!(!qdrant.collection_exists(&collection)?); - Ok(()) - } } diff --git a/epiphany-core/src/workspace_coverage_projector_service.rs b/epiphany-core/src/workspace_coverage_projector_service.rs index 70b1c8f9..4357ce0b 100644 --- a/epiphany-core/src/workspace_coverage_projector_service.rs +++ b/epiphany-core/src/workspace_coverage_projector_service.rs @@ -2,15 +2,16 @@ use crate::repository_body_observer::RepositoryBodyReadSession; use crate::semantic_backend::{OllamaConfig, OllamaEmbedder, QdrantBackend, QdrantConfig}; use crate::workspace_coverage_projector::{ WorkspaceCoverageAcquireResult, WorkspaceCoverageCurrentState, - acquire_workspace_coverage_projection, classify_current_workspace_coverage, - execute_workspace_coverage_projection, prepare_workspace_coverage_projection, - retire_workspace_coverage_collections, + WorkspaceCoverageProjectionExecutionAuthority, acquire_workspace_coverage_projection, + classify_current_workspace_coverage, execute_workspace_coverage_projection, + prepare_workspace_coverage_projection, retire_workspace_coverage_collections, }; use crate::{ - EPIPHANY_WORKSPACE_COVERAGE_PROJECTOR_DAEMON_ID, load_current_runtime_repository_body_basis, - runtime_repository_body_store_binding, + EPIPHANY_WORKSPACE_COVERAGE_PROJECTOR_DAEMON_ID, HostIncarnationIdentityEntry, + load_current_runtime_repository_body_basis, runtime_repository_body_store_binding, }; use anyhow::{Context, Result, anyhow, bail}; +use ed25519_dalek::SigningKey; use sha2::{Digest, Sha256}; #[cfg(windows)] use std::ffi::OsStr; @@ -172,6 +173,16 @@ pub struct WorkspaceCoverageProjectorServiceBody { managed_process_launch_id: String, embedder: OllamaEmbedder, qdrant: QdrantBackend, + execution_authority: Option, + operation_timeout_ms: u64, +} + +#[derive(Clone)] +struct WorkspaceCoverageProjectorExecutionAuthority { + local_verse_store: PathBuf, + runtime_id: String, + trusted_host: HostIncarnationIdentityEntry, + provider_signing_key: SigningKey, } impl WorkspaceCoverageProjectorServiceBody { @@ -212,6 +223,7 @@ impl WorkspaceCoverageProjectorServiceBody { let body_store = PathBuf::from(route.body_store_path); let singleton = acquire_workspace_coverage_projector_singleton(&runtime_store, &body_store)?; + let operation_timeout_ms = config.qdrant_timeout_ms; let qdrant = QdrantBackend::new(QdrantConfig { url: config.qdrant_url, api_key: config.qdrant_api_key, @@ -230,9 +242,36 @@ impl WorkspaceCoverageProjectorServiceBody { managed_process_launch_id, embedder, qdrant, + execution_authority: None, + operation_timeout_ms, }) } + pub fn install_execution_authority( + &mut self, + local_verse_store: impl AsRef, + runtime_id: impl Into, + trusted_host: HostIncarnationIdentityEntry, + provider_signing_key: SigningKey, + ) -> Result<()> { + let runtime_id = runtime_id.into(); + if runtime_id.trim().is_empty() { + bail!("workspace coverage execution authority requires its runtime id"); + } + let bound = runtime_repository_body_store_binding(&self.runtime_store)? + .ok_or_else(|| anyhow!("runtime has no repository Body-store binding"))?; + if bound.runtime_id != runtime_id { + bail!("workspace coverage execution authority runtime disagrees with Body route"); + } + self.execution_authority = Some(WorkspaceCoverageProjectorExecutionAuthority { + local_verse_store: local_verse_store.as_ref().to_path_buf(), + runtime_id, + trusted_host, + provider_signing_key, + }); + Ok(()) + } + pub fn provider_incarnation(&self) -> &str { &self.executor_incarnation } @@ -329,11 +368,21 @@ impl WorkspaceCoverageProjectorServiceBody { }) } WorkspaceCoverageAcquireResult::Acquired(acquisition) => { + let authority = self.execution_authority.as_ref().ok_or_else(|| { + anyhow!("workspace coverage projector has no installed execution authority") + })?; let receipt = execute_workspace_coverage_projection( &acquisition, &prepared, &self.embedder, &mut self.qdrant, + &WorkspaceCoverageProjectionExecutionAuthority { + local_verse_store: &authority.local_verse_store, + runtime_id: &authority.runtime_id, + trusted_host: &authority.trusted_host, + provider_signing_key: &authority.provider_signing_key, + operation_timeout_ms: self.operation_timeout_ms, + }, )?; Ok(WorkspaceCoverageProjectorServicePulse { status: WorkspaceCoverageProjectorPulseStatus::Executed,