talos-session 0.8.0

JSONL-based session logging for Talos agent conversations
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
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128

use rusqlite::{Connection, OptionalExtension, params};
use sha2::{Digest, Sha256};
use talos_core::submission::{
    PendingSubmissionState, SubmissionReceiptDisposition, SubmissionRejectionReason,
};

use super::{PendingSubmissionError, PendingSubmissionRecord};

pub(super) type RecordTuple = (String, String, String, String, Option<String>);

pub(super) struct IdentityRow {
    pub(super) receipt_id: String,
    pub(super) fingerprint: String,
    pub(super) json: Option<String>,
    pub(super) state: String,
    pub(super) turn_id: Option<String>,
}

pub(super) fn identity_matches(row: &IdentityRow, fingerprint: &str, encoded: &str) -> bool {
    row.fingerprint == fingerprint && row.json.as_deref().is_none_or(|json| json == encoded)
}

pub(super) fn lookup(
    connection: &Connection,
    submission_id: &str,
) -> Result<Option<IdentityRow>, rusqlite::Error> {
    let active = connection
        .query_row(
            "SELECT receipt_id, fingerprint, submission_json, state, turn_id
             FROM pending_submissions WHERE batch_id = ?1",
            params![submission_id],
            |row| {
                Ok(IdentityRow {
                    receipt_id: row.get(0)?,
                    fingerprint: row.get(1)?,
                    json: Some(row.get(2)?),
                    state: row.get(3)?,
                    turn_id: row.get(4)?,
                })
            },
        )
        .optional()?;
    if active.is_some() {
        return Ok(active);
    }
    connection
        .query_row(
            "SELECT receipt_id, fingerprint, state, turn_id
             FROM submission_idempotency WHERE batch_id = ?1",
            params![submission_id],
            |row| {
                Ok(IdentityRow {
                    receipt_id: row.get(0)?,
                    fingerprint: row.get(1)?,
                    json: None,
                    state: row.get(2)?,
                    turn_id: row.get(3)?,
                })
            },
        )
        .optional()
}

pub(super) fn read_record_tuple(row: &rusqlite::Row<'_>) -> rusqlite::Result<RecordTuple> {
    Ok((
        row.get(0)?,
        row.get(1)?,
        row.get(2)?,
        row.get(3)?,
        row.get(4)?,
    ))
}

pub(super) fn tuple_to_record(
    tuple: RecordTuple,
) -> Result<PendingSubmissionRecord, PendingSubmissionError> {
    let (receipt_id, payload_fingerprint, json, state, turn_id) = tuple;
    Ok(PendingSubmissionRecord {
        receipt_id,
        payload_fingerprint,
        submission: serde_json::from_str(&json)?,
        state: decode_state(&state)?,
        turn_id,
    })
}

pub(super) fn rejected(
    reason: SubmissionRejectionReason,
) -> (String, SubmissionReceiptDisposition) {
    (
        String::new(),
        SubmissionReceiptDisposition::Rejected { reason },
    )
}

pub(super) fn fingerprint(bytes: &[u8]) -> String {
    hex::encode(Sha256::digest(bytes))
}

pub(super) fn to_i64<T>(value: T) -> i64
where
    i64: TryFrom<T>,
{
    i64::try_from(value).unwrap_or(i64::MAX)
}

pub(super) fn encode_state(state: PendingSubmissionState) -> &'static str {
    match state {
        PendingSubmissionState::AcceptedPending => "accepted_pending",
        PendingSubmissionState::Running => "running",
        PendingSubmissionState::PausedPending => "paused_pending",
        PendingSubmissionState::TerminalCancelled => "terminal_cancelled",
        PendingSubmissionState::TerminalError => "terminal_error",
        PendingSubmissionState::Committed => "committed",
    }
}

pub(super) fn decode_state(state: &str) -> Result<PendingSubmissionState, PendingSubmissionError> {
    match state {
        "accepted_pending" => Ok(PendingSubmissionState::AcceptedPending),
        "running" => Ok(PendingSubmissionState::Running),
        "paused_pending" => Ok(PendingSubmissionState::PausedPending),
        "terminal_cancelled" => Ok(PendingSubmissionState::TerminalCancelled),
        "terminal_error" => Ok(PendingSubmissionState::TerminalError),
        "committed" => Ok(PendingSubmissionState::Committed),
        _ => Err(PendingSubmissionError::UnsupportedSchema(-1)),
    }
}