feat(worker): separate restored candidates, repair, and activation
This commit is contained in:
@@ -2301,9 +2301,28 @@ impl WorkerAggregateRecord {
|
||||
});
|
||||
}
|
||||
if let Some(request) = request {
|
||||
let mut validation_request = request.clone();
|
||||
// Pending admission may supply a missing Workspace connection
|
||||
// without applying it to the saved execution. Validate that
|
||||
// exact owner's intent against identity; do not widen the
|
||||
// Available-record rule or mutate the canonical request.
|
||||
if validation_request.workspace_api.is_none()
|
||||
&& let PersistedWorkerLifecycleOperation::Restore(restore) = &operation
|
||||
&& let Some(api) = restore_guard
|
||||
.pending_preparation(restore.operation_id)
|
||||
.and_then(|preparation| preparation.workspace_api.as_ref())
|
||||
{
|
||||
validation_request.workspace_api =
|
||||
Some(
|
||||
serde_json::from_value(serde_json::to_value(api).map_err(
|
||||
|error| runtime_store_corrupt(path, error.to_string()),
|
||||
)?)
|
||||
.map_err(|error| runtime_store_corrupt(path, error.to_string()))?,
|
||||
);
|
||||
}
|
||||
WorkerExecutionRecord {
|
||||
schema_version: SCHEMA_VERSION,
|
||||
request: request.clone(),
|
||||
request: validation_request,
|
||||
binding: binding.clone(),
|
||||
restore_intent,
|
||||
}
|
||||
|
||||
@@ -16,6 +16,9 @@ pub mod management;
|
||||
pub mod observation;
|
||||
pub mod profile_archive;
|
||||
pub mod resource;
|
||||
#[cfg(test)]
|
||||
#[path = "../../../tests/support/restore_environment.rs"]
|
||||
mod restore_test_environment;
|
||||
#[cfg(feature = "fs-store")]
|
||||
pub mod retention;
|
||||
mod runtime;
|
||||
|
||||
@@ -682,8 +682,10 @@ fn runtime_v5_rejects_broken_or_future_source_without_rewriting_evidence() {
|
||||
#[cfg(unix)]
|
||||
#[test]
|
||||
fn current_runtime_session_migration_still_rejects_live_worker_sockets() {
|
||||
let temp = tempfile::tempdir().unwrap();
|
||||
let root = temp.path().join("runtime");
|
||||
// The same TempDir owns the relative bind/connect reference. No cwd change
|
||||
// or alias is needed, and the real live-socket migration refusal stays tested.
|
||||
let temp = tempfile::tempdir_in(".").unwrap();
|
||||
let root = Path::new(temp.path().file_name().unwrap()).join("runtime");
|
||||
let (worker, _, _) = runtime_fixture(&root, 12);
|
||||
let before = tree_bytes(&root);
|
||||
let runs = root.join("workers").join(worker.to_string()).join("runs");
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,173 @@
|
||||
//! Explicit, operation-owned repair of saved Session material. This is not
|
||||
//! read-only candidate preparation: the canonical Session/attribution is the
|
||||
//! receipt for replay after a lost response, and effects are reported as such.
|
||||
|
||||
use super::*;
|
||||
|
||||
impl ProfileRuntimeWorkerFactory {
|
||||
pub(super) async fn recover_saved_restore_source(
|
||||
&self,
|
||||
request: &WorkerExecutionRestoreRequest,
|
||||
) -> Result<bool, RuntimeRestoreFailure> {
|
||||
let cause = |cause| RuntimeRestoreFailure::started(RuntimeRestoreStage::Recovery, cause);
|
||||
let name = Self::runtime_worker_name_for_ref(&request.worker_ref);
|
||||
let root = self
|
||||
.worker_aggregate_dir(&request.worker_ref)
|
||||
.map_err(|_| cause("restore recovery aggregate unavailable"))?;
|
||||
let metadata_store = WorkerAggregateStore::open_read_only(&root, &name)
|
||||
.map_err(|_| cause("restore recovery metadata open failed"))?;
|
||||
let metadata = metadata_store
|
||||
.read_by_name(&name)
|
||||
.map_err(|_| cause("restore recovery metadata read failed"))?
|
||||
.ok_or_else(|| cause("restore recovery metadata missing"))?;
|
||||
let active = metadata
|
||||
.active
|
||||
.as_ref()
|
||||
.ok_or_else(|| cause("restore recovery Session missing"))?;
|
||||
let pending = active.segment_id.is_none();
|
||||
let incomplete = request.request.subjektiv_attached
|
||||
&& !matches!(
|
||||
metadata.subjektiv_session_attribution,
|
||||
Some(session_store::SubjektivSessionAttributionState::Confirmed { .. })
|
||||
);
|
||||
if !pending && !incomplete {
|
||||
return Ok(false);
|
||||
}
|
||||
// The accepted Runtime operation already owns this Session. Its stable
|
||||
// Session identity and sole committed seed, not a new restore request,
|
||||
// are used to reconcile an interrupted first-head publication.
|
||||
let session_store = WorkerSessionStore::new_with_retention(
|
||||
root.join("session"),
|
||||
self.session_retention()
|
||||
.map_err(|_| cause("restore recovery retention unavailable"))?,
|
||||
)
|
||||
.map_err(|_| cause("restore recovery Session open failed"))?;
|
||||
let (manifest, mut loader) = Self::manifest_for_restore(&metadata, &request.request)
|
||||
.map_err(|_| cause("restore recovery saved manifest invalid"))?;
|
||||
let workspace = RuntimeWorkspaceBackendRef::from_worker_request(
|
||||
&request.request,
|
||||
self.runtime_id.as_deref(),
|
||||
);
|
||||
let request_client = request
|
||||
.request
|
||||
.workspace_api
|
||||
.as_ref()
|
||||
.and_then(|api| self.workspace_request_clients.get(&api.workspace_id));
|
||||
let context = workspace.worker_context(
|
||||
&request.worker_ref,
|
||||
request.workspace_scope.as_ref(),
|
||||
self.worker_mutation_identity.as_ref(),
|
||||
request_client,
|
||||
self.embedded_worker_mutation_dispatcher.as_ref(),
|
||||
Some(self.prompt_projection_cache.clone()),
|
||||
);
|
||||
if pending && context.workspace_id().is_some() {
|
||||
let bundle = request
|
||||
.config_bundle
|
||||
.as_ref()
|
||||
.ok_or_else(|| cause("pending recovery launch material missing"))?;
|
||||
let projection = self
|
||||
.observe_bundle_prompt_projection(
|
||||
bundle,
|
||||
request
|
||||
.request
|
||||
.workspace_api
|
||||
.as_ref()
|
||||
.map(|api| api.workspace_id.as_str()),
|
||||
)
|
||||
.map_err(|_| cause("pending recovery Prompt projection invalid"))?
|
||||
.ok_or_else(|| cause("pending recovery Prompt projection missing"))?;
|
||||
loader = loader.with_effective_catalog(projection.projection.catalog.clone());
|
||||
}
|
||||
self.failed_restore_resources
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.insert(
|
||||
request.worker_ref.clone(),
|
||||
FailedRestoreResources {
|
||||
operation_id: request.operation_id,
|
||||
sessions: Vec::new(),
|
||||
candidate_workdir_cleanup: None,
|
||||
allocation: None,
|
||||
feature_tasks: None,
|
||||
},
|
||||
);
|
||||
let store = CombinedStore::new(session_store, metadata_store);
|
||||
let worker_result = Worker::restore_pending_from_worker_metadata_with_context(
|
||||
&name,
|
||||
manifest,
|
||||
store,
|
||||
loader,
|
||||
context,
|
||||
WorkerFilesystemAuthority::None,
|
||||
)
|
||||
.await;
|
||||
let mut worker = worker_result.map_err(|error| {
|
||||
self.retain_construction_failure(request, &error);
|
||||
cause(restore_worker_cause(&error))
|
||||
})?;
|
||||
let tasks = worker.retain_startup_feature_cleanup();
|
||||
let sessions = worker.workdir_sessions();
|
||||
let allocation = worker.restore_allocation_cleanup_handle();
|
||||
self.failed_restore_resources
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.insert(
|
||||
request.worker_ref.clone(),
|
||||
FailedRestoreResources {
|
||||
operation_id: request.operation_id,
|
||||
sessions: vec![sessions.clone()],
|
||||
candidate_workdir_cleanup: None,
|
||||
allocation: allocation.clone(),
|
||||
feature_tasks: Some(tasks.clone()),
|
||||
},
|
||||
);
|
||||
let result = async {
|
||||
if let Some(binding) = request.request.backend_job.as_ref() {
|
||||
worker
|
||||
.bind_backend_job(binding.clone())
|
||||
.map_err(|_| cause("recovery Backend Job binding failed"))?;
|
||||
}
|
||||
worker
|
||||
.finalize_subjektiv_session_attribution(
|
||||
request.request.subjektiv_attached,
|
||||
SubjektivSessionAttributionLifecycle::RestoredSession,
|
||||
)
|
||||
.map_err(|error| cause(restore_worker_cause(&error)))?;
|
||||
if pending {
|
||||
worker::WorkerController::recover_pending_head(&mut worker, &root)
|
||||
.await
|
||||
.map_err(|_| cause("pending Session head recovery failed"))?;
|
||||
}
|
||||
Ok::<_, RuntimeRestoreFailure>(())
|
||||
}
|
||||
.await;
|
||||
// Never release allocation while the Worker guard is attached. Retained
|
||||
// handles in the same operation remain available on every cleanup error.
|
||||
drop(worker);
|
||||
let cleanup = async {
|
||||
tasks
|
||||
.shutdown()
|
||||
.await
|
||||
.map_err(|_| cause("recovery Feature settlement failed"))?;
|
||||
sessions
|
||||
.close_all()
|
||||
.await
|
||||
.map_err(|_| cause("recovery Workdir close failed"))?;
|
||||
if let Some(allocation) = allocation {
|
||||
allocation
|
||||
.release()
|
||||
.map_err(|_| cause("recovery allocation release failed"))?;
|
||||
}
|
||||
Ok::<_, RuntimeRestoreFailure>(())
|
||||
}
|
||||
.await;
|
||||
result?;
|
||||
cleanup?;
|
||||
self.failed_restore_resources
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.remove(&request.worker_ref);
|
||||
Ok(true)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,898 @@
|
||||
// Included inside worker_backend::tests, alongside the restore failure suite.
|
||||
// These integration tests connect the real Profile factory, canonical stores,
|
||||
// restore owner, and Controller. Only model I/O and the Workspace HTTP transport
|
||||
// are substituted. The loopback Host is not an authentication/policy test.
|
||||
// Run in the ordinary worker-runtime lib suite with the default product features
|
||||
// and the filter `restore_candidate::`. No live service is used. Like the
|
||||
// failure suite, this requires the existing Yoi path settings to resolve into
|
||||
// target/restore-test-environment; the common fixture rejects unsafe fallbacks
|
||||
// before any factory, Git, or allocation-table operation.
|
||||
|
||||
#[cfg(feature = "http-server")]
|
||||
mod restore_candidate {
|
||||
use super::*;
|
||||
|
||||
const FIRST: &str = "candidate-saved-first";
|
||||
const SECOND: &str = "candidate-saved-second";
|
||||
const NOTICE: &str = "candidate-saved-notice";
|
||||
const FRESH: &str = "candidate-fresh-input";
|
||||
const WAIT: Duration = Duration::from_secs(5);
|
||||
|
||||
struct HostProbe {
|
||||
calls: Arc<Mutex<Vec<(String, String)>>>,
|
||||
task: Option<tokio::task::JoinHandle<()>>,
|
||||
}
|
||||
|
||||
impl Drop for HostProbe {
|
||||
fn drop(&mut self) {
|
||||
if let Some(task) = self.task.take() {
|
||||
task.abort();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Aborting on unwind prevents a failed setup assertion from leaving its
|
||||
// hanging model turn alive. Successful setup uses the normal shutdown/join.
|
||||
struct SeedController(Option<RuntimeWorkerController>);
|
||||
|
||||
impl Drop for SeedController {
|
||||
fn drop(&mut self) {
|
||||
if let Some(controller) = self.0.as_ref() {
|
||||
controller.controller_task.abort();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn receive_matching(
|
||||
events: &mut tokio::sync::broadcast::Receiver<Event>,
|
||||
phase: &str,
|
||||
mut matches: impl FnMut(&Event) -> bool,
|
||||
) -> Event {
|
||||
let mut seen = Vec::new();
|
||||
tokio::time::timeout(WAIT, async {
|
||||
loop {
|
||||
let event = events.recv().await.expect("Controller event stream");
|
||||
seen.push(format!("{event:?}"));
|
||||
if matches(&event) {
|
||||
return event;
|
||||
}
|
||||
assert!(
|
||||
!matches!(
|
||||
event,
|
||||
Event::Error { .. }
|
||||
| Event::SubmissionRejected { .. }
|
||||
| Event::NotificationRejected { .. }
|
||||
),
|
||||
"unexpected Controller rejection: {event:?}"
|
||||
);
|
||||
}
|
||||
})
|
||||
.await
|
||||
.unwrap_or_else(|_| {
|
||||
panic!("Controller acknowledgement deadline during {phase}; observed {seen:?}")
|
||||
})
|
||||
}
|
||||
|
||||
async fn submit(handle: &WorkerHandle, id: &str) -> Event {
|
||||
let mut events = handle.subscribe();
|
||||
handle
|
||||
.send(Method::Submit {
|
||||
submission_request_id: id.into(),
|
||||
input: vec![protocol::Segment::Text { content: id.into() }],
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
receive_matching(&mut events, id, |event| {
|
||||
matches!(event, Event::SubmissionAccepted { submission_request_id, .. }
|
||||
if submission_request_id == id)
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
async fn notify(handle: &WorkerHandle) {
|
||||
let mut events = handle.subscribe();
|
||||
handle
|
||||
.send(Method::Notify {
|
||||
notification_request_id: NOTICE.into(),
|
||||
message: NOTICE.into(),
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
receive_matching(&mut events, "notification acceptance", |event| {
|
||||
matches!(event, Event::NotificationAccepted { notification_request_id }
|
||||
if notification_request_id == NOTICE)
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
async fn pending(handle: &WorkerHandle) -> protocol::PendingSubmissionsSnapshot {
|
||||
let mut events = handle.subscribe();
|
||||
handle.send(Method::ListPendingSubmissions).await.unwrap();
|
||||
match receive_matching(&mut events, "pending queue snapshot", |event| {
|
||||
matches!(event, Event::PendingSubmissionsChanged { .. })
|
||||
})
|
||||
.await
|
||||
{
|
||||
Event::PendingSubmissionsChanged { pending } => pending,
|
||||
_ => unreachable!(),
|
||||
}
|
||||
}
|
||||
|
||||
async fn finished(handle: &WorkerHandle, id: &str) {
|
||||
let mut events = handle.subscribe();
|
||||
let is_finished = |snapshot: &protocol::WorkerStateSnapshot| {
|
||||
snapshot.last_finished_submission_request_id.as_deref() == Some(id)
|
||||
&& snapshot.state == protocol::WorkerState::Idle
|
||||
};
|
||||
if !is_finished(&handle.shared_state.snapshot()) {
|
||||
receive_matching(
|
||||
&mut events,
|
||||
id,
|
||||
|event| matches!(event, Event::WorkerState { snapshot } if is_finished(snapshot)),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
fn queue_checkpoint(entries: &[LogEntry]) -> serde_json::Value {
|
||||
entries
|
||||
.iter()
|
||||
.rev()
|
||||
.find_map(|entry| match entry {
|
||||
LogEntry::Extension {
|
||||
domain, payload, ..
|
||||
} if domain == "worker.pending_activations.v1" => Some(payload.clone()),
|
||||
_ => None,
|
||||
})
|
||||
.expect("real Controller must have committed its pending activation checkpoint")
|
||||
}
|
||||
|
||||
struct Fixture {
|
||||
backend: Arc<WorkerRuntimeExecutionBackend>,
|
||||
request: WorkerExecutionRestoreRequest,
|
||||
model: MockClient,
|
||||
session: WorkerSessionStore,
|
||||
session_id: session_store::SessionId,
|
||||
segment_id: session_store::SegmentId,
|
||||
receipts: Vec<serde_json::Value>,
|
||||
host: HostProbe,
|
||||
_root: tempfile::TempDir,
|
||||
}
|
||||
|
||||
impl Fixture {
|
||||
fn new() -> Self {
|
||||
Self::with_transport(WorkerControllerTransport::InProcess)
|
||||
}
|
||||
|
||||
fn with_transport(transport: WorkerControllerTransport) -> Self {
|
||||
let root = restore_failure_tempdir();
|
||||
let model = MockClient::sequential(vec![
|
||||
MockResponse::Complete(vec![
|
||||
LlmEvent::tool_use_start(0, "candidate-task-probe", "TaskCreate"),
|
||||
LlmEvent::tool_input_delta(
|
||||
0,
|
||||
r#"{"subject":"Activated candidate","description":"Execution probe"}"#,
|
||||
),
|
||||
LlmEvent::tool_use_stop(0),
|
||||
LlmEvent::Status(StatusEvent {
|
||||
status: ResponseStatus::Completed,
|
||||
}),
|
||||
]),
|
||||
MockResponse::Complete(simple_text_events()),
|
||||
MockResponse::Complete(simple_text_events()),
|
||||
MockResponse::Complete(simple_text_events()),
|
||||
]);
|
||||
let injected = model.clone();
|
||||
let mut factory = ProfileRuntimeWorkerFactory::new(root.path())
|
||||
.with_runtime_store_dir(root.path())
|
||||
.with_runtime_id("candidate-runtime")
|
||||
.with_spawn_model_client_factory(move || Box::new(injected.clone()));
|
||||
factory.controller_transport = transport;
|
||||
let catalog = worker::PromptCatalog::builtins_only()
|
||||
.unwrap()
|
||||
.projection()
|
||||
.clone();
|
||||
factory
|
||||
.observe_workspace_prompt_projection(
|
||||
worker::WorkspacePromptProjection::new(
|
||||
"workspace-saved-subjektiv",
|
||||
"candidate-fixture",
|
||||
catalog.catalog_digest.clone(),
|
||||
catalog,
|
||||
)
|
||||
.unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
let backend = Arc::new(WorkerRuntimeExecutionBackend::new(factory).unwrap());
|
||||
let fixture_root = root.path().to_path_buf();
|
||||
let (request, session, session_id, segment_id, receipts, host) = backend
|
||||
.run_on_adapter_runtime(async move {
|
||||
let calls = Arc::new(Mutex::new(Vec::new()));
|
||||
let authorization_calls = calls.clone();
|
||||
let other_calls = calls.clone();
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let base_url = format!("http://{}", listener.local_addr().unwrap());
|
||||
let authorization_path =
|
||||
"/api/w/workspace-saved-subjektiv/authorization/check";
|
||||
let app = axum::Router::new()
|
||||
.route(
|
||||
authorization_path,
|
||||
axum::routing::post(move |axum::Json(input): axum::Json<
|
||||
server_api::authorization::OperationAuthorizationRequest,
|
||||
>| {
|
||||
let calls = authorization_calls.clone();
|
||||
async move {
|
||||
calls.lock().unwrap().push((
|
||||
"POST".into(), authorization_path.into(),
|
||||
));
|
||||
let response = if input.operation == "task"
|
||||
&& input.resource == server_api::authorization::PermissionResource::Feature
|
||||
{
|
||||
server_api::authorization::OperationAuthorizationResponse {
|
||||
scope: Some(server_api::authorization::PermissionResource::Feature),
|
||||
}
|
||||
} else {
|
||||
saved_subjektiv_authorization_fixture(input)
|
||||
.expect("only the saved fixture Feature grants are available")
|
||||
};
|
||||
axum::Json(response)
|
||||
}
|
||||
}),
|
||||
)
|
||||
.fallback(move |request: axum::extract::Request| {
|
||||
let calls = other_calls.clone();
|
||||
async move {
|
||||
calls.lock().unwrap().push((
|
||||
request.method().to_string(), request.uri().path().to_owned(),
|
||||
));
|
||||
axum::http::StatusCode::FORBIDDEN
|
||||
}
|
||||
});
|
||||
let host = HostProbe {
|
||||
calls,
|
||||
task: Some(tokio::spawn(async move { axum::serve(listener, app).await.unwrap() })),
|
||||
};
|
||||
let worker_ref = WorkerRef::new(WorkerId::now_v7());
|
||||
let (mut manifest, mut create) =
|
||||
saved_subjektiv_policy(&fixture_root, &worker_ref);
|
||||
// Normal restore deliberately excludes Subject repair. Keep
|
||||
// a real Workspace client so accidental Host effects are visible.
|
||||
manifest.feature.subjektiv = manifest::ResolvedSubjektivFeatureConfig::default();
|
||||
manifest.feature.task.enabled = true;
|
||||
create.subjektiv_attached = false;
|
||||
create.workspace_api.as_mut().unwrap().base_url = base_url;
|
||||
let aggregate = fixture_root.join("workers").join(worker_ref.worker_id.to_string());
|
||||
let session = WorkerSessionStore::new(aggregate.join("session")).unwrap();
|
||||
let metadata = WorkerAggregateStore::new(&aggregate, &manifest.worker.name).unwrap();
|
||||
let mut engine = Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated(
|
||||
MockClient::sequential(vec![MockResponse::Hang(simple_text_events().into_iter().take(2).collect())]),
|
||||
);
|
||||
engine.set_system_prompt("Committed restore candidate fixture prompt");
|
||||
let mut worker = Worker::new(
|
||||
manifest.clone(),
|
||||
engine,
|
||||
CombinedStore::new(session.clone(), metadata.clone()),
|
||||
RuntimeWorkspaceBackendRef::from_worker_request(&create, Some("candidate-runtime"))
|
||||
.worker_context(&worker_ref, Some(&crate::runtime::RuntimeWorkspaceScope::new(
|
||||
"workspace-saved-subjektiv", "server-main",
|
||||
)), None, None, None, None),
|
||||
WorkerFilesystemAuthority::None,
|
||||
Scope::empty(),
|
||||
).await.unwrap();
|
||||
worker.enable_worker_metadata_write_through().unwrap();
|
||||
worker.materialize_durable_session_head().await.unwrap();
|
||||
let (session_id, segment_id) = (worker.session_id(), worker.segment_id());
|
||||
let workspace_client = worker.workspace_client_handle();
|
||||
let started = PreparedWorker::new(
|
||||
worker,
|
||||
WorkerBootstrapLayout::RuntimeManagedRun { run_dir: fixture_root.join("seed") },
|
||||
WorkerControllerTransport::InProcess,
|
||||
).start().await.unwrap();
|
||||
let mut seed = SeedController(Some(RuntimeWorkerController {
|
||||
handle: started.handle,
|
||||
shutdown: Arc::new(tokio::sync::Mutex::new(Some(started.shutdown))),
|
||||
controller_task: started.controller_task,
|
||||
workspace_client,
|
||||
}));
|
||||
let handle = seed.0.as_ref().unwrap().handle.clone();
|
||||
let mut started_events = handle.subscribe();
|
||||
submit(&handle, "candidate-setup-turn").await;
|
||||
// Running and streaming observations can arrive in either
|
||||
// order. Do not discard TextDelta while awaiting WorkerState.
|
||||
let mut running = handle.shared_state.catalog_status() == WorkerStatus::Running;
|
||||
let mut streaming = false;
|
||||
receive_matching(&mut started_events, "seed running and streaming", |event| {
|
||||
if let Event::WorkerState { snapshot } = event {
|
||||
running |= snapshot.catalog_status() == protocol::WorkerStatus::Running;
|
||||
}
|
||||
streaming |= matches!(event, Event::TextDelta { .. });
|
||||
running && streaming
|
||||
}).await;
|
||||
let state = Arc::new(RwLock::new(handle.shared_state.snapshot()));
|
||||
let mut events = handle.subscribe();
|
||||
handle.send(Method::Pause { command: next_internal_command(&state).unwrap() })
|
||||
.await.unwrap();
|
||||
receive_matching(&mut events, "seed pause acknowledgement", |event| {
|
||||
matches!(event, Event::WorkerState { snapshot }
|
||||
if snapshot.catalog_status() == protocol::WorkerStatus::Paused)
|
||||
}).await;
|
||||
let mut receipts = Vec::new();
|
||||
for id in [FIRST, SECOND] {
|
||||
let receipt = submit(&handle, id).await;
|
||||
assert!(matches!(receipt, Event::SubmissionAccepted {
|
||||
disposition: protocol::SubmissionDisposition::Queued, ..
|
||||
}));
|
||||
receipts.push(serde_json::to_value(receipt).unwrap());
|
||||
if id == FIRST { notify(&handle).await; }
|
||||
}
|
||||
assert_eq!(pending(&handle).await.submissions.len(), 2);
|
||||
shutdown_profile_controller(seed.0.take().unwrap()).await;
|
||||
// This orphan pin would be removed by the normal activation
|
||||
// reconciliation. Preparation and discard must leave it alone.
|
||||
let orphan = session.write_uploaded_file(
|
||||
session_id, "orphan.txt", "text/plain", b"retained orphan",
|
||||
session_store::UploadedFileLimits::default(),
|
||||
).unwrap();
|
||||
session.pin_uploaded_file(session_id, &orphan, "uncommitted-acceptance").unwrap();
|
||||
let mut saved = metadata.read_by_name(&ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&worker_ref))
|
||||
.unwrap().unwrap();
|
||||
saved.resolved_manifest_snapshot = Some(manifest::write_persisted_worker_manifest_snapshot(&manifest).unwrap());
|
||||
saved.workspace_root = Some(fixture_root.join("stale-derived-root"));
|
||||
metadata.write(&saved).unwrap();
|
||||
let mut request = direct_restore_request(worker_ref);
|
||||
request.request = create;
|
||||
request.workspace_scope = Some(crate::runtime::RuntimeWorkspaceScope::new(
|
||||
"workspace-saved-subjektiv", "server-main",
|
||||
));
|
||||
Ok((request, session, session_id, segment_id, receipts, host))
|
||||
})
|
||||
.unwrap();
|
||||
let fixture = Self {
|
||||
backend,
|
||||
request,
|
||||
model,
|
||||
session,
|
||||
session_id,
|
||||
segment_id,
|
||||
receipts,
|
||||
host,
|
||||
_root: root,
|
||||
};
|
||||
let checkpoint = queue_checkpoint(&fixture.entries());
|
||||
assert_eq!(checkpoint["pending"].as_array().unwrap().len(), 2);
|
||||
assert_eq!(
|
||||
checkpoint["pending_notifications"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.len(),
|
||||
1
|
||||
);
|
||||
fixture.assert_quiet();
|
||||
fixture
|
||||
}
|
||||
|
||||
fn entries(&self) -> Vec<LogEntry> {
|
||||
self.session
|
||||
.read_all(self.session_id, self.segment_id)
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
fn canonical_files(&self) -> BTreeMap<PathBuf, Vec<u8>> {
|
||||
let aggregate = self
|
||||
.backend
|
||||
.factory
|
||||
.worker_aggregate_dir(&self.request.worker_ref)
|
||||
.unwrap();
|
||||
let mut files = persisted_files(&aggregate);
|
||||
files.retain(|path, _| !path.components().any(|part| part.as_os_str() == "runs"));
|
||||
files
|
||||
}
|
||||
|
||||
fn assert_no_modifying_host_calls(&self) {
|
||||
assert!(
|
||||
self.host
|
||||
.calls
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.all(|(method, path)| {
|
||||
method == "GET"
|
||||
|| method == "HEAD"
|
||||
|| (method == "POST"
|
||||
&& path == "/api/w/workspace-saved-subjektiv/authorization/check")
|
||||
}),
|
||||
"unexpected modifying Host call"
|
||||
);
|
||||
}
|
||||
|
||||
fn assert_quiet(&self) {
|
||||
assert_eq!(self.model.call_count.load(Ordering::SeqCst), 0);
|
||||
assert!(self.model.captured.lock().unwrap().is_empty());
|
||||
self.assert_no_modifying_host_calls();
|
||||
assert!(
|
||||
session_store::collect_state(&self.entries())
|
||||
.annotated_history
|
||||
.iter()
|
||||
.all(|entry| {
|
||||
!matches!(
|
||||
entry.item,
|
||||
LoggedItem::ToolCall { .. } | LoggedItem::ToolResult { .. }
|
||||
)
|
||||
}),
|
||||
"prepare committed tool execution"
|
||||
);
|
||||
}
|
||||
|
||||
fn prepare(&self) {
|
||||
assert!(matches!(
|
||||
self.backend.restore_worker(self.request.clone()),
|
||||
WorkerExecutionSpawnResult::Connected { .. }
|
||||
));
|
||||
// Connected here is the backend's prepared snapshot, not publication.
|
||||
assert!(
|
||||
self.backend
|
||||
.pending_restore(&self.request.worker_ref)
|
||||
.unwrap()
|
||||
.is_some()
|
||||
);
|
||||
assert!(self.backend.workers.lock().unwrap().is_empty());
|
||||
assert!(
|
||||
self.backend
|
||||
.factory
|
||||
.observation_hub
|
||||
.get(&self.request.worker_ref)
|
||||
.is_none()
|
||||
);
|
||||
assert!(
|
||||
!self
|
||||
.backend
|
||||
.factory
|
||||
.worker_restore_run_dir(&self.request.worker_ref, self.request.operation_id,)
|
||||
.unwrap()
|
||||
.exists()
|
||||
);
|
||||
let table = worker::runtime::worker_allocation::LockFileGuard::open(
|
||||
&worker::runtime::worker_allocation::default_allocation_path().unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(
|
||||
table
|
||||
.data()
|
||||
.find(&ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(
|
||||
&self.request.worker_ref,
|
||||
))
|
||||
.is_none()
|
||||
);
|
||||
}
|
||||
|
||||
fn assert_no_allocation(&self) {
|
||||
let table = worker::runtime::worker_allocation::LockFileGuard::open(
|
||||
&worker::runtime::worker_allocation::default_allocation_path().unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(
|
||||
table
|
||||
.data()
|
||||
.find(&ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(
|
||||
&self.request.worker_ref,
|
||||
))
|
||||
.is_none()
|
||||
);
|
||||
}
|
||||
|
||||
fn assert_released(&self) {
|
||||
self.assert_no_allocation();
|
||||
assert!(
|
||||
self.backend
|
||||
.pending_restore(&self.request.worker_ref)
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
assert!(self.backend.workers.lock().unwrap().is_empty());
|
||||
assert!(
|
||||
self.backend
|
||||
.factory
|
||||
.failed_restore_resources
|
||||
.lock()
|
||||
.unwrap()
|
||||
.is_empty()
|
||||
);
|
||||
assert!(
|
||||
self.backend
|
||||
.factory
|
||||
.observation_hub
|
||||
.get(&self.request.worker_ref)
|
||||
.is_none()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Fixture {
|
||||
fn drop(&mut self) {
|
||||
// Keep the storage and loopback transport alive until owned tasks stop.
|
||||
let _ = self.backend.stop_worker(&self.request.worker_ref);
|
||||
if let Some(task) = self.host.task.take() {
|
||||
task.abort();
|
||||
let _ = self.backend.run_on_adapter_runtime(async move {
|
||||
let _ = task.await;
|
||||
Ok(())
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial_test::serial(worker_allocation)]
|
||||
fn prepare_saved_inputs_rejects_new_input_and_discard_preserves_authority() {
|
||||
// Exercise the configured socket transport without ever publishing or
|
||||
// binding it; activation coverage below uses the normal InProcess path.
|
||||
let mut fixture = Fixture::with_transport(WorkerControllerTransport::UnixSocket);
|
||||
let opens = Arc::new(AtomicUsize::new(0));
|
||||
let counted = opens.clone();
|
||||
let raw = deferred_workdir::DeferredWorkdirSession::handle(
|
||||
"candidate-discard-resource",
|
||||
workdir::WorkdirSessionCapabilities::READ_ONLY,
|
||||
move || {
|
||||
counted.fetch_add(1, Ordering::SeqCst);
|
||||
Err("not opened during prepare".into())
|
||||
},
|
||||
);
|
||||
fixture.request.deferred_workdir_sessions.insert(
|
||||
WorkdirAttachmentAlias::new("checkout").unwrap(),
|
||||
raw.clone(),
|
||||
);
|
||||
let before = fixture.canonical_files();
|
||||
fixture.prepare();
|
||||
for input in [
|
||||
WorkerInput::user("must-not-accept"),
|
||||
WorkerInput::notify("must-not-notify"),
|
||||
] {
|
||||
assert!(
|
||||
!fixture
|
||||
.backend
|
||||
.dispatch_input(&fixture.request.worker_ref, input)
|
||||
.is_accepted()
|
||||
);
|
||||
}
|
||||
#[cfg(feature = "ws-server")]
|
||||
assert!(
|
||||
fixture
|
||||
.backend
|
||||
.clone()
|
||||
.attach_worker_protocol(&fixture.request.worker_ref)
|
||||
.is_err()
|
||||
);
|
||||
assert!(!matches!(
|
||||
fixture
|
||||
.backend
|
||||
.worker_session(WorkerSessionObservationRequest {
|
||||
worker_ref: fixture.request.worker_ref.clone(),
|
||||
}),
|
||||
runtime_api::WorkerSessionAvailability::LiveProtocol
|
||||
));
|
||||
fixture.assert_quiet();
|
||||
assert_eq!(fixture.canonical_files(), before);
|
||||
assert!(
|
||||
fixture
|
||||
.backend
|
||||
.factory
|
||||
.failed_restore_resources
|
||||
.lock()
|
||||
.unwrap()[&fixture.request.worker_ref]
|
||||
.candidate_workdir_cleanup
|
||||
.is_some(),
|
||||
"retain the real prepared Worker's close-only resource owner"
|
||||
);
|
||||
assert!(
|
||||
fixture
|
||||
.backend
|
||||
.stop_worker(&fixture.request.worker_ref)
|
||||
.is_accepted()
|
||||
);
|
||||
fixture.assert_released();
|
||||
|
||||
let factory = fixture.backend.factory.clone();
|
||||
let request = fixture.request.clone();
|
||||
fixture
|
||||
.backend
|
||||
.run_on_adapter_runtime(async move {
|
||||
// The adapter resolves and owns its own raw handles. Separately
|
||||
// exercise the factory seam with the exact supplied cleanup handle.
|
||||
let mut candidate = factory
|
||||
.prepare_restore(request)
|
||||
.await
|
||||
.map_err(|error| error.to_string())?;
|
||||
candidate
|
||||
.discard()
|
||||
.await
|
||||
.map_err(|error| error.to_string())?;
|
||||
assert!(matches!(
|
||||
raw.read(workdir::ReadRequest {
|
||||
path: workdir::WorkdirPath::new("probe.txt").unwrap(),
|
||||
offset: 0,
|
||||
limit: 1,
|
||||
max_bytes: 32,
|
||||
})
|
||||
.await,
|
||||
Err(workdir::WorkdirError::SessionClosed)
|
||||
));
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
assert_eq!(opens.load(Ordering::SeqCst), 0);
|
||||
fixture.assert_quiet();
|
||||
assert_eq!(fixture.canonical_files(), before);
|
||||
assert!(
|
||||
fixture
|
||||
.backend
|
||||
.stop_worker(&fixture.request.worker_ref)
|
||||
.is_accepted()
|
||||
);
|
||||
assert_eq!(fixture.canonical_files(), before);
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial_test::serial(worker_allocation)]
|
||||
fn prepare_failure_at_each_pre_activation_checkpoint_preserves_saved_authority() {
|
||||
// Feature and Controller are activation checkpoints, not preparation.
|
||||
for stage in [
|
||||
RuntimeRestoreStage::Metadata,
|
||||
RuntimeRestoreStage::Reconstruction,
|
||||
RuntimeRestoreStage::Binding,
|
||||
] {
|
||||
let fixture = Fixture::new();
|
||||
let before = fixture.canonical_files();
|
||||
*fixture
|
||||
.backend
|
||||
.factory
|
||||
.restore_faults
|
||||
.fail_at
|
||||
.lock()
|
||||
.unwrap() = Some(stage);
|
||||
let result = fixture.backend.restore_worker(fixture.request.clone());
|
||||
assert!(
|
||||
matches!(result, WorkerExecutionSpawnResult::RolledBack(_)),
|
||||
"{stage:?}: {result:?}"
|
||||
);
|
||||
fixture.assert_released();
|
||||
fixture.assert_quiet();
|
||||
assert_eq!(fixture.canonical_files(), before, "{stage:?}");
|
||||
let completed = fixture.backend.completed_restore_failures.lock().unwrap();
|
||||
let (_, failure) = &completed[&fixture.request.worker_ref];
|
||||
assert_eq!(failure.stage, stage);
|
||||
assert_eq!(failure.cleanup, RuntimeRestoreCleanup::Confirmed);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial_test::serial(worker_allocation)]
|
||||
fn activate_saved_inputs_drains_fifo_once_and_replays_durable_receipts() {
|
||||
let fixture = Fixture::new();
|
||||
let before = fixture.canonical_files();
|
||||
fixture.prepare();
|
||||
fixture.assert_quiet();
|
||||
assert_eq!(fixture.canonical_files(), before);
|
||||
fixture
|
||||
.backend
|
||||
.activate_restored_worker(fixture.request.operation_id, &fixture.request.worker_ref)
|
||||
.unwrap();
|
||||
let handle = fixture.backend.workers.lock().unwrap()[&fixture.request.worker_ref]
|
||||
.handle
|
||||
.clone();
|
||||
let model = fixture.model.clone();
|
||||
let receipts = fixture.receipts.clone();
|
||||
fixture
|
||||
.backend
|
||||
.run_on_adapter_runtime(async move {
|
||||
finished(&handle, SECOND).await;
|
||||
assert_eq!(model.call_count.load(Ordering::SeqCst), 3);
|
||||
{
|
||||
let requests = model.captured.lock().unwrap();
|
||||
let first = serde_json::to_string(&requests[0].items).unwrap();
|
||||
let second = serde_json::to_string(&requests[2].items).unwrap();
|
||||
assert!(
|
||||
requests[0]
|
||||
.tools
|
||||
.iter()
|
||||
.any(|tool| tool.name == "TaskCreate")
|
||||
);
|
||||
assert!(first.contains(FIRST));
|
||||
assert!(!first.contains(SECOND), "second Submit overtook the first");
|
||||
assert_eq!(first.matches(NOTICE).count(), 1);
|
||||
assert!(second.find(FIRST).unwrap() < second.find(SECOND).unwrap());
|
||||
assert_eq!(second.matches(NOTICE).count(), 1);
|
||||
}
|
||||
let committed = serde_json::to_value(handle.committed_entries()).unwrap();
|
||||
for (id, mut expected) in [FIRST, SECOND].into_iter().zip(receipts) {
|
||||
// Activation advances the durable receipt from Queued to Started;
|
||||
// replay must preserve its original submission identity.
|
||||
expected["data"]["disposition"] = serde_json::json!("started");
|
||||
assert_eq!(
|
||||
serde_json::to_value(submit(&handle, id).await).unwrap(),
|
||||
expected
|
||||
);
|
||||
}
|
||||
notify(&handle).await;
|
||||
assert!(pending(&handle).await.submissions.is_empty());
|
||||
assert_eq!(model.call_count.load(Ordering::SeqCst), 3);
|
||||
assert_eq!(
|
||||
serde_json::to_value(handle.committed_entries()).unwrap(),
|
||||
committed
|
||||
);
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
let history = session_store::collect_state(&fixture.entries()).annotated_history;
|
||||
assert!(
|
||||
history.iter().any(|entry| matches!(
|
||||
&entry.item,
|
||||
LoggedItem::ToolResult { call_id, is_error: false, .. }
|
||||
if call_id == "candidate-task-probe"
|
||||
)),
|
||||
"activation must execute the real TaskCreate tool, not just decode a call"
|
||||
);
|
||||
for marker in [FIRST, SECOND, NOTICE] {
|
||||
let count = history
|
||||
.iter()
|
||||
.filter(|entry| {
|
||||
match &entry.item {
|
||||
LoggedItem::Message { content, .. } => content.iter().any(|part| {
|
||||
matches!(part, LoggedContentPart::Text { text } if text.contains(marker))
|
||||
}),
|
||||
_ => false,
|
||||
}
|
||||
})
|
||||
.count();
|
||||
assert_eq!(count, 1, "{marker} must commit exactly once");
|
||||
}
|
||||
let mut fresh = WorkerInput::user(FRESH);
|
||||
fresh.submission_request_id = Some(FRESH.into());
|
||||
assert!(
|
||||
fixture
|
||||
.backend
|
||||
.dispatch_input(&fixture.request.worker_ref, fresh)
|
||||
.is_accepted()
|
||||
);
|
||||
let handle = fixture.backend.workers.lock().unwrap()[&fixture.request.worker_ref]
|
||||
.handle
|
||||
.clone();
|
||||
fixture
|
||||
.backend
|
||||
.run_on_adapter_runtime(async move {
|
||||
finished(&handle, FRESH).await;
|
||||
assert!(pending(&handle).await.submissions.is_empty());
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
assert_eq!(fixture.model.call_count.load(Ordering::SeqCst), 4);
|
||||
assert!(
|
||||
!fixture.host.calls.lock().unwrap().is_empty(),
|
||||
"activation must reach the configured Host transport"
|
||||
);
|
||||
fixture.assert_no_modifying_host_calls();
|
||||
assert!(
|
||||
fixture
|
||||
.backend
|
||||
.stop_worker(&fixture.request.worker_ref)
|
||||
.is_accepted()
|
||||
);
|
||||
fixture.assert_released();
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial_test::serial(worker_allocation)]
|
||||
fn explicit_repair_reconciles_lost_attribution_response_without_reseeding() {
|
||||
for materialized in [true, false] {
|
||||
let root = restore_failure_tempdir();
|
||||
let model = MockClient::new(Vec::new());
|
||||
let injected = model.clone();
|
||||
let backend = WorkerRuntimeExecutionBackend::new(
|
||||
ProfileRuntimeWorkerFactory::new(root.path())
|
||||
.with_runtime_store_dir(root.path())
|
||||
.with_runtime_id("repair-runtime")
|
||||
.with_spawn_model_client_factory(move || Box::new(injected.clone())),
|
||||
)
|
||||
.unwrap();
|
||||
let factory = backend.factory.clone();
|
||||
let fixture_root = root.path().to_path_buf();
|
||||
backend.run_on_adapter_runtime(async move {
|
||||
let mut request = saved_subjektiv_restore_fixture(&fixture_root, materialized).await;
|
||||
let name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&request.worker_ref);
|
||||
let aggregate = fixture_root.join("workers").join(request.worker_ref.worker_id.to_string());
|
||||
let metadata = WorkerAggregateStore::new(&aggregate, &name).unwrap();
|
||||
let mut saved = metadata.read_by_name(&name).unwrap().unwrap();
|
||||
let session_id = saved.active.as_ref().unwrap().session_id;
|
||||
saved.subjektiv_session_attribution = Some(session_store::SubjektivSessionAttributionState::OutcomeUnknown { session_id, diagnostic: "saved interrupted attribution".into() });
|
||||
metadata.write(&saved).unwrap();
|
||||
// Explicit repair may acquire artifact locks; compare canonical
|
||||
// contents separately from that lock-file side effect.
|
||||
let session_files = || {
|
||||
let mut files = persisted_files(&aggregate.join("session"));
|
||||
files.remove(Path::new("artifacts/paste/.aggregate.lock"));
|
||||
files
|
||||
};
|
||||
let original = session_files();
|
||||
let mut bundle = test_bundle();
|
||||
bundle.metadata.workspace_id = "workspace-saved-subjektiv".into();
|
||||
bundle.prompt_catalog = Some(worker::PromptCatalog::builtins_only().unwrap().projection().clone());
|
||||
request.config_bundle = Some(bundle.with_computed_digest());
|
||||
|
||||
// Substitute only the Host transport. A committed attribution
|
||||
// with an undecodable response models lost acknowledgement;
|
||||
// its durable Session-keyed receipt survives the retry. This
|
||||
// does not test the real Host's signature or Subject policy.
|
||||
let attribution_path = fixture_root.join("host-attribution.json");
|
||||
let stored = attribution_path.clone();
|
||||
let calls = Arc::new(AtomicUsize::new(0));
|
||||
let record_calls = calls.clone();
|
||||
let startup_calls = Arc::new(AtomicUsize::new(0));
|
||||
let startups = startup_calls.clone();
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
request.request.workspace_api.as_mut().unwrap().base_url = format!("http://{}", listener.local_addr().unwrap());
|
||||
let app = axum::Router::new()
|
||||
.route("/api/w/workspace-saved-subjektiv/authorization/check", axum::routing::post(
|
||||
|axum::Json(input): axum::Json<server_api::authorization::OperationAuthorizationRequest>| async move {
|
||||
axum::Json(saved_subjektiv_authorization_fixture(input).unwrap())
|
||||
}))
|
||||
.route("/api/w/workspace-saved-subjektiv/subjektiv/sessions", axum::routing::post(
|
||||
move |axum::Json(input): axum::Json<server_api::SubjektivRecordSessionRequest>| {
|
||||
let stored = stored.clone();
|
||||
let calls = record_calls.clone();
|
||||
async move {
|
||||
assert_eq!(input.session_id, session_id.to_string());
|
||||
let ordinal = calls.fetch_add(1, Ordering::SeqCst);
|
||||
if !stored.exists() {
|
||||
assert!(input.create_if_missing);
|
||||
fs::write(&stored, serde_json::to_vec(&server_api::SubjektivRecordSessionResponse {
|
||||
session_id: input.session_id, subject_id: "saved-subject".into(),
|
||||
}).unwrap()).unwrap();
|
||||
}
|
||||
(axum::http::StatusCode::OK, [(axum::http::header::CONTENT_TYPE, "application/json")],
|
||||
if ordinal == 0 { "{}".to_string() } else { fs::read_to_string(stored).unwrap() })
|
||||
}
|
||||
}))
|
||||
.route("/api/w/workspace-saved-subjektiv/subjektiv/journal/startup", axum::routing::get(move || {
|
||||
let startups = startups.clone();
|
||||
async move {
|
||||
startups.fetch_add(1, Ordering::SeqCst);
|
||||
axum::Json(serde_json::json!({ "subject_id": "saved-subject", "journals": [], "omitted": false }))
|
||||
}
|
||||
}));
|
||||
let host = HostProbe { calls: Arc::new(Mutex::new(Vec::new())), task: Some(tokio::spawn(async move { axum::serve(listener, app).await.unwrap() })) };
|
||||
// Same real factory and same operation for both repairs. No
|
||||
// candidate/controller is constructed by the test itself.
|
||||
assert!(factory.recover_saved_restore_source(&request).await.is_err());
|
||||
assert!(attribution_path.exists(), "Host committed before response loss");
|
||||
assert_eq!(session_files(), original);
|
||||
let failed = metadata.read_by_name(&name).unwrap().unwrap();
|
||||
assert!(matches!(failed.subjektiv_session_attribution,
|
||||
Some(session_store::SubjektivSessionAttributionState::OutcomeUnknown { session_id: id, .. }) if id == session_id));
|
||||
assert_eq!(startup_calls.load(Ordering::SeqCst), 0);
|
||||
factory.cleanup_failed_restore(&request).await.unwrap();
|
||||
assert!(factory.recover_saved_restore_source(&request).await.unwrap());
|
||||
let repaired = metadata.read_by_name(&name).unwrap().unwrap();
|
||||
assert_eq!(repaired.active.as_ref().unwrap().session_id, session_id);
|
||||
assert!(matches!(repaired.subjektiv_session_attribution,
|
||||
Some(session_store::SubjektivSessionAttributionState::Confirmed { session_id: id, .. }) if id == session_id));
|
||||
let segments = WorkerSessionStore::new(aggregate.join("session")).unwrap()
|
||||
.list_segments(session_id).unwrap();
|
||||
assert_eq!(segments.len(), 1, "retry must retain exactly one seed");
|
||||
if materialized {
|
||||
assert_eq!(session_files(), original);
|
||||
assert_eq!(startup_calls.load(Ordering::SeqCst), 0);
|
||||
} else {
|
||||
assert_eq!(startup_calls.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
let after_repair = persisted_files(&aggregate);
|
||||
let calls_after_repair = calls.load(Ordering::SeqCst);
|
||||
assert!(!factory.recover_saved_restore_source(&request).await.unwrap());
|
||||
assert_eq!(persisted_files(&aggregate), after_repair);
|
||||
assert_eq!(calls.load(Ordering::SeqCst), calls_after_repair);
|
||||
assert!(factory.failed_restore_resources.lock().unwrap().is_empty());
|
||||
drop(host);
|
||||
Ok(())
|
||||
}).unwrap();
|
||||
assert_eq!(model.call_count.load(Ordering::SeqCst), 0);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,8 +1,11 @@
|
||||
// Integration coverage through EmbeddedRuntime -> real Profile factory -> real
|
||||
// Worker/bootstrap, local Git Workdir provider, and allocation table. Model I/O
|
||||
// is not invoked; this is a deterministic ordinary suite, not product E2E.
|
||||
// Worker/bootstrap, local Git Workdir provider, and allocation table. External
|
||||
// model I/O is replaced for saved-input coverage; this is not product E2E.
|
||||
|
||||
use crate::restore_test_environment::require_restore_test_environment;
|
||||
|
||||
fn restore_failure_tempdir() -> tempfile::TempDir {
|
||||
require_restore_test_environment();
|
||||
// Explicitly stay inside this checkout, even when TMPDIR points elsewhere.
|
||||
// Keep fixture components short; no symlink/alias to evade socket limits.
|
||||
let target = Path::new(env!("CARGO_MANIFEST_DIR"))
|
||||
@@ -102,11 +105,16 @@ struct ProfileFailureFixture {
|
||||
backend: Arc<WorkerRuntimeExecutionBackend>,
|
||||
worker: crate::catalog::WorkerDetail,
|
||||
provider_opens: Arc<AtomicUsize>,
|
||||
seed_handle: Option<WorkerHandle>,
|
||||
root: tempfile::TempDir,
|
||||
}
|
||||
|
||||
impl ProfileFailureFixture {
|
||||
fn new(timeout: Duration) -> Self {
|
||||
Self::with_saved_inputs(timeout, None)
|
||||
}
|
||||
|
||||
fn with_saved_inputs(timeout: Duration, model: Option<MockClient>) -> Self {
|
||||
let root = restore_failure_tempdir();
|
||||
let repo = root.path().join("g");
|
||||
fs::create_dir(&repo).unwrap();
|
||||
@@ -118,16 +126,20 @@ impl ProfileFailureFixture {
|
||||
git(&repo, &["commit", "-m", "init"]);
|
||||
let store = root.path().join("s");
|
||||
let provider_opens = Arc::new(AtomicUsize::new(0));
|
||||
let mut factory =
|
||||
ProfileRuntimeWorkerFactory::new(root.path()).with_runtime_store_dir(&store);
|
||||
let queue_inputs = model.is_some();
|
||||
if let Some(model) = model {
|
||||
factory = factory.with_spawn_model_client_factory(move || Box::new(model.clone()));
|
||||
}
|
||||
let backend = Arc::new(
|
||||
WorkerRuntimeExecutionBackend::new(
|
||||
ProfileRuntimeWorkerFactory::new(root.path()).with_runtime_store_dir(&store),
|
||||
)
|
||||
.unwrap()
|
||||
.with_spawn_restore_timeout(timeout)
|
||||
.with_working_directory_materializer(RestoreOpenCountingMaterializer {
|
||||
inner: RuntimeGitMaterializer::new(root.path().join("w")),
|
||||
opens: provider_opens.clone(),
|
||||
}),
|
||||
WorkerRuntimeExecutionBackend::new(factory)
|
||||
.unwrap()
|
||||
.with_spawn_restore_timeout(timeout)
|
||||
.with_working_directory_materializer(RestoreOpenCountingMaterializer {
|
||||
inner: RuntimeGitMaterializer::new(root.path().join("w")),
|
||||
opens: provider_opens.clone(),
|
||||
}),
|
||||
);
|
||||
let runtime = EmbeddedRuntime::with_fs_store_and_execution_backend(
|
||||
crate::fs_store::FsRuntimeStoreOptions {
|
||||
@@ -146,12 +158,41 @@ impl ProfileFailureFixture {
|
||||
working_directory: working_directory_request(&repo),
|
||||
}];
|
||||
let worker = runtime.create_worker(request).unwrap();
|
||||
let seed_handle = queue_inputs.then(|| {
|
||||
backend.workers.lock().unwrap()[&worker.worker_ref]
|
||||
.handle
|
||||
.clone()
|
||||
});
|
||||
if let Some(handle) = seed_handle.clone() {
|
||||
backend.run_on_adapter_runtime(async move {
|
||||
async fn event(events: &mut tokio::sync::broadcast::Receiver<Event>, predicate: impl Fn(&Event) -> bool) {
|
||||
tokio::time::timeout(Duration::from_secs(5), async {
|
||||
loop {
|
||||
let next = events.recv().await.unwrap();
|
||||
assert!(!matches!(next, Event::Error { .. } | Event::SubmissionRejected { .. } | Event::NotificationRejected { .. }), "{next:?}");
|
||||
if predicate(&next) { break; }
|
||||
}
|
||||
}).await.expect("saved-input fixture acknowledgement");
|
||||
}
|
||||
let mut events = handle.subscribe();
|
||||
handle.send(Method::submit_text("revocation-seed", "hold for saved input setup")).await.unwrap();
|
||||
event(&mut events, |next| matches!(next, Event::TextDelta { .. })).await;
|
||||
handle.send(Method::Pause { command: protocol::WorkerCommandEnvelope::new(handle.shared_state.snapshot().last_command_id + 1) }).await.unwrap();
|
||||
event(&mut events, |next| matches!(next, Event::WorkerState { snapshot } if snapshot.catalog_status() == protocol::WorkerStatus::Paused)).await;
|
||||
handle.send(Method::submit_text("revocation-saved-submit", "must stay queued")).await.unwrap();
|
||||
event(&mut events, |next| matches!(next, Event::SubmissionAccepted { submission_request_id, disposition: protocol::SubmissionDisposition::Queued, .. } if submission_request_id == "revocation-saved-submit")).await;
|
||||
handle.send(Method::Notify { notification_request_id: "revocation-saved-notice".into(), message: "must also stay queued".into() }).await.unwrap();
|
||||
event(&mut events, |next| matches!(next, Event::NotificationAccepted { notification_request_id } if notification_request_id == "revocation-saved-notice")).await;
|
||||
Ok(())
|
||||
}).unwrap();
|
||||
}
|
||||
runtime.stop_worker(&worker.worker_ref, None).unwrap();
|
||||
// Count restore provider access, not the initial create/stop setup.
|
||||
provider_opens.store(0, Ordering::SeqCst);
|
||||
Self {
|
||||
root,
|
||||
provider_opens,
|
||||
seed_handle,
|
||||
runtime,
|
||||
backend,
|
||||
worker,
|
||||
@@ -241,6 +282,68 @@ impl ProfileFailureFixture {
|
||||
assert_eq!(self.provider_opens.load(Ordering::SeqCst), 0);
|
||||
}
|
||||
|
||||
fn resume_same_operation(
|
||||
&self,
|
||||
request: runtime_api::WorkerRestoreRequest,
|
||||
operation_id: crate::execution::WorkerLifecycleOperationId,
|
||||
) {
|
||||
let result = self
|
||||
.runtime
|
||||
.restore_worker_operation(&self.worker.worker_ref, request.clone())
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
result.state,
|
||||
server_api::WorkerRestoreState::Accepted,
|
||||
"{result:?}"
|
||||
);
|
||||
assert_eq!(
|
||||
self.backend.workers.lock().unwrap()[&self.worker.worker_ref].restore_operation_id,
|
||||
Some(operation_id)
|
||||
);
|
||||
let calls = self
|
||||
.backend
|
||||
.factory
|
||||
.restore_faults
|
||||
.executions
|
||||
.load(Ordering::SeqCst);
|
||||
let replay = self
|
||||
.runtime
|
||||
.restore_worker_operation(&self.worker.worker_ref, request)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
serde_json::to_value(replay).unwrap(),
|
||||
serde_json::to_value(result).unwrap()
|
||||
);
|
||||
assert_eq!(
|
||||
self.backend
|
||||
.factory
|
||||
.restore_faults
|
||||
.executions
|
||||
.load(Ordering::SeqCst),
|
||||
calls
|
||||
);
|
||||
self.runtime
|
||||
.stop_worker(&self.worker.worker_ref, None)
|
||||
.unwrap();
|
||||
assert!(
|
||||
self.backend
|
||||
.factory
|
||||
.observation_hub
|
||||
.get(&self.worker.worker_ref)
|
||||
.is_none()
|
||||
);
|
||||
assert!(self.backend.pending_restores.lock().unwrap().is_empty());
|
||||
assert!(self.backend.workers.lock().unwrap().is_empty());
|
||||
assert!(
|
||||
self.backend
|
||||
.factory
|
||||
.failed_restore_resources
|
||||
.lock()
|
||||
.unwrap()
|
||||
.is_empty()
|
||||
);
|
||||
}
|
||||
|
||||
fn fresh_restore_and_old_replay(
|
||||
&self,
|
||||
request: runtime_api::WorkerRestoreRequest,
|
||||
@@ -319,6 +422,171 @@ impl ProfileFailureFixture {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial_test::serial(worker_allocation)]
|
||||
fn profile_restore_timeout_attachment_revocation_cannot_reactivate_old_candidate() {
|
||||
let model = MockClient::sequential(vec![MockResponse::Hang(
|
||||
simple_text_events().into_iter().take(2).collect(),
|
||||
)]);
|
||||
let fixture =
|
||||
ProfileFailureFixture::with_saved_inputs(Duration::from_secs(2), Some(model.clone()));
|
||||
let body_calls = model.call_count.load(Ordering::SeqCst);
|
||||
assert_eq!(body_calls, 1);
|
||||
let aggregate = fixture
|
||||
.backend
|
||||
.factory
|
||||
.worker_aggregate_dir(&fixture.worker.worker_ref)
|
||||
.unwrap();
|
||||
let session_before = persisted_files(&aggregate.join("session"));
|
||||
let metadata = WorkerAggregateStore::open_read_only(
|
||||
&aggregate,
|
||||
ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&fixture.worker.worker_ref),
|
||||
)
|
||||
.unwrap();
|
||||
let saved = metadata
|
||||
.read_by_name(&ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(
|
||||
&fixture.worker.worker_ref,
|
||||
))
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
let release = Arc::new(tokio::sync::Notify::new());
|
||||
let (entered_tx, entered) = mpsc::channel();
|
||||
*fixture
|
||||
.backend
|
||||
.factory
|
||||
.restore_faults
|
||||
.factory_gate
|
||||
.lock()
|
||||
.unwrap() = Some(release.clone());
|
||||
*fixture
|
||||
.backend
|
||||
.factory
|
||||
.restore_faults
|
||||
.factory_entered
|
||||
.lock()
|
||||
.unwrap() = Some(entered_tx);
|
||||
let request = fixture
|
||||
.runtime
|
||||
.test_restore_request(&fixture.worker.worker_ref);
|
||||
let first = fixture
|
||||
.runtime
|
||||
.restore_worker_operation(&fixture.worker.worker_ref, request.clone())
|
||||
.unwrap();
|
||||
entered.recv_timeout(Duration::from_secs(5)).unwrap();
|
||||
assert_eq!(
|
||||
first.state,
|
||||
server_api::WorkerRestoreState::ReconciliationRequired
|
||||
);
|
||||
let pending = fixture
|
||||
.backend
|
||||
.pending_restore(&fixture.worker.worker_ref)
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert!(!pending.request.logical_workdir_attachments.is_empty());
|
||||
let updated = fixture
|
||||
.runtime
|
||||
.replace_worker_workdir_attachments(&fixture.worker.worker_ref, Vec::new())
|
||||
.unwrap();
|
||||
assert!(updated.workdir_attachments.is_empty());
|
||||
let authority = || -> serde_json::Value {
|
||||
serde_json::from_slice(
|
||||
&fs::read(
|
||||
fixture
|
||||
.root
|
||||
.path()
|
||||
.join("s/workers")
|
||||
.join(fixture.worker.worker_ref.worker_id.to_string())
|
||||
.join("worker.json"),
|
||||
)
|
||||
.unwrap(),
|
||||
)
|
||||
.unwrap()
|
||||
};
|
||||
assert!(authority().get("logical_workdir_attachments").is_none());
|
||||
release.notify_one();
|
||||
let replay = fixture
|
||||
.runtime
|
||||
.restore_worker_operation(&fixture.worker.worker_ref, request.clone())
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
replay.state,
|
||||
server_api::WorkerRestoreState::ReconciliationRequired,
|
||||
"current attachment revocation must fence the old candidate: {replay:?}"
|
||||
);
|
||||
assert!(
|
||||
replay
|
||||
.message
|
||||
.as_deref()
|
||||
.unwrap_or_default()
|
||||
.contains("current Worker authority differs"),
|
||||
"{replay:?}"
|
||||
);
|
||||
let record = authority();
|
||||
let owner = &record["restore_guard"]["owners"][&request.request_id];
|
||||
assert_eq!(
|
||||
owner["operation_id"],
|
||||
serde_json::to_value(pending.request.operation_id).unwrap()
|
||||
);
|
||||
assert!(owner["receipt"].is_null());
|
||||
assert!(fixture.backend.workers.lock().unwrap().is_empty());
|
||||
assert!(Arc::ptr_eq(
|
||||
&pending,
|
||||
&fixture
|
||||
.backend
|
||||
.pending_restore(&fixture.worker.worker_ref)
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
));
|
||||
assert_eq!(fixture.provider_opens.load(Ordering::SeqCst), 0);
|
||||
assert!(!pending.activation_started.load(Ordering::SeqCst));
|
||||
assert_eq!(model.call_count.load(Ordering::SeqCst), body_calls);
|
||||
assert_eq!(persisted_files(&aggregate.join("session")), session_before);
|
||||
assert_eq!(
|
||||
metadata.read_by_name(&saved.worker_name).unwrap().unwrap(),
|
||||
saved
|
||||
);
|
||||
let old = fixture.seed_handle.clone().unwrap();
|
||||
fixture
|
||||
.backend
|
||||
.run_on_adapter_runtime(async move {
|
||||
assert!(old.protocol_is_closed());
|
||||
let old_command = Method::Resume {
|
||||
command: protocol::WorkerCommandEnvelope::new(
|
||||
old.shared_state.snapshot().last_command_id + 1,
|
||||
),
|
||||
};
|
||||
assert!(
|
||||
old.send(old_command).await.is_err(),
|
||||
"old execution command must not enter the restore candidate"
|
||||
);
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
assert!(authority().get("logical_workdir_attachments").is_none());
|
||||
let input = fixture.runtime.send_input(
|
||||
&fixture.worker.worker_ref,
|
||||
WorkerInput::user("must-not-enter-revoked-candidate"),
|
||||
);
|
||||
assert!(
|
||||
input.is_err(),
|
||||
"pending candidate must not admit fresh input: {input:?}"
|
||||
);
|
||||
// Stop owns cleanup of the original operation, never a newly rebound alias.
|
||||
fixture
|
||||
.runtime
|
||||
.stop_worker(&fixture.worker.worker_ref, None)
|
||||
.unwrap();
|
||||
fixture.assert_closed_deferred_sessions(&pending);
|
||||
assert_eq!(fixture.provider_opens.load(Ordering::SeqCst), 0);
|
||||
assert!(fixture.backend.pending_restores.lock().unwrap().is_empty());
|
||||
let old = fixture
|
||||
.runtime
|
||||
.restore_worker_operation(&fixture.worker.worker_ref, request)
|
||||
.unwrap();
|
||||
assert_ne!(old.state, server_api::WorkerRestoreState::Accepted);
|
||||
assert!(fixture.backend.workers.lock().unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial_test::serial(worker_allocation)]
|
||||
fn profile_restore_multistage_failure_cleans_then_fresh_restore_without_stop() {
|
||||
@@ -327,8 +595,6 @@ fn profile_restore_multistage_failure_cleans_then_fresh_restore_without_stop() {
|
||||
for stage in [
|
||||
RuntimeRestoreStage::Reconstruction,
|
||||
RuntimeRestoreStage::Binding,
|
||||
RuntimeRestoreStage::Feature,
|
||||
RuntimeRestoreStage::Controller,
|
||||
] {
|
||||
let fixture = ProfileFailureFixture::new(Duration::from_secs(5));
|
||||
fixture.fail(stage);
|
||||
@@ -474,7 +740,7 @@ fn profile_restore_cleanup_failure_retains_owner_and_cause_without_opening_defer
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
result.state,
|
||||
server_api::WorkerRestoreState::RolledBack,
|
||||
server_api::WorkerRestoreState::ReconciliationRequired,
|
||||
"{result:?}"
|
||||
);
|
||||
assert!(
|
||||
@@ -488,12 +754,11 @@ fn profile_restore_cleanup_failure_retains_owner_and_cause_without_opening_defer
|
||||
.iter()
|
||||
.all(|sessions| sessions.aliases().is_empty())
|
||||
);
|
||||
fixture.assert_released();
|
||||
fixture.assert_closed_deferred_sessions(&pending);
|
||||
eprintln!(
|
||||
"restore-fault cleanup=Workdir owner-retained=true retry=same-operation outcome=RolledBack"
|
||||
"restore-fault cleanup=Workdir owner-retained=true retry=same-operation outcome=ReconciliationRequired"
|
||||
);
|
||||
fixture.fresh_restore_and_old_replay(request, result);
|
||||
fixture.resume_same_operation(request, pending.request.operation_id);
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -585,6 +850,7 @@ async fn failed_restore_raw_cleanup_is_operation_bound_and_does_not_follow_reuse
|
||||
FailedRestoreResources {
|
||||
operation_id: current.operation_id,
|
||||
sessions: Vec::new(),
|
||||
candidate_workdir_cleanup: None,
|
||||
allocation: None,
|
||||
feature_tasks: None,
|
||||
},
|
||||
@@ -715,14 +981,14 @@ fn profile_restore_cleanup_timeout_keeps_running_join_and_late_failure_reconcile
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
result.state,
|
||||
server_api::WorkerRestoreState::RolledBack,
|
||||
server_api::WorkerRestoreState::Accepted,
|
||||
"{result:?}"
|
||||
);
|
||||
fixture.assert_released();
|
||||
assert_eq!(fixture.backend.workers.lock().unwrap().len(), 1);
|
||||
eprintln!(
|
||||
"restore-fault cleanup=timeout retained-task=true late-completion=joined executions=1"
|
||||
);
|
||||
fixture.fresh_restore_and_old_replay(request, result);
|
||||
fixture.resume_same_operation(request, pending.request.operation_id);
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
@@ -745,7 +1011,7 @@ fn profile_restore_real_socket_start_failure_removes_partial_run_and_allocation(
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
result.state,
|
||||
server_api::WorkerRestoreState::RolledBack,
|
||||
server_api::WorkerRestoreState::ReconciliationRequired,
|
||||
"{result:?}"
|
||||
);
|
||||
let diagnostic = serde_json::to_string(&result).unwrap();
|
||||
@@ -757,8 +1023,12 @@ fn profile_restore_real_socket_start_failure_removes_partial_run_and_allocation(
|
||||
// This case deliberately exercises the real Unix socket path-limit error;
|
||||
// positive restore cases retain the factory's normal InProcess transport.
|
||||
// Short fixture roots must not accidentally turn it into a successful bind.
|
||||
let operation_id =
|
||||
fixture.backend.completed_restore_failures.lock().unwrap()[&fixture.worker.worker_ref].0;
|
||||
let pending = fixture
|
||||
.backend
|
||||
.pending_restore(&fixture.worker.worker_ref)
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
let operation_id = pending.request.operation_id;
|
||||
let socket = fixture
|
||||
.backend
|
||||
.factory
|
||||
@@ -767,7 +1037,25 @@ fn profile_restore_real_socket_start_failure_removes_partial_run_and_allocation(
|
||||
.join("worker.sock");
|
||||
assert!(socket.as_os_str().as_encoded_bytes().len() >= 108);
|
||||
assert!(!diagnostic.contains(&fixture.root.path().to_string_lossy().to_string()));
|
||||
fixture.assert_released();
|
||||
assert!(!socket.exists());
|
||||
assert!(fixture.backend.workers.lock().unwrap().is_empty());
|
||||
assert!(
|
||||
fixture
|
||||
.backend
|
||||
.factory
|
||||
.failed_restore_resources
|
||||
.lock()
|
||||
.unwrap()
|
||||
.is_empty()
|
||||
);
|
||||
assert!(
|
||||
fixture
|
||||
.backend
|
||||
.completed_restore_failures
|
||||
.lock()
|
||||
.unwrap()
|
||||
.is_empty()
|
||||
);
|
||||
fixture
|
||||
.backend
|
||||
.factory
|
||||
@@ -777,7 +1065,7 @@ fn profile_restore_real_socket_start_failure_removes_partial_run_and_allocation(
|
||||
eprintln!(
|
||||
"restore-fault stage=socket-start real-bootstrap=true partial-run=removed allocation=released"
|
||||
);
|
||||
fixture.fresh_restore_and_old_replay(request, result);
|
||||
fixture.resume_same_operation(request, operation_id);
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -1178,6 +1466,8 @@ fn profile_restore_missing_committed_system_prompt_rolls_back_and_replays_classi
|
||||
"#,
|
||||
)
|
||||
.unwrap();
|
||||
let mut manifest = manifest;
|
||||
manifest.worker.name = worker_name.clone();
|
||||
metadata.resolved_manifest_snapshot =
|
||||
Some(manifest::write_persisted_worker_manifest_snapshot(&manifest).unwrap());
|
||||
metadata_store.write(&metadata).unwrap();
|
||||
|
||||
@@ -3676,8 +3676,12 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn bound_ssh_workdir_uses_attachment_scoped_agent_and_cleans_it_up() {
|
||||
crate::restore_test_environment::require_restore_test_environment();
|
||||
let repo = create_clean_repo();
|
||||
let runtime_root = tempfile::tempdir().unwrap();
|
||||
// Both ends use a short reference into this TempDir-owned tree. Keep
|
||||
// real agent/broker sockets and the existing bind/cleanup assertions.
|
||||
let runtime_root = tempfile::tempdir_in(".").unwrap();
|
||||
let runtime_path = Path::new(runtime_root.path().file_name().unwrap());
|
||||
let key_root = tempfile::tempdir().unwrap();
|
||||
let key_path = key_root.path().join("id_ed25519");
|
||||
let fallback_key_path = key_root.path().join("id_ed25519_fallback");
|
||||
@@ -3689,7 +3693,7 @@ mod tests {
|
||||
.unwrap();
|
||||
assert!(status.success());
|
||||
}
|
||||
let materializer = RuntimeGitMaterializer::new(runtime_root.path());
|
||||
let materializer = RuntimeGitMaterializer::new(runtime_path);
|
||||
let mut request = request(repo.path());
|
||||
request.materialization = Some(crate::catalog::RepositoryMaterializationContext {
|
||||
workspace_id: "workspace-1".to_string(),
|
||||
@@ -3731,7 +3735,7 @@ mod tests {
|
||||
kind: server_api::RepositorySourceKind::Ssh,
|
||||
uri: "ssh://git@example.test/repo.git".to_string(),
|
||||
};
|
||||
let command_access = RepositoryCommandAccess::prepare(runtime_root.path(), &ssh_operation)
|
||||
let command_access = RepositoryCommandAccess::prepare(runtime_path, &ssh_operation)
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert!(!command_access.root.join("identity").exists());
|
||||
@@ -3873,18 +3877,17 @@ mod tests {
|
||||
.unwrap();
|
||||
// Restore preflight must validate without consuming this one-shot SSH
|
||||
// delivery or creating agent/session resources needed by actual bind.
|
||||
let record_before =
|
||||
fs::read(runtime_root.path().join(&id).join(MATERIALIZATION_RECORD)).unwrap();
|
||||
let record_before = fs::read(runtime_path.join(&id).join(MATERIALIZATION_RECORD)).unwrap();
|
||||
materializer
|
||||
.preflight_bind_working_directory(&id, None)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
fs::read(runtime_root.path().join(&id).join(MATERIALIZATION_RECORD)).unwrap(),
|
||||
fs::read(runtime_path.join(&id).join(MATERIALIZATION_RECORD)).unwrap(),
|
||||
record_before
|
||||
);
|
||||
assert_eq!(materializer.list_working_directories().unwrap().len(), 1);
|
||||
assert_eq!(
|
||||
fs::read_dir(runtime_root.path().join(".repository-agents"))
|
||||
fs::read_dir(runtime_path.join(".repository-agents"))
|
||||
.map(|entries| entries.count())
|
||||
.unwrap_or_default(),
|
||||
0
|
||||
@@ -3898,7 +3901,7 @@ mod tests {
|
||||
assert!(ssh_policy.contains("__repository-ssh"));
|
||||
assert!(!ssh_policy.contains("SSH_AUTH_SOCK"));
|
||||
assert_eq!(
|
||||
fs::read_dir(runtime_root.path().join(".repository-agents"))
|
||||
fs::read_dir(runtime_path.join(".repository-agents"))
|
||||
.map(|entries| entries.count())
|
||||
.unwrap_or_default(),
|
||||
1
|
||||
@@ -3916,7 +3919,7 @@ mod tests {
|
||||
"consumed delivery expiry must not revoke a live Workdir session"
|
||||
);
|
||||
assert_eq!(
|
||||
fs::read_dir(runtime_root.path().join(".repository-agents"))
|
||||
fs::read_dir(runtime_path.join(".repository-agents"))
|
||||
.map(|entries| entries.count())
|
||||
.unwrap_or_default(),
|
||||
1,
|
||||
@@ -3925,7 +3928,7 @@ mod tests {
|
||||
drop(binding);
|
||||
assert!(!ssh_command.exists());
|
||||
assert_eq!(
|
||||
fs::read_dir(runtime_root.path().join(".repository-agents"))
|
||||
fs::read_dir(runtime_path.join(".repository-agents"))
|
||||
.map(|entries| entries.count())
|
||||
.unwrap_or_default(),
|
||||
0
|
||||
@@ -3971,7 +3974,7 @@ mod tests {
|
||||
],
|
||||
&read_only_command_policy,
|
||||
));
|
||||
let broker_socket = fs::read_dir(runtime_root.path().join(REPOSITORY_ACCESS_DIR))
|
||||
let broker_socket = fs::read_dir(runtime_path.join(REPOSITORY_ACCESS_DIR))
|
||||
.unwrap()
|
||||
.filter_map(Result::ok)
|
||||
.map(|entry| entry.path().join("b.sock"))
|
||||
@@ -4104,7 +4107,7 @@ mod tests {
|
||||
.code,
|
||||
"working_directory_repository_access_expired"
|
||||
);
|
||||
let restored = RuntimeGitMaterializer::new(runtime_root.path());
|
||||
let restored = RuntimeGitMaterializer::new(runtime_path);
|
||||
assert_eq!(
|
||||
restored.bind_working_directory(&id, None).unwrap_err().code,
|
||||
"working_directory_remote_repository_access_required"
|
||||
|
||||
Reference in New Issue
Block a user