@@ -17,7 +17,7 @@ use serde::{Deserialize, Serialize};
1717use serde_json:: { json, Value } ;
1818use std:: {
1919 collections:: { HashMap , VecDeque } ,
20- path:: PathBuf ,
20+ path:: { Path , PathBuf } ,
2121 sync:: {
2222 atomic:: { AtomicBool , Ordering } ,
2323 Arc , Mutex ,
@@ -75,6 +75,33 @@ struct Running {
7575struct QueueState {
7676 jobs : HashMap < String , Arc < Running > > ,
7777 lanes : HashMap < PathBuf , VecDeque < Arc < Running > > > ,
78+ manual : HashMap < PathBuf , Arc < ManualState > > ,
79+ }
80+ struct ManualState {
81+ cancel : CancellationToken ,
82+ finished : AtomicBool ,
83+ done : Notify ,
84+ }
85+ /// Holds an exclusive manual-write reservation in the same registry as agent
86+ /// jobs. Dropping it releases the workspace even if its request is cancelled.
87+ pub struct WorkspaceReservation {
88+ engine : Engine ,
89+ workspace : PathBuf ,
90+ state : Arc < ManualState > ,
91+ }
92+ impl WorkspaceReservation {
93+ pub fn cancellation ( & self ) -> CancellationToken {
94+ self . state . cancel . clone ( )
95+ }
96+ }
97+ impl Drop for WorkspaceReservation {
98+ fn drop ( & mut self ) {
99+ if let Ok ( mut queues) = self . engine . 0 . queues . lock ( ) {
100+ queues. manual . remove ( & self . workspace ) ;
101+ }
102+ self . state . finished . store ( true , Ordering :: Release ) ;
103+ self . state . done . notify_waiters ( ) ;
104+ }
78105}
79106struct Inner {
80107 paths : AppPaths ,
@@ -119,6 +146,60 @@ impl Engine {
119146 pub fn paths ( & self ) -> & AppPaths {
120147 & self . 0 . paths
121148 }
149+ pub fn delete_session ( & self , id : & str ) -> Result < bool > {
150+ // Starting a job uses this same lock through session lookup and insert.
151+ // No filesystem lookup is needed to delete a session for a missing folder.
152+ let queues = self
153+ . 0
154+ . queues
155+ . lock ( )
156+ . map_err ( |_| anyhow ! ( "Task queue lock poisoned" ) ) ?;
157+ let session = self . 0 . store . session ( id) ?. context ( "Session not found" ) ?;
158+ let workspace = Path :: new (
159+ session[ "workspace" ]
160+ . as_str ( )
161+ . context ( "Missing session workspace" ) ?,
162+ ) ;
163+ ensure ! (
164+ !queues. manual. contains_key( workspace) ,
165+ "Wait for the manual operation to finish before deleting this session"
166+ ) ;
167+ self . 0 . store . delete_session ( id)
168+ }
169+ pub fn reserve_workspace ( & self , workspace : & Path ) -> Result < WorkspaceReservation > {
170+ let workspace = workspace. canonicalize ( ) ?;
171+ let mut queues = self
172+ . 0
173+ . queues
174+ . lock ( )
175+ . map_err ( |_| anyhow ! ( "Task queue lock poisoned" ) ) ?;
176+ ensure ! (
177+ !self . 0 . closing. load( Ordering :: Acquire ) ,
178+ "Application is shutting down"
179+ ) ;
180+ ensure ! (
181+ !queues. manual. contains_key( & workspace) ,
182+ "A manual operation is already using this workspace"
183+ ) ;
184+ ensure ! (
185+ !queues
186+ . lanes
187+ . get( & workspace)
188+ . is_some_and( |lane| lane. iter( ) . any( |job| !job. finished. load( Ordering :: Acquire ) ) ) ,
189+ "Stop the running task before making manual changes"
190+ ) ;
191+ let state = Arc :: new ( ManualState {
192+ cancel : CancellationToken :: new ( ) ,
193+ finished : AtomicBool :: new ( false ) ,
194+ done : Notify :: new ( ) ,
195+ } ) ;
196+ queues. manual . insert ( workspace. clone ( ) , state. clone ( ) ) ;
197+ Ok ( WorkspaceReservation {
198+ engine : self . clone ( ) ,
199+ workspace,
200+ state,
201+ } )
202+ }
122203 pub async fn start ( & self , request : StartRequest ) -> Result < Job > {
123204 ensure ! (
124205 !self . 0 . closing. load( Ordering :: Acquire ) ,
@@ -155,6 +236,10 @@ impl Engine {
155236 queues. jobs. len( ) < 64 ,
156237 "At most 64 tasks may be running or queued"
157238 ) ;
239+ ensure ! (
240+ !queues. manual. contains_key( & workspace. path) ,
241+ "Wait for the manual operation in this workspace to finish before starting a task"
242+ ) ;
158243 let has_lane = queues. lanes . contains_key ( & workspace. path ) ;
159244 let busy = queues
160245 . lanes
@@ -306,6 +391,18 @@ impl Engine {
306391 }
307392 pub async fn shutdown ( & self ) -> Result < ( ) > {
308393 self . 0 . closing . store ( true , Ordering :: Release ) ;
394+ let manual: Vec < _ > = self
395+ . 0
396+ . queues
397+ . lock ( )
398+ . map_err ( |_| anyhow ! ( "Task queue lock poisoned" ) ) ?
399+ . manual
400+ . values ( )
401+ . cloned ( )
402+ . collect ( ) ;
403+ for operation in & manual {
404+ operation. cancel . cancel ( ) ;
405+ }
309406 let jobs: Vec < _ > = self
310407 . 0
311408 . queues
@@ -320,6 +417,15 @@ impl Engine {
320417 self . 0 . approvals . deny_task ( & job. snapshot ( ) ?. task_id ) ;
321418 }
322419 tokio:: time:: timeout ( Duration :: from_secs ( 15 ) , async {
420+ for operation in manual {
421+ loop {
422+ let notified = operation. done . notified ( ) ;
423+ if operation. finished . load ( Ordering :: Acquire ) {
424+ break ;
425+ }
426+ notified. await ;
427+ }
428+ }
323429 for job in jobs {
324430 loop {
325431 let notified = job. done . notified ( ) ;
0 commit comments