actl_core/
signal_session.rs

1//! One publisher per invocation; a standard thread refreshes liveness even during COM calls.
2//! Writes never contaminate stdout; execution-side acknowledgement gates detect missing records.
3use crate::activity::{ActionPhase, ActionState, ActionTarget, InputEffects};
4use crate::state::{Phase, SessionState, SignalPaths, unix_ms};
5use std::cell::RefCell;
6use std::sync::{Arc, Mutex, Weak, mpsc};
7use std::time::Duration;
8
9struct Publisher {
10    paths: SignalPaths,
11    state: SessionState,
12    journal: Option<crate::history::Journal>,
13    task: Option<String>,
14}
15impl Publisher {
16    fn event(&mut self, event: &str, data: serde_json::Value) {
17        if let Some(journal) = &mut self.journal {
18            journal.record_context(event, data, self.state.workflow.as_ref());
19        }
20    }
21    fn action_event(&mut self) {
22        if let Some(a) = &self.state.action {
23            // Window titles, selectors, text and clipboard contents are deliberately omitted.
24            let data = serde_json::json!({"action_id":a.id,"kind":a.kind,"phase":a.phase,"effects":a.effects});
25            self.event("action", data);
26        }
27    }
28    fn publish(&mut self) {
29        self.state.ts_ms = unix_ms();
30        self.state.seq += 1;
31        self.paths.write_state(&self.state);
32    }
33}
34thread_local! { static CURRENT: RefCell<Weak<Mutex<Publisher>>> = const { RefCell::new(Weak::new()) }; }
35
36pub struct SignalSession {
37    publisher: Arc<Mutex<Publisher>>,
38    stop: mpsc::Sender<()>,
39    worker: Option<std::thread::JoinHandle<()>>,
40}
41impl SignalSession {
42    pub fn start(paths: SignalPaths, command: &str, app: Option<&str>) -> Self {
43        Self::start_with_task(paths, command, app, None)
44    }
45    pub fn start_with_task(
46        paths: SignalPaths,
47        command: &str,
48        app: Option<&str>,
49        task: Option<String>,
50    ) -> Self {
51        let _ = paths.read_calls();
52        let mut state = SessionState::now(std::process::id(), command, app, None, Phase::Running);
53        state.version = 2;
54        state.call_id = format!(
55            "{}-{}",
56            std::process::id(),
57            crate::snapshot::new_snapshot_id()
58        );
59        state.started_ms = state.ts_ms;
60        let journal = if command == "history" {
61            None
62        } else {
63            match crate::history::Journal::open(&paths.dir, &state.call_id, task.clone()) {
64                Ok(journal) => Some(journal),
65                Err(e) => {
66                    eprintln!("[actl-history] INTERNAL: history unavailable: {e}");
67                    None
68                }
69            }
70        };
71        let publisher = Arc::new(Mutex::new(Publisher {
72            paths,
73            state,
74            journal,
75            task,
76        }));
77        CURRENT.with(|c| *c.borrow_mut() = Arc::downgrade(&publisher));
78        if let Ok(mut p) = publisher.lock() {
79            p.event(
80                "call_started",
81                serde_json::json!({"command":command,"version":env!("CARGO_PKG_VERSION")}),
82            );
83            p.publish();
84        }
85        let (stop, rx) = mpsc::channel();
86        let shared = publisher.clone();
87        let worker = std::thread::Builder::new()
88            .name("actl-signal-heartbeat".into())
89            .spawn(move || {
90                while rx.recv_timeout(Duration::from_millis(1000))
91                    == Err(mpsc::RecvTimeoutError::Timeout)
92                {
93                    if let Ok(mut p) = shared.lock() {
94                        p.publish();
95                    }
96                }
97            })
98            .ok();
99        Self {
100            publisher,
101            stop,
102            worker,
103        }
104    }
105
106    pub fn finish(&mut self, phase: Phase, error: Option<&str>) {
107        self.stop_worker();
108        if let Ok(mut p) = self.publisher.lock() {
109            p.state.phase = phase;
110            p.state.finished_ms = Some(unix_ms());
111            p.state.error_code = error.map(str::to_owned);
112            let elapsed = unix_ms().saturating_sub(p.state.started_ms);
113            let mut data =
114                serde_json::json!({"phase":phase,"error_code":error,"duration_ms":elapsed});
115            if let Some(flow) = &p.state.workflow {
116                data["workflow_status"] = serde_json::json!(flow.status);
117                data["workflow_completed"] = serde_json::json!(flow.status == "completed");
118                data["workflow_reason"] = serde_json::json!(flow.reason);
119                data["workflow_error_code"] = serde_json::json!(flow.error_code);
120            }
121            p.event("call_finished", data);
122            p.publish();
123        }
124        CURRENT.with(|c| *c.borrow_mut() = Weak::new());
125        if let Ok(p) = self.publisher.lock() {
126            let _ = std::fs::remove_file(p.paths.dir.join(format!("ack-{}.json", p.state.call_id)));
127        }
128    }
129    fn stop_worker(&mut self) {
130        let _ = self.stop.send(());
131        if let Some(worker) = self.worker.take() {
132            let _ = worker.join();
133        }
134    }
135}
136impl Drop for SignalSession {
137    fn drop(&mut self) {
138        self.stop_worker();
139    }
140}
141
142pub fn step(command: &str, app: Option<&str>, index: usize, total: usize) {
143    if let Some(shared) = CURRENT.with(|c| c.borrow().upgrade())
144        && let Ok(mut p) = shared.lock()
145    {
146        p.state.command = command.into();
147        p.state.app = app.map(str::to_owned);
148        p.state.step = Some(format!("{index}/{total}"));
149        p.state.action = None;
150        p.event(
151            "step",
152            serde_json::json!({"command":command,"index":index,"total":total}),
153        );
154        p.publish();
155    }
156}
157
158pub fn workflow(progress: crate::state::WorkflowProgress) {
159    if let Some(shared) = CURRENT.with(|c| c.borrow().upgrade())
160        && let Ok(mut p) = shared.lock()
161    {
162        p.state.workflow = Some(progress.clone());
163        p.event(
164            "workflow",
165            serde_json::json!({"run_id":progress.run_id,"status":progress.status,
166                "step_id":progress.step_id,"reason":progress.reason,"error_code":progress.error_code}),
167        );
168        p.state.workflow = Some(progress);
169        p.publish();
170    }
171}
172
173/// Content-free workflow diagnostics; not an authorization channel.
174pub fn flow_event(event: &str, data: serde_json::Value) {
175    if let Some(shared) = CURRENT.with(|c| c.borrow().upgrade())
176        && let Ok(mut p) = shared.lock()
177    {
178        p.event(event, data);
179    }
180}
181
182/// Guard owns exactly one resolved action. Nested helpers share their enclosing action.
183pub struct ActionGuard {
184    shared: Option<Arc<Mutex<Publisher>>>,
185    id: u64,
186    ended: bool,
187}
188impl ActionGuard {
189    pub fn begin(kind: &str, effects: InputEffects, target: ActionTarget) -> Self {
190        let shared = CURRENT.with(|c| c.borrow().upgrade());
191        let mut id = 0;
192        if let Some(shared) = &shared
193            && let Ok(mut p) = shared.lock()
194        {
195            if p.state.action.as_ref().is_some_and(ActionState::active) {
196                return Self {
197                    shared: None,
198                    id,
199                    ended: false,
200                };
201            }
202            id = p.state.seq + 1;
203            let now = unix_ms();
204            p.state.action = Some(ActionState {
205                id,
206                seq: id,
207                kind: kind.into(),
208                phase: ActionPhase::Preparing,
209                effects,
210                target,
211                ts_ms: now,
212                expires_ms: now + 6000,
213            });
214            p.action_event();
215            p.publish();
216        }
217        Self {
218            shared,
219            id,
220            ended: false,
221        }
222    }
223    pub fn is_recording(&self) -> bool {
224        self.shared.is_some()
225    }
226    pub fn identity(&self) -> Option<(String, u64, u64)> {
227        let p = self.shared.as_ref()?.lock().ok()?;
228        Some((
229            p.state.call_id.clone(),
230            self.id,
231            p.state.action.as_ref()?.ts_ms,
232        ))
233    }
234    pub fn executing(&mut self) {
235        self.transition(ActionPhase::Executing);
236    }
237    pub fn delivered(&mut self) {
238        self.transition(ActionPhase::Delivered);
239        self.ended = true;
240    }
241    pub fn partial(&mut self) {
242        self.transition(ActionPhase::Partial);
243        self.ended = true;
244    }
245    fn transition(&mut self, phase: ActionPhase) {
246        let Some(shared) = &self.shared else { return };
247        let Ok(mut p) = shared.lock() else { return };
248        let seq = p.state.seq + 1;
249        let Some(action) = p.state.action.as_mut().filter(|a| a.id == self.id) else {
250            return;
251        };
252        action.phase = phase;
253        action.seq = seq;
254        action.ts_ms = unix_ms();
255        action.expires_ms = action.ts_ms + 2000;
256        let copy = action.clone();
257        if !copy.active() {
258            p.state.recent_actions.push(copy);
259            if p.state.recent_actions.len() > 32 {
260                p.state.recent_actions.remove(0);
261            }
262        }
263        p.action_event();
264        p.publish();
265    }
266}
267impl Drop for ActionGuard {
268    fn drop(&mut self) {
269        if !self.ended {
270            let cancelled = self
271                .shared
272                .as_ref()
273                .and_then(|s| s.lock().ok())
274                .is_some_and(|p| p.paths.stop_requested());
275            self.transition(if cancelled {
276                ActionPhase::Cancelled
277            } else {
278                ActionPhase::Failed
279            });
280        }
281    }
282}
283
284/// Identity shared by commands in ACTL_TASK_ID / one flow run.
285pub fn handoff_identity() -> Option<(String, String)> {
286    CURRENT.with(|c| {
287        let p = c.borrow().upgrade()?;
288        let p = p.lock().ok()?;
289        Some((
290            p.state.call_id.clone(),
291            p.task.clone().unwrap_or_else(|| p.state.call_id.clone()),
292        ))
293    })
294}
295
296/// Content-free handoff transitions in the existing per-call journal.
297pub fn handoff_event(reason: &str) {
298    CURRENT.with(|c| {
299        if let Some(shared) = c.borrow().upgrade()
300            && let Ok(mut publisher) = shared.lock()
301        {
302            publisher.state.handoff_reason = (reason != "allowed").then(|| reason.to_owned());
303            publisher.event("handoff", serde_json::json!({"reason":reason}));
304            publisher.publish();
305        }
306    });
307}
actl_core/signal_session.rs

actl_core/
signal_session.rs
