Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -133,8 +133,10 @@ changed would report a failed edit that was applied.
The `index_codebase` tool does not wait for the first index. It registers the
codebase, starts the scan and upload in the background, and returns at once with
the codebase id and the path. The `sync_status` tool then reports the first-index
phase: registering, syncing, embedding, ready, or failed. `semctl daemon status`
reports the same phase for each checkout.
phase: registering, syncing, embedding, ready, or failed. While a sync runs, it
also reports a `current sync:` line with the latest step, such as the number of
uploaded files. `semctl daemon status` reports the same phase and step for each
checkout.

### Coming from the old `semctx` CLI

Expand Down Expand Up @@ -181,7 +183,7 @@ A daemon started by hand skips the job check, and clients attach to it.
Two commands inspect and end a running daemon:

```sh
semctl daemon status # version, pid, uptime, sessions, permits, checkouts and their first-index phase
semctl daemon status # version, pid, uptime, sessions, permits, checkouts, their first-index phase, and the progress of a running sync
semctl daemon status --json # the same facts as one JSON object
semctl daemon stop # end every session and exit
```
Expand Down
7 changes: 4 additions & 3 deletions plugins/semctx/skills/codebase-retrieval/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -50,9 +50,10 @@ after explicit opt-in. That first-ever index runs in the background:
catalog, and graph tools never serve a partial index. They wait at most 5
seconds, then fail with a "still running" error. Call `sync_status` every 10 to
15 seconds to follow the first-index phase (registering, syncing, embedding,
ready, or failed), and use local Read/Grep meanwhile. If the phase is failed,
call `index_codebase` for the same path to retry. Later re-syncs do not block
use of the last complete index.
ready, or failed), and use local Read/Grep meanwhile. While the phase is
syncing, the `current sync:` line shows how many files the upload has sent. If
the phase is failed, call `index_codebase` for the same path to retry. Later
re-syncs do not block use of the last complete index.

## Rules

Expand Down
55 changes: 2 additions & 53 deletions src/commands/index.rs
Original file line number Diff line number Diff line change
Expand Up @@ -134,25 +134,7 @@ pub async fn run(args: IndexArgs, cli: &Cli) -> Result<()> {
}

fn report_sync_progress(progress: &SyncProgress) {
info!("{}", sync_progress_message(progress));
}

fn sync_progress_message(progress: &SyncProgress) -> String {
match progress {
SyncProgress::Preparing => "preparing index".to_string(),
SyncProgress::Scanning { root } => format!("scanning files in {}", root.display()),
SyncProgress::Planning {
files,
cached_files,
} => format!(
"scanned {files} files ({cached_files} filter decisions reused) — checking for changes"
),
SyncProgress::Uploading {
uploaded_files,
total_files,
} => format!("uploading: {uploaded_files}/{total_files} files"),
SyncProgress::Finalizing => "finalizing upload".to_string(),
}
info!("{progress}");
}

fn poll_interval(elapsed: std::time::Duration) -> std::time::Duration {
Expand Down Expand Up @@ -219,11 +201,7 @@ async fn wait_for_job(client: &Client, job_id: &str) -> Result<()> {

#[cfg(test)]
mod tests {
use std::path::PathBuf;

use super::{
JobProgress, SyncProgress, api, poll_interval, sync_progress_message, terminal_result,
};
use super::{JobProgress, api, poll_interval, terminal_result};

#[test]
fn polling_is_responsive_for_fast_jobs_then_backs_off() {
Expand Down Expand Up @@ -282,33 +260,4 @@ mod tests {
assert_eq!(progress.processed, 12);
assert_eq!(progress.total, 15);
}

#[test]
fn upload_progress_reports_completed_and_total_files() {
let message = sync_progress_message(&SyncProgress::Uploading {
uploaded_files: 3,
total_files: 8,
});
assert_eq!(message, "uploading: 3/8 files");
}

#[test]
fn scan_summary_reports_verified_filter_decisions() {
let message = sync_progress_message(&SyncProgress::Planning {
files: 80,
cached_files: 73,
});
assert_eq!(
message,
"scanned 80 files (73 filter decisions reused) — checking for changes"
);
}

#[test]
fn scan_progress_names_the_effective_root() {
let message = sync_progress_message(&SyncProgress::Scanning {
root: PathBuf::from("repo"),
});
assert_eq!(message, "scanning files in repo");
}
}
17 changes: 17 additions & 0 deletions src/daemon/status.rs
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,7 @@ mod tests {
use crate::engine::coordinator::WatcherState;
use crate::engine::scheduler::{PermitUsage, SchedulerUsage};
use crate::mcp::readiness::FirstIndexPhase;
use crate::sync::SyncProgress;

fn status() -> DaemonStatus {
DaemonStatus {
Expand Down Expand Up @@ -138,6 +139,10 @@ mod tests {
first_index: Some(FirstIndexPhase::Embedding {
job_id: Some("job-1".to_string()),
}),
sync_progress: Some(SyncProgress::Uploading {
uploaded_files: 3,
total_files: 10,
}),
}],
}
}
Expand Down Expand Up @@ -174,6 +179,13 @@ mod tests {
job_id: Some("job-1".to_string())
})
);
assert_eq!(
coordinator.sync_progress,
Some(SyncProgress::Uploading {
uploaded_files: 3,
total_files: 10
})
);
}

