acativa 0.3.0

Multi-Agent Collaboration Strategy Compiler — generic runtime, state machine and durable strategy store
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

use anyhow::{Result, anyhow, ensure};
use rusqlite::{Connection, OptionalExtension, Transaction, params};
use serde_json::Value;

use crate::ReducerEvent;

/// The durable bound on one stored receipt: a receipt is the host's opaque
/// record of one attempt outcome, not a payload channel, so an oversized one
/// is refused with `strategy_receipt_too_large` rather than growing the
/// database (and every read of it) without limit.
pub(super) const MAX_RECEIPT_JSON_BYTES: usize = 32 * 1024;

pub(super) fn initialize_schema(connection: &Connection) -> Result<()> {
    connection.execute_batch(
        "CREATE TABLE IF NOT EXISTS strategy_command_receipts(
           run_id TEXT NOT NULL REFERENCES strategy_runs(run_id) ON DELETE CASCADE,
           command_id TEXT NOT NULL,
           attempt_token TEXT NOT NULL,
           receipt_json TEXT NOT NULL,
           PRIMARY KEY(run_id, command_id, attempt_token)
         );",
    )?;
    Ok(())
}

pub(super) fn event_identity(event: &ReducerEvent) -> Result<(String, String)> {
    match event {
        ReducerEvent::CommandSucceeded {
            command_id,
            attempt_token,
            ..
        }
        | ReducerEvent::CommandFailed {
            command_id,
            attempt_token,
            ..
        }
        | ReducerEvent::CancellationAcknowledged {
            command_id,
            attempt_token,
        }
        | ReducerEvent::CancellationUnknown {
            command_id,
            attempt_token,
        } => Ok((command_id.clone(), attempt_token.clone())),
        _ => Err(anyhow!("strategy_receipt_event_invalid")),
    }
}

pub(super) fn lookup(
    connection: &Connection,
    run_id: &str,
    command_id: &str,
    attempt_token: &str,
) -> Result<Option<Value>> {
    let receipt: Option<String> = connection
        .query_row(
            "SELECT receipt_json FROM strategy_command_receipts
             WHERE run_id=?1 AND command_id=?2 AND attempt_token=?3",
            params![run_id, command_id, attempt_token],
            |row| row.get(0),
        )
        .optional()?;
    Ok(receipt
        .map(|json| serde_json::from_str(&json))
        .transpose()?)
}

pub(super) fn persist(
    transaction: &Transaction<'_>,
    run_id: &str,
    command_id: &str,
    attempt_token: &str,
    receipt: &Value,
    applied: bool,
) -> Result<()> {
    let encoded = serde_json::to_string(receipt)?;
    ensure!(
        encoded.len() <= MAX_RECEIPT_JSON_BYTES,
        "strategy_receipt_too_large"
    );
    if applied {
        let inserted = transaction.execute(
            "INSERT INTO strategy_command_receipts(run_id, command_id, attempt_token, receipt_json)
             VALUES (?1, ?2, ?3, ?4)
             ON CONFLICT(run_id, command_id, attempt_token) DO NOTHING",
            params![run_id, command_id, attempt_token, encoded],
        )?;
        if inserted == 1 {
            return Ok(());
        }
    }
    let existing = lookup(transaction, run_id, command_id, attempt_token)?
        .ok_or_else(|| anyhow!("strategy_receipt_missing"))?;
    ensure!(existing == *receipt, "strategy_receipt_conflict");
    Ok(())
}