aft/db/
bash_watches.rs

1use rusqlite::{params, Connection, OptionalExtension, Row};
2
3/// Durable async pattern-watch registration (bash_notify / bash_watch background).
4///
5/// Survives bridge/daemon restarts so a match that lands during the gap can still
6/// be delivered. `pending_match` + match fields implement at-least-once delivery
7/// until `bash_ack_completions` acks the task (same ack path as completions).
8#[derive(Debug, Clone)]
9pub struct BashPatternWatchRow {
10    pub harness: String,
11    pub session_id: String,
12    pub task_id: String,
13    pub watch_id: String,
14    /// `"substring"` or `"regex"`.
15    pub pattern_kind: String,
16    pub pattern: String,
17    pub once: bool,
18    pub created_at: i64,
19    pub stdout_offset: i64,
20    pub stderr_offset: i64,
21    pub pty_offset: i64,
22    /// When false, a once-watch already matched and is only kept for pending delivery.
23    pub scanning: bool,
24    pub pending_match: bool,
25    pub match_text: Option<String>,
26    pub match_offset: Option<i64>,
27    pub match_context: Option<String>,
28}
29
30pub fn upsert_bash_pattern_watch(
31    conn: &Connection,
32    row: &BashPatternWatchRow,
33) -> rusqlite::Result<()> {
34    conn.execute(
35        "INSERT INTO bash_pattern_watches (
36            harness, session_id, task_id, watch_id, pattern_kind, pattern, once,
37            created_at, stdout_offset, stderr_offset, pty_offset, scanning,
38            pending_match, match_text, match_offset, match_context
39         ) VALUES (
40            ?1, ?2, ?3, ?4, ?5, ?6, ?7,
41            ?8, ?9, ?10, ?11, ?12,
42            ?13, ?14, ?15, ?16
43         )
44         ON CONFLICT(harness, session_id, task_id, watch_id) DO UPDATE SET
45            pattern_kind = excluded.pattern_kind,
46            pattern = excluded.pattern,
47            once = excluded.once,
48            created_at = excluded.created_at,
49            stdout_offset = excluded.stdout_offset,
50            stderr_offset = excluded.stderr_offset,
51            pty_offset = excluded.pty_offset,
52            scanning = excluded.scanning,
53            pending_match = excluded.pending_match,
54            match_text = excluded.match_text,
55            match_offset = excluded.match_offset,
56            match_context = excluded.match_context",
57        params![
58            row.harness,
59            row.session_id,
60            row.task_id,
61            row.watch_id,
62            row.pattern_kind,
63            row.pattern,
64            row.once,
65            row.created_at,
66            row.stdout_offset,
67            row.stderr_offset,
68            row.pty_offset,
69            row.scanning,
70            row.pending_match,
71            row.match_text,
72            row.match_offset,
73            row.match_context,
74        ],
75    )?;
76    Ok(())
77}
78
79pub fn delete_bash_pattern_watch(
80    conn: &Connection,
81    harness: &str,
82    session_id: &str,
83    task_id: &str,
84    watch_id: &str,
85) -> rusqlite::Result<usize> {
86    conn.execute(
87        "DELETE FROM bash_pattern_watches
88         WHERE harness = ?1 AND session_id = ?2 AND task_id = ?3 AND watch_id = ?4",
89        params![harness, session_id, task_id, watch_id],
90    )
91}
92
93pub fn delete_bash_pattern_watches_for_task(
94    conn: &Connection,
95    harness: &str,
96    session_id: &str,
97    task_id: &str,
98) -> rusqlite::Result<usize> {
99    conn.execute(
100        "DELETE FROM bash_pattern_watches
101         WHERE harness = ?1 AND session_id = ?2 AND task_id = ?3",
102        params![harness, session_id, task_id],
103    )
104}
105
106pub fn list_bash_pattern_watches_for_task(
107    conn: &Connection,
108    harness: &str,
109    session_id: &str,
110    task_id: &str,
111) -> rusqlite::Result<Vec<BashPatternWatchRow>> {
112    let mut stmt = conn.prepare(
113        "SELECT harness, session_id, task_id, watch_id, pattern_kind, pattern, once,
114                created_at, stdout_offset, stderr_offset, pty_offset, scanning,
115                pending_match, match_text, match_offset, match_context
116         FROM bash_pattern_watches
117         WHERE harness = ?1 AND session_id = ?2 AND task_id = ?3
118         ORDER BY created_at ASC, watch_id ASC",
119    )?;
120    let rows = stmt
121        .query_map(params![harness, session_id, task_id], map_watch_row)?
122        .collect();
123    rows
124}
125
126pub fn list_bash_pattern_watches_for_session(
127    conn: &Connection,
128    harness: &str,
129    session_id: &str,
130) -> rusqlite::Result<Vec<BashPatternWatchRow>> {
131    let mut stmt = conn.prepare(
132        "SELECT harness, session_id, task_id, watch_id, pattern_kind, pattern, once,
133                created_at, stdout_offset, stderr_offset, pty_offset, scanning,
134                pending_match, match_text, match_offset, match_context
135         FROM bash_pattern_watches
136         WHERE harness = ?1 AND session_id = ?2
137         ORDER BY created_at ASC, task_id ASC, watch_id ASC",
138    )?;
139    let rows = stmt
140        .query_map(params![harness, session_id], map_watch_row)?
141        .collect();
142    rows
143}
144
145pub fn list_bash_pattern_watches(
146    conn: &Connection,
147    harness: &str,
148) -> rusqlite::Result<Vec<BashPatternWatchRow>> {
149    let mut stmt = conn.prepare(
150        "SELECT harness, session_id, task_id, watch_id, pattern_kind, pattern, once,
151                created_at, stdout_offset, stderr_offset, pty_offset, scanning,
152                pending_match, match_text, match_offset, match_context
153         FROM bash_pattern_watches
154         WHERE harness = ?1
155         ORDER BY created_at ASC, task_id ASC, watch_id ASC",
156    )?;
157    let rows = stmt.query_map(params![harness], map_watch_row)?.collect();
158    rows
159}
160
161pub fn list_bash_pattern_watches_by_task_id(
162    conn: &Connection,
163    harness: &str,
164    task_id: &str,
165) -> rusqlite::Result<Vec<BashPatternWatchRow>> {
166    let mut stmt = conn.prepare(
167        "SELECT harness, session_id, task_id, watch_id, pattern_kind, pattern, once,
168                created_at, stdout_offset, stderr_offset, pty_offset, scanning,
169                pending_match, match_text, match_offset, match_context
170         FROM bash_pattern_watches
171         WHERE harness = ?1 AND task_id = ?2
172         ORDER BY created_at ASC, session_id ASC, watch_id ASC",
173    )?;
174    let rows = stmt
175        .query_map(params![harness, task_id], map_watch_row)?
176        .collect();
177    rows
178}
179
180pub fn count_pending_bash_pattern_watches_for_session(
181    conn: &Connection,
182    harness: &str,
183    session_id: &str,
184) -> rusqlite::Result<usize> {
185    conn.query_row(
186        "SELECT COUNT(*)
187         FROM bash_pattern_watches
188         WHERE harness = ?1 AND session_id = ?2 AND pending_match = 1",
189        params![harness, session_id],
190        |row| row.get(0),
191    )
192}
193
194pub fn count_pending_bash_pattern_watches(
195    conn: &Connection,
196    harness: &str,
197) -> rusqlite::Result<usize> {
198    conn.query_row(
199        "SELECT COUNT(*)
200         FROM bash_pattern_watches
201         WHERE harness = ?1 AND pending_match = 1",
202        params![harness],
203        |row| row.get(0),
204    )
205}
206
207pub fn get_bash_pattern_watch(
208    conn: &Connection,
209    harness: &str,
210    session_id: &str,
211    task_id: &str,
212    watch_id: &str,
213) -> rusqlite::Result<Option<BashPatternWatchRow>> {
214    conn.query_row(
215        "SELECT harness, session_id, task_id, watch_id, pattern_kind, pattern, once,
216                created_at, stdout_offset, stderr_offset, pty_offset, scanning,
217                pending_match, match_text, match_offset, match_context
218         FROM bash_pattern_watches
219         WHERE harness = ?1 AND session_id = ?2 AND task_id = ?3 AND watch_id = ?4",
220        params![harness, session_id, task_id, watch_id],
221        map_watch_row,
222    )
223    .optional()
224}
225
226pub fn update_watch_offsets_for_task(
227    conn: &Connection,
228    harness: &str,
229    session_id: &str,
230    task_id: &str,
231    stdout_offset: i64,
232    stderr_offset: i64,
233    pty_offset: i64,
234) -> rusqlite::Result<usize> {
235    conn.execute(
236        "UPDATE bash_pattern_watches
237         SET stdout_offset = ?4, stderr_offset = ?5, pty_offset = ?6
238         WHERE harness = ?1 AND session_id = ?2 AND task_id = ?3",
239        params![
240            harness,
241            session_id,
242            task_id,
243            stdout_offset,
244            stderr_offset,
245            pty_offset
246        ],
247    )
248}
249
250fn map_watch_row(row: &Row<'_>) -> rusqlite::Result<BashPatternWatchRow> {
251    Ok(BashPatternWatchRow {
252        harness: row.get(0)?,
253        session_id: row.get(1)?,
254        task_id: row.get(2)?,
255        watch_id: row.get(3)?,
256        pattern_kind: row.get(4)?,
257        pattern: row.get(5)?,
258        once: row.get::<_, i64>(6)? != 0,
259        created_at: row.get(7)?,
260        stdout_offset: row.get(8)?,
261        stderr_offset: row.get(9)?,
262        pty_offset: row.get(10)?,
263        scanning: row.get::<_, i64>(11)? != 0,
264        pending_match: row.get::<_, i64>(12)? != 0,
265        match_text: row.get(13)?,
266        match_offset: row.get(14)?,
267        match_context: row.get(15)?,
268    })
269}
aft/db/bash_watches.rs

aft/db/
bash_watches.rs