/// A reader that is not this build must find the documented field names.
Expand All @@ -198,6 +210,10 @@ mod tests {
value["coordinators"][0]["first_index"],
serde_json::json!({"embedding": {"job_id": "job-1"}})
);
assert_eq!(
value["coordinators"][0]["sync_progress"],
serde_json::json!({"uploading": {"uploaded_files": 3, "total_files": 10}})
);
}

#[test]
Expand All @@ -223,6 +239,7 @@ mod tests {
lines[8].contains("first index embedding (job job-1)"),
"{rendered}"
);
assert!(lines[8].contains("sync uploading 3/10 files"), "{rendered}");
}

#[test]
Expand Down
99 changes: 36 additions & 63 deletions src/engine/coordinator.rs
Original file line number Diff line number Diff line change
Expand Up @@ -48,10 +48,16 @@ use super::registry::CheckoutKey;
use super::scheduler;
use super::watch_hub::{WatchBatch, WatchRegistration, can_change_tree};
use crate::client::{self, Client};
use crate::mcp::readiness::{FirstIndexPhase, InitialIndexGate};
use crate::mcp::readiness::InitialIndexGate;
use crate::sync::policy::{SourcePolicy, event_may_affect_policy};
use crate::sync::{self, LastJob, SyncCache, SyncLimits, SyncOutcome, blocking};

mod progress;
mod status;

use progress::ProgressSlot;
pub(crate) use status::CoordinatorStatus;

/// Reasons a reconcile can be due, before any coalescing.
const TRIGGER_CAPACITY: usize = 64;

Expand Down Expand Up @@ -100,6 +106,8 @@ pub(crate) struct ReconcileRun {
pub(crate) root: PathBuf,
pub(crate) cache: Arc<Mutex<SyncCache>>,
pub(crate) limits: SyncLimits,
/// Where the reconcile reports its milestones for status callers.
pub(crate) progress: Arc<ProgressSlot>,
}

/// A reconcile in progress. Boxed and `'static`, so an implementation must own
Expand Down Expand Up @@ -128,9 +136,15 @@ pub(crate) struct SyncReconciler;
impl Reconciler for SyncReconciler {
fn reconcile(&self, run: ReconcileRun) -> Reconciled {
Box::pin(async move {
sync::sync(&run.client, &run.root, &run.cache, &run.limits)
.await
.map_err(|error| format!("{error:#}"))
sync::sync_with_progress(
&run.client,
&run.root,
&run.cache,
&run.limits,
|progress| run.progress.publish(progress),
)
.await
.map_err(|error| format!("{error:#}"))
})
}

Expand Down Expand Up @@ -183,57 +197,6 @@ struct RunState {
last_error: Option<String>,
}

/// One checkout, as `sync_status` and the daemon status see it.
///
/// `semctl daemon status` reports one of these per checkout, so this is the
/// single source of truth for that part of the status line as well.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub(crate) struct CoordinatorStatus {
pub(crate) root: PathBuf,
pub(crate) codebase_id: Option<String>,
pub(crate) leases: usize,
pub(crate) watcher: WatcherState,
pub(crate) last_job_id: Option<String>,
pub(crate) running: bool,
pub(crate) pending_triggers: usize,
pub(crate) trigger_overflow: bool,
pub(crate) last_outcome: Option<String>,
pub(crate) last_error: Option<String>,
/// The phase of this checkout's first index. `None` when the checkout has
/// no first-index gate, and for a daemon that predates the field.
#[serde(default)]
pub(crate) first_index: Option<FirstIndexPhase>,
}

