GameCult/Epiphany
Archived
Watch
9
0
Fork
You've already forked Epiphany
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Checkpoint live workspace projection batches

Browse source
This commit is contained in:
Meta Vens 2026-07-17 01:14:41 +01:00
commit 7362e54b66
5 changed files with 928 additions and 150 deletions
Download patch file Download diff file Expand all files Collapse all files

4
epiphany-core/src/bin/epiphany-daemon-supervisor.rs
View file

@ -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",

206
epiphany-core/src/bin/epiphany-workspace-coverage-projector.rs
View file

@ -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<Mutex<HeartbeatProjection>>,
) -> 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<WorkspaceCoverageProviderHeartbeatEntry> {
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<WorkspaceCoverageProviderHeartbeatEntry> {
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::<u64>()?),
"--heartbeat-interval-seconds" => {
heartbeat_interval_seconds = Some(value()?.parse::<u64>()?)
}
"--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")?,

189
epiphany-core/src/workspace_coverage_projection_batch_checkpoint.rs
View file

@ -471,6 +471,151 @@ pub(crate) fn load_authenticated_current_checkpoint(
}))
}
pub(crate) fn load_authenticated_checkpoint_chain(
body_store: impl AsRef<Path>,
local_verse_store: impl AsRef<Path>,
trusted_host: &HostIncarnationIdentityEntry,
claim_id: &str,
claim_epoch: u64,
) -> Result<Vec<WorkspaceCoverageProjectionBatchCheckpointAdmission>> {
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,
&current.checkpoint.body_observation_id,
)?;
let manifest_env = find(
&opening,
BODY_MANIFEST_TYPE,
&current.checkpoint.manifest_root_sha256,
)?;
let obligation_env = find(&opening, OBLIGATION_TYPE, &current.checkpoint.obligation_id)?;
let plan_env = find(&opening, PLAN_TYPE, &current.checkpoint.plan_id)?;
let claim_env = find(&opening, CLAIM_TYPE, CLAIM_KEY)?;
let attempt_env = find(&opening, ATTEMPT_TYPE, &current.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(&current.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(),
&current.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);

620
epiphany-core/src/workspace_coverage_projector.rs
View file

@ -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<f32>,
}
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<WorkspaceCoveragePointPayload>],
) -> Result<()>;
fn observe_exact_ids(
&mut self,
collection: &str,
point_ids: &[String],
) -> Result<Vec<SemanticStoredPoint<WorkspaceCoveragePointPayload>>>;
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<WorkspaceCoveragePointPayload>],
) -> 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<Vec<SemanticStoredPoint<WorkspaceCoveragePointPayload>>> {
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<WorkspaceCoverageReceipt> {
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<WorkspaceCoverageReceipt> {
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::<Vec<_>>();
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::<Vec<_>>();
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<WorkspaceCoveragePointPayload>],
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<std::ops::Range<usize>> {
(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<WorkspaceCoverageProjectionInput>,
@ -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<SemanticStoredPoint<WorkspaceCoveragePointPayload>>,
chain: &[WorkspaceCoverageProjectionBatchCheckpointAdmission],
) -> Result<WorkspaceCoverageObservedBinding> {
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<SemanticStoredPoint<WorkspaceCoveragePointPayload>>,
expected_points: &[WorkspaceCoveragePointBinding],
expected_vectors: &[WorkspaceCoverageVectorBinding],
) -> Result<Vec<SemanticStoredPoint<WorkspaceCoveragePointPayload>>> {
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<WorkspaceCoveragePointPayload>],
) -> 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<Vec<SemanticStoredPoint<WorkspaceCoveragePointPayload>>> {
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::<Vec<_>>(),
(0..QDRANT_POINT_BATCH_MAX * 2 + 3).collect::<Vec<_>>()
);
}
#[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(())
}
}

59
epiphany-core/src/workspace_coverage_projector_service.rs
View file

@ -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<WorkspaceCoverageProjectorExecutionAuthority>,
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<Path>,
runtime_id: impl Into<String>,
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,