aft/
alert_records.rs

1//! Durable observation records for the alert channel.
2//!
3//! The alert engine owns diagnostics acceptance and lifecycle-id minting. This
4//! module receives those server-side events, constructs one row per represented
5//! identity, and persists rows only when the active sink is OpenCode.
6
7use crate::harness::Harness;
8use rusqlite::{params, Connection};
9use std::collections::{HashMap, HashSet};
10use std::fmt;
11
12pub const ALERT_RENDERED_TABLE: &str = "alert_rendered_records";
13pub const DISAPPEARANCE_TABLE: &str = "alert_disappearance_records";
14pub const FIVE_TURN_WINDOW: u64 = 5;
15
16/// A predefined offline query for the five-turn observation window. It joins
17/// lifecycle episodes rather than trying to infer them by parsing reminder prose.
18pub const FIVE_TURN_RESOLUTION_QUERY: &str = r#"
19SELECT
20    rendered.identity_fingerprint,
21    rendered.representation,
22    rendered.producer_key,
23    rendered.lifecycle_episode_id,
24    rendered.agent_visible_response_ordinal AS rendered_ordinal,
25    disappearance.agent_visible_response_ordinal AS disappearance_ordinal
26FROM alert_rendered_records AS rendered
27JOIN alert_disappearance_records AS disappearance
28  ON rendered.producer_key = disappearance.producer_key
29 AND rendered.lifecycle_episode_id = disappearance.lifecycle_episode_id
30WHERE disappearance.agent_visible_response_ordinal
31      BETWEEN rendered.agent_visible_response_ordinal + 1
32          AND rendered.agent_visible_response_ordinal + 5
33ORDER BY rendered.agent_visible_response_ordinal, rendered.identity_fingerprint
34"#;
35
36const ALERT_RECORD_SCHEMA: &str = r#"
37CREATE TABLE IF NOT EXISTS alert_rendered_records (
38    block_id                         TEXT NOT NULL,
39    session_id                       TEXT NOT NULL,
40    dispatch_root                    TEXT NOT NULL,
41    producer_key                     TEXT NOT NULL,
42    response_id                      TEXT NOT NULL,
43    identity_fingerprint             TEXT NOT NULL,
44    file_path                        TEXT NOT NULL,
45    line                             INTEGER NOT NULL CHECK (line > 0),
46    severity                         TEXT NOT NULL,
47    code                             TEXT,
48    wording_form                     TEXT NOT NULL CHECK (wording_form IN ('attributed', 'neutral')),
49    representation                   TEXT NOT NULL CHECK (representation IN ('shown', 'counted_only')),
50    agent_visible_response_ordinal   INTEGER NOT NULL CHECK (agent_visible_response_ordinal > 0),
51    lifecycle_episode_id             TEXT NOT NULL,
52    PRIMARY KEY (identity_fingerprint, lifecycle_episode_id)
53);
54CREATE INDEX IF NOT EXISTS idx_alert_rendered_producer_episode
55    ON alert_rendered_records (producer_key, lifecycle_episode_id);
56
57CREATE TABLE IF NOT EXISTS alert_disappearance_records (
58    session_id                       TEXT NOT NULL,
59    dispatch_root                    TEXT NOT NULL,
60    producer_key                     TEXT NOT NULL,
61    identity_fingerprint             TEXT NOT NULL,
62    lifecycle_episode_id             TEXT NOT NULL,
63    observation_ordinal              INTEGER NOT NULL CHECK (observation_ordinal > 0),
64    agent_visible_response_ordinal   INTEGER NOT NULL CHECK (agent_visible_response_ordinal > 0),
65    PRIMARY KEY (identity_fingerprint, lifecycle_episode_id)
66);
67CREATE INDEX IF NOT EXISTS idx_alert_disappearance_producer_episode
68    ON alert_disappearance_records (producer_key, lifecycle_episode_id);
69"#;
70
71#[derive(Debug)]
72pub enum AlertRecordError {
73    Sqlite(rusqlite::Error),
74    MissingOpenCodeSink,
75    DuplicateRenderedIdentity {
76        identity_fingerprint: String,
77        lifecycle_episode_id: String,
78    },
79}
80
81impl fmt::Display for AlertRecordError {
82    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
83        match self {
84            Self::Sqlite(error) => write!(f, "alert record database error: {error}"),
85            Self::MissingOpenCodeSink => {
86                write!(f, "an OpenCode session requires an alert record database sink")
87            }
88            Self::DuplicateRenderedIdentity {
89                identity_fingerprint,
90                lifecycle_episode_id,
91            } => write!(
92                f,
93                "finalized block represents {identity_fingerprint} more than once in lifecycle episode {lifecycle_episode_id}"
94            ),
95        }
96    }
97}
98
99impl std::error::Error for AlertRecordError {}
100
101impl From<rusqlite::Error> for AlertRecordError {
102    fn from(error: rusqlite::Error) -> Self {
103        Self::Sqlite(error)
104    }
105}
106
107/// Prepare the database tables used only by the OpenCode-hosted measurement sink.
108pub fn ensure_schema(connection: &Connection) -> Result<(), AlertRecordError> {
109    connection.execute_batch(ALERT_RECORD_SCHEMA)?;
110    Ok(())
111}
112
113/// A sink chosen by the host. Disabled sinks still receive constructed rows in
114/// [`FinalizationLog`], but intentionally perform no durable write.
115pub enum AlertRecordSink<'connection> {
116    OpenCode(&'connection mut Connection),
117    Disabled,
118}
119
120impl<'connection> AlertRecordSink<'connection> {
121    pub fn for_harness(
122        harness: &Harness,
123        connection: Option<&'connection mut Connection>,
124    ) -> Result<Self, AlertRecordError> {
125        if !matches!(harness, Harness::Opencode) {
126            return Ok(Self::Disabled);
127        }
128
129        let connection = connection.ok_or(AlertRecordError::MissingOpenCodeSink)?;
130        ensure_schema(connection)?;
131        Ok(Self::OpenCode(connection))
132    }
133
134    pub fn is_durable(&self) -> bool {
135        matches!(self, Self::OpenCode(_))
136    }
137
138    fn persist(
139        &mut self,
140        rendered_rows: &[AlertRenderedRecord],
141        disappearance_rows: &[DisappearanceRecord],
142    ) -> Result<(), AlertRecordError> {
143        let Self::OpenCode(connection) = self else {
144            return Ok(());
145        };
146
147        let transaction = connection.transaction()?;
148        for row in rendered_rows {
149            transaction.execute(
150                r#"
151                INSERT OR IGNORE INTO alert_rendered_records (
152                    block_id, session_id, dispatch_root, producer_key, response_id,
153                    identity_fingerprint, file_path, line, severity, code, wording_form,
154                    representation, agent_visible_response_ordinal, lifecycle_episode_id
155                )
156                VALUES (
157                    ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14
158                )
159                "#,
160                params![
161                    row.block_id,
162                    row.session_id,
163                    row.dispatch_root,
164                    row.producer_key,
165                    row.response_id,
166                    row.identity_fingerprint,
167                    row.file_path,
168                    row.line,
169                    row.severity,
170                    row.code,
171                    row.wording_form.as_str(),
172                    row.representation.as_str(),
173                    row.agent_visible_response_ordinal,
174                    row.lifecycle_episode_id,
175                ],
176            )?;
177        }
178
179        for row in disappearance_rows {
180            transaction.execute(
181                r#"
182                INSERT OR IGNORE INTO alert_disappearance_records (
183                    session_id, dispatch_root, producer_key, identity_fingerprint,
184                    lifecycle_episode_id, observation_ordinal,
185                    agent_visible_response_ordinal
186                )
187                VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
188                "#,
189                params![
190                    row.session_id,
191                    row.dispatch_root,
192                    row.producer_key,
193                    row.identity_fingerprint,
194                    row.lifecycle_episode_id,
195                    row.observation_ordinal,
196                    row.agent_visible_response_ordinal,
197                ],
198            )?;
199        }
200        transaction.commit()?;
201        Ok(())
202    }
203}
204
205#[derive(Debug, Clone, PartialEq, Eq)]
206pub struct DiagnosticIdentity {
207    pub fingerprint: String,
208    pub file_path: String,
209    pub line: u32,
210    pub severity: String,
211    pub code: Option<String>,
212}
213
214#[derive(Debug, Clone, Copy, PartialEq, Eq)]
215pub enum WordingForm {
216    Attributed,
217    Neutral,
218}
219
220impl WordingForm {
221    fn as_str(self) -> &'static str {
222        match self {
223            Self::Attributed => "attributed",
224            Self::Neutral => "neutral",
225        }
226    }
227}
228
229#[derive(Debug, Clone, Copy, PartialEq, Eq)]
230pub enum Representation {
231    Shown,
232    CountedOnly,
233}
234
235impl Representation {
236    fn as_str(self) -> &'static str {
237        match self {
238            Self::Shown => "shown",
239            Self::CountedOnly => "counted_only",
240        }
241    }
242}
243
244/// An identity represented by a finalized alert block. Every entry, including a
245/// `CountedOnly` entry represented solely by `(+N more)`, becomes its own row.
246#[derive(Debug, Clone, PartialEq, Eq)]
247pub struct RenderedAlertIdentity {
248    pub producer_key: String,
249    pub diagnostic: DiagnosticIdentity,
250    pub wording_form: WordingForm,
251    pub representation: Representation,
252    pub lifecycle_episode_id: String,
253}
254
255#[derive(Debug, Clone, PartialEq, Eq)]
256pub struct RenderedAlertBlock {
257    pub block_id: String,
258    pub dispatch_root: String,
259    pub identities: Vec<RenderedAlertIdentity>,
260}
261
262/// A finalized agent-visible response. The logger receives one of these for
263/// every agent-visible response, including responses with no alert block, so the
264/// session-global response ordinal remains contiguous.
265#[derive(Debug, Clone, PartialEq, Eq)]
266pub struct AgentVisibleFinalization {
267    pub session_id: String,
268    pub response_id: String,
269    pub rendered_block: Option<RenderedAlertBlock>,
270}
271
272/// The accepted-observation event supplied by the alert delta engine when an
273/// identity that it had already rendered leaves its producer partition.
274#[derive(Debug, Clone, PartialEq, Eq)]
275pub struct RenderedIdentityDisappearance {
276    pub session_id: String,
277    pub dispatch_root: String,
278    pub producer_key: String,
279    pub identity_fingerprint: String,
280    pub lifecycle_episode_id: String,
281    pub observation_ordinal: u64,
282}
283
284#[derive(Debug, Clone, PartialEq, Eq)]
285pub struct AlertRenderedRecord {
286    pub block_id: String,
287    pub session_id: String,
288    pub dispatch_root: String,
289    pub producer_key: String,
290    pub response_id: String,
291    pub identity_fingerprint: String,
292    pub file_path: String,
293    pub line: u32,
294    pub severity: String,
295    pub code: Option<String>,
296    pub wording_form: WordingForm,
297    pub representation: Representation,
298    pub agent_visible_response_ordinal: u64,
299    pub lifecycle_episode_id: String,
300}
301
302#[derive(Debug, Clone, PartialEq, Eq)]
303pub struct DisappearanceRecord {
304    pub session_id: String,
305    pub dispatch_root: String,
306    pub producer_key: String,
307    pub identity_fingerprint: String,
308    pub lifecycle_episode_id: String,
309    pub observation_ordinal: u64,
310    pub agent_visible_response_ordinal: u64,
311}
312
313#[derive(Debug, Clone, PartialEq, Eq)]
314pub struct FinalizationLog {
315    pub agent_visible_response_ordinal: u64,
316    pub rendered_rows: Vec<AlertRenderedRecord>,
317    pub disappearance_rows: Vec<DisappearanceRecord>,
318    pub durably_written: bool,
319}
320
321#[derive(Debug, Clone, PartialEq, Eq, Hash)]
322struct RenderedEpisodeKey {
323    session_id: String,
324    producer_key: String,
325    identity_fingerprint: String,
326    lifecycle_episode_id: String,
327}
328
329/// Session-local bookkeeping that turns accepted observation and finalization
330/// events into record rows. It intentionally has no response rendering logic.
331#[derive(Default)]
332pub struct AlertRecordLogger {
333    response_ordinals: HashMap<String, u64>,
334    rendered_episodes: HashSet<RenderedEpisodeKey>,
335    pending_disappearances: Vec<RenderedIdentityDisappearance>,
336}
337
338impl AlertRecordLogger {
339    /// Queue a disappearance only when this exact lifecycle episode was rendered.
340    /// The durable row is deliberately deferred until the next agent-visible
341    /// finalization supplies a response ordinal.
342    pub fn note_authoritative_disappearance(
343        &mut self,
344        disappearance: RenderedIdentityDisappearance,
345    ) -> bool {
346        let key = RenderedEpisodeKey {
347            session_id: disappearance.session_id.clone(),
348            producer_key: disappearance.producer_key.clone(),
349            identity_fingerprint: disappearance.identity_fingerprint.clone(),
350            lifecycle_episode_id: disappearance.lifecycle_episode_id.clone(),
351        };
352        if !self.rendered_episodes.remove(&key) {
353            return false;
354        }
355
356        self.pending_disappearances.push(disappearance);
357        true
358    }
359
360    /// Finalize an agent-visible response, incrementing the session-global
361    /// ordinal even when there is no alert block. Pending disappearances for the
362    /// session receive this ordinal; a session closed before this call receives none.
363    pub fn finalize_agent_visible_response(
364        &mut self,
365        finalization: AgentVisibleFinalization,
366        sink: &mut AlertRecordSink<'_>,
367    ) -> Result<FinalizationLog, AlertRecordError> {
368        let agent_visible_response_ordinal = self
369            .response_ordinals
370            .get(&finalization.session_id)
371            .copied()
372            .unwrap_or_default()
373            + 1;
374
375        let disappearance_rows = self
376            .pending_disappearances
377            .iter()
378            .filter(|pending| pending.session_id == finalization.session_id)
379            .map(|pending| DisappearanceRecord {
380                session_id: pending.session_id.clone(),
381                dispatch_root: pending.dispatch_root.clone(),
382                producer_key: pending.producer_key.clone(),
383                identity_fingerprint: pending.identity_fingerprint.clone(),
384                lifecycle_episode_id: pending.lifecycle_episode_id.clone(),
385                observation_ordinal: pending.observation_ordinal,
386                agent_visible_response_ordinal,
387            })
388            .collect::<Vec<_>>();
389
390        let rendered_rows = finalization
391            .rendered_block
392            .as_ref()
393            .map(|block| {
394                block
395                    .identities
396                    .iter()
397                    .map(|identity| AlertRenderedRecord {
398                        block_id: block.block_id.clone(),
399                        session_id: finalization.session_id.clone(),
400                        dispatch_root: block.dispatch_root.clone(),
401                        producer_key: identity.producer_key.clone(),
402                        response_id: finalization.response_id.clone(),
403                        identity_fingerprint: identity.diagnostic.fingerprint.clone(),
404                        file_path: identity.diagnostic.file_path.clone(),
405                        line: identity.diagnostic.line,
406                        severity: identity.diagnostic.severity.clone(),
407                        code: identity.diagnostic.code.clone(),
408                        wording_form: identity.wording_form,
409                        representation: identity.representation,
410                        agent_visible_response_ordinal,
411                        lifecycle_episode_id: identity.lifecycle_episode_id.clone(),
412                    })
413                    .collect::<Vec<_>>()
414            })
415            .unwrap_or_default();
416
417        validate_rendered_rows(&rendered_rows)?;
418        sink.persist(&rendered_rows, &disappearance_rows)?;
419
420        self.response_ordinals.insert(
421            finalization.session_id.clone(),
422            agent_visible_response_ordinal,
423        );
424        self.pending_disappearances
425            .retain(|pending| pending.session_id != finalization.session_id);
426        self.rendered_episodes
427            .extend(rendered_rows.iter().map(|row| RenderedEpisodeKey {
428                session_id: row.session_id.clone(),
429                producer_key: row.producer_key.clone(),
430                identity_fingerprint: row.identity_fingerprint.clone(),
431                lifecycle_episode_id: row.lifecycle_episode_id.clone(),
432            }));
433
434        Ok(FinalizationLog {
435            agent_visible_response_ordinal,
436            rendered_rows,
437            disappearance_rows,
438            durably_written: sink.is_durable(),
439        })
440    }
441
442    /// Discard in-memory session state at session close. Pending disappearances
443    /// are not fabricated into rows because no agent-visible response supplied an ordinal.
444    pub fn close_session(&mut self, session_id: &str) {
445        self.response_ordinals.remove(session_id);
446        self.pending_disappearances
447            .retain(|pending| pending.session_id != session_id);
448        self.rendered_episodes
449            .retain(|key| key.session_id != session_id);
450    }
451}
452
453fn validate_rendered_rows(rows: &[AlertRenderedRecord]) -> Result<(), AlertRecordError> {
454    let mut keys = HashSet::with_capacity(rows.len());
455    for row in rows {
456        let key = (
457            row.identity_fingerprint.as_str(),
458            row.lifecycle_episode_id.as_str(),
459        );
460        if !keys.insert(key) {
461            return Err(AlertRecordError::DuplicateRenderedIdentity {
462                identity_fingerprint: row.identity_fingerprint.clone(),
463                lifecycle_episode_id: row.lifecycle_episode_id.clone(),
464            });
465        }
466    }
467    Ok(())
468}
469
470#[derive(Debug, Clone, PartialEq, Eq)]
471pub struct FiveTurnResolutionRow {
472    pub identity_fingerprint: String,
473    pub representation: String,
474    pub producer_key: String,
475    pub lifecycle_episode_id: String,
476    pub rendered_ordinal: u64,
477    pub disappearance_ordinal: u64,
478}
479
480/// Execute the predefined five-turn measurement query and return its rows. This
481/// function intentionally does not calculate an efficacy rate.
482pub fn five_turn_resolution_rows(
483    connection: &Connection,
484) -> Result<Vec<FiveTurnResolutionRow>, AlertRecordError> {
485    let mut statement = connection.prepare(FIVE_TURN_RESOLUTION_QUERY)?;
486    let rows = statement.query_map([], |row| {
487        Ok(FiveTurnResolutionRow {
488            identity_fingerprint: row.get(0)?,
489            representation: row.get(1)?,
490            producer_key: row.get(2)?,
491            lifecycle_episode_id: row.get(3)?,
492            rendered_ordinal: row.get(4)?,
493            disappearance_ordinal: row.get(5)?,
494        })
495    })?;
496    rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
497}
aft/alert_records.rs

aft/
alert_records.rs
