fix(daemon): require authoritative Claude job removal at stop

crates/cccc-daemon/src/ops/codex_voice_analyst/claude.rs

@@ -489,6 +489,19 @@ async fn rollback_started_job(config_dir: &Path, short: &str) -> io::Result<()>
 }
 
 async fn kill_and_confirm(endpoint: &control::Endpoint, short: &str) -> io::Result<()> {
+    confirm_job_removal(endpoint, short)
+        .await
+        .map_err(unconfirmed_stop)
+}
+
+fn unconfirmed_stop(error: io::Error) -> io::Error {
+    io::Error::new(
+        error.kind(),
+        format!("Claude Agent View stop unconfirmed; ownership retained: {error}"),
+    )
+}
+
+async fn confirm_job_removal(endpoint: &control::Endpoint, short: &str) -> io::Result<()> {
     let mut kill_error = None;
     match control::list(endpoint).await {
         Ok(jobs)
@@ -507,6 +520,9 @@ async fn kill_and_confirm(endpoint: &control::Endpoint, short: &str) -> io::Resu
     }
     let deadline = tokio::time::Instant::now() + STOP_TIMEOUT;
     let mut last_error: Option<io::Error>;
+    // Agent View jobs can outlive their control harness. This adapter owns a
+    // job identifier, not a process identity it can verify has exited, so only
+    // a successful authoritative listing without that job confirms removal.
     loop {
         match control::list(endpoint).await {
             Ok(jobs)
@@ -516,8 +532,12 @@ async fn kill_and_confirm(endpoint: &control::Endpoint, short: &str) -> io::Resu
             {
                 return Ok(());
             }
-            Ok(_) => last_error = None,
-            Err(error) if retryable_control_error(&error) => last_error = Some(error),
+            Ok(_) => {
+                last_error = None;
+            }
+            Err(error) if retryable_control_error(&error) => {
+                last_error = Some(error);
+            }
             Err(error) => return Err(error),
         }
         if tokio::time::Instant::now() >= deadline {
@@ -529,7 +549,7 @@ async fn kill_and_confirm(endpoint: &control::Endpoint, short: &str) -> io::Resu
                 .unwrap_or_default();
             return Err(io::Error::new(
                 io::ErrorKind::TimedOut,
-                format!("Claude Agent View task did not stop{observation}{kill}"),
+                format!("job removal was not confirmed{observation}{kill}"),
             ));
         }
         tokio::time::sleep(Duration::from_millis(50)).await;
@@ -909,12 +929,12 @@ impl ClaudeClient {
             tokio::time::timeout(STOP_TIMEOUT + Duration::from_secs(1), receiver)
                 .await
                 .map_err(|_| {
-                    io::Error::new(
+                    unconfirmed_stop(io::Error::new(
                         io::ErrorKind::TimedOut,
-                        "Claude managed-session stop confirmation timed out",
-                    )
+                        "stop confirmation timed out",
+                    ))
                 })?
-                .map_err(|_| closed_error())??;
+                .map_err(|_| unconfirmed_stop(closed_error()))??;
         } else {
             kill_and_confirm(&self.endpoint, &self.short).await?;
         }
@@ -2494,6 +2514,154 @@ mod tests {
             fixture.operations()
         );
     }
+
+    /// A missing endpoint cannot establish whether the managed job exited.
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn stop_is_unconfirmed_when_the_endpoint_stays_missing() {
+        use sha2::{Digest, Sha256};
+        use std::os::unix::fs::{MetadataExt, PermissionsExt};
+
+        let temp = tempfile::tempdir().expect("tempdir");
+        let config_dir = temp.path().join("config");
+        std::fs::create_dir_all(config_dir.join("daemon")).expect("daemon config");
+        let control_key = config_dir.join("daemon/control.key");
+        std::fs::write(&control_key, "0123456789abcdef0123456789abcdef\n").expect("key");
+        std::fs::set_permissions(&control_key, std::fs::Permissions::from_mode(0o600))
+            .expect("key permissions");
+        let config_dir = config_dir.canonicalize().expect("canonical config");
+        let digest = format!(
+            "{:x}",
+            Sha256::digest(config_dir.to_string_lossy().as_bytes())
+        );
+        let control_dir = Path::new("/tmp")
+            .join(format!(
+                "cc-daemon-{}",
+                config_dir.metadata().expect("metadata").uid()
+            ))
+            .join(&digest[..8]);
+        std::fs::create_dir_all(&control_dir).expect("control directory");
+        std::fs::set_permissions(&control_dir, std::fs::Permissions::from_mode(0o700))
+            .expect("control directory permissions");
+        let _guard = ControlDirectory(control_dir); // no control.sock is bound
+
+        let endpoint = control::Endpoint::resolve(&config_dir).expect("endpoint resolves");
+        let started = std::time::Instant::now();
+        let error = kill_and_confirm(&endpoint, "deadbeef")
+            .await
+            .expect_err("a missing endpoint cannot confirm job removal");
+        assert_eq!(error.kind(), io::ErrorKind::TimedOut);
+        assert!(error.to_string().contains("stop unconfirmed"), "{error}");
+        assert!(
+            started.elapsed() >= STOP_TIMEOUT,
+            "a missing endpoint must allow time for authoritative confirmation: {:?}",
+            started.elapsed()
+        );
+    }
+
+    /// A task that stays listed keeps the full timeout: a stuck Agent View
+    /// job must still fail the stop, not masquerade as gone.
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn stop_still_times_out_when_the_task_stays_listed() {
+        let temp = tempfile::tempdir().expect("tempdir");
+        let config_dir = temp.path().join("config");
+        let (config_dir, listener, _guard) = bind_fake_control(&config_dir);
+        let server = tokio::spawn(async move {
+            while let Ok((mut stream, _)) = listener.accept().await {
+                let mut line = Vec::new();
+                let mut byte = [0u8; 1];
+                while tokio::io::AsyncReadExt::read(&mut stream, &mut byte)
+                    .await
+                    .expect("read")
+                    > 0
+                {
+                    if byte[0] == b'\n' {
+                        break;
+                    }
+                    line.push(byte[0]);
+                }
+                let request: Value = serde_json::from_slice(&line).expect("request json");
+                let op = request.get("op").and_then(Value::as_str).unwrap_or("");
+                let response = match op {
+                    "list" => json!({"ok":true,"op":"list","jobs":[{"short":"deadbeef"}]}),
+                    "kill" => json!({"ok":true,"op":"kill"}),
+                    _ => json!({"ok":false,"op":op,"code":"EPROTO","error":"unexpected op"}),
+                };
+                let mut bytes = serde_json::to_vec(&response).expect("response");
+                bytes.push(b'\n');
+                tokio::io::AsyncWriteExt::write_all(&mut stream, &bytes)
+                    .await
+                    .expect("write");
+            }
+        });
+        let endpoint = control::Endpoint::resolve(&config_dir).expect("endpoint resolves");
+        let error = kill_and_confirm(&endpoint, "deadbeef")
+            .await
+            .expect_err("a stuck task still fails the stop");
+        assert_eq!(error.kind(), io::ErrorKind::TimedOut);
+        assert!(error.to_string().contains("stop unconfirmed"), "{error}");
+        server.abort();
+    }
+
+    /// A stale socket is not proof that the managed job has exited. Its owned
+    /// directory is still cleaned by the test's guard after the failed stop.
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn stop_is_unconfirmed_when_a_stale_socket_refuses_connections() {
+        let temp = tempfile::tempdir().expect("tempdir");
+        let (config_dir, listener, _guard) = bind_fake_control(&temp.path().join("config"));
+        // The socket file outlives its listener, so connecting is refused.
+        drop(listener);
+        let endpoint = control::Endpoint::resolve(&config_dir).expect("endpoint resolves");
+        let started = std::time::Instant::now();
+        let error = kill_and_confirm(&endpoint, "deadbeef")
+            .await
+            .expect_err("a stale socket cannot confirm job removal");
+        assert_eq!(error.kind(), io::ErrorKind::TimedOut);
+        assert!(error.to_string().contains("stop unconfirmed"), "{error}");
+        assert!(started.elapsed() >= STOP_TIMEOUT, "{:?}", started.elapsed());
+        let socket = _guard.0.join("control.sock");
+        assert!(socket.exists(), "unconfirmed stop preserves the socket");
+        drop(_guard);
+        assert!(!socket.exists(), "the owner can still clean a stale socket");
+    }
+
+    /// Neither missing nor refused observations confirm managed job removal.
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn stop_is_unconfirmed_when_the_endpoint_alternates_missing_and_refused() {
+        let temp = tempfile::tempdir().expect("tempdir");
+        let (config_dir, listener, guard) = bind_fake_control(&temp.path().join("config"));
+        let socket = guard.0.join("control.sock");
+        drop(listener);
+        let alternating = Arc::new(AtomicBool::new(true));
+        let alternate = {
+            let alternating = Arc::clone(&alternating);
+            let socket = socket.clone();
+            tokio::spawn(async move {
+                while alternating.load(Ordering::Acquire) {
+                    if socket.exists() {
+                        // NotFound from here on.
+                        let _ = std::fs::remove_file(&socket);
+                    } else {
+                        // ConnectionRefused from here on: bound, never accepted.
+                        drop(std::os::unix::net::UnixListener::bind(&socket).expect("rebind"));
+                    }
+                    tokio::time::sleep(Duration::from_millis(100)).await;
+                }
+            })
+        };
+        let endpoint = control::Endpoint::resolve(&config_dir).expect("endpoint resolves");
+        let started = std::time::Instant::now();
+        let stopped = kill_and_confirm(&endpoint, "deadbeef").await;
+        alternating.store(false, Ordering::Release);
+        alternate.await.expect("alternating endpoint");
+        let error = stopped.expect_err("endpoint absence cannot confirm job removal");
+        assert_eq!(error.kind(), io::ErrorKind::TimedOut);
+        assert!(error.to_string().contains("stop unconfirmed"), "{error}");
+        assert!(started.elapsed() >= STOP_TIMEOUT, "{:?}", started.elapsed());
+    }
 }
 
 #[cfg(test)]