impl std::fmt::Display for CoordinatorStatus {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
formatter,
"root {} codebase {} leases {} watcher {} job {} running {} first index {} pending {}{} outcome {} error {}",
self.root.display(),
self.codebase_id.as_deref().unwrap_or("(unbound)"),
self.leases,
match &self.watcher {
WatcherState::Active => "active",
WatcherState::Unavailable(reason) => reason,
},
self.last_job_id.as_deref().unwrap_or("(none)"),
self.running,
self.first_index
.as_ref()
.map_or_else(|| "(none)".to_string(), ToString::to_string),
self.pending_triggers,
if self.trigger_overflow {
" (overflowed)"
} else {
""
},
self.last_outcome.as_deref().unwrap_or("(none)"),
self.last_error.as_deref().unwrap_or("(none)"),
)
}
}

/// Everything one coordinator needs to exist. The registry fills it; the watch
/// registration is already decided, because registering is blocking work that
/// must not happen under the registry's lock.
Expand Down Expand Up @@ -270,6 +233,8 @@ pub(crate) struct CheckoutCoordinator {
/// unconditional, because what was dropped cannot be judged.
overflow: Arc<AtomicBool>,
run: Mutex<RunState>,
/// The latest milestone of the reconcile that is running.
progress: Arc<ProgressSlot>,
leases: AtomicUsize,
/// When the lease count last reached zero, for the registry's sweeper.
released_at: StdMutex<Option<Instant>>,
Expand Down Expand Up @@ -309,6 +274,7 @@ impl CheckoutCoordinator {
triggers,
overflow: setup.overflow,
run: Mutex::new(RunState::default()),
progress: Arc::new(ProgressSlot::new()),
leases: AtomicUsize::new(0),
released_at: StdMutex::new(None),
task: StdMutex::new(None),
Expand Down Expand Up @@ -415,6 +381,7 @@ impl CheckoutCoordinator {
last_job_id,
running: run.running,
first_index,
sync_progress: self.progress.latest(),
pending_triggers: self.triggers.max_capacity() - self.triggers.capacity(),
trigger_overflow: self.overflow.load(Ordering::Acquire),
last_outcome: run.last_outcome.clone(),
Expand Down Expand Up @@ -558,14 +525,20 @@ impl CheckoutCoordinator {
// process. It is held for the whole reconcile, including its uploads,
// because the scan's retained content is what the bound protects.
let permit = scheduler::permit(scan_permits).await;
let result = reconciler
.reconcile(ReconcileRun {
client: client.clone(),
root: self.root.clone(),
cache: self.cache.clone(),
limits: limits.clone(),
})
.await;
let result = {
// The scope clears the milestone when the reconcile ends, fails,
// or is cancelled by dropping this future.
let _progress = self.progress.scope();
reconciler
.reconcile(ReconcileRun {
client: client.clone(),
root: self.root.clone(),
cache: self.cache.clone(),
limits: limits.clone(),
progress: self.progress.clone(),
})
.await
};
drop(permit);
self.record(trigger, &result).await;
// Any run can report a first index, because a run only starts once the
Expand Down
53 changes: 53 additions & 0 deletions src/engine/coordinator/progress.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
//! The latest milestone of the reconcile that is running.
//!
//! A reconciler publishes each milestone from a synchronous callback, and a
//! status caller reads the latest one. The slot is a `watch` channel because
//! publishing and reading never wait and never poison, so the callback cannot
//! stall an upload and a status call cannot stall a reconcile.

use tokio::sync::watch;

use crate::sync::SyncProgress;

/// One coordinator's progress slot.
pub(crate) struct ProgressSlot {
latest: watch::Sender<Option<SyncProgress>>,
}

impl ProgressSlot {
pub(super) fn new() -> Self {
Self {
latest: watch::Sender::new(None),
}
}

/// Record a milestone of the reconcile that is running. Only the latest
/// one is kept.
pub(crate) fn publish(&self, progress: &SyncProgress) {
self.latest.send_replace(Some(progress.clone()));
}

/// The latest milestone, or `None` when no reconcile is running or none
/// reported one yet.
pub(super) fn latest(&self) -> Option<SyncProgress> {
self.latest.borrow().clone()
}

/// Hold this guard for the length of one reconcile. Its drop clears the
/// slot, so a reconcile that ends, fails, panics, or is cancelled leaves
/// no stale milestone behind.
pub(super) fn scope(&self) -> ProgressScope<'_> {
ProgressScope { slot: self }
}
}

/// Clears its slot when dropped. See [`ProgressSlot::scope`].
pub(super) struct ProgressScope<'slot> {
slot: &'slot ProgressSlot,
}

impl Drop for ProgressScope<'_> {
fn drop(&mut self) {
self.slot.latest.send_replace(None);
}
}
Loading
Loading