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 get_bash_pattern_watch(
181    conn: &Connection,
182    harness: &str,
183    session_id: &str,
184    task_id: &str,
185    watch_id: &str,
186) -> rusqlite::Result<Option<BashPatternWatchRow>> {
187    conn.query_row(
188        "SELECT harness, session_id, task_id, watch_id, pattern_kind, pattern, once,
189                created_at, stdout_offset, stderr_offset, pty_offset, scanning,
190                pending_match, match_text, match_offset, match_context
191         FROM bash_pattern_watches
192         WHERE harness = ?1 AND session_id = ?2 AND task_id = ?3 AND watch_id = ?4",
193        params![harness, session_id, task_id, watch_id],
194        map_watch_row,
195    )
196    .optional()
197}
198
199pub fn update_watch_offsets_for_task(
200    conn: &Connection,
201    harness: &str,
202    session_id: &str,
203    task_id: &str,
204    stdout_offset: i64,
205    stderr_offset: i64,
206    pty_offset: i64,
207) -> rusqlite::Result<usize> {
208    conn.execute(
209        "UPDATE bash_pattern_watches
210         SET stdout_offset = ?4, stderr_offset = ?5, pty_offset = ?6
211         WHERE harness = ?1 AND session_id = ?2 AND task_id = ?3",
212        params![
213            harness,
214            session_id,
215            task_id,
216            stdout_offset,
217            stderr_offset,
218            pty_offset
219        ],
220    )
221}
222
223fn map_watch_row(row: &Row<'_>) -> rusqlite::Result<BashPatternWatchRow> {
224    Ok(BashPatternWatchRow {
225        harness: row.get(0)?,
226        session_id: row.get(1)?,
227        task_id: row.get(2)?,
228        watch_id: row.get(3)?,
229        pattern_kind: row.get(4)?,
230        pattern: row.get(5)?,
231        once: row.get::<_, i64>(6)? != 0,
232        created_at: row.get(7)?,
233        stdout_offset: row.get(8)?,
234        stderr_offset: row.get(9)?,
235        pty_offset: row.get(10)?,
236        scanning: row.get::<_, i64>(11)? != 0,
237        pending_match: row.get::<_, i64>(12)? != 0,
238        match_text: row.get(13)?,
239        match_offset: row.get(14)?,
240        match_context: row.get(15)?,
241    })
242}
aft/db/bash_watches.rs

aft/db/
bash_watches.rs
