merge: manual worker workdir cleanup

# Conflicts:
#	.yoi/tickets/00001KX6CRVBE/item.md
#	.yoi/tickets/00001KX6CRVBE/thread.md

crates/workspace-server/src/hosts.rs

@@ -242,6 +242,10 @@ pub struct WorkerSummary {
     pub state: String,
     pub status: String,
     pub last_seen_at: Option<String>,
+    #[serde(default)]
+    pub pinned: bool,
+    #[serde(default)]
+    pub retention_state: String,
     pub implementation: WorkerImplementationSummary,
     pub capabilities: WorkerCapabilitySummary,
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -1264,6 +1268,8 @@ impl EmbeddedWorkerRuntime {
             status: embedded_worker_execution_status_label(summary.status, &summary.execution)
                 .to_string(),
             last_seen_at: None,
+            pinned: false,
+            retention_state: "transient".to_string(),
             implementation: WorkerImplementationSummary {
                 kind: "embedded_worker_runtime".to_string(),
                 display_hint: "backend-internal worker-runtime Worker".to_string(),
@@ -1299,6 +1305,8 @@ impl EmbeddedWorkerRuntime {
             status: embedded_worker_execution_status_label(detail.status, &detail.execution)
                 .to_string(),
             last_seen_at: None,
+            pinned: false,
+            retention_state: "transient".to_string(),
             implementation: WorkerImplementationSummary {
                 kind: "embedded_worker_runtime".to_string(),
                 display_hint: "backend-internal worker-runtime Worker".to_string(),
@@ -2037,6 +2045,8 @@ impl RemoteWorkerRuntime {
             status: embedded_worker_execution_status_label(summary.status, &summary.execution)
                 .to_string(),
             last_seen_at: None,
+            pinned: false,
+            retention_state: "transient".to_string(),
             implementation: WorkerImplementationSummary {
                 kind: "remote_worker_runtime".to_string(),
                 display_hint: "Backend-proxied remote worker-runtime Worker".to_string(),
@@ -2071,6 +2081,8 @@ impl RemoteWorkerRuntime {
             status: embedded_worker_execution_status_label(detail.status, &detail.execution)
                 .to_string(),
             last_seen_at: None,
+            pinned: false,
+            retention_state: "transient".to_string(),
             implementation: WorkerImplementationSummary {
                 kind: "remote_worker_runtime".to_string(),
                 display_hint: "Backend-proxied remote worker-runtime Worker".to_string(),
@@ -3278,6 +3290,8 @@ pub fn placeholder_worker(host_id: impl Into<String>) -> WorkerSummary {
         state: "unsupported".to_string(),
         status: "Worker runtime control is not wired yet".to_string(),
         last_seen_at: None,
+        pinned: false,
+        retention_state: "transient".to_string(),
         implementation: WorkerImplementationSummary {
             kind: "placeholder".to_string(),
             display_hint: "unsupported".to_string(),
@@ -3604,6 +3618,8 @@ mod tests {
                     state: "running".to_string(),
                     status: "available".to_string(),
                     last_seen_at: None,
+                    pinned: false,
+                    retention_state: "transient".to_string(),
                     implementation: WorkerImplementationSummary {
                         kind: "fixture".to_string(),
                         display_hint: "test fixture".to_string(),

crates/workspace-server/src/store.rs

@@ -118,6 +118,14 @@ pub trait ControlPlaneStore: Send + Sync {
         workspace_id: &str,
         limit: usize,
     ) -> Result<Vec<WorkerRegistryRecord>>;
+    fn update_worker_retention(
+        &self,
+        workspace_id: &str,
+        worker_id: &str,
+        retention_state: &str,
+        updated_at: &str,
+    ) -> Result<bool>;
+    fn delete_worker_registry(&self, workspace_id: &str, worker_id: &str) -> Result<bool>;
 
     fn upsert_workdir_registry(&self, record: &WorkdirRegistryRecord) -> Result<()>;
     fn get_workdir_registry(
@@ -135,6 +143,7 @@ pub trait ControlPlaneStore: Send + Sync {
         workspace_id: &str,
         limit: usize,
     ) -> Result<Vec<WorkdirRegistryRecord>>;
+    fn delete_workdir_registry(&self, workspace_id: &str, workdir_id: &str) -> Result<bool>;
 
     fn upsert_worker_workdir_link(&self, record: &WorkerWorkdirLinkRecord) -> Result<()>;
     fn list_worker_workdir_links(
@@ -142,6 +151,11 @@ pub trait ControlPlaneStore: Send + Sync {
         workspace_id: &str,
         worker_id: &str,
     ) -> Result<Vec<WorkerWorkdirLinkRecord>>;
+    fn list_workdir_worker_links(
+        &self,
+        workspace_id: &str,
+        workdir_id: &str,
+    ) -> Result<Vec<WorkerWorkdirLinkRecord>>;
 }
 
 #[derive(Clone)]
@@ -325,6 +339,34 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
         })
     }
 
+    fn update_worker_retention(
+        &self,
+        workspace_id: &str,
+        worker_id: &str,
+        retention_state: &str,
+        updated_at: &str,
+    ) -> Result<bool> {
+        self.with_conn(|conn| {
+            let changed = conn.execute(
+                r#"UPDATE worker_registry
+                   SET retention_state = ?3, updated_at = ?4
+                   WHERE workspace_id = ?1 AND worker_id = ?2"#,
+                params![workspace_id, worker_id, retention_state, updated_at],
+            )?;
+            Ok(changed > 0)
+        })
+    }
+
+    fn delete_worker_registry(&self, workspace_id: &str, worker_id: &str) -> Result<bool> {
+        self.with_conn(|conn| {
+            let changed = conn.execute(
+                "DELETE FROM worker_registry WHERE workspace_id = ?1 AND worker_id = ?2",
+                params![workspace_id, worker_id],
+            )?;
+            Ok(changed > 0)
+        })
+    }
+
     fn upsert_workdir_registry(&self, record: &WorkdirRegistryRecord) -> Result<()> {
         self.with_conn(|conn| {
             conn.execute(
@@ -410,6 +452,16 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
         })
     }
 
+    fn delete_workdir_registry(&self, workspace_id: &str, workdir_id: &str) -> Result<bool> {
+        self.with_conn(|conn| {
+            let changed = conn.execute(
+                "DELETE FROM workdir_registry WHERE workspace_id = ?1 AND workdir_id = ?2",
+                params![workspace_id, workdir_id],
+            )?;
+            Ok(changed > 0)
+        })
+    }
+
     fn upsert_worker_workdir_link(&self, record: &WorkerWorkdirLinkRecord) -> Result<()> {
         self.with_conn(|conn| {
             conn.execute(
@@ -444,7 +496,40 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
                    WHERE workspace_id = ?1 AND worker_id = ?2 AND unlinked_at IS NULL
                    ORDER BY linked_at DESC"#,
             )?;
-            let rows = stmt.query_map(params![workspace_id, worker_id], |row| {
+            let rows = stmt.query_map(
+                params![workspace_id, worker_id],
+                read_worker_workdir_link_record,
+            )?;
+            rows.collect::<std::result::Result<Vec<_>, _>>()
+                .map_err(Error::from)
+        })
+    }
+
+    fn list_workdir_worker_links(
+        &self,
+        workspace_id: &str,
+        workdir_id: &str,
+    ) -> Result<Vec<WorkerWorkdirLinkRecord>> {
+        self.with_conn(|conn| {
+            let mut stmt = conn.prepare(
+                r#"SELECT workspace_id, worker_id, workdir_id, role, linked_at, unlinked_at
+                   FROM worker_workdir_links
+                   WHERE workspace_id = ?1 AND workdir_id = ?2 AND unlinked_at IS NULL
+                   ORDER BY linked_at DESC"#,
+            )?;
+            let rows = stmt.query_map(
+                params![workspace_id, workdir_id],
+                read_worker_workdir_link_record,
+            )?;
+            rows.collect::<std::result::Result<Vec<_>, _>>()
+                .map_err(Error::from)
+        })
+    }
+}
+
+fn read_worker_workdir_link_record(
+    row: &rusqlite::Row<'_>,
+) -> rusqlite::Result<WorkerWorkdirLinkRecord> {
     Ok(WorkerWorkdirLinkRecord {
         workspace_id: row.get(0)?,
         worker_id: row.get(1)?,
@@ -453,11 +538,6 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
         linked_at: row.get(4)?,
         unlinked_at: row.get(5)?,
     })
-            })?;
-            rows.collect::<std::result::Result<Vec<_>, _>>()
-                .map_err(Error::from)
-        })
-    }
 }
 
 fn worker_registry_select_sql(where_clause: &str) -> String {

web/workspace/src/lib/workspace-sidebar/types.ts

@@ -85,6 +85,8 @@ export type Worker = {
   workspace: { visibility: string; identity: string };
   state: string;
   status: string;
+  pinned?: boolean;
+  retention_state?: string;
   last_seen_at?: string | null;
   implementation: { kind: string; display_hint: string };
   capabilities: WorkerCapabilities;
@@ -145,6 +147,70 @@ export type BrowserWorkingDirectoryListResponse = {
   diagnostics: Diagnostic[];
 };
 
+export type CleanupTargetKind =
+  | "worker_delete"
+  | "workdir_clean_cleanup"
+  | "workdir_dirty_discard"
+  | "workdir_record_delete";
+
+export type CleanupWorkerCandidate = {
+  target_id: string;
+  action: CleanupTargetKind;
+  worker_id: string;
+  runtime_worker_id: string;
+  runtime_id: string;
+  reason: string;
+  blocking_reason?: string | null;
+  pinned: boolean;
+  retention_state: string;
+  lifecycle_state: string;
+  linked_workdir_ids: string[];
+  running_linked: boolean;
+  estimated_reclaim_bytes?: number | null;
+};
+
+export type CleanupWorkdirCandidate = {
+  target_id: string;
+  action: CleanupTargetKind;
+  workdir_id: string;
+  runtime_id: string;
+  repository_id: string;
+  reason: string;
+  blocking_reason?: string | null;
+  linked_worker_ids: string[];
+  linked_running_worker_ids: string[];
+  running_linked: boolean;
+  pinned_linked: boolean;
+  file_status: string;
+  cleanliness: string;
+  estimated_reclaim_bytes?: number | null;
+};
+
+export type RuntimeCleanupPlanResponse = {
+  workspace_id: string;
+  runtime_id: string;
+  generated_at: string;
+  revision: string;
+  digest: string;
+  workers: CleanupWorkerCandidate[];
+  workdirs: CleanupWorkdirCandidate[];
+  diagnostics: Diagnostic[];
+};
+
+export type RuntimeCleanupExecutionResponse = {
+  workspace_id: string;
+  runtime_id: string;
+  executed_at: string;
+  results: {
+    target_id: string;
+    action: CleanupTargetKind;
+    status: string;
+    message: string;
+  }[];
+  plan_after: RuntimeCleanupPlanResponse;
+  diagnostics: Diagnostic[];
+};
+
 export type BrowserWorkerWorkingDirectorySelection = {
   working_directory_id: string;
   relative_cwd?: string | null;

web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workdirs/+page.svelte

@@ -1,8 +1,16 @@
 <script lang="ts">
-  import type { WorkingDirectorySummary } from '$lib/workspace-sidebar/types';
+  import { workspaceApiPath } from '$lib/workspace-api/http';
+  import type { CleanupWorkdirCandidate, WorkingDirectorySummary } from '$lib/workspace-sidebar/types';
   import type { PageProps } from './$types';
 
   let { data }: PageProps = $props();
+  let selectedCleanupTargets = $state(new Set<string>());
+  let selectedWorkerCleanupTargets = $state(new Set<string>());
+  let confirmedDirtyTargets = $state(new Set<string>());
+  let cleanupStatus = $state<string | null>(null);
+  let cleanupBusy = $state(false);
+  let cleanupCandidates = $derived(data.cleanupPlan?.workdirs ?? []);
+  let workerCleanupCandidates = $derived(data.cleanupPlan?.workers ?? []);
   let runtimeLabel = $derived(
     data.runtimes?.items.find((runtime) => runtime.runtime_id === data.runtimeId)?.label ?? data.runtimeId,
   );
@@ -14,6 +22,67 @@
   function selectorLabel(workdir: WorkingDirectorySummary): string {
     return workdir.requested_selector ?? 'HEAD';
   }
+
+  function cleanupLabel(candidate: CleanupWorkdirCandidate): string {
+    if (candidate.action === 'workdir_dirty_discard') {
+      return candidate.cleanliness === 'dirty' ? 'Discard dirty workdir' : 'Discard unknown-state workdir';
+    }
+    if (candidate.action === 'workdir_record_delete') return 'Delete missing/removed record';
+    return 'Clean up verified-clean workdir';
+  }
+
+  function toggleSelected(targetId: string): void {
+    const next = new Set(selectedCleanupTargets);
+    if (next.has(targetId)) next.delete(targetId);
+    else next.add(targetId);
+    selectedCleanupTargets = next;
+  }
+
+  function toggleWorkerSelected(targetId: string): void {
+    const next = new Set(selectedWorkerCleanupTargets);
+    if (next.has(targetId)) next.delete(targetId);
+    else next.add(targetId);
+    selectedWorkerCleanupTargets = next;
+  }
+
+  function toggleDirtyConfirmation(targetId: string): void {
+    const next = new Set(confirmedDirtyTargets);
+    if (next.has(targetId)) next.delete(targetId);
+    else next.add(targetId);
+    confirmedDirtyTargets = next;
+  }
+
+  async function executeCleanup(): Promise<void> {
+    if (!data.cleanupPlan || (selectedCleanupTargets.size === 0 && selectedWorkerCleanupTargets.size === 0)) return;
+    cleanupBusy = true;
+    cleanupStatus = null;
+    try {
+      const response = await fetch(
+        workspaceApiPath(data.workspaceId, `/runtimes/${encodeURIComponent(data.runtimeId)}/cleanup-executions`),
+        {
+          method: 'POST',
+          headers: { 'content-type': 'application/json' },
+          body: JSON.stringify({
+            expected_plan_revision: data.cleanupPlan.revision,
+            expected_plan_digest: data.cleanupPlan.digest,
+            worker_target_ids: Array.from(selectedWorkerCleanupTargets),
+            workdir_target_ids: Array.from(selectedCleanupTargets),
+            confirm_dirty_discard_target_ids: Array.from(confirmedDirtyTargets),
+          }),
+        },
+      );
+      const payload = await response.json().catch(() => null);
+      if (!response.ok) throw new Error(payload?.message ?? payload?.error ?? response.statusText);
+      cleanupStatus = `Executed ${payload?.results?.length ?? 0} cleanup action(s). Refresh to see the latest plan.`;
+      selectedCleanupTargets = new Set();
+      selectedWorkerCleanupTargets = new Set();
+      confirmedDirtyTargets = new Set();
+    } catch (error) {
+      cleanupStatus = error instanceof Error ? error.message : 'Cleanup failed';
+    } finally {
+      cleanupBusy = false;
+    }
+  }
 </script>
 
 <svelte:head>
@@ -37,6 +106,68 @@
   {:else if data.workdirs.items.length === 0}
     <p class="section-state">No workdirs are visible for this Runtime.</p>
   {:else}
+    <section class="cleanup-panel">
+      <div>
+        <h2>Manual cleanup preview</h2>
+        <p class="muted">Select explicit Workdir targets. Raw Runtime paths are intentionally not shown.</p>
+      </div>
+      {#if data.cleanupPlanError}
+        <p class="section-state error">{data.cleanupPlanError}</p>
+      {:else if cleanupCandidates.length === 0 && workerCleanupCandidates.length === 0}
+        <p>No cleanup candidates.</p>
+      {:else}
+        <div class="cleanup-list">
+          {#each workerCleanupCandidates as candidate (candidate.target_id)}
+            <label class:blocked={!!candidate.blocking_reason}>
+              <input
+                type="checkbox"
+                checked={selectedWorkerCleanupTargets.has(candidate.target_id)}
+                disabled={!!candidate.blocking_reason}
+                onchange={() => toggleWorkerSelected(candidate.target_id)}
+              />
+              <span>
+                <strong>Delete Worker registry row:</strong> <code>{candidate.runtime_worker_id}</code>
+                <small>{candidate.lifecycle_state}; {candidate.retention_state}; linked Workdirs {candidate.linked_workdir_ids.length}</small>
+                {#if candidate.blocking_reason}<small class="error">Blocked: {candidate.blocking_reason}</small>{/if}
+              </span>
+            </label>
+          {/each}
+          {#each cleanupCandidates as candidate (candidate.target_id)}
+            <label class:blocked={!!candidate.blocking_reason} class:dirty={candidate.action === 'workdir_dirty_discard'}>
+              <input
+                type="checkbox"
+                checked={selectedCleanupTargets.has(candidate.target_id)}
+                disabled={!!candidate.blocking_reason}
+                onchange={() => toggleSelected(candidate.target_id)}
+              />
+              <span>
+                <strong>{cleanupLabel(candidate)}:</strong> <code>{candidate.workdir_id}</code>
+                <small>
+                  file {candidate.file_status}; {candidate.cleanliness}; linked Workers {candidate.linked_worker_ids.length}
+                </small>
+                {#if candidate.blocking_reason}
+                  <small class="error">Blocked: {candidate.blocking_reason}</small>
+                {:else if candidate.action === 'workdir_dirty_discard'}
+                  <label class="confirm-dirty">
+                    <input
+                      type="checkbox"
+                      checked={confirmedDirtyTargets.has(candidate.target_id)}
+                      onchange={() => toggleDirtyConfirmation(candidate.target_id)}
+                    />
+                    Confirm discard
+                  </label>
+                {/if}
+              </span>
+            </label>
+          {/each}
+        </div>
+        <button type="button" onclick={executeCleanup} disabled={cleanupBusy || (selectedCleanupTargets.size === 0 && selectedWorkerCleanupTargets.size === 0)}>
+          {cleanupBusy ? 'Executing…' : 'Execute selected cleanup'}
+        </button>
+        {#if cleanupStatus}<p>{cleanupStatus}</p>{/if}
+      {/if}
+    </section>
+
     <div class="table-wrap">
       <table class="workdirs-table">
         <thead>

web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workdirs/+page.ts

@@ -3,12 +3,13 @@ import type {
   BrowserWorkingDirectoryListResponse,
   ListResponse,
   Runtime,
+  RuntimeCleanupPlanResponse,
 } from "$lib/workspace-sidebar/types";
 import type { PageLoad } from "./$types";
 
 export const load: PageLoad = async ({ fetch, params }) => {
   const runtimeId = params.runtimeId;
-  const [runtimes, workdirs] = await Promise.all([
+  const [runtimes, workdirs, cleanupPlan] = await Promise.all([
     loadJson<ListResponse<Runtime>>(fetch, workspaceApiPath(params.workspaceId, "/runtimes")),
     loadJson<BrowserWorkingDirectoryListResponse>(
       fetch,
@@ -17,6 +18,10 @@ export const load: PageLoad = async ({ fetch, params }) => {
         `/runtimes/${encodeURIComponent(runtimeId)}/working-directories`,
       ),
     ),
+    loadJson<RuntimeCleanupPlanResponse>(
+      fetch,
+      workspaceApiPath(params.workspaceId, `/runtimes/${encodeURIComponent(runtimeId)}/cleanup-plan`),
+    ),
   ]);
 
   return {
@@ -26,5 +31,7 @@ export const load: PageLoad = async ({ fetch, params }) => {
     runtimesError: runtimes.error,
     workdirs: workdirs.data,
     workdirsError: workdirs.error,
+    cleanupPlan: cleanupPlan.data,
+    cleanupPlanError: cleanupPlan.error,
   };
 };

web/workspace/src/routes/w/[workspaceId]/workers/+page.svelte

@@ -1,9 +1,30 @@
 <script lang="ts">
+  import { workspaceApiPath } from '$lib/workspace-api/http';
   import { workerConsoleHref } from '$lib/workspace-console/model';
   import type { Worker } from '$lib/workspace-sidebar/types';
   import type { PageProps } from './$types';
 
   let { data }: PageProps = $props();
+  let retentionStatus = $state<string | null>(null);
+
+  async function setPinned(worker: Worker, pinned: boolean): Promise<void> {
+    retentionStatus = null;
+    const response = await fetch(
+      workspaceApiPath(
+        data.workspaceId,
+        `/runtimes/${encodeURIComponent(worker.runtime_id)}/workers/${encodeURIComponent(worker.worker_id)}/pin`,
+      ),
+      { method: pinned ? 'PUT' : 'DELETE' },
+    );
+    const payload = await response.json().catch(() => null);
+    if (!response.ok) {
+      retentionStatus = payload?.message ?? payload?.error ?? response.statusText;
+      return;
+    }
+    worker.pinned = Boolean(payload?.pinned);
+    worker.retention_state = payload?.retention_state ?? (worker.pinned ? 'pinned' : 'normal');
+    retentionStatus = `${worker.label} ${worker.pinned ? 'pinned' : 'unpinned'}.`;
+  }
 
   function workerStatus(worker: Worker): string {
     return `${worker.state} · ${worker.status}`;
@@ -31,7 +52,8 @@
   <header class="workers-page-header">
     <div>
       <h1 id="workers-heading">Workers</h1>
-      <p>Workers running or persisted for this workspace.</p>
+      <p>Workers running or persisted for this workspace. Pinning only updates Backend retention.</p>
+      {#if retentionStatus}<p>{retentionStatus}</p>{/if}
     </div>
     <a class="section-action" href={`/w/${data.workspaceId}/workers/new`}>New Worker</a>
   </header>
@@ -51,6 +73,7 @@
             <th>Runtime</th>
             <th>Profile</th>
             <th>Status</th>
+            <th>Retention</th>
             <th>Workdir</th>
             <th>Action</th>
           </tr>
@@ -65,9 +88,13 @@
               <td><code>{worker.runtime_id}</code></td>
               <td>{workerProfile(worker)}</td>
               <td>{workerStatus(worker)}</td>
+              <td><span class="pill {worker.pinned ? 'success' : 'muted'}">{worker.retention_state ?? 'normal'}</span></td>
               <td>{workerDirectory(worker)}</td>
               <td>
                 <a class="inline-link" href={workerConsoleHref(worker, data.workspaceId)}>Open Console</a>
+                <button type="button" onclick={() => setPinned(worker, !worker.pinned)}>
+                  {worker.pinned ? 'Unpin' : 'Pin'}
+                </button>
               </td>
             </tr>
           {/each}