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 get_bash_pattern_watch(
146    conn: &Connection,
147    harness: &str,
148    session_id: &str,
149    task_id: &str,
150    watch_id: &str,
151) -> rusqlite::Result<Option<BashPatternWatchRow>> {
152    conn.query_row(
153        "SELECT harness, session_id, task_id, watch_id, pattern_kind, pattern, once,
154                created_at, stdout_offset, stderr_offset, pty_offset, scanning,
155                pending_match, match_text, match_offset, match_context
156         FROM bash_pattern_watches
157         WHERE harness = ?1 AND session_id = ?2 AND task_id = ?3 AND watch_id = ?4",
158        params![harness, session_id, task_id, watch_id],
159        map_watch_row,
160    )
161    .optional()
162}
163
164pub fn update_watch_offsets_for_task(
165    conn: &Connection,
166    harness: &str,
167    session_id: &str,
168    task_id: &str,
169    stdout_offset: i64,
170    stderr_offset: i64,
171    pty_offset: i64,
172) -> rusqlite::Result<usize> {
173    conn.execute(
174        "UPDATE bash_pattern_watches
175         SET stdout_offset = ?4, stderr_offset = ?5, pty_offset = ?6
176         WHERE harness = ?1 AND session_id = ?2 AND task_id = ?3",
177        params![
178            harness,
179            session_id,
180            task_id,
181            stdout_offset,
182            stderr_offset,
183            pty_offset
184        ],
185    )
186}
187
188fn map_watch_row(row: &Row<'_>) -> rusqlite::Result<BashPatternWatchRow> {
189    Ok(BashPatternWatchRow {
190        harness: row.get(0)?,
191        session_id: row.get(1)?,
192        task_id: row.get(2)?,
193        watch_id: row.get(3)?,
194        pattern_kind: row.get(4)?,
195        pattern: row.get(5)?,
196        once: row.get::<_, i64>(6)? != 0,
197        created_at: row.get(7)?,
198        stdout_offset: row.get(8)?,
199        stderr_offset: row.get(9)?,
200        pty_offset: row.get(10)?,
201        scanning: row.get::<_, i64>(11)? != 0,
202        pending_match: row.get::<_, i64>(12)? != 0,
203        match_text: row.get(13)?,
204        match_offset: row.get(14)?,
205        match_context: row.get(15)?,
206    })
207}
aft/db/bash_watches.rs

aft/db/
bash_watches.rs
