objectiveai_cli/db/logs/
plugin_messages.rs

1//! `objectiveai.plugin_messages` — captured plugin stderr.
2//!
3//! Written line-by-line by the `plugins run` stderr writer task (which
4//! owns the child's stderr pipe); read by `plugins logs list`, filtered
5//! by the plugin coordinate (owner/name/version) and paginated by the
6//! BIGSERIAL `"index"` cursor. Unlike `objectiveai.messages`, this is a
7//! flat single-table log — no `messages`/`messages_queue` bookkeeping.
8
9use objectiveai_sdk::cli::command::plugins::logs::list::ResponseItem;
10use sqlx::Row as _;
11
12use super::super::{Error, Pool};
13
14/// INSERT one captured stderr line for a `plugins run` invocation.
15#[allow(clippy::too_many_arguments)]
16pub async fn insert_plugin_message(
17    pool: &Pool,
18    owner: &str,
19    name: &str,
20    version: &str,
21    agent_instance_hierarchy: &str,
22    response_id: Option<&str>,
23    created_at: i64,
24    line: &str,
25) -> Result<(), Error> {
26    sqlx::query(
27        "INSERT INTO objectiveai.plugin_messages \
28            (owner, name, version, agent_instance_hierarchy, response_id, created_at, line) \
29         VALUES ($1, $2, $3, $4, $5, $6, $7)",
30    )
31    .bind(owner)
32    .bind(name)
33    .bind(version)
34    .bind(agent_instance_hierarchy)
35    .bind(response_id)
36    .bind(created_at)
37    .bind(line)
38    .execute(&**pool)
39    .await?;
40    Ok(())
41}
42
43/// Read captured stderr lines for one plugin coordinate, ascending by
44/// `"index"`, skipping rows with `"index" <= after_id` and capping at
45/// `limit` (`None` = unlimited).
46pub async fn read_plugin_messages(
47    pool: &Pool,
48    owner: &str,
49    name: &str,
50    version: &str,
51    after_id: Option<i64>,
52    limit: Option<i64>,
53) -> Result<Vec<ResponseItem>, Error> {
54    let rows = sqlx::query(
55        "SELECT \"index\", owner, name, version, agent_instance_hierarchy, \
56                response_id, created_at, line \
57         FROM objectiveai.plugin_messages \
58         WHERE owner = $1 AND name = $2 AND version = $3 \
59           AND \"index\" > COALESCE($4, 0) \
60         ORDER BY \"index\" ASC \
61         LIMIT $5",
62    )
63    .bind(owner)
64    .bind(name)
65    .bind(version)
66    .bind(after_id)
67    .bind(limit)
68    .fetch_all(&**pool)
69    .await?;
70
71    rows.iter()
72        .map(|row| {
73            Ok(ResponseItem {
74                index: row.try_get("index")?,
75                owner: row.try_get("owner")?,
76                name: row.try_get("name")?,
77                version: row.try_get("version")?,
78                agent_instance_hierarchy: row.try_get("agent_instance_hierarchy")?,
79                response_id: row.try_get("response_id")?,
80                created_at: row.try_get("created_at")?,
81                line: row.try_get("line")?,
82            })
83        })
84        .collect()
85}
objectiveai_cli/db/logs/plugin_messages.rs

objectiveai_cli/db/logs/
plugin_messages.rs
