From b5b189498c212418c28672508e798a97526b446d Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 26 Sep 2026 14:12:07 +0000 Subject: [PATCH] fix(feedback): answer the first advisory cycle after a reopen --- .../src/invocation.rs | 14 ++--- .../src/invocation/dispatch.rs | 12 +++++ .../src/invocation/feedback.rs | 50 ++++++++++++++++++ crates/tracedecay-daemon-service/src/lib.rs | 42 +++++++-------- .../src/project_runtime.rs | 38 ++++++++++++++ .../advisory_cycle_language_journey_test.rs | 51 +++++++++++++++++++ .../project_open_owners/advisory_runtime.rs | 23 +++++++++ 7 files changed, 203 insertions(+), 27 deletions(-) diff --git a/crates/tracedecay-daemon-service/src/invocation.rs b/crates/tracedecay-daemon-service/src/invocation.rs index b1c8399a70..600cfb2ef2 100644 --- a/crates/tracedecay-daemon-service/src/invocation.rs +++ b/crates/tracedecay-daemon-service/src/invocation.rs @@ -74,9 +74,10 @@ use tracedecay_policy::{ use tracedecay_tool_catalog::{CapabilityId, EffectClass, SortContractId, UseCaseId}; use crate::project_runtime::{ - FeedbackCyclePublicationError, ProjectRuntimeAlreadyRegistered, ProjectRuntimeRegistryError, - ProjectRuntimeRegistryV1, RegisteredObservabilityProducerV1, StoreObservabilityMountErrorV1, - StoreObservabilityMountV1, StoreObservabilityRegistryV1, + FeedbackCyclePublicationError, ProjectRuntimeAlreadyRegistered, + ProjectRuntimePublicationStateV1, ProjectRuntimeRegistryError, ProjectRuntimeRegistryV1, + RegisteredObservabilityProducerV1, StoreObservabilityMountErrorV1, StoreObservabilityMountV1, + StoreObservabilityRegistryV1, }; use tracedecay_agent_hosts::agents::context_scout::address_registry::{ AdmittedContextScoutHookV1, ContextScoutLifecycleAddressV1, @@ -230,9 +231,10 @@ pub use work_routing::DaemonWorkProposalRoutingAuthorityV1; pub use feedback::{ DaemonAdvisoryCycleInvocationFuture, DaemonAdvisoryCycleInvocationOwner, DaemonAdvisoryCycleInvocationPort, DaemonAdvisoryCycleInvocationRequest, - DaemonFeedbackInvocationOwner, DaemonFeedbackProximityInvocationFuture, - DaemonFeedbackProximityInvocationRequest, advisory_cycle_invocation_result, - daemon_operation_event_authority, feedback_proximity_invocation_result, + DaemonAdvisoryCycleMountFuture, DaemonAdvisoryCycleMountV1, DaemonFeedbackInvocationOwner, + DaemonFeedbackProximityInvocationFuture, DaemonFeedbackProximityInvocationRequest, + advisory_cycle_invocation_result, daemon_operation_event_authority, + feedback_proximity_invocation_result, }; pub use primitive::{ DaemonContextScoutRuntimeRegistrar, DaemonContextScoutRuntimeRegistrationError, diff --git a/crates/tracedecay-daemon-service/src/invocation/dispatch.rs b/crates/tracedecay-daemon-service/src/invocation/dispatch.rs index 961c22123c..8f18490667 100644 --- a/crates/tracedecay-daemon-service/src/invocation/dispatch.rs +++ b/crates/tracedecay-daemon-service/src/invocation/dispatch.rs @@ -512,6 +512,18 @@ impl DaemonInvocationService { deadline, cancellation, } => { + let advisory_cycle = match advisory_cycle { + Some(owner) if owner.service.mount().await == DaemonAdvisoryCycleMountV1::Answers => { + Some(owner) + } + _ => { + self.answering_advisory_cycle_owner( + registered_project_root.as_deref(), + &deadline, + ) + .await + } + }; execute_feedback_advisory_cycle( request_id, advisory_cycle, diff --git a/crates/tracedecay-daemon-service/src/invocation/feedback.rs b/crates/tracedecay-daemon-service/src/invocation/feedback.rs index 90c07d3729..9fbb9ad3c5 100644 --- a/crates/tracedecay-daemon-service/src/invocation/feedback.rs +++ b/crates/tracedecay-daemon-service/src/invocation/feedback.rs @@ -108,12 +108,28 @@ pub type DaemonFeedbackProximityInvocationFuture<'a> = Pin< >, >; +/// Whether an advisory-cycle owner answers for its project now, or stands in +/// for a full cycle a ready sealed generation is mounting. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum DaemonAdvisoryCycleMountV1 { + Answers, + Mounting, +} + +pub type DaemonAdvisoryCycleMountFuture<'a> = + Pin + Send + 'a>>; + pub trait DaemonAdvisoryCycleInvocationPort: Send + Sync { fn invoke( &self, request: DaemonAdvisoryCycleInvocationRequest, ) -> DaemonAdvisoryCycleInvocationFuture<'_>; + /// A mounted cycle answers; only a pre-mount owner reports `Mounting`. + fn mount(&self) -> DaemonAdvisoryCycleMountFuture<'_> { + Box::pin(async { DaemonAdvisoryCycleMountV1::Answers }) + } + fn invoke_proximity( &self, request: DaemonFeedbackProximityInvocationRequest, @@ -941,6 +957,40 @@ impl DaemonInvocationService { .await } + /// The owner that answers an advisory-cycle request for `project_root`. + /// + /// A reopened project serves its first requests while project open is + /// still publishing owners, and then while a ready sealed generation + /// mounts the full cycle behind a placeholder. Those requests wait for the + /// publication within their own deadline instead of failing a call a + /// retry would answer. A finished publication without an owner, or an + /// owner that answers, returns at once. + #[hotpath::measure(label = "daemon.service.feedback.advisory_owner_wait", future = true)] + pub(super) async fn answering_advisory_cycle_owner( + &self, + project_root: Option<&Path>, + deadline: &Deadline, + ) -> Option { + let project_root = project_root?; + loop { + let (owner, publication, mut changed) = + self.project_runtimes.advisory_cycle_view(project_root); + let waits = match &owner { + Some(owner) => owner.service.mount().await == DaemonAdvisoryCycleMountV1::Mounting, + None => publication == Some(ProjectRuntimePublicationStateV1::Warming), + }; + let remaining_micros = deadline.expires_at.0.saturating_sub(now_micros().0); + if !waits || remaining_micros <= 0 { + return owner; + } + let remaining = Duration::from_micros(remaining_micros.unsigned_abs()); + match tokio::time::timeout(remaining, changed.changed()).await { + Ok(Ok(())) => {} + Ok(Err(_)) | Err(_) => return owner, + } + } + } + #[hotpath::skip] pub(super) async fn feedback_cycle_input( &self, diff --git a/crates/tracedecay-daemon-service/src/lib.rs b/crates/tracedecay-daemon-service/src/lib.rs index e95c2d493a..585addab7d 100644 --- a/crates/tracedecay-daemon-service/src/lib.rs +++ b/crates/tracedecay-daemon-service/src/lib.rs @@ -91,27 +91,27 @@ pub use invocation::{ BoundedHookOrchestratorV1, ConfigurationRuntimeRefreshFuture, ConfigurationRuntimeRefreshPort, DaemonAdvisoryCycleInvocationFuture, DaemonAdvisoryCycleInvocationOwner, DaemonAdvisoryCycleInvocationPort, DaemonAdvisoryCycleInvocationRequest, - DaemonAdvisoryRuntimeRegistrar, DaemonAdvisoryRuntimeRegistrationError, - DaemonConfigurationGrantAuthority, DaemonConfigurationRuntimeRegistrar, - DaemonContextScoutRuntimeRegistrar, DaemonContextScoutRuntimeRegistrationError, - DaemonFeedbackInvocationOwner, DaemonFeedbackProximityInvocationFuture, - DaemonFeedbackProximityInvocationRequest, DaemonFeedbackRuntimeRegistrar, - DaemonFeedbackRuntimeRegistrationError, DaemonGraphToolOwnerRegistrationError, - DaemonInvocationService, DaemonLspInvocationOwner, DaemonLspOwnerRegistrar, - DaemonPrimitiveRuntimeRegistrar, DaemonPrimitiveRuntimeRegistrationError, - DaemonRetainedRuntimeRegistrar, DaemonSourceEditOwnerRegistrationError, - DaemonWorkProposalRoutingAuthorityV1, DaemonWorkRuntimeRegistrar, - FeedbackCycleRuntimeBuilderV1, GraphToolFuture, GraphToolInvocationV1, - HookOrchestrationAdmissionV1, HookOrchestrationRequestV1, HookOrchestrationTriggerV1, - HookOrchestrationWorkOutcomeV1, LSP_WORKSPACE_CAPABILITY_ID_V1, LSP_WORKSPACE_USE_CASE_ID_V1, - LspDeliverySettlementAdmissionV1, MAX_COALESCED_HOOK_COMPLETIONS, ProjectGraphToolPortV1, - RegisteredCallableCodeRuntime, RegisteredConfigurationRuntime, RegisteredFeedbackRuntime, - RegisteredGraphToolOwnerV1, RegisteredRetainedRequestContextError, RegisteredRetainedRuntime, - RegisteredWorkRuntime, SwitchableFeedbackCycleRuntimeV1, UnavailableFeedbackCycleRuntimeV1, - admit_registered_hook_orchestration, advisory_cycle_invocation_result, - callable_code_request_context, daemon_operation_event_authority, - feedback_proximity_invocation_result, register_hook_orchestration_runtime, - unregister_hook_orchestration_runtime, + DaemonAdvisoryCycleMountFuture, DaemonAdvisoryCycleMountV1, DaemonAdvisoryRuntimeRegistrar, + DaemonAdvisoryRuntimeRegistrationError, DaemonConfigurationGrantAuthority, + DaemonConfigurationRuntimeRegistrar, DaemonContextScoutRuntimeRegistrar, + DaemonContextScoutRuntimeRegistrationError, DaemonFeedbackInvocationOwner, + DaemonFeedbackProximityInvocationFuture, DaemonFeedbackProximityInvocationRequest, + DaemonFeedbackRuntimeRegistrar, DaemonFeedbackRuntimeRegistrationError, + DaemonGraphToolOwnerRegistrationError, DaemonInvocationService, DaemonLspInvocationOwner, + DaemonLspOwnerRegistrar, DaemonPrimitiveRuntimeRegistrar, + DaemonPrimitiveRuntimeRegistrationError, DaemonRetainedRuntimeRegistrar, + DaemonSourceEditOwnerRegistrationError, DaemonWorkProposalRoutingAuthorityV1, + DaemonWorkRuntimeRegistrar, FeedbackCycleRuntimeBuilderV1, GraphToolFuture, + GraphToolInvocationV1, HookOrchestrationAdmissionV1, HookOrchestrationRequestV1, + HookOrchestrationTriggerV1, HookOrchestrationWorkOutcomeV1, LSP_WORKSPACE_CAPABILITY_ID_V1, + LSP_WORKSPACE_USE_CASE_ID_V1, LspDeliverySettlementAdmissionV1, MAX_COALESCED_HOOK_COMPLETIONS, + ProjectGraphToolPortV1, RegisteredCallableCodeRuntime, RegisteredConfigurationRuntime, + RegisteredFeedbackRuntime, RegisteredGraphToolOwnerV1, RegisteredRetainedRequestContextError, + RegisteredRetainedRuntime, RegisteredWorkRuntime, SwitchableFeedbackCycleRuntimeV1, + UnavailableFeedbackCycleRuntimeV1, admit_registered_hook_orchestration, + advisory_cycle_invocation_result, callable_code_request_context, + daemon_operation_event_authority, feedback_proximity_invocation_result, + register_hook_orchestration_runtime, unregister_hook_orchestration_runtime, }; pub use mcp_project_registry::DaemonProjectRegistryReadService; pub use mcp_workflow_index::DaemonWorkflowIndexReadService; diff --git a/crates/tracedecay-daemon-service/src/project_runtime.rs b/crates/tracedecay-daemon-service/src/project_runtime.rs index 61ead05582..751afd4caa 100644 --- a/crates/tracedecay-daemon-service/src/project_runtime.rs +++ b/crates/tracedecay-daemon-service/src/project_runtime.rs @@ -642,6 +642,9 @@ pub struct ProjectRuntimeRegistryV1 { root_fences: Arc>, reservation_changed: watch::Sender, reservation_blocking_changed: Arc<(StdMutex, Condvar)>, + /// Bumped whenever a published owner or a project's publication state + /// changes, so a request waiting on an owner re-reads the runtime. + published_changed: watch::Sender, /// The blocking drain is retained independently of whichever async /// shutdown caller first requested it. A retry can therefore join the /// same work after that caller is cancelled. @@ -658,6 +661,7 @@ pub struct ProjectRuntimeRegistryV1 { impl Default for ProjectRuntimeRegistryV1 { fn default() -> Self { let (reservation_changed, _) = watch::channel(0); + let (published_changed, _) = watch::channel(0); let (shutdown_complete, _) = watch::channel(ShutdownState::Pending); Self { runtimes: Arc::new(hotpath::mutex!( @@ -670,6 +674,7 @@ impl Default for ProjectRuntimeRegistryV1 { )), reservation_changed, reservation_blocking_changed: Arc::new((StdMutex::new(0), Condvar::new())), + published_changed, shutdown_task: Arc::new(AsyncMutex::new(None)), closed: Arc::new(AtomicBool::new(false)), shutdown_started: Arc::new(AtomicBool::new(false)), @@ -1096,6 +1101,8 @@ impl ProjectRuntimeRegistryV1 { let runtime = runtimes.entry(project_root.clone()).or_default(); if !runtime.reservations.contains(&TypeId::of::()) { *C::slot(runtime) = Some(component); + drop(runtimes); + self.signal_published_changed(); return Ok(()); } } @@ -1243,6 +1250,9 @@ impl ProjectRuntimeRegistryV1 { .map_err(|_| FeedbackCyclePublicationError::RouterUnavailable)?; runtime.advisory = Some(advisory); runtime.advisory_cycle = Some(advisory_cycle); + drop(runtimes); + drop(root_fences); + self.signal_published_changed(); return Ok(()); } } @@ -1349,9 +1359,37 @@ impl ProjectRuntimeRegistryV1 { } runtime.publication = state; runtime.publication_attempt = None; + drop(runtimes); + self.signal_published_changed(); true } + fn signal_published_changed(&self) { + self.published_changed + .send_modify(|version| *version = version.wrapping_add(1)); + } + + /// The advisory-cycle owner and publication state held for + /// `project_root`, with a receiver subscribed before the read so no later + /// publication is missed. + pub(crate) fn advisory_cycle_view( + &self, + project_root: &Path, + ) -> ( + Option, + Option, + watch::Receiver, + ) { + let changed = self.published_changed.subscribe(); + let runtimes = self.lock_runtimes(); + let runtime = runtimes.get(project_root); + ( + runtime.and_then(|runtime| runtime.advisory_cycle.clone()), + runtime.map(|runtime| runtime.publication), + changed, + ) + } + /// Record terminal owner failure only for the attempt that is still current. pub fn mark_publication_failed(&self, attempt: &ProjectRuntimePublicationAttemptV1) -> bool { self.finish_publication(attempt, ProjectRuntimePublicationStateV1::Failed) diff --git a/crates/tracedecay/src/daemon/production_harness/advisory_cycle_language_journey_test.rs b/crates/tracedecay/src/daemon/production_harness/advisory_cycle_language_journey_test.rs index 5f6f74018d..06601a05b5 100644 --- a/crates/tracedecay/src/daemon/production_harness/advisory_cycle_language_journey_test.rs +++ b/crates/tracedecay/src/daemon/production_harness/advisory_cycle_language_journey_test.rs @@ -105,6 +105,57 @@ async fn typescript_only_checkout_runs_the_pull_request_advisory_cycle() { harness.shutdown().await; } +/// A reopened project already holds a ready sealed generation, so the first +/// advisory-cycle call after the reopen answers instead of the retryable +/// pre-mount state a second call would get past. +#[tokio::test(flavor = "multi_thread")] +#[hotpath::skip] +async fn first_advisory_cycle_after_a_reopen_answers_without_a_retry() { + let isolation = tempfile::TempDir::new().expect("production harness isolation"); + let project = isolation.path().join("project"); + std::fs::create_dir_all(project.join("src")).expect("project source dir"); + std::fs::write( + project.join("src/lib.rs"), + "pub fn reopened_cycle(value: u32) -> u32 {\n value + 1\n}\n", + ) + .expect("rust source"); + git(&project, &["init", "--quiet", "-b", "feature/reopen"]); + git(&project, &["add", "."]); + git( + &project, + &["commit", "--quiet", "-m", "seed reopened cycle"], + ); + + let harness = ProductionProjectCompositionHarnessV1::open(isolation.path(), [project.clone()]) + .await + .expect("first production composition"); + let (refused, settled) = settled_advisory_cycle(&harness, &project, "src/lib.rs").await; + assert!(!refused, "the first open must mount the cycle: {settled}"); + harness.shutdown().await; + + let harness = ProductionProjectCompositionHarnessV1::open(isolation.path(), [project.clone()]) + .await + .expect("reopened production composition"); + let document_uri = url::Url::from_file_path(project.join("src/lib.rs")) + .expect("document uri") + .to_string(); + let response = harness + .call_tool( + &project, + "tracedecay_feedback_advisory_cycle", + json!({"document_uri": document_uri, "format": "json"}), + ) + .await + .expect("advisory cycle call"); + let (refused, answer) = tool_answer(&response); + assert!( + !refused, + "the first call after a reopen was refused: {answer}" + ); + assert_eq!(answer["outcome"]["outcome"], json!("evidence"), "{answer}"); + harness.shutdown().await; +} + #[tokio::test(flavor = "multi_thread")] #[hotpath::skip] async fn checkout_without_indexable_source_names_why_the_advisory_cycle_cannot_run() { diff --git a/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime.rs b/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime.rs index 88fa9c91ad..2d9a531581 100644 --- a/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime.rs +++ b/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime.rs @@ -100,6 +100,7 @@ use tracedecay_daemon_service::{ BoundedHookOrchestratorV1, ConfigurationRuntimeRefreshFuture, ConfigurationRuntimeRefreshPort, DaemonAdvisoryCycleInvocationFuture, DaemonAdvisoryCycleInvocationOwner, DaemonAdvisoryCycleInvocationPort, DaemonAdvisoryCycleInvocationRequest, + DaemonAdvisoryCycleMountFuture, DaemonAdvisoryCycleMountV1, DaemonFeedbackProximityInvocationFuture, DaemonFeedbackProximityInvocationRequest, HookOrchestrationRequestV1, HookOrchestrationTriggerV1, HookOrchestrationWorkOutcomeV1, advisory_cycle_invocation_result, daemon_operation_event_authority, @@ -2085,6 +2086,28 @@ impl DaemonAdvisoryCycleInvocationPort for ProjectOpenProximityReadOwnerV1 { }) } + /// With a ready sealed generation the deferred mount is already upgrading + /// this owner, so a request waits for that publication instead of taking + /// the retryable warming answer. + // ponytail: a deferred mount that fails terminally leaves this owner in + // place, so such a request waits out its own deadline before the warming + // answer; surfacing the terminal mount failure here would end it early. + fn mount(&self) -> DaemonAdvisoryCycleMountFuture<'_> { + Box::pin(async move { + if code_index_disabled_for_scope(&self.code_index_schedulers, &self.scope) { + return DaemonAdvisoryCycleMountV1::Answers; + } + match self + .code_index_schedulers + .latest_feedback_generation_for_scope(&self.project_root, &self.scope) + .await + { + Some(_) => DaemonAdvisoryCycleMountV1::Mounting, + None => DaemonAdvisoryCycleMountV1::Answers, + } + }) + } + fn invoke_proximity( &self, request: DaemonFeedbackProximityInvocationRequest,