Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

feat: integrate asynchronous Subject chat with permission management

Browse Source
This commit is contained in:
Hare
2026-10-30 05:05:08 +09:00
parent 57b57aed67 55ad8c4265
commit 70ce8479b2
66 changed files with 7856 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/subjektiv/src/chat.rs
+419
View File
@@ -0,0 +1,419 @@
//! Public conversation storage only; no delivery queue, Worker ownership or implicit publication.
//!
//! Hosts supply authenticated principals and hold their execution/authorization guard through
//! commit. Membership, exact replay and the user-input fence are checked in the same IMMEDIATE
//! FeatureStorage transaction as append. A saved message proves publication, not Runtime acceptance.
use crate::{SubjektivError, SubjektivStore, require_active_subject};
use feature_storage::FeatureStorageError;
use rusqlite::{Connection, OptionalExtension, Transaction, params};
use server_api::{
CHAT_DEFAULT_PAGE_LIMIT, CHAT_MAX_COUNTER, CHAT_MAX_PAGE_LIMIT, CHAT_MAX_TEXT_BYTES,
ChatConversation, ChatCreateConversationRequest, ChatMessage, ChatMessagePage,
ChatMessagePageQuery, ChatMessageSender, ChatSendMessageRequest, ChatUserMessageRequest,
};
use uuid::Uuid;
pub type Result<T> = std::result::Result<T, ChatError>;
#[derive(Debug, thiserror::Error)]
pub enum ChatError {
#[error(transparent)]
Storage(#[from] FeatureStorageError),
#[error(transparent)]
Subject(#[from] SubjektivError),
#[error("conversation not found or not accessible")]
ConversationNotFound,
#[error("chat operation `{0}` was reused with different input")]
OperationConflict(String),
#[error("chat user input changed: expected {expected:?}, current {actual:?}")]
UserInputChanged {
expected: Option<String>,
actual: Option<String>,
},
#[error("chat authorization changed")]
AuthorizationChanged,
#[error("invalid chat input: {0}")]
InvalidInput(String),
}
impl From<rusqlite::Error> for ChatError {
fn from(error: rusqlite::Error) -> Self {
Self::Storage(error.into())
}
}
pub(crate) fn migrate(tx: &Transaction<'_>) -> feature_storage::Result<()> {
tx.execute_batch(include_str!("chat/schema.sql"))?;
Ok(())
}
#[derive(Clone, Copy)]
enum Principal<'a> {
Account(&'a str),
Subject(&'a str),
}
impl Principal<'_> {
fn sender(self) -> (&'static str, String) {
match self {
Self::Account(id) => ("user", id.into()),
Self::Subject(id) => ("subject", id.into()),
}
}
fn member_of(self, conversation: &ChatConversation) -> bool {
match self {
Self::Account(id) => id == conversation.participant_account_id,
Self::Subject(id) => id == conversation.subject_id,
}
}
}
impl SubjektivStore {
/// Exact account-scoped replay returns the original creation result (no user input).
pub fn chat_create_conversation(
&self,
authenticated_account_id: &str,
request: &ChatCreateConversationRequest,
) -> Result<ChatConversation> {
identifier(authenticated_account_id)?;
identifier(&request.operation_id)?;
identifier(&request.subject_id)?;
self.database.try_transaction(|tx| {
let saved = tx.query_row(
"SELECT id,subject_id,participant_account_id,last_user_message_id FROM chat_conversations
WHERE participant_account_id=?1 AND create_operation_id=?2",
params![authenticated_account_id, request.operation_id], conversation_row,
).optional()?;
if let Some(mut conversation) = saved {
if conversation.subject_id != request.subject_id {
return Err(ChatError::OperationConflict(request.operation_id.clone()));
}
require_active_subject(tx, &conversation.subject_id)?;
conversation.last_user_message_id = None;
return Ok(conversation);
}
require_active_subject(tx, &request.subject_id)?;
let conversation = ChatConversation {
id: Uuid::now_v7().to_string(),
subject_id: request.subject_id.clone(),
participant_account_id: authenticated_account_id.into(),
last_user_message_id: None,
};
tx.execute(
"INSERT INTO chat_conversations
(id,subject_id,participant_account_id,last_user_message_id,create_operation_id)
VALUES (?1,?2,?3,NULL,?4)",
params![
conversation.id,
conversation.subject_id,
authenticated_account_id,
request.operation_id
],
)?;
Ok(conversation)
})
}
pub fn chat_conversation_for_account(
&self,
account: &str,
conversation_id: &str,
) -> Result<ChatConversation> {
self.database
.try_transaction(|tx| conversation(tx, Principal::Account(account), conversation_id))
}
pub fn chat_conversation_for_subject(
&self,
subject: &str,
conversation_id: &str,
) -> Result<ChatConversation> {
self.database
.try_transaction(|tx| conversation(tx, Principal::Subject(subject), conversation_id))
}
pub fn chat_messages_for_account(
&self,
account: &str,
conversation_id: &str,
query: &ChatMessagePageQuery,
) -> Result<ChatMessagePage> {
self.chat_messages(Principal::Account(account), conversation_id, query, None)
}
pub fn chat_messages_for_subject(
&self,
subject: &str,
conversation_id: &str,
query: &ChatMessagePageQuery,
) -> Result<ChatMessagePage> {
self.chat_messages(Principal::Subject(subject), conversation_id, query, None)
}
/// The Host must authorize `chat.read`, then pin that policy's existing change ID.
/// It must hold current Subject body ownership through this FeatureStorage transaction.
pub fn chat_messages_for_subject_authorized(
&self,
subject: &str,
conversation_id: &str,
query: &ChatMessagePageQuery,
expected_permission_change_id: &str,
) -> Result<ChatMessagePage> {
self.chat_messages(
Principal::Subject(subject),
conversation_id,
query,
Some(expected_permission_change_id),
)
}
fn chat_messages(
&self,
principal: Principal<'_>,
conversation_id: &str,
query: &ChatMessagePageQuery,
expected_permission_change_id: Option<&str>,
) -> Result<ChatMessagePage> {
if query.after_sequence.is_some_and(|s| s > CHAT_MAX_COUNTER) {
return Err(ChatError::InvalidInput(
"message cursor exceeds JSON safe-integer range".into(),
));
}
let limit = query.limit.unwrap_or(CHAT_DEFAULT_PAGE_LIMIT);
if !(1..=CHAT_MAX_PAGE_LIMIT).contains(&limit) {
return Err(ChatError::InvalidInput(format!(
"page limit must be in 1..={CHAT_MAX_PAGE_LIMIT}"
)));
}
self.database.try_transaction(|tx| {
check_authorization(tx, principal, expected_permission_change_id)?;
let conversation = conversation(tx, principal, conversation_id)?;
let after = query.after_sequence.unwrap_or(0) as i64;
let mut messages = tx.prepare(&format!("SELECT {MESSAGE_COLUMNS} FROM chat_messages WHERE conversation_id=?1 AND sequence>?2 ORDER BY sequence LIMIT ?3"))?
.query_map(params![conversation_id, after, limit+1], message_row)?
.collect::<std::result::Result<Vec<_>, _>>()?;
let has_more = messages.len() > limit as usize;
messages.truncate(limit as usize);
Ok(ChatMessagePage { conversation, messages, has_more })
})
}
pub fn chat_append_user_message(
&self,
account: &str,
conversation_id: &str,
request: &ChatUserMessageRequest,
) -> Result<ChatMessage> {
self.chat_append(
Principal::Account(account),
conversation_id,
Append {
operation_id: &request.operation_id,
text: &request.text,
reply_to: &request.reply_to,
work_ref: &request.work_ref,
expected_last_user_message_id: None,
},
None,
)
}
/// The authenticated Subject is supplied by the Host, never accepted from the DTO.
pub fn chat_send_message(
&self,
subject: &str,
request: &ChatSendMessageRequest,
) -> Result<ChatMessage> {
self.chat_send(subject, request, None)
}
/// The Host must authorize `chat.send`, then pin that policy's existing change ID.
/// The ownership guard must span commit; its callback must not reenter the Backend Store.
pub fn chat_send_message_authorized(
&self,
subject: &str,
request: &ChatSendMessageRequest,
expected_permission_change_id: &str,
) -> Result<ChatMessage> {
self.chat_send(subject, request, Some(expected_permission_change_id))
}
fn chat_send(
&self,
subject: &str,
request: &ChatSendMessageRequest,
expected_permission_change_id: Option<&str>,
) -> Result<ChatMessage> {
self.chat_append(
Principal::Subject(subject),
&request.conversation_id,
Append {
operation_id: &request.operation_id,
text: &request.text,
reply_to: &request.reply_to,
work_ref: &request.work_ref,
expected_last_user_message_id: Some(&request.expected_last_user_message_id),
},
expected_permission_change_id,
)
}
fn chat_append(
&self,
principal: Principal<'_>,
conversation_id: &str,
input: Append<'_>,
expected_permission_change_id: Option<&str>,
) -> Result<ChatMessage> {
identifier(input.operation_id)?;
if input.text.trim().is_empty() || input.text.len() > CHAT_MAX_TEXT_BYTES {
return Err(ChatError::InvalidInput(format!(
"text must be nonblank and at most {CHAT_MAX_TEXT_BYTES} UTF-8 bytes"
)));
}
for reference in [input.reply_to, input.work_ref]
.into_iter()
.flatten()
.chain(input.expected_last_user_message_id.into_iter().flatten())
{
identifier(reference)?;
}
self.database.try_transaction(|tx| {
check_authorization(tx, principal, expected_permission_change_id)?;
let mut conversation = conversation(tx, principal, conversation_id)?;
let (kind, sender_id) = principal.sender();
let saved = tx.query_row(&format!("SELECT {MESSAGE_COLUMNS} FROM chat_messages
WHERE conversation_id=?1 AND sender_kind=?2 AND sender_id=?3 AND operation_id=?4"),
params![conversation_id, kind, sender_id, input.operation_id], message_row).optional()?;
// Replay precedes the current-input fence: ACK loss must remain recoverable after
// another user input. The pinned observed user identity remains part of the full request.
if let Some(saved) = saved {
if saved.text != input.text || &saved.reply_to != input.reply_to || &saved.work_ref != input.work_ref
|| input.expected_last_user_message_id.is_some_and(|r| r != &saved.last_user_message_id) {
return Err(ChatError::OperationConflict(input.operation_id.into()));
}
return Ok(saved);
}
if let Some(expected) = input.expected_last_user_message_id {
if expected != &conversation.last_user_message_id {
return Err(ChatError::UserInputChanged { expected: expected.clone(), actual: conversation.last_user_message_id });
}
}
if let Some(reply_to) = input.reply_to {
let exists: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM chat_messages WHERE conversation_id=?1 AND id=?2)", params![conversation_id,reply_to], |r| r.get(0))?;
if !exists { return Err(ChatError::InvalidInput("reply_to must reference a message in this conversation".into())); }
}
let message_id = Uuid::now_v7().to_string();
if matches!(principal, Principal::Account(_)) {
conversation.last_user_message_id = Some(message_id.clone());
tx.execute("UPDATE chat_conversations SET last_user_message_id=?1 WHERE id=?2", params![message_id,conversation_id])?;
}
let last_sequence: u64 = tx.query_row("SELECT COALESCE(MAX(sequence),0) FROM chat_messages WHERE conversation_id=?1", [conversation_id], |r| r.get(0))?;
let sequence = last_sequence.checked_add(1).filter(|s| *s <= CHAT_MAX_COUNTER)
.ok_or_else(|| ChatError::InvalidInput("message sequence exhausted".into()))?;
let message = ChatMessage {
id: message_id, conversation_id: conversation_id.into(), sequence,
sender: match principal { Principal::Account(id) => ChatMessageSender::User { account_id: id.into() }, Principal::Subject(id) => ChatMessageSender::Subject { subject_id: id.into() } },
text: input.text.into(), reply_to: input.reply_to.clone(), work_ref: input.work_ref.clone(),
operation_id: input.operation_id.into(), last_user_message_id: conversation.last_user_message_id,
};
tx.execute("INSERT INTO chat_messages (id,conversation_id,sequence,sender_kind,sender_id,text,reply_to,work_ref,operation_id,last_user_message_id)
VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10)", params![message.id,conversation_id,sequence,kind,sender_id,message.text,message.reply_to,message.work_ref,message.operation_id,message.last_user_message_id])?;
Ok(message)
})
}
}
struct Append<'a> {
operation_id: &'a str,
text: &'a str,
reply_to: &'a Option<String>,
work_ref: &'a Option<String>,
expected_last_user_message_id: Option<&'a Option<String>>,
}
fn check_authorization(
tx: &Transaction<'_>,
principal: Principal<'_>,
expected: Option<&str>,
) -> Result<()> {
if let Some(expected) = expected {
identifier(expected)?;
let Principal::Subject(subject) = principal else {
return Err(ChatError::AuthorizationChanged);
};
let actual: Option<String> = tx
.query_row(
"SELECT change_id FROM subject_permissions WHERE subject_id=?1",
[subject],
|row| row.get(0),
)
.optional()?;
if actual.as_deref() != Some(expected) {
return Err(ChatError::AuthorizationChanged);
}
}
Ok(())
}
fn identifier(value: &str) -> Result<()> {
if value.trim().is_empty() || value.len() > 512 || value.chars().any(char::is_control) {
return Err(ChatError::InvalidInput(
"identifiers must be nonblank, control-free and at most 512 UTF-8 bytes".into(),
));
}
Ok(())
}
fn conversation(
tx: &Transaction<'_>,
principal: Principal<'_>,
id: &str,
) -> Result<ChatConversation> {
let conversation = tx.query_row("SELECT id,subject_id,participant_account_id,last_user_message_id FROM chat_conversations WHERE id=?1", [id], conversation_row).optional()?
.filter(|c| principal.member_of(c)).ok_or(ChatError::ConversationNotFound)?;
require_active_subject(tx, &conversation.subject_id)?;
Ok(conversation)
}
fn conversation_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<ChatConversation> {
Ok(ChatConversation {
id: row.get(0)?,
subject_id: row.get(1)?,
participant_account_id: row.get(2)?,
last_user_message_id: row.get(3)?,
})
}
const MESSAGE_COLUMNS: &str = "id,conversation_id,sequence,sender_kind,sender_id,text,reply_to,work_ref,operation_id,last_user_message_id";
fn message_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<ChatMessage> {
let kind: String = row.get(3)?;
let id = row.get(4)?;
let sender = match kind.as_str() {
"user" => ChatMessageSender::User { account_id: id },
"subject" => ChatMessageSender::Subject { subject_id: id },
_ => return Err(rusqlite::Error::InvalidQuery),
};
Ok(ChatMessage {
id: row.get(0)?,
conversation_id: row.get(1)?,
sequence: row.get(2)?,
sender,
text: row.get(5)?,
reply_to: row.get(6)?,
work_ref: row.get(7)?,
operation_id: row.get(8)?,
last_user_message_id: row.get(9)?,
})
}
pub(crate) fn erase_subject(tx: &Connection, subject: &str) -> crate::Result<()> {
tx.execute("DELETE FROM chat_messages WHERE conversation_id IN (SELECT id FROM chat_conversations WHERE subject_id=?1)", [subject])?;
tx.execute(
"DELETE FROM chat_conversations WHERE subject_id=?1",
[subject],
)?;
Ok(())
}
#[cfg(test)]
mod tests;
crates/subjektiv/src/chat/schema.sql
+81
View File
@@ -0,0 +1,81 @@
CREATE TABLE chat_conversations (
id TEXT PRIMARY KEY,
subject_id TEXT NOT NULL REFERENCES subjects(subject_id) ON DELETE RESTRICT,
participant_account_id TEXT NOT NULL,
last_user_message_id TEXT,
create_operation_id TEXT NOT NULL,
UNIQUE(participant_account_id, create_operation_id),
FOREIGN KEY(id, last_user_message_id) REFERENCES chat_messages(conversation_id, id)
DEFERRABLE INITIALLY DEFERRED
);
CREATE TABLE chat_messages (
id TEXT PRIMARY KEY,
conversation_id TEXT NOT NULL REFERENCES chat_conversations(id) ON DELETE RESTRICT,
sequence INTEGER NOT NULL CHECK(sequence BETWEEN 1 AND 9007199254740991),
sender_kind TEXT NOT NULL CHECK(sender_kind IN ('user','subject')),
sender_id TEXT NOT NULL,
text TEXT NOT NULL CHECK(length(CAST(text AS BLOB)) BETWEEN 1 AND 16384),
reply_to TEXT,
work_ref TEXT,
operation_id TEXT NOT NULL,
last_user_message_id TEXT,
CHECK(sender_kind<>'user' OR (last_user_message_id IS NOT NULL AND last_user_message_id=id)),
UNIQUE(conversation_id, sequence),
UNIQUE(conversation_id, id),
UNIQUE(conversation_id, sender_kind, sender_id, operation_id),
FOREIGN KEY(conversation_id, reply_to) REFERENCES chat_messages(conversation_id, id)
DEFERRABLE INITIALLY DEFERRED,
FOREIGN KEY(conversation_id, last_user_message_id) REFERENCES chat_messages(conversation_id, id)
DEFERRABLE INITIALLY DEFERRED
);
CREATE TRIGGER chat_conversation_initial_input BEFORE INSERT ON chat_conversations
WHEN NEW.last_user_message_id IS NOT NULL
BEGIN SELECT RAISE(ABORT, 'chat conversations start without user input'); END;
-- The deferred FK requires this fresh identity to be inserted in the same transaction.
-- Membership below requires that row to be user-authored with its own ID as baseline.
CREATE TRIGGER chat_conversation_identity BEFORE UPDATE ON chat_conversations
WHEN OLD.id<>NEW.id OR OLD.subject_id<>NEW.subject_id
OR OLD.participant_account_id<>NEW.participant_account_id
OR OLD.create_operation_id<>NEW.create_operation_id
OR NEW.last_user_message_id IS NULL
OR NEW.last_user_message_id IS OLD.last_user_message_id
OR EXISTS(SELECT 1 FROM chat_messages WHERE id=NEW.last_user_message_id)
BEGIN SELECT RAISE(ABORT, 'chat conversation identity or new user identity is invalid'); END;
CREATE TRIGGER chat_messages_immutable BEFORE UPDATE ON chat_messages
BEGIN SELECT RAISE(ABORT, 'chat messages are immutable'); END;
CREATE TRIGGER chat_messages_member BEFORE INSERT ON chat_messages
WHEN NOT EXISTS(SELECT 1 FROM chat_conversations c WHERE c.id=NEW.conversation_id
AND ((NEW.sender_kind='user' AND c.participant_account_id=NEW.sender_id)
OR (NEW.sender_kind='subject' AND c.subject_id=NEW.sender_id))
AND c.last_user_message_id IS NEW.last_user_message_id)
OR (NEW.sender_kind='subject' AND NEW.last_user_message_id IS NOT NULL
AND NOT EXISTS(SELECT 1 FROM chat_messages m WHERE m.conversation_id=NEW.conversation_id
AND m.id=NEW.last_user_message_id AND m.sender_kind='user'))
BEGIN SELECT RAISE(ABORT, 'chat message membership or observed user identity mismatch'); END;
CREATE TRIGGER chat_conversations_deletion_fence BEFORE INSERT ON chat_conversations
WHEN EXISTS(SELECT 1 FROM subject_deletions WHERE subject_id=NEW.subject_id)
BEGIN SELECT RAISE(ABORT, 'Subject deletion has begun'); END;
CREATE TRIGGER chat_conversations_update_fence BEFORE UPDATE ON chat_conversations
WHEN EXISTS(SELECT 1 FROM subject_deletions WHERE subject_id=OLD.subject_id OR subject_id=NEW.subject_id)
BEGIN SELECT RAISE(ABORT, 'Subject deletion has begun'); END;
CREATE TRIGGER chat_messages_deletion_fence BEFORE INSERT ON chat_messages
WHEN EXISTS(SELECT 1 FROM subject_deletions d JOIN chat_conversations c ON c.subject_id=d.subject_id
WHERE c.id=NEW.conversation_id)
BEGIN SELECT RAISE(ABORT, 'Subject deletion has begun'); END;
CREATE TRIGGER chat_messages_retained BEFORE DELETE ON chat_messages
WHEN NOT EXISTS(SELECT 1 FROM subject_deletions d JOIN chat_conversations c ON c.subject_id=d.subject_id
WHERE c.id=OLD.conversation_id)
BEGIN SELECT RAISE(ABORT, 'chat messages are retained until Subject erasure'); END;
CREATE TRIGGER chat_conversations_retained BEFORE DELETE ON chat_conversations
WHEN NOT EXISTS(SELECT 1 FROM subject_deletions WHERE subject_id=OLD.subject_id)
BEGIN SELECT RAISE(ABORT, 'chat conversations are retained until Subject erasure'); END;
crates/subjektiv/src/chat/tests.rs
+938
View File
@@ -0,0 +1,938 @@
//! Real SQLite component tests through FeatureStorage and the chat APIs. Host authentication,
//! operation authorization and Backend singleton ownership are outside this suite; policy IDs
//! supplied here model the Host's observed existing policy, not a new authority source.
use super::*;
use crate::tests::{open_store, role};
use feature_storage::{FeatureRegistration, FeatureStorage};
use std::sync::{Arc, Barrier};
fn create(
store: &SubjektivStore,
account: &str,
subject: &str,
operation: &str,
) -> ChatConversation {
store
.chat_create_conversation(
account,
&ChatCreateConversationRequest {
operation_id: operation.into(),
subject_id: subject.into(),
},
)
.unwrap()
}
fn user(operation: &str, text: &str) -> ChatUserMessageRequest {
ChatUserMessageRequest {
operation_id: operation.into(),
text: text.into(),
reply_to: None,
work_ref: None,
}
}
fn send(
conversation: &ChatConversation,
operation: &str,
last_user_message_id: Option<&str>,
) -> ChatSendMessageRequest {
ChatSendMessageRequest {
conversation_id: conversation.id.clone(),
operation_id: operation.into(),
text: "public answer".into(),
reply_to: None,
work_ref: None,
expected_last_user_message_id: last_user_message_id.map(str::to_owned),
}
}
fn page(store: &SubjektivStore, conversation: &ChatConversation) -> ChatMessagePage {
store
.chat_messages_for_account(
&conversation.participant_account_id,
&conversation.id,
&ChatMessagePageQuery::default(),
)
.unwrap()
}
fn sql(store: &SubjektivStore, sql: &str) {
store
.database
.with_connection(|c| {
c.execute_batch(sql)?;
Ok(())
})
.unwrap();
}
#[test]
fn membership_and_workspace_isolation_cover_reads_writes_and_replay() {
let temp = tempfile::tempdir().unwrap();
let (_manager, _scope, store) = open_store(temp.path(), "workspace-a");
let a = store.create_subject(role()).unwrap().id;
let b = store.create_subject(role()).unwrap().id;
let c = create(&store, "alice", &a, "create");
let input = user("message", "private user message");
store
.chat_append_user_message("alice", &c.id, &input)
.unwrap();
for result in [
store.chat_conversation_for_account("bob", &c.id),
store.chat_conversation_for_subject(&b, &c.id),
] {
assert!(matches!(result, Err(ChatError::ConversationNotFound)));
}
assert!(matches!(
store.chat_messages_for_account("bob", &c.id, &ChatMessagePageQuery::default()),
Err(ChatError::ConversationNotFound)
));
assert!(matches!(
store.chat_messages_for_subject(&b, &c.id, &ChatMessagePageQuery::default()),
Err(ChatError::ConversationNotFound)
));
assert!(matches!(
store.chat_append_user_message("bob", &c.id, &input),
Err(ChatError::ConversationNotFound)
));
assert!(matches!(
store.chat_send_message(&b, &send(&c, "message", None)),
Err(ChatError::ConversationNotFound)
));
let (_other_manager, _other_scope, other) = open_store(temp.path(), "workspace-b");
assert!(matches!(
other.chat_conversation_for_account("alice", &c.id),
Err(ChatError::ConversationNotFound)
));
assert!(matches!(
other.chat_append_user_message("alice", &c.id, &input),
Err(ChatError::ConversationNotFound)
));
assert!(
other
.chat_create_conversation(
"alice",
&ChatCreateConversationRequest {
operation_id: "create".into(),
subject_id: a
}
)
.is_err()
);
assert_eq!(page(&store, &c).messages.len(), 1);
}
#[test]
fn ack_loss_reopens_and_replays_original_results_after_new_input() {
let temp = tempfile::tempdir().unwrap();
let (c, first, answer, request) = {
let (_manager, _scope, store) = open_store(temp.path(), "workspace-a");
let subject = store.create_subject(role()).unwrap().id;
let c = create(&store, "alice", &subject, "create");
let first = store
.chat_append_user_message("alice", &c.id, &user("same", "first"))
.unwrap();
let mut request = send(&c, "same", Some(&first.id)); // Sender namespaces are independent.
request.reply_to = Some(first.id.clone());
request.work_ref = Some("T-766".into());
let answer = store.chat_send_message(&subject, &request).unwrap();
store
.chat_append_user_message("alice", &c.id, &user("next", "later input"))
.unwrap();
(c, first, answer, request)
}; // Drop every connection: simulate committed storage with a lost acknowledgement.
let (_manager, _scope, store) = open_store(temp.path(), "workspace-a");
assert_eq!(create(&store, "alice", &c.subject_id, "create"), c);
assert_eq!(
store
.chat_append_user_message("alice", &c.id, &user("same", "first"))
.unwrap(),
first
);
assert_eq!(
store.chat_send_message(&c.subject_id, &request).unwrap(),
answer
);
let p = page(&store, &c);
let last = &p.messages[2];
assert_eq!(
p.conversation.last_user_message_id.as_deref(),
Some(last.id.as_str())
);
assert_eq!(
first.last_user_message_id.as_deref(),
Some(first.id.as_str())
);
assert_eq!(
answer.last_user_message_id.as_deref(),
Some(first.id.as_str())
);
assert_eq!(last.last_user_message_id.as_deref(), Some(last.id.as_str()));
assert_eq!(
p.messages.iter().map(|m| m.sequence).collect::<Vec<_>>(),
vec![1, 2, 3]
);
assert!(!p.has_more);
assert_eq!(
store
.chat_messages_for_subject(&c.subject_id, &c.id, &ChatMessagePageQuery::default())
.unwrap(),
p
);
}
#[test]
fn operation_reuse_checks_each_payload_field_and_pinned_observed_user_identity() {
let temp = tempfile::tempdir().unwrap();
let (_manager, _scope, store) = open_store(temp.path(), "workspace-a");
let subject = store.create_subject(role()).unwrap().id;
let c = create(&store, "alice", &subject, "create");
let other_subject = store.create_subject(role()).unwrap().id;
assert!(matches!(
store.chat_create_conversation(
"alice",
&ChatCreateConversationRequest {
operation_id: "create".into(),
subject_id: other_subject
}
),
Err(ChatError::OperationConflict(_))
));
// Same creation operation in another account namespace is a different operation.
assert_ne!(create(&store, "bob", &subject, "create").id, c.id);
let input = user("input", "first");
let first = store
.chat_append_user_message("alice", &c.id, &input)
.unwrap();
let mut variants = vec![input.clone(); 3];
variants[0].text = "changed".into();
variants[1].reply_to = Some(first.id.clone());
variants[2].work_ref = Some("work".into());
for variant in variants {
assert!(matches!(
store.chat_append_user_message("alice", &c.id, &variant),
Err(ChatError::OperationConflict(_))
));
}
let request = send(&c, "answer", Some(&first.id));
store.chat_send_message(&subject, &request).unwrap();
let mut variants = vec![request.clone(); 4];
variants[0].text = "changed".into();
variants[1].reply_to = Some(first.id.clone());
variants[2].work_ref = Some("work".into());
variants[3].expected_last_user_message_id = None;
for variant in variants {
assert!(matches!(
store.chat_send_message(&subject, &variant),
Err(ChatError::OperationConflict(_))
));
}
assert_eq!(page(&store, &c).messages.len(), 2);
assert_eq!(
page(&store, &c)
.conversation
.last_user_message_id
.as_deref(),
Some(first.id.as_str())
);
}
#[test]
fn reply_edges_cannot_cross_conversations_and_operations_are_conversation_scoped() {
let temp = tempfile::tempdir().unwrap();
let (_manager, _scope, store) = open_store(temp.path(), "workspace-a");
let subject = store.create_subject(role()).unwrap().id;
let c = create(&store, "alice", &subject, "create-a");
let d = create(&store, "alice", &subject, "create-b");
let first = store
.chat_append_user_message("alice", &c.id, &user("same", "a"))
.unwrap();
let mut bad = user("bad", "not a valid reply");
bad.reply_to = Some(first.id.clone());
assert!(matches!(
store.chat_append_user_message("alice", &d.id, &bad),
Err(ChatError::InvalidInput(_))
));
let mut bad_send = send(&d, "bad", None);
bad_send.reply_to = Some(first.id.clone());
assert!(matches!(
store.chat_send_message(&subject, &bad_send),
Err(ChatError::InvalidInput(_))
));
bad.reply_to = Some("missing".into());
assert!(
store
.chat_append_user_message("alice", &c.id, &bad)
.is_err()
);
assert!(page(&store, &d).messages.is_empty());
assert_eq!(page(&store, &d).conversation.last_user_message_id, None);
let other_input = store
.chat_append_user_message("alice", &d.id, &user("same", "b"))
.unwrap();
// The composite FK rejects bypassing the repository too, not merely the ID lookup.
let result = store.database.transaction(|tx| {
tx.execute("INSERT INTO chat_messages (id,conversation_id,sequence,sender_kind,sender_id,text,reply_to,operation_id,last_user_message_id) VALUES ('bad',?1,2,'subject',?2,'bad',?3,'direct',?4)",params![d.id,subject,first.id,other_input.id])?;
Ok(())
});
assert!(result.is_err());
assert_eq!(page(&store, &d).messages.len(), 1);
}
#[test]
fn page_and_utf8_text_bounds_are_enforced_without_silent_clamping() {
let temp = tempfile::tempdir().unwrap();
let (_manager, _scope, store) = open_store(temp.path(), "workspace-a");
let subject = store.create_subject(role()).unwrap().id;
let c = create(&store, "alice", &subject, "create");
let exactly = "é".repeat(CHAT_MAX_TEXT_BYTES / 2);
let first = store
.chat_append_user_message("alice", &c.id, &user("limit", &exactly))
.unwrap();
for text in [String::new(), " \n".into(), format!("{exactly}x")] {
assert!(matches!(
store.chat_append_user_message("alice", &c.id, &user("bad", &text)),
Err(ChatError::InvalidInput(_))
));
let mut request = send(&c, "bad", Some(&first.id));
request.text = text;
assert!(matches!(
store.chat_send_message(&subject, &request),
Err(ChatError::InvalidInput(_))
));
}
for n in 2..=102 {
store
.chat_send_message(&subject, &send(&c, &format!("answer-{n}"), Some(&first.id)))
.unwrap();
}
let query = |after, limit| ChatMessagePageQuery {
after_sequence: after,
limit,
};
for limit in [0, 101, u32::MAX] {
assert!(matches!(
store.chat_messages_for_account("alice", &c.id, &query(None, Some(limit))),
Err(ChatError::InvalidInput(_))
));
}
let first = store
.chat_messages_for_account("alice", &c.id, &query(None, Some(100)))
.unwrap();
assert_eq!(first.messages.len(), 100);
assert!(first.has_more);
let tail = store
.chat_messages_for_account("alice", &c.id, &query(Some(100), Some(100)))
.unwrap();
assert_eq!(
tail.messages.iter().map(|m| m.sequence).collect::<Vec<_>>(),
vec![101, 102]
);
assert!(!tail.has_more);
assert_eq!(
page(&store, &c).messages.len(),
CHAT_DEFAULT_PAGE_LIMIT as usize
);
assert!(matches!(
store.chat_messages_for_account("alice", &c.id, &query(Some(u64::MAX), None)),
Err(ChatError::InvalidInput(_))
));
}
#[test]
fn observed_user_identity_fence_and_append_share_one_sqlite_transaction_across_connections() {
let temp = tempfile::tempdir().unwrap();
let (_manager, _scope, store) = open_store(temp.path(), "workspace-a");
let subject = store.create_subject(role()).unwrap().id;
let c = create(&store, "alice", &subject, "create");
let (_second_manager, _second_scope, second) = open_store(temp.path(), "workspace-a");
let barrier = Arc::new(Barrier::new(2));
let (input, output) = std::thread::scope(|s| {
let barrier = &barrier;
let c = &c;
let subject = &subject;
let first = s.spawn(move || {
barrier.wait();
store.chat_append_user_message("alice", &c.id, &user("input", "new input"))
});
let other = s.spawn(move || {
barrier.wait();
second.chat_send_message(subject, &send(c, "answer", None))
});
(first.join().unwrap().unwrap(), other.join().unwrap())
});
// A subject send that won the SQLite write lock must precede user input. If input won,
// no stale subject message can commit. Both connections must observe the same ordering.
match output {
Ok(answer) => {
assert_eq!(answer.sequence, 1);
assert_eq!(answer.last_user_message_id, None);
assert_eq!(input.sequence, 2);
}
Err(ChatError::UserInputChanged {
expected: None,
actual,
}) => {
assert_eq!(actual.as_deref(), Some(input.id.as_str()));
assert_eq!(input.sequence, 1);
}
other => panic!("unexpected race outcome: {other:?}"),
}
let (_manager, _scope, reopened) = open_store(temp.path(), "workspace-a");
assert!(matches!(
reopened.chat_send_message(&subject, &send(&c, "stale", None)),
Err(ChatError::UserInputChanged {
expected: None,
actual: Some(ref actual)
}) if actual == &input.id
));
assert_eq!(
page(&reopened, &c)
.conversation
.last_user_message_id
.as_deref(),
Some(input.id.as_str())
);
reopened
.chat_send_message(&subject, &send(&c, "fresh", Some(&input.id)))
.unwrap();
}
#[test]
fn concurrent_exact_user_retries_publish_one_identity() {
let temp = tempfile::tempdir().unwrap();
let (_manager, _scope, store) = open_store(temp.path(), "workspace-a");
let subject = store.create_subject(role()).unwrap().id;
let c = create(&store, "alice", &subject, "create");
let (_second_manager, _second_scope, second) = open_store(temp.path(), "workspace-a");
let barrier = Barrier::new(2);
let (a, b) = std::thread::scope(|s| {
let a = s.spawn(|| {
barrier.wait();
store.chat_append_user_message("alice", &c.id, &user("same", "hello"))
});
let b = s.spawn(|| {
barrier.wait();
second.chat_append_user_message("alice", &c.id, &user("same", "hello"))
});
(a.join().unwrap().unwrap(), b.join().unwrap().unwrap())
});
assert_eq!(a, b);
assert_eq!(page(&store, &c).messages.len(), 1);
assert_eq!(
page(&store, &c)
.conversation
.last_user_message_id
.as_deref(),
Some(a.id.as_str())
);
}
#[test]
fn insert_and_commit_failures_rollback_message_baseline_and_operation_identity() {
let temp = tempfile::tempdir().unwrap();
let (_manager, _scope, store) = open_store(temp.path(), "workspace-a");
let subject = store.create_subject(role()).unwrap().id;
let c = create(&store, "alice", &subject, "create");
sql(
&store,
"CREATE TRIGGER fail_chat_save BEFORE INSERT ON chat_messages BEGIN SELECT RAISE(ABORT,'injected save failure'); END;",
);
assert!(matches!(
store.chat_append_user_message("alice", &c.id, &user("retry", "hello")),
Err(ChatError::Storage(_))
));
assert_eq!(page(&store, &c).conversation.last_user_message_id, None);
assert!(page(&store, &c).messages.is_empty());
sql(&store,"DROP TRIGGER fail_chat_save;
CREATE TABLE chat_commit_failure (subject_id TEXT REFERENCES subjects(subject_id) DEFERRABLE INITIALLY DEFERRED);
CREATE TRIGGER fail_chat_commit AFTER INSERT ON chat_messages BEGIN INSERT INTO chat_commit_failure VALUES ('missing'); END;");
assert!(matches!(
store.chat_append_user_message("alice", &c.id, &user("retry", "hello")),
Err(ChatError::Storage(_))
));
assert_eq!(page(&store, &c).conversation.last_user_message_id, None);
assert!(page(&store, &c).messages.is_empty());
assert!(matches!(
store.chat_send_message(&subject, &send(&c, "retry", None)),
Err(ChatError::Storage(_))
));
sql(
&store,
"DROP TRIGGER fail_chat_commit; DROP TABLE chat_commit_failure;",
);
let first = store
.chat_append_user_message("alice", &c.id, &user("retry", "hello"))
.unwrap();
assert_eq!(first.sequence, 1);
assert_eq!(
first.last_user_message_id.as_deref(),
Some(first.id.as_str())
);
let answer = store
.chat_send_message(&subject, &send(&c, "retry", Some(&first.id)))
.unwrap();
assert_eq!(answer.sequence, 2);
assert_eq!(
answer.last_user_message_id.as_deref(),
Some(first.id.as_str())
);
}
#[test]
fn subject_erasure_fences_replay_then_purges_chat_without_touching_other_subject() {
let temp = tempfile::tempdir().unwrap();
let (_manager, _scope, store) = open_store(temp.path(), "workspace-a");
let subject = store.create_subject(role()).unwrap().id;
let other = store.create_subject(role()).unwrap().id;
let c = create(&store, "alice", &subject, "create");
let d = create(&store, "alice", &other, "other");
let first = store
.chat_append_user_message("alice", &c.id, &user("input", "erase me"))
.unwrap();
let mut request = send(&c, "reply", Some(&first.id));
request.reply_to = Some(first.id);
store.chat_send_message(&subject, &request).unwrap();
let retained = store
.chat_append_user_message("alice", &d.id, &user("input", "keep me"))
.unwrap();
store.begin_subject_deletion(&subject, "erase").unwrap();
assert!(store.chat_conversation_for_account("alice", &c.id).is_err());
assert!(
store
.chat_append_user_message("alice", &c.id, &user("input", "erase me"))
.is_err()
);
assert!(store.chat_send_message(&subject, &request).is_err());
assert!(
store
.chat_create_conversation(
"alice",
&ChatCreateConversationRequest {
operation_id: "create".into(),
subject_id: subject.clone()
}
)
.is_err()
);
store.finish_subject_deletion(&subject, "erase").unwrap();
store
.database
.with_connection(|db| {
assert_eq!(
db.query_row("SELECT COUNT(*) FROM chat_conversations", [], |r| r
.get::<_, u64>(0))?,
1
);
assert_eq!(
db.query_row("SELECT COUNT(*) FROM chat_messages", [], |r| r
.get::<_, u64>(0))?,
1
);
Ok(())
})
.unwrap();
assert_eq!(page(&store, &d).messages, vec![retained]);
}
fn schema(store: &SubjektivStore) -> Vec<(String, String, String)> {
store.database.with_connection(|db| {
Ok(db.prepare("SELECT type,name,sql FROM sqlite_master WHERE sql IS NOT NULL AND name NOT LIKE 'sqlite_%' ORDER BY type,name")?
.query_map([],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?)))?.collect::<std::result::Result<Vec<_>,_>>()?)
}).unwrap()
}
#[test]
fn oldest_supported_store_upgrades_to_fresh_schema_preserving_subjects_and_constraints() {
let temp = tempfile::tempdir().unwrap();
let subject = {
let manager = FeatureStorage::new(temp.path());
let scope = manager.workspace("upgraded").unwrap();
let registered = scope
.register(FeatureRegistration::new(
crate::SUBJEKTIV_FEATURE_ID,
&crate::MIGRATIONS[..1],
))
.unwrap();
let old = SubjektivStore::open(&scope, &registered).unwrap();
let subject = crate::SubjectRecord {
schema_version: crate::SUBJEKTIV_SCHEMA_VERSION,
id: "preserved-subject".into(),
role: role(),
behavior_md: String::new(),
state: crate::SubjectState::Active,
created_at: crate::now(),
updated_at: crate::now(),
};
old.database
.with_connection(|db| {
db.execute(
"INSERT INTO subjects VALUES (?1,?2,'active',0,?3,?4,?5)",
params![
subject.id,
subject.role.as_str(),
serde_json::to_string(&subject).unwrap(),
subject.created_at,
subject.updated_at
],
)?;
Ok(())
})
.unwrap();
subject
};
let (_manager, _scope, upgraded) = open_store(temp.path(), "upgraded");
let (_fresh_manager, _fresh_scope, fresh) = open_store(temp.path(), "fresh");
assert_eq!(
upgraded.subject(&subject.id).unwrap(),
Some(subject.clone())
);
assert_eq!(schema(&upgraded), schema(&fresh));
let c = create(&upgraded, "alice", &subject.id, "create");
upgraded
.chat_append_user_message("alice", &c.id, &user("input", "preserved subject"))
.unwrap();
let write = upgraded.database.with_connection(|db| {
db.execute("UPDATE chat_messages SET text='tampered'", [])?;
Ok(())
});
assert!(write.is_err());
let delete = upgraded.database.with_connection(|db| {
db.execute("DELETE FROM chat_messages", [])?;
Ok(())
});
assert!(delete.is_err());
assert_eq!(page(&upgraded, &c).messages[0].text, "preserved subject");
}
#[test]
fn failed_chat_migration_rolls_back_all_ddl_and_can_be_retried() {
let connection = Connection::open_in_memory().unwrap();
connection.execute_batch("PRAGMA foreign_keys=ON; CREATE TABLE subjects(subject_id TEXT PRIMARY KEY); CREATE TABLE subject_deletions(subject_id TEXT PRIMARY KEY); CREATE TABLE chat_messages(blocker TEXT);").unwrap();
{
let tx = connection.unchecked_transaction().unwrap();
assert!(migrate(&tx).is_err());
}
let exists: bool = connection
.query_row(
"SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE name='chat_conversations')",
[],
|r| r.get(0),
)
.unwrap();
assert!(!exists);
connection
.execute_batch("DROP TABLE chat_messages;")
.unwrap();
let tx = connection.unchecked_transaction().unwrap();
migrate(&tx).unwrap();
tx.commit().unwrap();
assert!(
connection
.execute(
"INSERT INTO chat_conversations VALUES ('c','missing','a',NULL,'op')",
[]
)
.is_err()
);
}
fn policy(
store: &SubjektivStore,
subject: &str,
operation: &str,
expected: Option<&str>,
) -> String {
store
.set_subject_permissions(
subject,
&server_api::authorization::SubjectPermissionsUpdate {
operation_id: operation.into(),
expected_change_id: expected.map(str::to_owned),
execution: Vec::new(),
delegation: Vec::new(),
},
)
.unwrap()
.change_id
}
#[test]
fn missing_or_replaced_policy_refuses_read_append_and_saved_replay() {
let temp = tempfile::tempdir().unwrap();
let (_manager, _scope, store) = open_store(temp.path(), "workspace-a");
let subject = store.create_subject(role()).unwrap().id;
let c = create(&store, "alice", &subject, "create");
let request = send(&c, "answer", None);
let query = ChatMessagePageQuery::default();
assert!(matches!(
store.chat_messages_for_subject_authorized(&subject, &c.id, &query, "missing"),
Err(ChatError::AuthorizationChanged)
));
assert!(matches!(
store.chat_send_message_authorized(&subject, &request, "missing"),
Err(ChatError::AuthorizationChanged)
));
let observed = policy(&store, &subject, "initial", None);
let answer = store
.chat_send_message_authorized(&subject, &request, &observed)
.unwrap();
assert_eq!(
store
.chat_messages_for_subject_authorized(&subject, &c.id, &query, &observed)
.unwrap()
.messages,
vec![answer.clone()]
);
let (_other_manager, _other_scope, other) = open_store(temp.path(), "workspace-a");
// An identical replacement still invalidates the Host's earlier observation.
let current = policy(&other, &subject, "replace", Some(&observed));
assert_ne!(current, observed);
assert!(matches!(
store.chat_messages_for_subject_authorized(&subject, &c.id, &query, &observed),
Err(ChatError::AuthorizationChanged)
));
assert!(matches!(
store.chat_send_message_authorized(&subject, &request, &observed),
Err(ChatError::AuthorizationChanged)
));
assert!(matches!(
store.chat_send_message_authorized(&subject, &send(&c, "new", None), &observed),
Err(ChatError::AuthorizationChanged)
));
assert_eq!(page(&store, &c).messages, vec![answer.clone()]);
assert_eq!(
store
.chat_send_message_authorized(&subject, &request, &current)
.unwrap(),
answer
);
}
#[test]
fn policy_replacement_and_send_serialize_across_sqlite_connections() {
let temp = tempfile::tempdir().unwrap();
let (_manager, _scope, store) = open_store(temp.path(), "workspace-a");
let subject = store.create_subject(role()).unwrap().id;
let c = create(&store, "alice", &subject, "create");
let observed = policy(&store, &subject, "initial", None);
let (_other_manager, _other_scope, other) = open_store(temp.path(), "workspace-a");
let barrier = Barrier::new(2);
let (result, current) = std::thread::scope(|s| {
let send = s.spawn(|| {
barrier.wait();
store.chat_send_message_authorized(&subject, &send(&c, "answer", None), &observed)
});
let replace = s.spawn(|| {
barrier.wait();
policy(&other, &subject, "replace", Some(&observed))
});
(send.join().unwrap(), replace.join().unwrap())
});
let saved = page(&store, &c).messages;
match result {
Ok(answer) => assert_eq!(saved, vec![answer]),
Err(ChatError::AuthorizationChanged) => assert!(saved.is_empty()),
other => panic!("unexpected policy race outcome: {other:?}"),
}
assert!(matches!(
store.chat_send_message_authorized(&subject, &send(&c, "answer", None), &observed),
Err(ChatError::AuthorizationChanged)
));
assert!(matches!(
store.chat_messages_for_subject_authorized(
&subject,
&c.id,
&ChatMessagePageQuery::default(),
&observed
),
Err(ChatError::AuthorizationChanged)
));
assert_eq!(
store
.chat_messages_for_subject_authorized(
&subject,
&c.id,
&ChatMessagePageQuery::default(),
&current
)
.unwrap()
.messages,
saved
);
}
#[test]
fn null_baseline_replay_survives_user_input_and_reply_target_is_independent() {
let temp = tempfile::tempdir().unwrap();
let (_manager, _scope, store) = open_store(temp.path(), "workspace-a");
let subject = store.create_subject(role()).unwrap().id;
let c = create(&store, "alice", &subject, "create");
let opening = send(&c, "opening", None);
let published = store.chat_send_message(&subject, &opening).unwrap();
let first = store
.chat_append_user_message("alice", &c.id, &user("first", "first"))
.unwrap();
let latest = store
.chat_append_user_message("alice", &c.id, &user("latest", "new input"))
.unwrap();
assert_eq!(
store.chat_send_message(&subject, &opening).unwrap(),
published
);
let mut changed = opening.clone();
changed.expected_last_user_message_id = Some(latest.id.clone());
assert!(matches!(
store.chat_send_message(&subject, &changed),
Err(ChatError::OperationConflict(_))
));
let mut request = send(&c, "reply", Some(&latest.id));
request.reply_to = Some(first.id.clone());
let answer = store.chat_send_message(&subject, &request).unwrap();
assert_eq!(answer.reply_to, Some(first.id));
assert_eq!(answer.last_user_message_id, Some(latest.id));
// A subject-authored ID is never a user-input baseline.
let bad = send(&c, "wrong", Some(&answer.id));
assert!(
matches!(store.chat_send_message(&subject, &bad), Err(ChatError::UserInputChanged { expected, actual }) if expected == Some(answer.id.clone()) && actual == answer.last_user_message_id)
);
assert_eq!(page(&store, &c).messages.len(), 4);
}
#[test]
fn schema_rejects_dangling_cross_conversation_or_subject_authored_user_identity() {
let temp = tempfile::tempdir().unwrap();
let (_manager, _scope, store) = open_store(temp.path(), "workspace-a");
let subject = store.create_subject(role()).unwrap().id;
let c = create(&store, "alice", &subject, "create");
let other = create(&store, "alice", &subject, "other");
let input = store
.chat_append_user_message("alice", &other.id, &user("input", "other conversation"))
.unwrap();
let answer = store
.chat_send_message(&subject, &send(&c, "answer", None))
.unwrap();
for target in [&input.id, &answer.id, "missing"] {
let result = store.database.transaction(|tx| {
tx.execute(
"UPDATE chat_conversations SET last_user_message_id=?1 WHERE id=?2",
params![target, c.id],
)?;
Ok(())
});
assert!(result.is_err(), "must reject user identity {target}");
}
let result = store.database.transaction(|tx| {
tx.execute("UPDATE chat_conversations SET last_user_message_id='fresh-subject-message' WHERE id=?1", [&c.id])?;
tx.execute("INSERT INTO chat_messages (id,conversation_id,sequence,sender_kind,sender_id,text,operation_id,last_user_message_id) VALUES ('fresh-subject-message',?1,2,'subject',?2,'bad','direct','fresh-subject-message')", params![c.id, subject])?;
Ok(())
});
assert!(result.is_err());
let result = store.database.transaction(|tx| {
tx.execute("INSERT INTO chat_messages (id,conversation_id,sequence,sender_kind,sender_id,text,operation_id,last_user_message_id) VALUES ('bad-user',?1,2,'user','alice','bad','direct',NULL)", [&c.id])?;
Ok(())
});
assert!(result.is_err());
assert_eq!(page(&store, &c).conversation.last_user_message_id, None);
assert_eq!(page(&store, &c).messages, vec![answer]);
}
#[test]
fn previous_schema_upgrade_preserves_policy_and_matches_fresh_database() {
let temp = tempfile::tempdir().unwrap();
let (subject, observed) = {
let manager = FeatureStorage::new(temp.path());
let scope = manager.workspace("upgraded").unwrap();
let registered = scope
.register(FeatureRegistration::new(
crate::SUBJEKTIV_FEATURE_ID,
&crate::MIGRATIONS[..14],
))
.unwrap();
let old = SubjektivStore::open(&scope, &registered).unwrap();
let subject = old
.create_subject_with_behavior(role(), "preserved behavior".into())
.unwrap();
let observed = policy(&old, &subject.id, "initial", None);
(subject, observed)
};
let (_manager, _scope, upgraded) = open_store(temp.path(), "upgraded");
let (_fresh_manager, _fresh_scope, fresh) = open_store(temp.path(), "fresh");
assert_eq!(schema(&upgraded), schema(&fresh));
assert_eq!(
upgraded.subject(&subject.id).unwrap(),
Some(subject.clone())
);
assert_eq!(
upgraded
.subject_permissions(&subject.id)
.unwrap()
.unwrap()
.change_id,
observed
);
let c = create(&upgraded, "alice", &subject.id, "create");
let input = upgraded
.chat_append_user_message("alice", &c.id, &user("input", "preserved subject"))
.unwrap();
upgraded
.chat_send_message_authorized(&subject.id, &send(&c, "answer", Some(&input.id)), &observed)
.unwrap();
assert_eq!(
page(&upgraded, &c).conversation.last_user_message_id,
Some(input.id)
);
}
#[test]
fn policy_replacement_and_read_serialize_across_sqlite_connections() {
let temp = tempfile::tempdir().unwrap();
let (_manager, _scope, store) = open_store(temp.path(), "workspace-a");
let subject = store.create_subject(role()).unwrap().id;
let c = create(&store, "alice", &subject, "create");
let input = store
.chat_append_user_message("alice", &c.id, &user("input", "private input"))
.unwrap();
let observed = policy(&store, &subject, "initial", None);
let (_other_manager, _other_scope, other) = open_store(temp.path(), "workspace-a");
let barrier = Barrier::new(2);
let (result, current) = std::thread::scope(|s| {
let read = s.spawn(|| {
barrier.wait();
store.chat_messages_for_subject_authorized(
&subject,
&c.id,
&ChatMessagePageQuery::default(),
&observed,
)
});
let replace = s.spawn(|| {
barrier.wait();
policy(&other, &subject, "replace", Some(&observed))
});
(read.join().unwrap(), replace.join().unwrap())
});
match result {
Ok(page) => assert_eq!(page.messages, vec![input.clone()]),
Err(ChatError::AuthorizationChanged) => {}
other => panic!("unexpected policy race outcome: {other:?}"),
}
assert!(matches!(
store.chat_messages_for_subject_authorized(
&subject,
&c.id,
&ChatMessagePageQuery::default(),
&observed
),
Err(ChatError::AuthorizationChanged)
));
assert_eq!(
store
.chat_messages_for_subject_authorized(
&subject,
&c.id,
&ChatMessagePageQuery::default(),
&current
)
.unwrap()
.messages,
vec![input]
);
}
crates/subjektiv/src/deletion.rs
+1
View File
@@ -391,6 +391,7 @@ impl SubjektivStore {
return Ok(saved);
}
tx.execute_batch("PRAGMA defer_foreign_keys=ON;")?;
crate::chat::erase_subject(tx, subject)?;
for (table, predicate) in INDIRECT {
tx.execute(
&format!(
crates/subjektiv/src/deletion/tests.rs
+5 -1
View File
@@ -273,12 +273,16 @@ fn every_subject_owned_table_has_old_and_new_sql_fences_and_retention_is_scoped(
for table in tables {
let columns = c.prepare(&format!("PRAGMA table_info({table})"))?.query_map([], |r| r.get::<_,String>(1))?.collect::<std::result::Result<Vec<_>,_>>()?;
if !columns.iter().any(|v| v == "subject_id") || table.starts_with("subject_delet") { continue; }
assert!(DIRECT.contains(&table.as_str()) || table == "journal_erased_sessions", "unclassified Subject table: {table}");
assert!(DIRECT.contains(&table.as_str()) || table == "journal_erased_sessions" || table == "chat_conversations", "unclassified Subject table: {table}");
}
for table in DIRECT.iter().copied().chain(INDIRECT.iter().map(|v| v.0)).chain(["journal_erased_sessions"]) {
let sql: String = c.query_row("SELECT sql FROM sqlite_master WHERE name=?1", [format!("deletion_fence_{table}_UPDATE")], |r| r.get(0))?;
assert!(sql.contains("OLD.") && sql.contains("NEW."), "{table}: {sql}");
}
// Chat owns its later migration; keep the earlier deletion migration unchanged.
let chat_guard: String = c.query_row("SELECT sql FROM sqlite_master WHERE name='chat_conversations_update_fence'", [], |r| r.get(0))?;
assert!(chat_guard.contains("OLD.subject_id") && chat_guard.contains("NEW.subject_id"));
assert!(c.execute("INSERT INTO chat_conversations VALUES ('late',?1,'alice',NULL,'late')", [&a]).is_err());
// OLD fenced -> NEW unfenced must not evade a guard.
assert!(c.execute("UPDATE subjects SET subject_id=?2 WHERE subject_id=?1", params![a, "new-id"]).is_err());
assert!(c.execute("UPDATE subjects SET subject_id=?2 WHERE subject_id=?1", params![b, a]).is_err());
crates/subjektiv/src/lib.rs
+2
View File
@@ -1,6 +1,7 @@
#![doc = include_str!("../README.md")]
pub mod api;
pub mod chat;
pub mod deletion;
pub mod job;
pub use deletion::{DeletionSubjectJob, SubjectDeletion};
@@ -656,6 +657,7 @@ static MIGRATIONS: &[FeatureMigration] = &[
"add permanent Subject deletion fences and explicit erasure",
deletion::migrate,
),
FeatureMigration::new(15, "add retained text-only conversations", chat::migrate),
];
pub const REGISTRATION: FeatureRegistration =