crates/cccc-daemon/src/ops/codex_voice_analyst/claude/control.rs

@@ -11,6 +11,9 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
 const REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
 const MAX_FRAME_BYTES: usize = 512 * 1024;
 
+#[cfg(all(test, unix))]
+mod stop_tests;
+
 #[cfg(unix)]
 pub(super) type PlatformStream = tokio::net::UnixStream;
 #[cfg(windows)]

crates/cccc-daemon/src/ops/codex_voice_analyst/claude/control/stop_tests.rs

@@ -0,0 +1,106 @@
+use super::{Endpoint, list};
+use crate::ops::codex_voice_analyst::{AnalystSession, LaunchConfig};
+use cccc_contracts::ActorRuntime;
+use cccc_core::HomeLayout;
+use serde_json::json;
+use std::time::Duration;
+
+/// Exercise the real CLI and its real background job, without an API request.
+/// Moving the bound socket leaves its listener and the managed job alive.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn live_claude_empty_session_resumes_without_a_prompt_after_control_gap() {
+    if std::env::var("CCCC_CLAUDE_EMPTY_LIVE").as_deref() != Ok("1") {
+        return;
+    }
+    let temp = tempfile::tempdir().expect("tempdir");
+    let home = HomeLayout::from_path(temp.path().join("home")).expect("home");
+    home.initialize().expect("initialize home");
+    let root = temp.path().join("project");
+    std::fs::create_dir(&root).expect("project");
+    let config_dir = temp.path().join("claude");
+    std::fs::create_dir(&config_dir).expect("isolated CLI config");
+    cccc_core::fs::write_json(
+        &config_dir.join(".claude.json"),
+        &json!({"bypassPermissionsModeAccepted":true,"hasCompletedOnboarding":true}),
+    )
+    .expect("isolated CLI onboarding");
+    cccc_core::fs::write_json(
+        &config_dir.join("settings.json"),
+        &json!({"skipDangerousModePermissionPrompt":true}),
+    )
+    .expect("isolated CLI settings");
+    let mut config = LaunchConfig::new(&root);
+    config.runtime = ActorRuntime::Claude;
+    config.environment.extend([
+        (
+            "CLAUDE_CONFIG_DIR".into(),
+            config_dir.to_string_lossy().into_owned(),
+        ),
+        ("ANTHROPIC_BASE_URL".into(), "http://127.0.0.1:9".into()),
+        (
+            "ANTHROPIC_AUTH_TOKEN".into(),
+            "cccc-local-input-test".into(),
+        ),
+    ]);
+    let mut session = AnalystSession::launch(&home, config)
+        .await
+        .expect("launch real background job");
+    let endpoint = Endpoint::resolve(&config_dir).expect("resolve real endpoint");
+    let before = list(&endpoint).await.expect("capture live jobs before gap");
+    let job = before
+        .iter()
+        .find(|job| job["sessionId"].as_str() == Some(session.thread_id()))
+        .expect("owned session is listed");
+    let short = job["short"]
+        .as_str()
+        .expect("owned job identity")
+        .to_owned();
+    let marker = temp.path().join("owned-resource");
+    std::fs::write(&marker, []).expect("owned resource");
+    session.cleanup_paths.push(marker.clone());
+
+    let socket = std::path::PathBuf::from(&endpoint.address);
+    let hidden = socket.with_extension("hidden");
+    std::fs::rename(&socket, &hidden).expect("hide bound endpoint");
+    let started = std::time::Instant::now();
+    let restore = async {
+        // Longer than the former two-second absence confirmation window.
+        tokio::time::sleep(Duration::from_secs(4)).await;
+        std::fs::rename(&hidden, &socket).expect("restore same bound endpoint");
+        started.elapsed()
+    };
+    let (stopped, gap) = tokio::join!(session.stop(session.generation()), restore);
+    let after = list(&endpoint)
+        .await
+        .expect("capture live jobs after return");
+    let still_listed = after.iter().any(|job| {
+        job["short"].as_str() == Some(&short)
+            && job["sessionId"].as_str() == Some(session.thread_id())
+    });
+    let ownership_retained = session.process_running() && marker.exists();
+    eprintln!(
+        "real endpoint-gap capture: gap_ms={}, stop_success={}, same_job_listed={}, ownership_retained={}",
+        gap.as_millis(),
+        stopped.is_ok(),
+        still_listed,
+        ownership_retained,
+    );
+
+    // Clean up the actual job even on the regression's failing implementation.
+    session
+        .stop(session.generation())
+        .await
+        .expect("confirm actual job removal after endpoint recovery");
+    let final_jobs = list(&endpoint).await.expect("capture confirmed removal");
+    assert!(!final_jobs.iter().any(|job| job["short"] == short));
+    assert!(!marker.exists(), "confirmed stop cleans owned resources");
+
+    assert!(gap >= Duration::from_secs(4));
+    assert!(still_listed, "the same real job survived the endpoint gap");
+    let error = stopped.expect_err(
+        "a still-running task whose endpoint disappears and returns must not be reported stopped",
+    );
+    assert_eq!(error.kind(), std::io::ErrorKind::TimedOut);
+    assert!(error.to_string().contains("stop unconfirmed"), "{error}");
+    assert!(ownership_retained, "unconfirmed stop must retain ownership");
+}