Skip to content

fix(agent-org): isolate runtime persistence namespace #836

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
ShiboSheng wants to merge 1 commit into
base: codex/issue-757-pr2-selected-rebuild
Choose a base branch
from
Draft

fix(agent-org): isolate runtime persistence namespace #836

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 7 additions & 7 deletions src-tauri/crates/agent-core/src/core/coordination/agent_inbox/message.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -683,7 +683,7 @@ mod tests {
fn seed_minimal_running_run_for_delivery_resolution(run_id: &str) {
let conn = get_connection().expect("open sandbox database");
conn.execute_batch(
"CREATE TABLE IF NOT EXISTS agent_org_runs (
"CREATE TABLE IF NOT EXISTS agent_org_runtime_runs (
id TEXT PRIMARY KEY,
status TEXT NOT NULL,
org_snapshot_json TEXT,
Expand All @@ -705,7 +705,7 @@ mod tests {
org_member_id TEXT,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS agent_org_tasks (
CREATE TABLE IF NOT EXISTS agent_org_runtime_tasks (
id TEXT PRIMARY KEY,
org_run_id TEXT NOT NULL
);",
Expand All @@ -721,7 +721,7 @@ mod tests {
)
.expect("seed coordinator session");
conn.execute(
"INSERT INTO agent_org_runs (id, status, org_snapshot_json, root_session_id)
"INSERT INTO agent_org_runtime_runs (id, status, org_snapshot_json, root_session_id)
VALUES (?1, 'running', NULL, ?2)",
params![run_id, &root_session_id],
)
Expand All @@ -736,7 +736,7 @@ mod tests {
let payload_json = serde_json::to_string(&message).expect("serialize legacy payload");
let conn = get_connection().expect("open sandbox database");
conn.execute(
"INSERT INTO agent_inbox (
"INSERT INTO agent_org_runtime_inbox (
recipient_agent_id, recipient_member_id,
sender_agent_id, sender_member_id, org_run_id,
payload_kind, payload_json, request_id,
Expand Down Expand Up @@ -1384,7 +1384,7 @@ mod tests {
);
let conn = get_connection().expect("open sandbox database");
conn.execute(
"INSERT INTO agent_inbox_materializations (
"INSERT INTO agent_org_runtime_inbox_materializations (
inbox_id, session_id, transcript_message_id,
transcript_intent_id, materialized_at
) VALUES (?1, 'old-session', 'message-1', 'intent-1', ?2)",
Expand Down Expand Up @@ -1457,7 +1457,7 @@ mod tests {
let source = seed_legacy_orphan_inbox_row(run_id, "Original", "Original work");
let conn = get_connection().expect("open sandbox database");
conn.execute(
"INSERT INTO agent_org_tasks (id, org_run_id) VALUES ('replacement-task', ?1)",
"INSERT INTO agent_org_runtime_tasks (id, org_run_id) VALUES ('replacement-task', ?1)",
params![run_id],
)
.expect("seed replacement task");
Expand Down Expand Up @@ -1532,7 +1532,7 @@ mod tests {
};
let conn = get_connection().expect("open sandbox database");
conn.execute(
"UPDATE agent_org_runs SET org_snapshot_json=?1 WHERE id=?2",
"UPDATE agent_org_runtime_runs SET org_snapshot_json=?1 WHERE id=?2",
params![
serde_json::to_string(&crate::definitions::orgs::AgentOrgLaunchSnapshot::from(
&org
Expand Down
1 change: 1 addition & 0 deletions src-tauri/crates/agent-core/src/core/coordination/agent_inbox/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ pub use record::{
InsertInboxParams, ResolveInboxDeliveryError, ResolveInboxDeliveryParams,
};
pub use schema::init_schema;
pub(crate) use schema::{create_schema, repair_dangling_materializations};

/// Reserved sender id for system-generated agent inbox rows.
///
Expand Down
129 changes: 56 additions & 73 deletions src-tauri/crates/agent-core/src/core/coordination/agent_inbox/schema.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,18 +12,21 @@ use crate::coordination::agent_org_payload_limits as limits;
/// - `(org_run_id, created_at)` — bounded debug / E2E history pages.
/// - `(request_id)` — RPC correlation lookups.
pub fn init_schema(conn: &Connection) -> SqliteResult<()> {
create_schema(conn)?;
repair_dangling_materializations(conn)
}

pub(crate) fn create_schema(conn: &Connection) -> SqliteResult<()> {
create_agent_inbox_table(conn)?;
ensure_agent_inbox_column(conn, "causation_inbox_id", "INTEGER")?;
ensure_agent_inbox_column(conn, "display_text", "TEXT")?;
let schema = format!(
"CREATE TABLE IF NOT EXISTS agent_inbox_materializations (
"CREATE TABLE IF NOT EXISTS agent_org_runtime_inbox_materializations (
inbox_id INTEGER PRIMARY KEY,
session_id TEXT NOT NULL,
transcript_message_id TEXT NOT NULL,
transcript_intent_id TEXT NOT NULL,
materialized_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS agent_inbox_delivery_resolutions (
CREATE TABLE IF NOT EXISTS agent_org_runtime_inbox_delivery_resolutions (
inbox_id INTEGER PRIMARY KEY,
org_run_id TEXT NOT NULL,
resolution_kind TEXT NOT NULL
Expand All @@ -43,25 +46,25 @@ pub fn init_schema(conn: &Connection) -> SqliteResult<()> {
<> (replacement_task_id IS NOT NULL)))
)
);
CREATE INDEX IF NOT EXISTS idx_agent_inbox_delivery_resolutions_run
ON agent_inbox_delivery_resolutions(org_run_id, inbox_id);
CREATE INDEX IF NOT EXISTS idx_agent_inbox_materializations_session
ON agent_inbox_materializations(session_id, inbox_id);
CREATE INDEX IF NOT EXISTS idx_agent_inbox_recipient_member_unread
ON agent_inbox(recipient_member_id, read_at, created_at);
CREATE INDEX IF NOT EXISTS idx_agent_inbox_recipient_unread
ON agent_inbox(recipient_agent_id, read_at, created_at);
CREATE INDEX IF NOT EXISTS idx_agent_inbox_org_run
ON agent_inbox(org_run_id, created_at);
CREATE INDEX IF NOT EXISTS idx_agent_inbox_org_run_id
ON agent_inbox(org_run_id, id);
CREATE INDEX IF NOT EXISTS idx_agent_inbox_run_unread_recipient
ON agent_inbox(org_run_id, recipient_member_id, recipient_agent_id, id)
CREATE INDEX IF NOT EXISTS idx_agent_org_runtime_inbox_delivery_resolutions_run
ON agent_org_runtime_inbox_delivery_resolutions(org_run_id, inbox_id);
CREATE INDEX IF NOT EXISTS idx_agent_org_runtime_inbox_materializations_session
ON agent_org_runtime_inbox_materializations(session_id, inbox_id);
CREATE INDEX IF NOT EXISTS idx_agent_org_runtime_inbox_recipient_member_unread
ON agent_org_runtime_inbox(recipient_member_id, read_at, created_at);
CREATE INDEX IF NOT EXISTS idx_agent_org_runtime_inbox_recipient_unread
ON agent_org_runtime_inbox(recipient_agent_id, read_at, created_at);
CREATE INDEX IF NOT EXISTS idx_agent_org_runtime_inbox_org_run
ON agent_org_runtime_inbox(org_run_id, created_at);
CREATE INDEX IF NOT EXISTS idx_agent_org_runtime_inbox_org_run_id
ON agent_org_runtime_inbox(org_run_id, id);
CREATE INDEX IF NOT EXISTS idx_agent_org_runtime_inbox_run_unread_recipient
ON agent_org_runtime_inbox(org_run_id, recipient_member_id, recipient_agent_id, id)
WHERE read_at IS NULL;
CREATE INDEX IF NOT EXISTS idx_agent_inbox_run_kind_id
ON agent_inbox(org_run_id, payload_kind, id);
CREATE INDEX IF NOT EXISTS idx_agent_inbox_run_task_assignment_v4
ON agent_inbox(
CREATE INDEX IF NOT EXISTS idx_agent_org_runtime_inbox_run_kind_id
ON agent_org_runtime_inbox(org_run_id, payload_kind, id);
CREATE INDEX IF NOT EXISTS idx_agent_org_runtime_inbox_run_task_assignment_v4
ON agent_org_runtime_inbox(
org_run_id,
recipient_member_id,
json_extract(
Expand All @@ -80,13 +83,13 @@ pub fn init_schema(conn: &Connection) -> SqliteResult<()> {
THEN payload_json ELSE '{{}}' END,
'$.task_id'
)='text';
DROP INDEX IF EXISTS idx_agent_inbox_run_task_assignment_v3;
DROP INDEX IF EXISTS idx_agent_inbox_run_task_assignment_v2;
CREATE INDEX IF NOT EXISTS idx_agent_inbox_request_id
ON agent_inbox(request_id);
DROP INDEX IF EXISTS idx_agent_inbox_causation_once;
CREATE UNIQUE INDEX IF NOT EXISTS idx_agent_inbox_causation_recipient_once
ON agent_inbox(
DROP INDEX IF EXISTS idx_agent_org_runtime_inbox_run_task_assignment_v3;
DROP INDEX IF EXISTS idx_agent_org_runtime_inbox_run_task_assignment_v2;
CREATE INDEX IF NOT EXISTS idx_agent_org_runtime_inbox_request_id
ON agent_org_runtime_inbox(request_id);
DROP INDEX IF EXISTS idx_agent_org_runtime_inbox_causation_once;
CREATE UNIQUE INDEX IF NOT EXISTS idx_agent_org_runtime_inbox_causation_recipient_once
ON agent_org_runtime_inbox(
causation_inbox_id,
payload_kind,
recipient_agent_id,
Expand All @@ -95,7 +98,10 @@ pub fn init_schema(conn: &Connection) -> SqliteResult<()> {
WHERE causation_inbox_id IS NOT NULL;",
payload_max = limits::AGENT_INBOX_PAYLOAD_MAX_BYTES,
);
conn.execute_batch(&schema)?;
conn.execute_batch(&schema)
}

pub(crate) fn repair_dangling_materializations(conn: &Connection) -> SqliteResult<()> {
// Self-heal only provably dangling receipts. Source Inbox rows remain
// unread, allowing a healthy replacement Session to materialize them.
let transcript_tables_exist: bool = conn.query_row(
Expand All @@ -106,13 +112,13 @@ pub fn init_schema(conn: &Connection) -> SqliteResult<()> {
)?;
if transcript_tables_exist {
conn.execute(
"DELETE FROM agent_inbox_materializations AS receipt
"DELETE FROM agent_org_runtime_inbox_materializations AS receipt
WHERE NOT EXISTS (
SELECT 1 FROM agent_inbox inbox
SELECT 1 FROM agent_org_runtime_inbox inbox
WHERE inbox.id=receipt.inbox_id
AND inbox.read_at IS NULL
AND NOT EXISTS (
SELECT 1 FROM agent_inbox_delivery_resolutions resolution
SELECT 1 FROM agent_org_runtime_inbox_delivery_resolutions resolution
WHERE resolution.inbox_id=inbox.id
)
)
Expand All @@ -131,28 +137,9 @@ pub fn init_schema(conn: &Connection) -> SqliteResult<()> {
Ok(())
}

fn ensure_agent_inbox_column(
conn: &Connection,
column_name: &str,
column_definition: &str,
) -> SqliteResult<()> {
let mut stmt = conn.prepare("PRAGMA table_info(agent_inbox)")?;
let columns = stmt.query_map([], |row| row.get::<_, String>(1))?;
for column in columns {
if column? == column_name {
return Ok(());
}
}
conn.execute(
&format!("ALTER TABLE agent_inbox ADD COLUMN {column_name} {column_definition}"),
[],
)?;
Ok(())
}

fn create_agent_inbox_table(conn: &Connection) -> SqliteResult<()> {
conn.execute_batch(
"CREATE TABLE IF NOT EXISTS agent_inbox (
"CREATE TABLE IF NOT EXISTS agent_org_runtime_inbox (
id INTEGER PRIMARY KEY AUTOINCREMENT,
recipient_agent_id TEXT NOT NULL,
recipient_member_id TEXT,
Expand All @@ -175,29 +162,12 @@ mod tests {
use super::*;

#[test]
fn init_schema_adds_group_chat_display_text_to_legacy_inbox() {
fn canonical_schema_contains_current_columns_and_indexes() {
let conn = rusqlite::Connection::open_in_memory().expect("open in-memory database");
conn.execute_batch(
"CREATE TABLE agent_inbox (
id INTEGER PRIMARY KEY AUTOINCREMENT,
recipient_agent_id TEXT NOT NULL,
recipient_member_id TEXT,
sender_agent_id TEXT NOT NULL,
sender_member_id TEXT,
org_run_id TEXT,
payload_kind TEXT NOT NULL,
payload_json TEXT NOT NULL,
request_id TEXT,
created_at TEXT NOT NULL,
read_at TEXT
);",
)
.expect("create legacy inbox table");

init_schema(&conn).expect("upgrade legacy inbox schema");
init_schema(&conn).expect("create canonical inbox schema");

let mut stmt = conn
.prepare("PRAGMA table_info(agent_inbox)")
.prepare("PRAGMA table_info(agent_org_runtime_inbox)")
.expect("inspect inbox schema");
let columns = stmt
.query_map([], |row| row.get::<_, String>(1))
Expand All @@ -206,5 +176,18 @@ mod tests {
.expect("collect inbox columns");
assert!(columns.iter().any(|column| column == "causation_inbox_id"));
assert!(columns.iter().any(|column| column == "display_text"));
let required_index_count: i64 = conn
.query_row(
"SELECT COUNT(*) FROM sqlite_master
WHERE type='index' AND name IN (
'idx_agent_org_runtime_inbox_run_unread_recipient',
'idx_agent_org_runtime_inbox_run_task_assignment_v4',
'idx_agent_org_runtime_inbox_causation_recipient_once'
)",
[],
|row| row.get(0),
)
.expect("inspect canonical inbox indexes");
assert_eq!(required_index_count, 3);
}
}
Loading
Loading