//! Repository pattern for file database operations.
use chrono::Utc;
use serde::Deserialize;
use sqlx::PgPool;
use uuid::Uuid;
use super::models::{
// Core types
CheckpointPatch, CheckpointPatchInfo, Contract, ContractChatConversation,
ContractChatMessageRecord, ContractEvent, ContractRepository, ContractSummary,
ContractTypeTemplateRecord, ConversationMessage, ConversationSnapshot,
CreateContractRequest, CreateFileRequest, CreateTaskRequest, CreateTemplateRequest,
Daemon, DaemonTaskAssignment, DaemonWithCapacity, DeliverableDefinition,
File, FileSummary, FileVersion, HistoryEvent, HistoryQueryFilters,
MeshChatConversation, MeshChatMessageRecord, PhaseChangeResult, PhaseConfig,
PhaseDefinition, SupervisorHeartbeatRecord, SupervisorState, Task, TaskCheckpoint,
TaskEvent, TaskSummary, UpdateContractRequest, UpdateFileRequest, UpdateTaskRequest,
UpdateTemplateRequest,
// Directive types
AddStepRequest, ChainStep, CreateDirectiveRequest, Directive, DirectiveApproval,
DirectiveChain, DirectiveChainGraphEdge, DirectiveChainGraphNode, DirectiveChainGraphResponse,
DirectiveEvaluation, DirectiveEvent, DirectiveSummary, DirectiveVerifier,
DirectiveWithProgress, UpdateDirectiveRequest, UpdateStepRequest,
};
/// Repository error types.
#[derive(Debug)]
pub enum RepositoryError {
/// Database error
Database(sqlx::Error),
/// Version conflict (optimistic locking failure)
VersionConflict {
/// The version the client expected
expected: i32,
/// The actual current version in the database
actual: i32,
},
}
impl From<sqlx::Error> for RepositoryError {
fn from(e: sqlx::Error) -> Self {
RepositoryError::Database(e)
}
}
impl std::fmt::Display for RepositoryError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
RepositoryError::Database(e) => write!(f, "Database error: {}", e),
RepositoryError::VersionConflict { expected, actual } => {
write!(
f,
"Version conflict: expected {}, actual {}",
expected, actual
)
}
}
}
}
impl std::error::Error for RepositoryError {}
/// Generate a default name based on current timestamp.
fn generate_default_name() -> String {
let now = Utc::now();
now.format("Recording - %b %d %Y %H:%M:%S").to_string()
}
/// Internal request for creating files without contract association (e.g., audio transcription).
/// User-facing file creation should use CreateFileRequest which requires contract_id.
pub struct InternalCreateFileRequest {
pub name: Option<String>,
pub description: Option<String>,
pub transcript: Vec<super::models::TranscriptEntry>,
pub location: Option<String>,
}
/// Create a new file record (internal use, no contract required).
/// For user-facing file creation, use create_file_for_owner which requires a contract.
pub async fn create_file(pool: &PgPool, req: InternalCreateFileRequest) -> Result<File, sqlx::Error> {
let name = req.name.unwrap_or_else(generate_default_name);
let transcript_json = serde_json::to_value(&req.transcript).unwrap_or_default();
let body_json = serde_json::to_value::<Vec<super::models::BodyElement>>(vec![]).unwrap();
sqlx::query_as::<_, File>(
r#"
INSERT INTO files (name, description, transcript, location, summary, body)
VALUES ($1, $2, $3, $4, NULL, $5)
RETURNING id, owner_id, contract_id, contract_phase, name, description, transcript, location, summary, body, version, repo_file_path, repo_synced_at, repo_sync_status, created_at, updated_at
"#,
)
.bind(&name)
.bind(&req.description)
.bind(&transcript_json)
.bind(&req.location)
.bind(&body_json)
.fetch_one(pool)
.await
}
/// Get a file by ID.
pub async fn get_file(pool: &PgPool, id: Uuid) -> Result<Option<File>, sqlx::Error> {
sqlx::query_as::<_, File>(
r#"
SELECT id, owner_id, contract_id, contract_phase, name, description, transcript, location, summary, body, version, repo_file_path, repo_synced_at, repo_sync_status, created_at, updated_at
FROM files
WHERE id = $1
"#,
)
.bind(id)
.fetch_optional(pool)
.await
}
/// List all files, ordered by created_at DESC.
pub async fn list_files(pool: &PgPool) -> Result<Vec<File>, sqlx::Error> {
sqlx::query_as::<_, File>(
r#"
SELECT id, owner_id, contract_id, contract_phase, name, description, transcript, location, summary, body, version, repo_file_path, repo_synced_at, repo_sync_status, created_at, updated_at
FROM files
ORDER BY created_at DESC
"#,
)
.fetch_all(pool)
.await
}
/// Update a file by ID with optimistic locking.
///
/// If `req.version` is provided, the update will only succeed if the current
/// version matches. Returns `RepositoryError::VersionConflict` if there's a mismatch.
///
/// If `req.version` is None (e.g., internal system updates), version checking is skipped.
pub async fn update_file(
pool: &PgPool,
id: Uuid,
req: UpdateFileRequest,
) -> Result<Option<File>, RepositoryError> {
// Get the existing file first
let existing = get_file(pool, id).await?;
let Some(existing) = existing else {
return Ok(None);
};
// Check version if provided (optimistic locking)
if let Some(expected_version) = req.version {
if existing.version != expected_version {
return Err(RepositoryError::VersionConflict {
expected: expected_version,
actual: existing.version,
});
}
}
// Apply updates
let name = req.name.unwrap_or(existing.name);
let description = req.description.or(existing.description);
let transcript = req.transcript.unwrap_or(existing.transcript);
let transcript_json = serde_json::to_value(&transcript).unwrap_or_default();
let summary = req.summary.or(existing.summary);
let body = req.body.unwrap_or(existing.body);
let body_json = serde_json::to_value(&body).unwrap_or_default();
// Update with version check in WHERE clause for race condition safety
let result = if req.version.is_some() {
sqlx::query_as::<_, File>(
r#"
UPDATE files
SET name = $2, description = $3, transcript = $4, summary = $5, body = $6, updated_at = NOW()
WHERE id = $1 AND version = $7
RETURNING id, owner_id, contract_id, contract_phase, name, description, transcript, location, summary, body, version, repo_file_path, repo_synced_at, repo_sync_status, created_at, updated_at
"#,
)
.bind(id)
.bind(&name)
.bind(&description)
.bind(&transcript_json)
.bind(&summary)
.bind(&body_json)
.bind(req.version.unwrap())
.fetch_optional(pool)
.await?
} else {
// No version check for internal updates
sqlx::query_as::<_, File>(
r#"
UPDATE files
SET name = $2, description = $3, transcript = $4, summary = $5, body = $6, updated_at = NOW()
WHERE id = $1
RETURNING id, owner_id, contract_id, contract_phase, name, description, transcript, location, summary, body, version, repo_file_path, repo_synced_at, repo_sync_status, created_at, updated_at
"#,
)
.bind(id)
.bind(&name)
.bind(&description)
.bind(&transcript_json)
.bind(&summary)
.bind(&body_json)
.fetch_optional(pool)
.await?
};
// If versioned update returned None, there was a race condition
if result.is_none() && req.version.is_some() {
// Re-fetch to get the actual version
if let Some(current) = get_file(pool, id).await? {
return Err(RepositoryError::VersionConflict {
expected: req.version.unwrap(),
actual: current.version,
});
}
}
Ok(result)
}
/// Delete a file by ID.
pub async fn delete_file(pool: &PgPool, id: Uuid) -> Result<bool, sqlx::Error> {
let result = sqlx::query(
r#"
DELETE FROM files
WHERE id = $1
"#,
)
.bind(id)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Count total files.
pub async fn count_files(pool: &PgPool) -> Result<i64, sqlx::Error> {
let result: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM files")
.fetch_one(pool)
.await?;
Ok(result.0)
}
// =============================================================================
// Owner-Scoped File Functions
// =============================================================================
/// Create a new file record for a specific owner.
/// Files must belong to a contract - the contract_id is required and the phase is looked up.
pub async fn create_file_for_owner(
pool: &PgPool,
owner_id: Uuid,
req: CreateFileRequest,
) -> Result<File, sqlx::Error> {
let name = req.name.unwrap_or_else(generate_default_name);
let transcript_json = serde_json::to_value(&req.transcript).unwrap_or_default();
// Use body from request (may be empty or contain template elements)
let body_json = serde_json::to_value(&req.body).unwrap_or_default();
// Use provided contract_phase, or look up from contract's current phase
let contract_phase: Option<String> = if req.contract_phase.is_some() {
req.contract_phase
} else {
sqlx::query_scalar(
"SELECT phase FROM contracts WHERE id = $1 AND owner_id = $2",
)
.bind(req.contract_id)
.bind(owner_id)
.fetch_optional(pool)
.await?
};
sqlx::query_as::<_, File>(
r#"
INSERT INTO files (owner_id, contract_id, contract_phase, name, description, transcript, location, summary, body, repo_file_path)
VALUES ($1, $2, $3, $4, $5, $6, $7, NULL, $8, $9)
RETURNING id, owner_id, contract_id, contract_phase, name, description, transcript, location, summary, body, version, repo_file_path, repo_synced_at, repo_sync_status, created_at, updated_at
"#,
)
.bind(owner_id)
.bind(req.contract_id)
.bind(&contract_phase)
.bind(&name)
.bind(&req.description)
.bind(&transcript_json)
.bind(&req.location)
.bind(&body_json)
.bind(&req.repo_file_path)
.fetch_one(pool)
.await
}
/// Get a file by ID, scoped to owner.
pub async fn get_file_for_owner(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
) -> Result<Option<File>, sqlx::Error> {
sqlx::query_as::<_, File>(
r#"
SELECT id, owner_id, contract_id, contract_phase, name, description, transcript, location, summary, body, version, repo_file_path, repo_synced_at, repo_sync_status, created_at, updated_at
FROM files
WHERE id = $1 AND owner_id = $2
"#,
)
.bind(id)
.bind(owner_id)
.fetch_optional(pool)
.await
}
/// List all files for an owner, ordered by created_at DESC.
pub async fn list_files_for_owner(pool: &PgPool, owner_id: Uuid) -> Result<Vec<File>, sqlx::Error> {
sqlx::query_as::<_, File>(
r#"
SELECT id, owner_id, contract_id, contract_phase, name, description, transcript, location, summary, body, version, repo_file_path, repo_synced_at, repo_sync_status, created_at, updated_at
FROM files
WHERE owner_id = $1
ORDER BY created_at DESC
"#,
)
.bind(owner_id)
.fetch_all(pool)
.await
}
/// Database row type for file summary with contract info
#[derive(Debug, sqlx::FromRow)]
struct FileSummaryRow {
id: Uuid,
contract_id: Option<Uuid>,
contract_name: Option<String>,
contract_phase: Option<String>,
name: String,
description: Option<String>,
#[sqlx(json)]
transcript: Vec<crate::db::models::TranscriptEntry>,
version: i32,
repo_file_path: Option<String>,
repo_sync_status: Option<String>,
created_at: chrono::DateTime<chrono::Utc>,
updated_at: chrono::DateTime<chrono::Utc>,
}
/// List file summaries for an owner with contract info (joined).
pub async fn list_file_summaries_for_owner(
pool: &PgPool,
owner_id: Uuid,
) -> Result<Vec<FileSummary>, sqlx::Error> {
let rows = sqlx::query_as::<_, FileSummaryRow>(
r#"
SELECT
f.id, f.contract_id, c.name as contract_name, f.contract_phase,
f.name, f.description, f.transcript, f.version,
f.repo_file_path, f.repo_sync_status, f.created_at, f.updated_at
FROM files f
LEFT JOIN contracts c ON f.contract_id = c.id
WHERE f.owner_id = $1
ORDER BY f.created_at DESC
"#,
)
.bind(owner_id)
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|row| {
let duration = row
.transcript
.iter()
.map(|t| t.end)
.fold(0.0_f32, f32::max);
FileSummary {
id: row.id,
contract_id: row.contract_id,
contract_name: row.contract_name,
contract_phase: row.contract_phase,
name: row.name,
description: row.description,
transcript_count: row.transcript.len(),
duration: if duration > 0.0 { Some(duration) } else { None },
version: row.version,
repo_file_path: row.repo_file_path,
repo_sync_status: row.repo_sync_status,
created_at: row.created_at,
updated_at: row.updated_at,
}
})
.collect())
}
/// Update a file by ID with optimistic locking, scoped to owner.
pub async fn update_file_for_owner(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
req: UpdateFileRequest,
) -> Result<Option<File>, RepositoryError> {
// Get the existing file first (scoped to owner)
let existing = get_file_for_owner(pool, id, owner_id).await?;
let Some(existing) = existing else {
return Ok(None);
};
// Check version if provided (optimistic locking)
if let Some(expected_version) = req.version {
if existing.version != expected_version {
return Err(RepositoryError::VersionConflict {
expected: expected_version,
actual: existing.version,
});
}
}
// Apply updates
let name = req.name.unwrap_or(existing.name);
let description = req.description.or(existing.description);
let transcript = req.transcript.unwrap_or(existing.transcript);
let transcript_json = serde_json::to_value(&transcript).unwrap_or_default();
let summary = req.summary.or(existing.summary);
let body = req.body.unwrap_or(existing.body);
let body_json = serde_json::to_value(&body).unwrap_or_default();
// Update with version check in WHERE clause for race condition safety
let result = if req.version.is_some() {
sqlx::query_as::<_, File>(
r#"
UPDATE files
SET name = $3, description = $4, transcript = $5, summary = $6, body = $7, updated_at = NOW()
WHERE id = $1 AND owner_id = $2 AND version = $8
RETURNING id, owner_id, contract_id, contract_phase, name, description, transcript, location, summary, body, version, repo_file_path, repo_synced_at, repo_sync_status, created_at, updated_at
"#,
)
.bind(id)
.bind(owner_id)
.bind(&name)
.bind(&description)
.bind(&transcript_json)
.bind(&summary)
.bind(&body_json)
.bind(req.version.unwrap())
.fetch_optional(pool)
.await?
} else {
// No version check for internal updates
sqlx::query_as::<_, File>(
r#"
UPDATE files
SET name = $3, description = $4, transcript = $5, summary = $6, body = $7, updated_at = NOW()
WHERE id = $1 AND owner_id = $2
RETURNING id, owner_id, contract_id, contract_phase, name, description, transcript, location, summary, body, version, repo_file_path, repo_synced_at, repo_sync_status, created_at, updated_at
"#,
)
.bind(id)
.bind(owner_id)
.bind(&name)
.bind(&description)
.bind(&transcript_json)
.bind(&summary)
.bind(&body_json)
.fetch_optional(pool)
.await?
};
// If versioned update returned None, there was a race condition
if result.is_none() && req.version.is_some() {
// Re-fetch to get the actual version
if let Some(current) = get_file_for_owner(pool, id, owner_id).await? {
return Err(RepositoryError::VersionConflict {
expected: req.version.unwrap(),
actual: current.version,
});
}
}
Ok(result)
}
/// Delete a file by ID, scoped to owner.
pub async fn delete_file_for_owner(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
) -> Result<bool, sqlx::Error> {
let result = sqlx::query(
r#"
DELETE FROM files
WHERE id = $1 AND owner_id = $2
"#,
)
.bind(id)
.bind(owner_id)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
// =============================================================================
// Version History Functions
// =============================================================================
/// Set the version source for the current transaction.
/// This is used by the trigger to record who made the change.
pub async fn set_version_source(pool: &PgPool, source: &str) -> Result<(), sqlx::Error> {
sqlx::query(&format!("SET LOCAL app.version_source = '{}'", source))
.execute(pool)
.await?;
Ok(())
}
/// Set the change description for the current transaction.
pub async fn set_change_description(pool: &PgPool, description: &str) -> Result<(), sqlx::Error> {
// Escape single quotes for SQL
let escaped = description.replace('\'', "''");
sqlx::query(&format!("SET LOCAL app.change_description = '{}'", escaped))
.execute(pool)
.await?;
Ok(())
}
/// List all versions of a file, ordered by version DESC.
pub async fn list_file_versions(pool: &PgPool, file_id: Uuid) -> Result<Vec<FileVersion>, sqlx::Error> {
// First get the current version from the files table
let current = get_file(pool, file_id).await?;
let mut versions = sqlx::query_as::<_, FileVersion>(
r#"
SELECT id, file_id, version, name, description, summary, body, source, change_description, created_at
FROM file_versions
WHERE file_id = $1
ORDER BY version DESC
"#,
)
.bind(file_id)
.fetch_all(pool)
.await?;
// Add the current version as the first entry if it exists
if let Some(file) = current {
let current_version = FileVersion {
id: file.id,
file_id: file.id,
version: file.version,
name: file.name,
description: file.description,
summary: file.summary,
body: file.body,
source: "user".to_string(), // Current version source
change_description: None,
created_at: file.updated_at,
};
versions.insert(0, current_version);
}
Ok(versions)
}
/// Get a specific version of a file.
pub async fn get_file_version(
pool: &PgPool,
file_id: Uuid,
version: i32,
) -> Result<Option<FileVersion>, sqlx::Error> {
// First check if this is the current version
if let Some(file) = get_file(pool, file_id).await? {
if file.version == version {
return Ok(Some(FileVersion {
id: file.id,
file_id: file.id,
version: file.version,
name: file.name,
description: file.description,
summary: file.summary,
body: file.body,
source: "user".to_string(),
change_description: None,
created_at: file.updated_at,
}));
}
}
// Otherwise, look in the versions table
sqlx::query_as::<_, FileVersion>(
r#"
SELECT id, file_id, version, name, description, summary, body, source, change_description, created_at
FROM file_versions
WHERE file_id = $1 AND version = $2
"#,
)
.bind(file_id)
.bind(version)
.fetch_optional(pool)
.await
}
/// Restore a file to a previous version.
/// This creates a new version with the content from the target version.
pub async fn restore_file_version(
pool: &PgPool,
file_id: Uuid,
target_version: i32,
current_version: i32,
) -> Result<Option<File>, RepositoryError> {
// Get the target version content
let target = get_file_version(pool, file_id, target_version).await?;
let Some(target) = target else {
return Ok(None);
};
// Set version source and description for the trigger
set_version_source(pool, "system").await?;
set_change_description(pool, &format!("Restored from version {}", target_version)).await?;
// Update the file with the target version's content
// This will trigger the save_file_version trigger to save the current state first
let update_req = UpdateFileRequest {
name: Some(target.name),
description: target.description,
transcript: None,
summary: target.summary,
body: Some(target.body),
version: Some(current_version),
repo_file_path: None,
};
update_file(pool, file_id, update_req).await
}
/// Count versions for a file.
pub async fn count_file_versions(pool: &PgPool, file_id: Uuid) -> Result<i64, sqlx::Error> {
let result: (i64,) = sqlx::query_as(
"SELECT COUNT(*) + 1 FROM file_versions WHERE file_id = $1", // +1 for current version
)
.bind(file_id)
.fetch_one(pool)
.await?;
Ok(result.0)
}
// =============================================================================
// Task Functions
// =============================================================================
/// Create a new task.
///
/// If creating a subtask (parent_task_id is set) and repository settings are not provided,
/// the subtask will inherit repository_url, base_branch, target_branch, merge_mode,
/// and target_repo_path from the parent task. Depth is calculated from parent and limited
/// to max 1 (2 levels: orchestrator at depth 0, subtasks at depth 1).
///
/// NOTE: completion_action is NOT inherited - subtasks should not auto-merge unless
/// explicitly configured. The supervisor controls when completion steps happen.
///
/// Task spawning is now controlled by supervisors at the application level.
/// Depth is no longer constrained in the database.
pub async fn create_task(pool: &PgPool, req: CreateTaskRequest) -> Result<Task, sqlx::Error> {
// Calculate depth and inherit settings from parent if applicable
let (depth, contract_id, repo_url, base_branch, target_branch, merge_mode, target_repo_path, completion_action) =
if let Some(parent_id) = req.parent_task_id {
// Fetch parent task to get depth and inherit settings
let parent = get_task(pool, parent_id).await?
.ok_or_else(|| sqlx::Error::RowNotFound)?;
let new_depth = parent.depth + 1;
// Subtasks inherit contract_id from parent (or use request contract_id if parent has none)
let contract_id = parent.contract_id.or(req.contract_id);
// Inherit repo settings if not provided
let repo_url = req.repository_url.clone().or(parent.repository_url);
let base_branch = req.base_branch.clone().or(parent.base_branch);
let target_branch = req.target_branch.clone().or(parent.target_branch);
let merge_mode = req.merge_mode.clone().or(parent.merge_mode);
let target_repo_path = req.target_repo_path.clone().or(parent.target_repo_path);
// NOTE: completion_action is NOT inherited - subtasks should not auto-merge.
// The supervisor integrates subtask work from their worktrees.
let completion_action = req.completion_action.clone();
(new_depth, contract_id, repo_url, base_branch, target_branch, merge_mode, target_repo_path, completion_action)
} else {
// Top-level task: depth 0, use contract_id from request (may be None for branched tasks)
(
0,
req.contract_id,
req.repository_url.clone(),
req.base_branch.clone(),
req.target_branch.clone(),
req.merge_mode.clone(),
req.target_repo_path.clone(),
req.completion_action.clone(),
)
};
let copy_files_json = req.copy_files.as_ref().map(|f| serde_json::to_value(f).unwrap_or_default());
sqlx::query_as::<_, Task>(
r#"
INSERT INTO tasks (
contract_id, parent_task_id, depth, name, description, plan, priority,
is_supervisor, repository_url, base_branch, target_branch, merge_mode,
target_repo_path, completion_action, continue_from_task_id, copy_files,
branched_from_task_id, conversation_state, supervisor_worktree_task_id
)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)
RETURNING *
"#,
)
.bind(contract_id)
.bind(req.parent_task_id)
.bind(depth)
.bind(&req.name)
.bind(&req.description)
.bind(&req.plan)
.bind(req.priority)
.bind(req.is_supervisor)
.bind(&repo_url)
.bind(&base_branch)
.bind(&target_branch)
.bind(&merge_mode)
.bind(&target_repo_path)
.bind(&completion_action)
.bind(&req.continue_from_task_id)
.bind(©_files_json)
.bind(&req.branched_from_task_id)
.bind(&req.conversation_history)
.bind(&req.supervisor_worktree_task_id)
.fetch_one(pool)
.await
}
/// Get a task by ID.
pub async fn get_task(pool: &PgPool, id: Uuid) -> Result<Option<Task>, sqlx::Error> {
sqlx::query_as::<_, Task>(
r#"
SELECT *
FROM tasks
WHERE id = $1
"#,
)
.bind(id)
.fetch_optional(pool)
.await
}
/// List all top-level tasks (no parent), ordered by created_at DESC.
/// Hidden tasks are excluded by default.
pub async fn list_tasks(pool: &PgPool) -> Result<Vec<TaskSummary>, sqlx::Error> {
sqlx::query_as::<_, TaskSummary>(
r#"
SELECT
t.id, t.contract_id, c.name as contract_name, c.phase as contract_phase,
c.status as contract_status,
t.parent_task_id, t.depth, t.name, t.status, t.priority,
t.progress_summary,
(SELECT COUNT(*) FROM tasks WHERE parent_task_id = t.id) as subtask_count,
t.version, t.is_supervisor, COALESCE(t.hidden, false) as hidden, t.created_at, t.updated_at
FROM tasks t
LEFT JOIN contracts c ON t.contract_id = c.id
WHERE t.parent_task_id IS NULL AND COALESCE(t.hidden, false) = false
ORDER BY t.priority DESC, t.created_at DESC
"#,
)
.fetch_all(pool)
.await
}
/// List subtasks of a parent task.
pub async fn list_subtasks(pool: &PgPool, parent_id: Uuid) -> Result<Vec<TaskSummary>, sqlx::Error> {
sqlx::query_as::<_, TaskSummary>(
r#"
SELECT
t.id, t.contract_id, c.name as contract_name, c.phase as contract_phase,
c.status as contract_status,
t.parent_task_id, t.depth, t.name, t.status, t.priority,
t.progress_summary,
(SELECT COUNT(*) FROM tasks WHERE parent_task_id = t.id) as subtask_count,
t.version, t.is_supervisor, COALESCE(t.hidden, false) as hidden, t.created_at, t.updated_at
FROM tasks t
LEFT JOIN contracts c ON t.contract_id = c.id
WHERE t.parent_task_id = $1
ORDER BY t.priority DESC, t.created_at DESC
"#,
)
.bind(parent_id)
.fetch_all(pool)
.await
}
/// List all tasks in a contract (for supervisor tree view).
pub async fn list_tasks_by_contract(
pool: &PgPool,
contract_id: Uuid,
owner_id: Uuid,
) -> Result<Vec<Task>, sqlx::Error> {
sqlx::query_as::<_, Task>(
r#"
SELECT * FROM tasks
WHERE contract_id = $1 AND owner_id = $2
ORDER BY is_supervisor DESC, depth ASC, created_at ASC
"#,
)
.bind(contract_id)
.bind(owner_id)
.fetch_all(pool)
.await
}
/// Get pending tasks for a contract (non-supervisor tasks only).
/// Includes tasks that were interrupted (retry candidates).
/// Prioritizes interrupted tasks and excludes those that exceeded max_retries.
pub async fn get_pending_tasks_for_contract(
pool: &PgPool,
contract_id: Uuid,
owner_id: Uuid,
) -> Result<Vec<Task>, sqlx::Error> {
sqlx::query_as::<_, Task>(
r#"
SELECT * FROM tasks
WHERE contract_id = $1 AND owner_id = $2
AND status = 'pending'
AND is_supervisor = false
AND retry_count < max_retries
ORDER BY
interrupted_at DESC NULLS LAST,
priority DESC,
created_at ASC
"#,
)
.bind(contract_id)
.bind(owner_id)
.fetch_all(pool)
.await
}
/// Get all contracts that have pending tasks awaiting retry.
/// Returns tuples of (contract_id, owner_id) for contracts with retryable tasks.
pub async fn get_all_pending_task_contracts(
pool: &PgPool,
) -> Result<Vec<(Uuid, Uuid)>, sqlx::Error> {
sqlx::query_as::<_, (Uuid, Uuid)>(
r#"
SELECT DISTINCT contract_id, owner_id
FROM tasks
WHERE contract_id IS NOT NULL
AND status = 'pending'
AND is_supervisor = false
AND retry_count < max_retries
ORDER BY owner_id, contract_id
"#,
)
.fetch_all(pool)
.await
}
/// Mark a task as pending for retry after daemon failure.
/// Increments retry count and adds the failed daemon to exclusion list.
pub async fn mark_task_for_retry(
pool: &PgPool,
task_id: Uuid,
failed_daemon_id: Uuid,
) -> Result<Option<Task>, sqlx::Error> {
sqlx::query_as::<_, Task>(
r#"
UPDATE tasks
SET status = 'pending',
daemon_id = NULL,
retry_count = retry_count + 1,
failed_daemon_ids = array_append(COALESCE(failed_daemon_ids, '{}'), $2),
last_active_daemon_id = $2,
interrupted_at = NOW(),
error_message = 'Daemon disconnected, awaiting retry',
updated_at = NOW()
WHERE id = $1
AND retry_count < max_retries
RETURNING *
"#,
)
.bind(task_id)
.bind(failed_daemon_id)
.fetch_optional(pool)
.await
}
/// Mark a task as permanently failed (exceeded retry limit).
pub async fn mark_task_permanently_failed(
pool: &PgPool,
task_id: Uuid,
failed_daemon_id: Uuid,
) -> Result<(), sqlx::Error> {
sqlx::query(
r#"
UPDATE tasks
SET status = 'failed',
daemon_id = NULL,
retry_count = retry_count + 1,
failed_daemon_ids = array_append(COALESCE(failed_daemon_ids, '{}'), $2),
last_active_daemon_id = $2,
error_message = 'Task failed: exceeded maximum retry attempts',
updated_at = NOW()
WHERE id = $1
"#,
)
.bind(task_id)
.bind(failed_daemon_id)
.execute(pool)
.await?;
Ok(())
}
/// Update a task by ID with optimistic locking.
pub async fn update_task(
pool: &PgPool,
id: Uuid,
req: UpdateTaskRequest,
) -> Result<Option<Task>, RepositoryError> {
// Get the existing task first
let existing = get_task(pool, id).await?;
let Some(existing) = existing else {
return Ok(None);
};
// Check version if provided (optimistic locking)
if let Some(expected_version) = req.version {
if existing.version != expected_version {
return Err(RepositoryError::VersionConflict {
expected: expected_version,
actual: existing.version,
});
}
}
// Apply updates
let name = req.name.unwrap_or(existing.name);
let description = req.description.or(existing.description);
let plan = req.plan.unwrap_or(existing.plan);
let status = req.status.unwrap_or(existing.status);
let priority = req.priority.unwrap_or(existing.priority);
let progress_summary = req.progress_summary.or(existing.progress_summary);
let last_output = req.last_output.or(existing.last_output);
let error_message = req.error_message.or(existing.error_message);
let merge_mode = req.merge_mode.or(existing.merge_mode);
let pr_url = req.pr_url.or(existing.pr_url);
let target_repo_path = req.target_repo_path.or(existing.target_repo_path);
let completion_action = req.completion_action.or(existing.completion_action);
// Handle clear_daemon_id: if true, set to NULL; otherwise use provided value or keep existing
let daemon_id = if req.clear_daemon_id {
None
} else {
req.daemon_id.or(existing.daemon_id)
};
// Update with version check in WHERE clause for race condition safety
let result = if req.version.is_some() {
sqlx::query_as::<_, Task>(
r#"
UPDATE tasks
SET name = $2, description = $3, plan = $4, status = $5, priority = $6,
progress_summary = $7, last_output = $8, error_message = $9,
merge_mode = $10, pr_url = $11, daemon_id = $12,
target_repo_path = $13, completion_action = $14, updated_at = NOW()
WHERE id = $1 AND version = $15
RETURNING *
"#,
)
.bind(id)
.bind(&name)
.bind(&description)
.bind(&plan)
.bind(&status)
.bind(priority)
.bind(&progress_summary)
.bind(&last_output)
.bind(&error_message)
.bind(&merge_mode)
.bind(&pr_url)
.bind(daemon_id)
.bind(&target_repo_path)
.bind(&completion_action)
.bind(req.version.unwrap())
.fetch_optional(pool)
.await?
} else {
sqlx::query_as::<_, Task>(
r#"
UPDATE tasks
SET name = $2, description = $3, plan = $4, status = $5, priority = $6,
progress_summary = $7, last_output = $8, error_message = $9,
merge_mode = $10, pr_url = $11, daemon_id = $12,
target_repo_path = $13, completion_action = $14, updated_at = NOW()
WHERE id = $1
RETURNING *
"#,
)
.bind(id)
.bind(&name)
.bind(&description)
.bind(&plan)
.bind(&status)
.bind(priority)
.bind(&progress_summary)
.bind(&last_output)
.bind(&error_message)
.bind(&merge_mode)
.bind(&pr_url)
.bind(daemon_id)
.bind(&target_repo_path)
.bind(&completion_action)
.fetch_optional(pool)
.await?
};
// If versioned update returned None, there was a race condition
if result.is_none() && req.version.is_some() {
if let Some(current) = get_task(pool, id).await? {
return Err(RepositoryError::VersionConflict {
expected: req.version.unwrap(),
actual: current.version,
});
}
}
Ok(result)
}
/// Delete a task by ID.
pub async fn delete_task(pool: &PgPool, id: Uuid) -> Result<bool, sqlx::Error> {
let result = sqlx::query(
r#"
DELETE FROM tasks
WHERE id = $1
"#,
)
.bind(id)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Count total tasks.
pub async fn count_tasks(pool: &PgPool) -> Result<i64, sqlx::Error> {
let result: (i64,) = sqlx::query_as(
"SELECT COUNT(*) FROM tasks WHERE parent_task_id IS NULL",
)
.fetch_one(pool)
.await?;
Ok(result.0)
}
// =============================================================================
// Owner-Scoped Task Functions
// =============================================================================
/// Create a new task for a specific owner.
pub async fn create_task_for_owner(
pool: &PgPool,
owner_id: Uuid,
req: CreateTaskRequest,
) -> Result<Task, sqlx::Error> {
// Calculate depth and inherit settings from parent if applicable
let (depth, contract_id, repo_url, base_branch, target_branch, merge_mode, target_repo_path, completion_action) =
if let Some(parent_id) = req.parent_task_id {
// Fetch parent task to get depth and inherit settings (must belong to same owner)
let parent = get_task_for_owner(pool, parent_id, owner_id).await?
.ok_or_else(|| sqlx::Error::RowNotFound)?;
let new_depth = parent.depth + 1;
// Validate max depth
if new_depth >= 2 {
return Err(sqlx::Error::Protocol(format!(
"Maximum task depth exceeded. Cannot create subtask at depth {} (max is 1). Subtasks cannot have children.",
new_depth
)));
}
// Subtasks inherit contract_id from parent (or use request contract_id if parent has none)
let contract_id = parent.contract_id.or(req.contract_id);
// Inherit repo settings if not provided
let repo_url = req.repository_url.clone().or(parent.repository_url);
let base_branch = req.base_branch.clone().or(parent.base_branch);
let target_branch = req.target_branch.clone().or(parent.target_branch);
let merge_mode = req.merge_mode.clone().or(parent.merge_mode);
let target_repo_path = req.target_repo_path.clone().or(parent.target_repo_path);
// NOTE: completion_action is NOT inherited - subtasks should not auto-merge.
// The orchestrator integrates subtask work from their worktrees.
let completion_action = req.completion_action.clone();
(new_depth, contract_id, repo_url, base_branch, target_branch, merge_mode, target_repo_path, completion_action)
} else {
// Top-level task: depth 0, use contract_id from request (may be None for branched tasks)
(
0,
req.contract_id,
req.repository_url.clone(),
req.base_branch.clone(),
req.target_branch.clone(),
req.merge_mode.clone(),
req.target_repo_path.clone(),
req.completion_action.clone(),
)
};
let copy_files_json = req.copy_files.as_ref().map(|f| serde_json::to_value(f).unwrap_or_default());
sqlx::query_as::<_, Task>(
r#"
INSERT INTO tasks (
owner_id, contract_id, parent_task_id, depth, name, description, plan, priority,
is_supervisor, repository_url, base_branch, target_branch, merge_mode,
target_repo_path, completion_action, continue_from_task_id, copy_files,
branched_from_task_id, conversation_state, supervisor_worktree_task_id
)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20)
RETURNING *
"#,
)
.bind(owner_id)
.bind(contract_id)
.bind(req.parent_task_id)
.bind(depth)
.bind(&req.name)
.bind(&req.description)
.bind(&req.plan)
.bind(req.priority)
.bind(req.is_supervisor)
.bind(&repo_url)
.bind(&base_branch)
.bind(&target_branch)
.bind(&merge_mode)
.bind(&target_repo_path)
.bind(&completion_action)
.bind(&req.continue_from_task_id)
.bind(©_files_json)
.bind(&req.branched_from_task_id)
.bind(&req.conversation_history)
.bind(&req.supervisor_worktree_task_id)
.fetch_one(pool)
.await
}
/// Get a task by ID, scoped to owner.
pub async fn get_task_for_owner(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
) -> Result<Option<Task>, sqlx::Error> {
sqlx::query_as::<_, Task>(
r#"
SELECT *
FROM tasks
WHERE id = $1 AND owner_id = $2
"#,
)
.bind(id)
.bind(owner_id)
.fetch_optional(pool)
.await
}
/// List all top-level tasks (no parent) for an owner, ordered by created_at DESC.
/// Hidden tasks are excluded by default.
pub async fn list_tasks_for_owner(
pool: &PgPool,
owner_id: Uuid,
) -> Result<Vec<TaskSummary>, sqlx::Error> {
sqlx::query_as::<_, TaskSummary>(
r#"
SELECT
t.id, t.contract_id, c.name as contract_name, c.phase as contract_phase,
c.status as contract_status,
t.parent_task_id, t.depth, t.name, t.status, t.priority,
t.progress_summary,
(SELECT COUNT(*) FROM tasks WHERE parent_task_id = t.id) as subtask_count,
t.version, t.is_supervisor, COALESCE(t.hidden, false) as hidden, t.created_at, t.updated_at
FROM tasks t
LEFT JOIN contracts c ON t.contract_id = c.id
WHERE t.owner_id = $1 AND t.parent_task_id IS NULL AND COALESCE(t.hidden, false) = false
ORDER BY t.priority DESC, t.created_at DESC
"#,
)
.bind(owner_id)
.fetch_all(pool)
.await
}
/// List subtasks of a parent task, scoped to owner.
pub async fn list_subtasks_for_owner(
pool: &PgPool,
parent_id: Uuid,
owner_id: Uuid,
) -> Result<Vec<TaskSummary>, sqlx::Error> {
sqlx::query_as::<_, TaskSummary>(
r#"
SELECT
t.id, t.contract_id, c.name as contract_name, c.phase as contract_phase,
c.status as contract_status,
t.parent_task_id, t.depth, t.name, t.status, t.priority,
t.progress_summary,
(SELECT COUNT(*) FROM tasks WHERE parent_task_id = t.id) as subtask_count,
t.version, t.is_supervisor, COALESCE(t.hidden, false) as hidden, t.created_at, t.updated_at
FROM tasks t
LEFT JOIN contracts c ON t.contract_id = c.id
WHERE t.owner_id = $1 AND t.parent_task_id = $2
ORDER BY t.priority DESC, t.created_at DESC
"#,
)
.bind(owner_id)
.bind(parent_id)
.fetch_all(pool)
.await
}
/// Update a task by ID with optimistic locking, scoped to owner.
pub async fn update_task_for_owner(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
req: UpdateTaskRequest,
) -> Result<Option<Task>, RepositoryError> {
// Get the existing task first (scoped to owner)
let existing = get_task_for_owner(pool, id, owner_id).await?;
let Some(existing) = existing else {
return Ok(None);
};
// Check version if provided (optimistic locking)
if let Some(expected_version) = req.version {
if existing.version != expected_version {
return Err(RepositoryError::VersionConflict {
expected: expected_version,
actual: existing.version,
});
}
}
// Apply updates
let name = req.name.unwrap_or(existing.name);
let description = req.description.or(existing.description);
let plan = req.plan.unwrap_or(existing.plan);
let status = req.status.unwrap_or(existing.status);
let priority = req.priority.unwrap_or(existing.priority);
let progress_summary = req.progress_summary.or(existing.progress_summary);
let last_output = req.last_output.or(existing.last_output);
let error_message = req.error_message.or(existing.error_message);
let merge_mode = req.merge_mode.or(existing.merge_mode);
let pr_url = req.pr_url.or(existing.pr_url);
let repository_url = req.repository_url.or(existing.repository_url);
let target_repo_path = req.target_repo_path.or(existing.target_repo_path);
let completion_action = req.completion_action.or(existing.completion_action);
let hidden = req.hidden.unwrap_or(existing.hidden);
let daemon_id = if req.clear_daemon_id {
None
} else {
req.daemon_id.or(existing.daemon_id)
};
// Update with version check in WHERE clause for race condition safety
let result = if req.version.is_some() {
sqlx::query_as::<_, Task>(
r#"
UPDATE tasks
SET name = $3, description = $4, plan = $5, status = $6, priority = $7,
progress_summary = $8, last_output = $9, error_message = $10,
merge_mode = $11, pr_url = $12, daemon_id = $13,
target_repo_path = $14, completion_action = $15, repository_url = $16,
hidden = $17, updated_at = NOW()
WHERE id = $1 AND owner_id = $2 AND version = $18
RETURNING *
"#,
)
.bind(id)
.bind(owner_id)
.bind(&name)
.bind(&description)
.bind(&plan)
.bind(&status)
.bind(priority)
.bind(&progress_summary)
.bind(&last_output)
.bind(&error_message)
.bind(&merge_mode)
.bind(&pr_url)
.bind(daemon_id)
.bind(&target_repo_path)
.bind(&completion_action)
.bind(&repository_url)
.bind(hidden)
.bind(req.version.unwrap())
.fetch_optional(pool)
.await?
} else {
sqlx::query_as::<_, Task>(
r#"
UPDATE tasks
SET name = $3, description = $4, plan = $5, status = $6, priority = $7,
progress_summary = $8, last_output = $9, error_message = $10,
merge_mode = $11, pr_url = $12, daemon_id = $13,
target_repo_path = $14, completion_action = $15, repository_url = $16,
hidden = $17, updated_at = NOW()
WHERE id = $1 AND owner_id = $2
RETURNING *
"#,
)
.bind(id)
.bind(owner_id)
.bind(&name)
.bind(&description)
.bind(&plan)
.bind(&status)
.bind(priority)
.bind(&progress_summary)
.bind(&last_output)
.bind(&error_message)
.bind(&merge_mode)
.bind(&pr_url)
.bind(daemon_id)
.bind(&target_repo_path)
.bind(&completion_action)
.bind(&repository_url)
.bind(hidden)
.fetch_optional(pool)
.await?
};
// If versioned update returned None, there was a race condition
if result.is_none() && req.version.is_some() {
if let Some(current) = get_task_for_owner(pool, id, owner_id).await? {
return Err(RepositoryError::VersionConflict {
expected: req.version.unwrap(),
actual: current.version,
});
}
}
Ok(result)
}
/// Delete a task by ID, scoped to owner.
pub async fn delete_task_for_owner(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
) -> Result<bool, sqlx::Error> {
let result = sqlx::query(
r#"
DELETE FROM tasks
WHERE id = $1 AND owner_id = $2
"#,
)
.bind(id)
.bind(owner_id)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Update task status and record event.
pub async fn update_task_status(
pool: &PgPool,
id: Uuid,
new_status: &str,
event_data: Option<serde_json::Value>,
) -> Result<Option<Task>, sqlx::Error> {
// Get existing status
let existing = get_task(pool, id).await?;
let Some(existing) = existing else {
return Ok(None);
};
let previous_status = existing.status.clone();
// Update task status
let task = sqlx::query_as::<_, Task>(
r#"
UPDATE tasks
SET status = $2, updated_at = NOW(),
started_at = CASE WHEN $2 = 'running' AND started_at IS NULL THEN NOW() ELSE started_at END,
completed_at = CASE WHEN $2 IN ('done', 'failed', 'merged') THEN NOW() ELSE completed_at END
WHERE id = $1
RETURNING *
"#,
)
.bind(id)
.bind(new_status)
.fetch_optional(pool)
.await?;
// Record event
if task.is_some() {
let _ = create_task_event(
pool,
id,
"status_change",
Some(&previous_status),
Some(new_status),
event_data,
)
.await;
}
Ok(task)
}
// =============================================================================
// Task Event Functions
// =============================================================================
/// Create a task event.
pub async fn create_task_event(
pool: &PgPool,
task_id: Uuid,
event_type: &str,
previous_status: Option<&str>,
new_status: Option<&str>,
event_data: Option<serde_json::Value>,
) -> Result<TaskEvent, sqlx::Error> {
sqlx::query_as::<_, TaskEvent>(
r#"
INSERT INTO task_events (task_id, event_type, previous_status, new_status, event_data)
VALUES ($1, $2, $3, $4, $5)
RETURNING *
"#,
)
.bind(task_id)
.bind(event_type)
.bind(previous_status)
.bind(new_status)
.bind(event_data)
.fetch_one(pool)
.await
}
/// List events for a task.
pub async fn list_task_events(
pool: &PgPool,
task_id: Uuid,
limit: Option<i64>,
) -> Result<Vec<TaskEvent>, sqlx::Error> {
let limit = limit.unwrap_or(100);
sqlx::query_as::<_, TaskEvent>(
r#"
SELECT *
FROM task_events
WHERE task_id = $1
ORDER BY created_at DESC
LIMIT $2
"#,
)
.bind(task_id)
.bind(limit)
.fetch_all(pool)
.await
}
// =============================================================================
// Daemon Functions
// =============================================================================
/// Register a new daemon connection.
pub async fn register_daemon(
pool: &PgPool,
owner_id: Uuid,
connection_id: &str,
hostname: Option<&str>,
machine_id: Option<&str>,
max_concurrent_tasks: i32,
) -> Result<Daemon, sqlx::Error> {
sqlx::query_as::<_, Daemon>(
r#"
INSERT INTO daemons (owner_id, connection_id, hostname, machine_id, max_concurrent_tasks)
VALUES ($1, $2, $3, $4, $5)
RETURNING *
"#,
)
.bind(owner_id)
.bind(connection_id)
.bind(hostname)
.bind(machine_id)
.bind(max_concurrent_tasks)
.fetch_one(pool)
.await
}
/// Get a daemon by ID.
pub async fn get_daemon(pool: &PgPool, id: Uuid) -> Result<Option<Daemon>, sqlx::Error> {
sqlx::query_as::<_, Daemon>(
r#"
SELECT *
FROM daemons
WHERE id = $1
"#,
)
.bind(id)
.fetch_optional(pool)
.await
}
/// Get a daemon by connection ID.
pub async fn get_daemon_by_connection(
pool: &PgPool,
connection_id: &str,
) -> Result<Option<Daemon>, sqlx::Error> {
sqlx::query_as::<_, Daemon>(
r#"
SELECT *
FROM daemons
WHERE connection_id = $1
"#,
)
.bind(connection_id)
.fetch_optional(pool)
.await
}
/// List all daemons.
pub async fn list_daemons(pool: &PgPool) -> Result<Vec<Daemon>, sqlx::Error> {
sqlx::query_as::<_, Daemon>(
r#"
SELECT *
FROM daemons
ORDER BY connected_at DESC
"#,
)
.fetch_all(pool)
.await
}
/// List daemons for a specific owner.
pub async fn list_daemons_for_owner(pool: &PgPool, owner_id: Uuid) -> Result<Vec<Daemon>, sqlx::Error> {
sqlx::query_as::<_, Daemon>(
r#"
SELECT *
FROM daemons
WHERE owner_id = $1
ORDER BY connected_at DESC
"#,
)
.bind(owner_id)
.fetch_all(pool)
.await
}
/// Get a daemon by ID for a specific owner.
pub async fn get_daemon_for_owner(pool: &PgPool, id: Uuid, owner_id: Uuid) -> Result<Option<Daemon>, sqlx::Error> {
sqlx::query_as::<_, Daemon>(
r#"
SELECT *
FROM daemons
WHERE id = $1 AND owner_id = $2
"#,
)
.bind(id)
.bind(owner_id)
.fetch_optional(pool)
.await
}
/// Update daemon heartbeat.
pub async fn update_daemon_heartbeat(pool: &PgPool, id: Uuid) -> Result<bool, sqlx::Error> {
let result = sqlx::query(
r#"
UPDATE daemons
SET last_heartbeat_at = NOW(), status = 'connected'
WHERE id = $1
"#,
)
.bind(id)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Update daemon status.
pub async fn update_daemon_status(
pool: &PgPool,
id: Uuid,
status: &str,
) -> Result<bool, sqlx::Error> {
let result = sqlx::query(
r#"
UPDATE daemons
SET status = $2,
disconnected_at = CASE WHEN $2 = 'disconnected' THEN NOW() ELSE disconnected_at END
WHERE id = $1
"#,
)
.bind(id)
.bind(status)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Mark daemon as disconnected by connection_id.
pub async fn disconnect_daemon_by_connection(
pool: &PgPool,
connection_id: &str,
) -> Result<bool, sqlx::Error> {
let result = sqlx::query(
r#"
UPDATE daemons
SET status = 'disconnected',
disconnected_at = NOW()
WHERE connection_id = $1
"#,
)
.bind(connection_id)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Update daemon task count.
pub async fn update_daemon_task_count(
pool: &PgPool,
id: Uuid,
delta: i32,
) -> Result<bool, sqlx::Error> {
let result = sqlx::query(
r#"
UPDATE daemons
SET current_task_count = GREATEST(0, current_task_count + $2)
WHERE id = $1
"#,
)
.bind(id)
.bind(delta)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Delete a daemon by ID.
pub async fn delete_daemon(pool: &PgPool, id: Uuid) -> Result<bool, sqlx::Error> {
let result = sqlx::query(
r#"
DELETE FROM daemons
WHERE id = $1
"#,
)
.bind(id)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Delete a daemon by connection ID.
pub async fn delete_daemon_by_connection(
pool: &PgPool,
connection_id: &str,
) -> Result<bool, sqlx::Error> {
let result = sqlx::query(
r#"
DELETE FROM daemons
WHERE connection_id = $1
"#,
)
.bind(connection_id)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Count connected daemons.
pub async fn count_daemons(pool: &PgPool) -> Result<i64, sqlx::Error> {
let result: (i64,) = sqlx::query_as(
"SELECT COUNT(*) FROM daemons WHERE status = 'connected'",
)
.fetch_one(pool)
.await?;
Ok(result.0)
}
/// Delete stale daemons that haven't sent a heartbeat within the timeout.
/// Returns the number of deleted daemons.
pub async fn delete_stale_daemons(
pool: &PgPool,
timeout_seconds: i64,
) -> Result<u64, sqlx::Error> {
let result = sqlx::query(
r#"
DELETE FROM daemons
WHERE last_heartbeat_at < NOW() - INTERVAL '1 second' * $1
"#,
)
.bind(timeout_seconds)
.execute(pool)
.await?;
Ok(result.rows_affected())
}
// =============================================================================
// Sibling Awareness Functions
// =============================================================================
/// List sibling tasks (tasks with the same parent, excluding the given task).
pub async fn list_sibling_tasks(
pool: &PgPool,
task_id: Uuid,
parent_id: Option<Uuid>,
) -> Result<Vec<TaskSummary>, sqlx::Error> {
match parent_id {
Some(parent) => {
sqlx::query_as::<_, TaskSummary>(
r#"
SELECT
t.id, t.contract_id, c.name as contract_name, c.phase as contract_phase,
c.status as contract_status,
t.parent_task_id, t.depth, t.name, t.status, t.priority,
t.progress_summary,
(SELECT COUNT(*) FROM tasks WHERE parent_task_id = t.id) as subtask_count,
t.version, t.is_supervisor, COALESCE(t.hidden, false) as hidden, t.created_at, t.updated_at
FROM tasks t
LEFT JOIN contracts c ON t.contract_id = c.id
WHERE t.parent_task_id = $1 AND t.id != $2
ORDER BY t.priority DESC, t.created_at DESC
"#,
)
.bind(parent)
.bind(task_id)
.fetch_all(pool)
.await
}
None => {
// Top-level tasks (no parent) - siblings are other top-level tasks
sqlx::query_as::<_, TaskSummary>(
r#"
SELECT
t.id, t.contract_id, c.name as contract_name, c.phase as contract_phase,
c.status as contract_status,
t.parent_task_id, t.depth, t.name, t.status, t.priority,
t.progress_summary,
(SELECT COUNT(*) FROM tasks WHERE parent_task_id = t.id) as subtask_count,
t.version, t.is_supervisor, COALESCE(t.hidden, false) as hidden, t.created_at, t.updated_at
FROM tasks t
LEFT JOIN contracts c ON t.contract_id = c.id
WHERE t.parent_task_id IS NULL AND t.id != $1
ORDER BY t.priority DESC, t.created_at DESC
"#,
)
.bind(task_id)
.fetch_all(pool)
.await
}
}
}
/// Get running sibling tasks (for context injection).
pub async fn get_running_siblings(
pool: &PgPool,
owner_id: Uuid,
task_id: Uuid,
parent_id: Option<Uuid>,
) -> Result<Vec<Task>, sqlx::Error> {
match parent_id {
Some(parent) => {
sqlx::query_as::<_, Task>(
r#"
SELECT *
FROM tasks t
WHERE t.owner_id = $1
AND t.parent_task_id = $2
AND t.id != $3
AND t.status = 'running'
ORDER BY t.priority DESC
"#,
)
.bind(owner_id)
.bind(parent)
.bind(task_id)
.fetch_all(pool)
.await
}
None => {
sqlx::query_as::<_, Task>(
r#"
SELECT *
FROM tasks t
WHERE t.owner_id = $1
AND t.parent_task_id IS NULL
AND t.id != $2
AND t.status = 'running'
ORDER BY t.priority DESC
"#,
)
.bind(owner_id)
.bind(task_id)
.fetch_all(pool)
.await
}
}
}
/// Get task with its siblings for context awareness.
pub async fn get_task_with_siblings(
pool: &PgPool,
id: Uuid,
) -> Result<Option<(Task, Vec<TaskSummary>)>, sqlx::Error> {
let task = get_task(pool, id).await?;
let Some(task) = task else {
return Ok(None);
};
let siblings = list_sibling_tasks(pool, id, task.parent_task_id).await?;
Ok(Some((task, siblings)))
}
// =============================================================================
// Task Output Persistence Functions
// =============================================================================
/// Save task output to the database.
/// This stores output in the task_events table with event_type='output'.
pub async fn save_task_output(
pool: &PgPool,
task_id: Uuid,
message_type: &str,
content: &str,
tool_name: Option<&str>,
tool_input: Option<serde_json::Value>,
is_error: Option<bool>,
cost_usd: Option<f64>,
duration_ms: Option<u64>,
) -> Result<TaskEvent, sqlx::Error> {
let event_data = serde_json::json!({
"messageType": message_type,
"content": content,
"toolName": tool_name,
"toolInput": tool_input,
"isError": is_error,
"costUsd": cost_usd,
"durationMs": duration_ms,
});
create_task_event(pool, task_id, "output", None, None, Some(event_data)).await
}
/// Get task output from the database.
/// Retrieves all output events for a task, ordered by creation time.
pub async fn get_task_output(
pool: &PgPool,
task_id: Uuid,
limit: Option<i64>,
) -> Result<Vec<TaskEvent>, sqlx::Error> {
let limit = limit.unwrap_or(1000);
sqlx::query_as::<_, TaskEvent>(
r#"
SELECT *
FROM task_events
WHERE task_id = $1 AND event_type = 'output'
ORDER BY created_at ASC
LIMIT $2
"#,
)
.bind(task_id)
.bind(limit)
.fetch_all(pool)
.await
}
/// Update task completion status with error message.
/// Sets the task status to 'done' or 'failed' and records completion time.
pub async fn complete_task(
pool: &PgPool,
task_id: Uuid,
success: bool,
error_message: Option<&str>,
) -> Result<Option<Task>, sqlx::Error> {
let status = if success { "done" } else { "failed" };
let task = sqlx::query_as::<_, Task>(
r#"
UPDATE tasks
SET status = $2,
error_message = COALESCE($3, error_message),
completed_at = NOW(),
updated_at = NOW()
WHERE id = $1
RETURNING *
"#,
)
.bind(task_id)
.bind(status)
.bind(error_message)
.fetch_optional(pool)
.await?;
// Record completion event
if task.is_some() {
let event_data = serde_json::json!({
"success": success,
"errorMessage": error_message,
});
let _ = create_task_event(
pool,
task_id,
"complete",
Some("running"),
Some(status),
Some(event_data),
)
.await;
}
Ok(task)
}
// =============================================================================
// Mesh Chat History Functions
// =============================================================================
/// Get or create the active conversation for an owner.
pub async fn get_or_create_active_conversation(
pool: &PgPool,
owner_id: Uuid,
) -> Result<MeshChatConversation, sqlx::Error> {
// Try to get existing active conversation for this owner
let existing = sqlx::query_as::<_, MeshChatConversation>(
r#"
SELECT *
FROM mesh_chat_conversations
WHERE is_active = true AND owner_id = $1
LIMIT 1
"#,
)
.bind(owner_id)
.fetch_optional(pool)
.await?;
if let Some(conv) = existing {
return Ok(conv);
}
// Create new conversation
sqlx::query_as::<_, MeshChatConversation>(
r#"
INSERT INTO mesh_chat_conversations (owner_id, is_active)
VALUES ($1, true)
RETURNING *
"#,
)
.bind(owner_id)
.fetch_one(pool)
.await
}
/// List messages for a conversation.
pub async fn list_chat_messages(
pool: &PgPool,
conversation_id: Uuid,
limit: Option<i32>,
) -> Result<Vec<MeshChatMessageRecord>, sqlx::Error> {
let limit = limit.unwrap_or(100);
sqlx::query_as::<_, MeshChatMessageRecord>(
r#"
SELECT *
FROM mesh_chat_messages
WHERE conversation_id = $1
ORDER BY created_at ASC
LIMIT $2
"#,
)
.bind(conversation_id)
.bind(limit)
.fetch_all(pool)
.await
}
/// Add a message to a conversation.
#[allow(clippy::too_many_arguments)]
pub async fn add_chat_message(
pool: &PgPool,
conversation_id: Uuid,
role: &str,
content: &str,
context_type: &str,
context_task_id: Option<Uuid>,
tool_calls: Option<serde_json::Value>,
pending_questions: Option<serde_json::Value>,
) -> Result<MeshChatMessageRecord, sqlx::Error> {
sqlx::query_as::<_, MeshChatMessageRecord>(
r#"
INSERT INTO mesh_chat_messages
(conversation_id, role, content, context_type, context_task_id, tool_calls, pending_questions)
VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING *
"#,
)
.bind(conversation_id)
.bind(role)
.bind(content)
.bind(context_type)
.bind(context_task_id)
.bind(tool_calls)
.bind(pending_questions)
.fetch_one(pool)
.await
}
/// Clear conversation (archive existing and create new).
pub async fn clear_conversation(pool: &PgPool, owner_id: Uuid) -> Result<MeshChatConversation, sqlx::Error> {
// Mark existing as inactive for this owner
sqlx::query(
r#"
UPDATE mesh_chat_conversations
SET is_active = false, updated_at = NOW()
WHERE is_active = true AND owner_id = $1
"#,
)
.bind(owner_id)
.execute(pool)
.await?;
// Create new active conversation
get_or_create_active_conversation(pool, owner_id).await
}
// =============================================================================
// Contract Chat History Functions
// =============================================================================
/// Get or create the active conversation for a contract.
pub async fn get_or_create_contract_conversation(
pool: &PgPool,
contract_id: Uuid,
owner_id: Uuid,
) -> Result<ContractChatConversation, sqlx::Error> {
// Try to get existing active conversation for this contract
let existing = sqlx::query_as::<_, ContractChatConversation>(
r#"
SELECT *
FROM contract_chat_conversations
WHERE is_active = true AND contract_id = $1 AND owner_id = $2
LIMIT 1
"#,
)
.bind(contract_id)
.bind(owner_id)
.fetch_optional(pool)
.await?;
if let Some(conv) = existing {
return Ok(conv);
}
// Create new conversation
sqlx::query_as::<_, ContractChatConversation>(
r#"
INSERT INTO contract_chat_conversations (contract_id, owner_id, is_active)
VALUES ($1, $2, true)
RETURNING *
"#,
)
.bind(contract_id)
.bind(owner_id)
.fetch_one(pool)
.await
}
/// List messages for a contract conversation.
pub async fn list_contract_chat_messages(
pool: &PgPool,
conversation_id: Uuid,
limit: Option<i32>,
) -> Result<Vec<ContractChatMessageRecord>, sqlx::Error> {
let limit = limit.unwrap_or(100);
sqlx::query_as::<_, ContractChatMessageRecord>(
r#"
SELECT *
FROM contract_chat_messages
WHERE conversation_id = $1
ORDER BY created_at ASC
LIMIT $2
"#,
)
.bind(conversation_id)
.bind(limit)
.fetch_all(pool)
.await
}
/// Add a message to a contract conversation.
pub async fn add_contract_chat_message(
pool: &PgPool,
conversation_id: Uuid,
role: &str,
content: &str,
tool_calls: Option<serde_json::Value>,
pending_questions: Option<serde_json::Value>,
) -> Result<ContractChatMessageRecord, sqlx::Error> {
sqlx::query_as::<_, ContractChatMessageRecord>(
r#"
INSERT INTO contract_chat_messages
(conversation_id, role, content, tool_calls, pending_questions)
VALUES ($1, $2, $3, $4, $5)
RETURNING *
"#,
)
.bind(conversation_id)
.bind(role)
.bind(content)
.bind(tool_calls)
.bind(pending_questions)
.fetch_one(pool)
.await
}
/// Clear contract conversation (archive existing and create new).
pub async fn clear_contract_conversation(
pool: &PgPool,
contract_id: Uuid,
owner_id: Uuid,
) -> Result<ContractChatConversation, sqlx::Error> {
// Mark existing as inactive for this contract
sqlx::query(
r#"
UPDATE contract_chat_conversations
SET is_active = false, updated_at = NOW()
WHERE is_active = true AND contract_id = $1 AND owner_id = $2
"#,
)
.bind(contract_id)
.bind(owner_id)
.execute(pool)
.await?;
// Create new active conversation
get_or_create_contract_conversation(pool, contract_id, owner_id).await
}
// =============================================================================
// Contract Type Template Functions (Owner-Scoped)
// =============================================================================
/// Create a new contract type template for a specific owner.
pub async fn create_template_for_owner(
pool: &PgPool,
owner_id: Uuid,
req: CreateTemplateRequest,
) -> Result<ContractTypeTemplateRecord, sqlx::Error> {
sqlx::query_as::<_, ContractTypeTemplateRecord>(
r#"
INSERT INTO contract_type_templates (owner_id, name, description, phases, default_phase, deliverables)
VALUES ($1, $2, $3, $4, $5, $6)
RETURNING *
"#,
)
.bind(owner_id)
.bind(&req.name)
.bind(&req.description)
.bind(serde_json::to_value(&req.phases).unwrap_or_default())
.bind(&req.default_phase)
.bind(match &req.deliverables {
Some(d) => serde_json::to_value(d).ok(),
None => None,
})
.fetch_one(pool)
.await
}
/// Get a contract type template by ID, scoped to owner.
pub async fn get_template_for_owner(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
) -> Result<Option<ContractTypeTemplateRecord>, sqlx::Error> {
sqlx::query_as::<_, ContractTypeTemplateRecord>(
r#"
SELECT *
FROM contract_type_templates
WHERE id = $1 AND owner_id = $2
"#,
)
.bind(id)
.bind(owner_id)
.fetch_optional(pool)
.await
}
/// Get a contract type template by ID (internal use, no owner scoping).
pub async fn get_template_by_id(
pool: &PgPool,
id: Uuid,
) -> Result<Option<ContractTypeTemplateRecord>, sqlx::Error> {
sqlx::query_as::<_, ContractTypeTemplateRecord>(
r#"
SELECT *
FROM contract_type_templates
WHERE id = $1
"#,
)
.bind(id)
.fetch_optional(pool)
.await
}
/// List all contract type templates for an owner, ordered by name.
pub async fn list_templates_for_owner(
pool: &PgPool,
owner_id: Uuid,
) -> Result<Vec<ContractTypeTemplateRecord>, sqlx::Error> {
sqlx::query_as::<_, ContractTypeTemplateRecord>(
r#"
SELECT *
FROM contract_type_templates
WHERE owner_id = $1
ORDER BY name ASC
"#,
)
.bind(owner_id)
.fetch_all(pool)
.await
}
/// Update a contract type template for an owner.
pub async fn update_template_for_owner(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
req: UpdateTemplateRequest,
) -> Result<Option<ContractTypeTemplateRecord>, RepositoryError> {
// Build dynamic update query
let mut query = String::from("UPDATE contract_type_templates SET updated_at = NOW()");
let mut param_idx = 3; // $1 = id, $2 = owner_id
if req.name.is_some() {
query.push_str(&format!(", name = ${}", param_idx));
param_idx += 1;
}
if req.description.is_some() {
query.push_str(&format!(", description = ${}", param_idx));
param_idx += 1;
}
if req.phases.is_some() {
query.push_str(&format!(", phases = ${}", param_idx));
param_idx += 1;
}
if req.default_phase.is_some() {
query.push_str(&format!(", default_phase = ${}", param_idx));
param_idx += 1;
}
if req.deliverables.is_some() {
query.push_str(&format!(", deliverables = ${}", param_idx));
param_idx += 1;
}
// Optimistic locking
if req.version.is_some() {
query.push_str(&format!(", version = version + 1 WHERE id = $1 AND owner_id = $2 AND version = ${}", param_idx));
} else {
query.push_str(", version = version + 1 WHERE id = $1 AND owner_id = $2");
}
query.push_str(" RETURNING *");
let mut sql_query = sqlx::query_as::<_, ContractTypeTemplateRecord>(&query);
sql_query = sql_query.bind(id).bind(owner_id);
if let Some(ref name) = req.name {
sql_query = sql_query.bind(name);
}
if let Some(ref description) = req.description {
sql_query = sql_query.bind(description);
}
if let Some(ref phases) = req.phases {
sql_query = sql_query.bind(serde_json::to_value(phases).unwrap_or_default());
}
if let Some(ref default_phase) = req.default_phase {
sql_query = sql_query.bind(default_phase);
}
if let Some(ref deliverables) = req.deliverables {
sql_query = sql_query.bind(serde_json::to_value(deliverables).unwrap_or_default());
}
if let Some(version) = req.version {
sql_query = sql_query.bind(version);
}
match sql_query.fetch_optional(pool).await {
Ok(result) => {
if result.is_none() && req.version.is_some() {
// Check if it's a version conflict
if let Some(current) = get_template_for_owner(pool, id, owner_id).await? {
return Err(RepositoryError::VersionConflict {
expected: req.version.unwrap(),
actual: current.version,
});
}
}
Ok(result)
}
Err(e) => Err(RepositoryError::Database(e)),
}
}
/// Delete a contract type template for an owner.
pub async fn delete_template_for_owner(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
) -> Result<bool, sqlx::Error> {
let result = sqlx::query(
r#"
DELETE FROM contract_type_templates
WHERE id = $1 AND owner_id = $2
"#,
)
.bind(id)
.bind(owner_id)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Helper function to build PhaseConfig from a template.
pub fn build_phase_config_from_template(template: &ContractTypeTemplateRecord) -> PhaseConfig {
PhaseConfig {
phases: template.phases.clone(),
default_phase: template.default_phase.clone(),
deliverables: template.deliverables.clone().unwrap_or_default(),
}
}
/// Helper function to build PhaseConfig for built-in contract types.
pub fn build_phase_config_for_builtin(contract_type: &str) -> PhaseConfig {
match contract_type {
"simple" => PhaseConfig {
phases: vec![
PhaseDefinition { id: "plan".to_string(), name: "Plan".to_string(), order: 0 },
PhaseDefinition { id: "execute".to_string(), name: "Execute".to_string(), order: 1 },
],
default_phase: "plan".to_string(),
deliverables: [
("plan".to_string(), vec![DeliverableDefinition {
id: "plan-document".to_string(),
name: "Plan".to_string(),
priority: "required".to_string(),
}]),
("execute".to_string(), vec![DeliverableDefinition {
id: "pull-request".to_string(),
name: "Pull Request".to_string(),
priority: "required".to_string(),
}]),
].into_iter().collect(),
},
"specification" => PhaseConfig {
phases: vec![
PhaseDefinition { id: "research".to_string(), name: "Research".to_string(), order: 0 },
PhaseDefinition { id: "specify".to_string(), name: "Specify".to_string(), order: 1 },
PhaseDefinition { id: "plan".to_string(), name: "Plan".to_string(), order: 2 },
PhaseDefinition { id: "execute".to_string(), name: "Execute".to_string(), order: 3 },
PhaseDefinition { id: "review".to_string(), name: "Review".to_string(), order: 4 },
],
default_phase: "research".to_string(),
deliverables: [
("research".to_string(), vec![DeliverableDefinition {
id: "research-notes".to_string(),
name: "Research Notes".to_string(),
priority: "required".to_string(),
}]),
("specify".to_string(), vec![DeliverableDefinition {
id: "requirements-document".to_string(),
name: "Requirements Document".to_string(),
priority: "required".to_string(),
}]),
("plan".to_string(), vec![DeliverableDefinition {
id: "plan-document".to_string(),
name: "Plan".to_string(),
priority: "required".to_string(),
}]),
("execute".to_string(), vec![DeliverableDefinition {
id: "pull-request".to_string(),
name: "Pull Request".to_string(),
priority: "required".to_string(),
}]),
("review".to_string(), vec![DeliverableDefinition {
id: "release-notes".to_string(),
name: "Release Notes".to_string(),
priority: "required".to_string(),
}]),
].into_iter().collect(),
},
"execute" | _ => PhaseConfig {
phases: vec![
PhaseDefinition { id: "execute".to_string(), name: "Execute".to_string(), order: 0 },
],
default_phase: "execute".to_string(),
deliverables: std::collections::HashMap::new(),
},
}
}
// =============================================================================
// Contract Functions (Owner-Scoped)
// =============================================================================
/// Create a new contract for a specific owner.
/// Supports both built-in contract types (simple, specification, execute) and custom templates.
pub async fn create_contract_for_owner(
pool: &PgPool,
owner_id: Uuid,
req: CreateContractRequest,
) -> Result<Contract, sqlx::Error> {
// Determine phase configuration based on template_id or contract_type
let (phase_config, contract_type_str, default_phase): (PhaseConfig, String, String) =
if let Some(template_id) = req.template_id {
// Look up the custom template
let template = get_template_by_id(pool, template_id)
.await?
.ok_or_else(|| {
sqlx::Error::Protocol(format!("Template not found: {}", template_id))
})?;
let config = build_phase_config_from_template(&template);
let default = config.default_phase.clone();
// For custom templates, store the template name as the contract_type
(config, template.name.clone(), default)
} else {
// Use built-in contract type
let contract_type = req.contract_type.as_deref().unwrap_or("simple");
// Validate contract type
let valid_types = ["simple", "specification", "execute"];
if !valid_types.contains(&contract_type) {
return Err(sqlx::Error::Protocol(format!(
"Invalid contract_type '{}'. Must be one of: {} or provide a template_id",
contract_type,
valid_types.join(", ")
)));
}
let config = build_phase_config_for_builtin(contract_type);
let default = config.default_phase.clone();
(config, contract_type.to_string(), default)
};
// Get valid phase IDs from the configuration
let valid_phase_ids: Vec<String> = phase_config.phases.iter().map(|p| p.id.clone()).collect();
// Use provided initial_phase or default based on contract type/template
let phase = req.initial_phase.as_deref().unwrap_or(&default_phase);
// Validate the phase is valid for this contract type/template
if !valid_phase_ids.contains(&phase.to_string()) {
return Err(sqlx::Error::Protocol(format!(
"Invalid initial_phase '{}' for contract type '{}'. Must be one of: {}",
phase,
contract_type_str,
valid_phase_ids.join(", ")
)));
}
let autonomous_loop = req.autonomous_loop.unwrap_or(false);
let phase_guard = req.phase_guard.unwrap_or(false);
let local_only = req.local_only.unwrap_or(false);
let auto_merge_local = req.auto_merge_local.unwrap_or(false);
// Serialize phase_config to JSON
let phase_config_json = serde_json::to_value(&phase_config).ok();
sqlx::query_as::<_, Contract>(
r#"
INSERT INTO contracts (owner_id, name, description, contract_type, phase, autonomous_loop, phase_guard, local_only, auto_merge_local, phase_config)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
RETURNING *
"#,
)
.bind(owner_id)
.bind(&req.name)
.bind(&req.description)
.bind(&contract_type_str)
.bind(phase)
.bind(autonomous_loop)
.bind(phase_guard)
.bind(local_only)
.bind(auto_merge_local)
.bind(phase_config_json)
.fetch_one(pool)
.await
}
/// Get a contract by ID, scoped to owner.
pub async fn get_contract_for_owner(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
) -> Result<Option<Contract>, sqlx::Error> {
sqlx::query_as::<_, Contract>(
r#"
SELECT *
FROM contracts
WHERE id = $1 AND owner_id = $2
"#,
)
.bind(id)
.bind(owner_id)
.fetch_optional(pool)
.await
}
/// List all contracts for an owner, ordered by created_at DESC.
pub async fn list_contracts_for_owner(
pool: &PgPool,
owner_id: Uuid,
) -> Result<Vec<ContractSummary>, sqlx::Error> {
sqlx::query_as::<_, ContractSummary>(
r#"
SELECT
c.id, c.name, c.description, c.contract_type, c.phase, c.status,
c.supervisor_task_id, c.local_only, c.auto_merge_local, c.version, c.created_at,
(SELECT COUNT(*) FROM files WHERE contract_id = c.id) as file_count,
(SELECT COUNT(*) FROM tasks WHERE contract_id = c.id) as task_count,
(SELECT COUNT(*) FROM contract_repositories WHERE contract_id = c.id) as repository_count
FROM contracts c
WHERE c.owner_id = $1
ORDER BY c.created_at DESC
"#,
)
.bind(owner_id)
.fetch_all(pool)
.await
}
/// Get contract summary by ID.
pub async fn get_contract_summary_for_owner(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
) -> Result<Option<ContractSummary>, sqlx::Error> {
sqlx::query_as::<_, ContractSummary>(
r#"
SELECT
c.id, c.name, c.description, c.contract_type, c.phase, c.status,
c.supervisor_task_id, c.local_only, c.auto_merge_local, c.version, c.created_at,
(SELECT COUNT(*) FROM files WHERE contract_id = c.id) as file_count,
(SELECT COUNT(*) FROM tasks WHERE contract_id = c.id) as task_count,
(SELECT COUNT(*) FROM contract_repositories WHERE contract_id = c.id) as repository_count
FROM contracts c
WHERE c.id = $1 AND c.owner_id = $2
"#,
)
.bind(id)
.bind(owner_id)
.fetch_optional(pool)
.await
}
/// Update a contract by ID with optimistic locking, scoped to owner.
pub async fn update_contract_for_owner(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
req: UpdateContractRequest,
) -> Result<Option<Contract>, RepositoryError> {
let existing = get_contract_for_owner(pool, id, owner_id).await?;
let Some(existing) = existing else {
return Ok(None);
};
// Check version if provided (optimistic locking)
if let Some(expected_version) = req.version {
if existing.version != expected_version {
return Err(RepositoryError::VersionConflict {
expected: expected_version,
actual: existing.version,
});
}
}
// Apply updates
let name = req.name.unwrap_or(existing.name);
let description = req.description.or(existing.description);
let phase = req.phase.unwrap_or(existing.phase);
let status = req.status.unwrap_or(existing.status);
let supervisor_task_id = req.supervisor_task_id.or(existing.supervisor_task_id);
let autonomous_loop = req.autonomous_loop.unwrap_or(existing.autonomous_loop);
let phase_guard = req.phase_guard.unwrap_or(existing.phase_guard);
let local_only = req.local_only.unwrap_or(existing.local_only);
let auto_merge_local = req.auto_merge_local.unwrap_or(existing.auto_merge_local);
let result = if req.version.is_some() {
sqlx::query_as::<_, Contract>(
r#"
UPDATE contracts
SET name = $3, description = $4, phase = $5, status = $6,
supervisor_task_id = $7, autonomous_loop = $8, phase_guard = $9, local_only = $10, auto_merge_local = $11, version = version + 1, updated_at = NOW()
WHERE id = $1 AND owner_id = $2 AND version = $12
RETURNING *
"#,
)
.bind(id)
.bind(owner_id)
.bind(&name)
.bind(&description)
.bind(&phase)
.bind(&status)
.bind(supervisor_task_id)
.bind(autonomous_loop)
.bind(phase_guard)
.bind(local_only)
.bind(auto_merge_local)
.bind(req.version.unwrap())
.fetch_optional(pool)
.await?
} else {
sqlx::query_as::<_, Contract>(
r#"
UPDATE contracts
SET name = $3, description = $4, phase = $5, status = $6,
supervisor_task_id = $7, autonomous_loop = $8, phase_guard = $9, local_only = $10, auto_merge_local = $11, version = version + 1, updated_at = NOW()
WHERE id = $1 AND owner_id = $2
RETURNING *
"#,
)
.bind(id)
.bind(owner_id)
.bind(&name)
.bind(&description)
.bind(&phase)
.bind(&status)
.bind(supervisor_task_id)
.bind(autonomous_loop)
.bind(phase_guard)
.bind(local_only)
.bind(auto_merge_local)
.fetch_optional(pool)
.await?
};
// If versioned update returned None, there was a race condition
if result.is_none() && req.version.is_some() {
if let Some(current) = get_contract_for_owner(pool, id, owner_id).await? {
return Err(RepositoryError::VersionConflict {
expected: req.version.unwrap(),
actual: current.version,
});
}
}
Ok(result)
}
/// Delete a contract by ID, scoped to owner.
pub async fn delete_contract_for_owner(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
) -> Result<bool, sqlx::Error> {
let result = sqlx::query(
r#"
DELETE FROM contracts
WHERE id = $1 AND owner_id = $2
"#,
)
.bind(id)
.bind(owner_id)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Change contract phase and record event.
///
/// This is the simple version without version checking. Use `change_contract_phase_with_version`
/// for explicit version conflict detection.
pub async fn change_contract_phase_for_owner(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
new_phase: &str,
) -> Result<Option<Contract>, sqlx::Error> {
// Get current phase
let existing = get_contract_for_owner(pool, id, owner_id).await?;
let Some(existing) = existing else {
return Ok(None);
};
let previous_phase = existing.phase.clone();
// Update phase
let contract = sqlx::query_as::<_, Contract>(
r#"
UPDATE contracts
SET phase = $3, version = version + 1, updated_at = NOW()
WHERE id = $1 AND owner_id = $2
RETURNING *
"#,
)
.bind(id)
.bind(owner_id)
.bind(new_phase)
.fetch_optional(pool)
.await?;
// Record event
if contract.is_some() {
sqlx::query(
r#"
INSERT INTO contract_events (contract_id, event_type, previous_phase, new_phase)
VALUES ($1, 'phase_change', $2, $3)
"#,
)
.bind(id)
.bind(&previous_phase)
.bind(new_phase)
.execute(pool)
.await?;
}
Ok(contract)
}
/// Change contract phase with explicit version checking for conflict detection.
///
/// Uses `SELECT ... FOR UPDATE` to lock the row and prevent race conditions.
/// Returns `PhaseChangeResult::VersionConflict` if the expected version doesn't match.
pub async fn change_contract_phase_with_version(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
new_phase: &str,
expected_version: Option<i32>,
) -> Result<PhaseChangeResult, sqlx::Error> {
// Start a transaction to ensure atomicity with row locking
let mut tx = pool.begin().await?;
// Lock the row with SELECT FOR UPDATE and get current state
let existing: Option<Contract> = sqlx::query_as::<_, Contract>(
r#"
SELECT *
FROM contracts
WHERE id = $1 AND owner_id = $2
FOR UPDATE
"#,
)
.bind(id)
.bind(owner_id)
.fetch_optional(&mut *tx)
.await?;
let Some(existing) = existing else {
tx.rollback().await?;
return Ok(PhaseChangeResult::NotFound);
};
// Check version if provided (optimistic locking)
if let Some(expected) = expected_version {
if existing.version != expected {
tx.rollback().await?;
return Ok(PhaseChangeResult::VersionConflict {
expected,
actual: existing.version,
current_phase: existing.phase,
});
}
}
// Validate the phase transition is allowed
let valid_phases = existing.valid_phase_ids();
if !valid_phases.contains(&new_phase.to_string()) {
tx.rollback().await?;
return Ok(PhaseChangeResult::ValidationFailed {
reason: format!(
"Invalid phase '{}' for contract type '{}'",
new_phase, existing.contract_type
),
missing_requirements: vec![format!(
"Phase must be one of: {}",
valid_phases.join(", ")
)],
});
}
let previous_phase = existing.phase.clone();
// Update phase with version increment
let contract = sqlx::query_as::<_, Contract>(
r#"
UPDATE contracts
SET phase = $3, version = version + 1, updated_at = NOW()
WHERE id = $1 AND owner_id = $2
RETURNING *
"#,
)
.bind(id)
.bind(owner_id)
.bind(new_phase)
.fetch_one(&mut *tx)
.await?;
// Record event
sqlx::query(
r#"
INSERT INTO contract_events (contract_id, event_type, previous_phase, new_phase)
VALUES ($1, 'phase_change', $2, $3)
"#,
)
.bind(id)
.bind(&previous_phase)
.bind(new_phase)
.execute(&mut *tx)
.await?;
// Commit the transaction
tx.commit().await?;
Ok(PhaseChangeResult::Success(contract))
}
// =============================================================================
// Contract Repository Functions
// =============================================================================
/// List repositories for a contract.
pub async fn list_contract_repositories(
pool: &PgPool,
contract_id: Uuid,
) -> Result<Vec<ContractRepository>, sqlx::Error> {
sqlx::query_as::<_, ContractRepository>(
r#"
SELECT *
FROM contract_repositories
WHERE contract_id = $1
ORDER BY is_primary DESC, created_at ASC
"#,
)
.bind(contract_id)
.fetch_all(pool)
.await
}
/// Add a remote repository to a contract.
pub async fn add_remote_repository(
pool: &PgPool,
contract_id: Uuid,
name: &str,
repository_url: &str,
is_primary: bool,
) -> Result<ContractRepository, sqlx::Error> {
// If is_primary, clear other primaries first
if is_primary {
sqlx::query(
r#"
UPDATE contract_repositories
SET is_primary = false, updated_at = NOW()
WHERE contract_id = $1 AND is_primary = true
"#,
)
.bind(contract_id)
.execute(pool)
.await?;
}
sqlx::query_as::<_, ContractRepository>(
r#"
INSERT INTO contract_repositories (contract_id, name, repository_url, source_type, status, is_primary)
VALUES ($1, $2, $3, 'remote', 'ready', $4)
RETURNING *
"#,
)
.bind(contract_id)
.bind(name)
.bind(repository_url)
.bind(is_primary)
.fetch_one(pool)
.await
}
/// Add a local repository to a contract.
pub async fn add_local_repository(
pool: &PgPool,
contract_id: Uuid,
name: &str,
local_path: &str,
is_primary: bool,
) -> Result<ContractRepository, sqlx::Error> {
// If is_primary, clear other primaries first
if is_primary {
sqlx::query(
r#"
UPDATE contract_repositories
SET is_primary = false, updated_at = NOW()
WHERE contract_id = $1 AND is_primary = true
"#,
)
.bind(contract_id)
.execute(pool)
.await?;
}
sqlx::query_as::<_, ContractRepository>(
r#"
INSERT INTO contract_repositories (contract_id, name, local_path, source_type, status, is_primary)
VALUES ($1, $2, $3, 'local', 'ready', $4)
RETURNING *
"#,
)
.bind(contract_id)
.bind(name)
.bind(local_path)
.bind(is_primary)
.fetch_one(pool)
.await
}
/// Create a managed repository (daemon will create it).
pub async fn create_managed_repository(
pool: &PgPool,
contract_id: Uuid,
name: &str,
is_primary: bool,
) -> Result<ContractRepository, sqlx::Error> {
// If is_primary, clear other primaries first
if is_primary {
sqlx::query(
r#"
UPDATE contract_repositories
SET is_primary = false, updated_at = NOW()
WHERE contract_id = $1 AND is_primary = true
"#,
)
.bind(contract_id)
.execute(pool)
.await?;
}
sqlx::query_as::<_, ContractRepository>(
r#"
INSERT INTO contract_repositories (contract_id, name, source_type, status, is_primary)
VALUES ($1, $2, 'managed', 'pending', $3)
RETURNING *
"#,
)
.bind(contract_id)
.bind(name)
.bind(is_primary)
.fetch_one(pool)
.await
}
/// Delete a repository from a contract.
pub async fn delete_contract_repository(
pool: &PgPool,
repo_id: Uuid,
contract_id: Uuid,
) -> Result<bool, sqlx::Error> {
let result = sqlx::query(
r#"
DELETE FROM contract_repositories
WHERE id = $1 AND contract_id = $2
"#,
)
.bind(repo_id)
.bind(contract_id)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Set a repository as primary (and clear others).
pub async fn set_repository_primary(
pool: &PgPool,
repo_id: Uuid,
contract_id: Uuid,
) -> Result<bool, sqlx::Error> {
// Clear other primaries
sqlx::query(
r#"
UPDATE contract_repositories
SET is_primary = false, updated_at = NOW()
WHERE contract_id = $1 AND is_primary = true
"#,
)
.bind(contract_id)
.execute(pool)
.await?;
// Set this one as primary
let result = sqlx::query(
r#"
UPDATE contract_repositories
SET is_primary = true, updated_at = NOW()
WHERE id = $1 AND contract_id = $2
"#,
)
.bind(repo_id)
.bind(contract_id)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Update managed repository status (used by daemon).
pub async fn update_managed_repository_status(
pool: &PgPool,
repo_id: Uuid,
status: &str,
repository_url: Option<&str>,
) -> Result<Option<ContractRepository>, sqlx::Error> {
sqlx::query_as::<_, ContractRepository>(
r#"
UPDATE contract_repositories
SET status = $2, repository_url = COALESCE($3, repository_url), updated_at = NOW()
WHERE id = $1
RETURNING *
"#,
)
.bind(repo_id)
.bind(status)
.bind(repository_url)
.fetch_optional(pool)
.await
}
// =============================================================================
// Contract Task Association Functions
// =============================================================================
/// Add a task to a contract.
pub async fn add_task_to_contract(
pool: &PgPool,
contract_id: Uuid,
task_id: Uuid,
owner_id: Uuid,
) -> Result<bool, sqlx::Error> {
let result = sqlx::query(
r#"
UPDATE tasks
SET contract_id = $2, updated_at = NOW()
WHERE id = $1 AND owner_id = $3
"#,
)
.bind(task_id)
.bind(contract_id)
.bind(owner_id)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Remove a task from a contract.
pub async fn remove_task_from_contract(
pool: &PgPool,
contract_id: Uuid,
task_id: Uuid,
owner_id: Uuid,
) -> Result<bool, sqlx::Error> {
let result = sqlx::query(
r#"
UPDATE tasks
SET contract_id = NULL, updated_at = NOW()
WHERE id = $1 AND contract_id = $2 AND owner_id = $3
"#,
)
.bind(task_id)
.bind(contract_id)
.bind(owner_id)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// List files in a contract.
pub async fn list_files_in_contract(
pool: &PgPool,
contract_id: Uuid,
owner_id: Uuid,
) -> Result<Vec<FileSummary>, sqlx::Error> {
// Use a manual query since FileSummary doesn't have a FromRow derive with all the computed fields
let files = sqlx::query_as::<_, File>(
r#"
SELECT id, owner_id, contract_id, contract_phase, name, description, transcript, location, summary, body, version, repo_file_path, repo_synced_at, repo_sync_status, created_at, updated_at
FROM files
WHERE contract_id = $1 AND owner_id = $2
ORDER BY created_at DESC
"#,
)
.bind(contract_id)
.bind(owner_id)
.fetch_all(pool)
.await?;
Ok(files.into_iter().map(FileSummary::from).collect())
}
/// List tasks in a contract.
pub async fn list_tasks_in_contract(
pool: &PgPool,
contract_id: Uuid,
owner_id: Uuid,
) -> Result<Vec<TaskSummary>, sqlx::Error> {
sqlx::query_as::<_, TaskSummary>(
r#"
SELECT
t.id, t.contract_id, c.name as contract_name, c.phase as contract_phase,
c.status as contract_status,
t.parent_task_id, t.depth, t.name, t.status, t.priority,
t.progress_summary,
(SELECT COUNT(*) FROM tasks WHERE parent_task_id = t.id) as subtask_count,
t.version, t.is_supervisor, COALESCE(t.hidden, false) as hidden, t.created_at, t.updated_at
FROM tasks t
LEFT JOIN contracts c ON t.contract_id = c.id
WHERE t.contract_id = $1 AND t.owner_id = $2
ORDER BY t.priority DESC, t.created_at DESC
"#,
)
.bind(contract_id)
.bind(owner_id)
.fetch_all(pool)
.await
}
/// Minimal task info for worktree cleanup operations.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct TaskWorktreeInfo {
pub id: Uuid,
pub daemon_id: Option<Uuid>,
pub overlay_path: Option<String>,
/// If set, this task shares the worktree of the specified supervisor task.
/// Should NOT have its worktree deleted during cleanup.
pub supervisor_worktree_task_id: Option<Uuid>,
}
/// List tasks in a contract with their daemon/worktree info.
/// Used for cleaning up worktrees when a contract is completed or deleted.
pub async fn list_contract_tasks_with_worktree_info(
pool: &PgPool,
contract_id: Uuid,
) -> Result<Vec<TaskWorktreeInfo>, sqlx::Error> {
sqlx::query_as::<_, TaskWorktreeInfo>(
r#"
SELECT id, daemon_id, overlay_path, supervisor_worktree_task_id
FROM tasks
WHERE contract_id = $1 AND (daemon_id IS NOT NULL OR overlay_path IS NOT NULL)
"#,
)
.bind(contract_id)
.fetch_all(pool)
.await
}
// =============================================================================
// Contract Events
// =============================================================================
/// List events for a contract.
pub async fn list_contract_events(
pool: &PgPool,
contract_id: Uuid,
) -> Result<Vec<ContractEvent>, sqlx::Error> {
sqlx::query_as::<_, ContractEvent>(
r#"
SELECT *
FROM contract_events
WHERE contract_id = $1
ORDER BY created_at DESC
"#,
)
.bind(contract_id)
.fetch_all(pool)
.await
}
/// Record a contract event.
pub async fn record_contract_event(
pool: &PgPool,
contract_id: Uuid,
event_type: &str,
event_data: Option<serde_json::Value>,
) -> Result<ContractEvent, sqlx::Error> {
sqlx::query_as::<_, ContractEvent>(
r#"
INSERT INTO contract_events (contract_id, event_type, event_data)
VALUES ($1, $2, $3)
RETURNING *
"#,
)
.bind(contract_id)
.bind(event_type)
.bind(event_data)
.fetch_one(pool)
.await
}
// ============================================================================
// Task Checkpoints
// ============================================================================
/// Create a checkpoint for a task.
pub async fn create_task_checkpoint(
pool: &PgPool,
task_id: Uuid,
commit_sha: &str,
branch_name: &str,
message: &str,
files_changed: Option<serde_json::Value>,
lines_added: Option<i32>,
lines_removed: Option<i32>,
) -> Result<TaskCheckpoint, sqlx::Error> {
// Get current checkpoint count and increment
let checkpoint_number: i32 = sqlx::query_scalar(
"SELECT COALESCE(MAX(checkpoint_number), 0) + 1 FROM task_checkpoints WHERE task_id = $1",
)
.bind(task_id)
.fetch_one(pool)
.await?;
// Update task's checkpoint tracking
sqlx::query(
r#"
UPDATE tasks
SET last_checkpoint_sha = $1,
checkpoint_count = $2,
checkpoint_message = $3,
updated_at = NOW()
WHERE id = $4
"#,
)
.bind(commit_sha)
.bind(checkpoint_number)
.bind(message)
.bind(task_id)
.execute(pool)
.await?;
sqlx::query_as::<_, TaskCheckpoint>(
r#"
INSERT INTO task_checkpoints (
task_id, checkpoint_number, commit_sha, branch_name, message,
files_changed, lines_added, lines_removed
)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
RETURNING *
"#,
)
.bind(task_id)
.bind(checkpoint_number)
.bind(commit_sha)
.bind(branch_name)
.bind(message)
.bind(files_changed)
.bind(lines_added)
.bind(lines_removed)
.fetch_one(pool)
.await
}
/// Get a checkpoint by ID.
pub async fn get_task_checkpoint(
pool: &PgPool,
id: Uuid,
) -> Result<Option<TaskCheckpoint>, sqlx::Error> {
sqlx::query_as::<_, TaskCheckpoint>("SELECT * FROM task_checkpoints WHERE id = $1")
.bind(id)
.fetch_optional(pool)
.await
}
/// Get a checkpoint by commit SHA.
pub async fn get_task_checkpoint_by_sha(
pool: &PgPool,
commit_sha: &str,
) -> Result<Option<TaskCheckpoint>, sqlx::Error> {
sqlx::query_as::<_, TaskCheckpoint>("SELECT * FROM task_checkpoints WHERE commit_sha = $1")
.bind(commit_sha)
.fetch_optional(pool)
.await
}
/// List checkpoints for a task.
pub async fn list_task_checkpoints(
pool: &PgPool,
task_id: Uuid,
) -> Result<Vec<TaskCheckpoint>, sqlx::Error> {
sqlx::query_as::<_, TaskCheckpoint>(
"SELECT * FROM task_checkpoints WHERE task_id = $1 ORDER BY checkpoint_number DESC",
)
.bind(task_id)
.fetch_all(pool)
.await
}
// ============================================================================
// Supervisor State
// ============================================================================
/// Create or update supervisor state for a contract.
pub async fn upsert_supervisor_state(
pool: &PgPool,
contract_id: Uuid,
task_id: Uuid,
conversation_history: serde_json::Value,
pending_task_ids: &[Uuid],
phase: &str,
) -> Result<SupervisorState, sqlx::Error> {
sqlx::query_as::<_, SupervisorState>(
r#"
INSERT INTO supervisor_states (contract_id, task_id, conversation_history, pending_task_ids, phase, last_activity)
VALUES ($1, $2, $3, $4, $5, NOW())
ON CONFLICT (contract_id) DO UPDATE SET
task_id = EXCLUDED.task_id,
conversation_history = EXCLUDED.conversation_history,
pending_task_ids = EXCLUDED.pending_task_ids,
phase = EXCLUDED.phase,
last_activity = NOW(),
updated_at = NOW()
RETURNING *
"#,
)
.bind(contract_id)
.bind(task_id)
.bind(conversation_history)
.bind(pending_task_ids)
.bind(phase)
.fetch_one(pool)
.await
}
/// Get supervisor state for a contract.
pub async fn get_supervisor_state(
pool: &PgPool,
contract_id: Uuid,
) -> Result<Option<SupervisorState>, sqlx::Error> {
sqlx::query_as::<_, SupervisorState>("SELECT * FROM supervisor_states WHERE contract_id = $1")
.bind(contract_id)
.fetch_optional(pool)
.await
}
/// Get supervisor state by task ID.
pub async fn get_supervisor_state_by_task(
pool: &PgPool,
task_id: Uuid,
) -> Result<Option<SupervisorState>, sqlx::Error> {
sqlx::query_as::<_, SupervisorState>("SELECT * FROM supervisor_states WHERE task_id = $1")
.bind(task_id)
.fetch_optional(pool)
.await
}
/// Update supervisor conversation history.
pub async fn update_supervisor_conversation(
pool: &PgPool,
contract_id: Uuid,
conversation_history: serde_json::Value,
) -> Result<SupervisorState, sqlx::Error> {
sqlx::query_as::<_, SupervisorState>(
r#"
UPDATE supervisor_states
SET conversation_history = $1,
last_activity = NOW(),
updated_at = NOW()
WHERE contract_id = $2
RETURNING *
"#,
)
.bind(conversation_history)
.bind(contract_id)
.fetch_one(pool)
.await
}
/// Update supervisor pending tasks.
pub async fn update_supervisor_pending_tasks(
pool: &PgPool,
contract_id: Uuid,
pending_task_ids: &[Uuid],
) -> Result<SupervisorState, sqlx::Error> {
sqlx::query_as::<_, SupervisorState>(
r#"
UPDATE supervisor_states
SET pending_task_ids = $1,
last_activity = NOW(),
updated_at = NOW()
WHERE contract_id = $2
RETURNING *
"#,
)
.bind(pending_task_ids)
.bind(contract_id)
.fetch_one(pool)
.await
}
/// Update supervisor state with detailed activity tracking.
/// Called at key save points: LLM response, task spawn, question asked, phase change.
pub async fn update_supervisor_detailed_state(
pool: &PgPool,
contract_id: Uuid,
state: &str,
current_activity: Option<&str>,
progress: i32,
error_message: Option<&str>,
) -> Result<SupervisorState, sqlx::Error> {
sqlx::query_as::<_, SupervisorState>(
r#"
UPDATE supervisor_states
SET state = $1,
current_activity = $2,
progress = $3,
error_message = $4,
last_activity = NOW(),
updated_at = NOW()
WHERE contract_id = $5
RETURNING *
"#,
)
.bind(state)
.bind(current_activity)
.bind(progress)
.bind(error_message)
.bind(contract_id)
.fetch_one(pool)
.await
}
/// Add a spawned task ID to the supervisor's list.
pub async fn add_supervisor_spawned_task(
pool: &PgPool,
contract_id: Uuid,
task_id: Uuid,
) -> Result<SupervisorState, sqlx::Error> {
sqlx::query_as::<_, SupervisorState>(
r#"
UPDATE supervisor_states
SET spawned_task_ids = array_append(spawned_task_ids, $1),
last_activity = NOW(),
updated_at = NOW()
WHERE contract_id = $2
RETURNING *
"#,
)
.bind(task_id)
.bind(contract_id)
.fetch_one(pool)
.await
}
/// Add a pending question to the supervisor state.
pub async fn add_supervisor_pending_question(
pool: &PgPool,
contract_id: Uuid,
question: serde_json::Value,
) -> Result<SupervisorState, sqlx::Error> {
sqlx::query_as::<_, SupervisorState>(
r#"
UPDATE supervisor_states
SET pending_questions = pending_questions || $1::jsonb,
state = 'waiting_for_user',
last_activity = NOW(),
updated_at = NOW()
WHERE contract_id = $2
RETURNING *
"#,
)
.bind(question)
.bind(contract_id)
.fetch_one(pool)
.await
}
/// Remove a pending question by ID.
pub async fn remove_supervisor_pending_question(
pool: &PgPool,
contract_id: Uuid,
question_id: Uuid,
) -> Result<SupervisorState, sqlx::Error> {
sqlx::query_as::<_, SupervisorState>(
r#"
UPDATE supervisor_states
SET pending_questions = (
SELECT COALESCE(jsonb_agg(elem), '[]'::jsonb)
FROM jsonb_array_elements(pending_questions) elem
WHERE (elem->>'id')::uuid != $1
),
last_activity = NOW(),
updated_at = NOW()
WHERE contract_id = $2
RETURNING *
"#,
)
.bind(question_id)
.bind(contract_id)
.fetch_one(pool)
.await
}
/// Comprehensive state save - used at major save points.
pub async fn save_supervisor_state_full(
pool: &PgPool,
contract_id: Uuid,
task_id: Uuid,
conversation_history: serde_json::Value,
pending_task_ids: &[Uuid],
phase: &str,
state: &str,
current_activity: Option<&str>,
progress: i32,
error_message: Option<&str>,
spawned_task_ids: &[Uuid],
pending_questions: serde_json::Value,
) -> Result<SupervisorState, sqlx::Error> {
sqlx::query_as::<_, SupervisorState>(
r#"
INSERT INTO supervisor_states (
contract_id, task_id, conversation_history, pending_task_ids, phase,
state, current_activity, progress, error_message, spawned_task_ids,
pending_questions, last_activity
)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, NOW())
ON CONFLICT (contract_id) DO UPDATE SET
task_id = EXCLUDED.task_id,
conversation_history = EXCLUDED.conversation_history,
pending_task_ids = EXCLUDED.pending_task_ids,
phase = EXCLUDED.phase,
state = EXCLUDED.state,
current_activity = EXCLUDED.current_activity,
progress = EXCLUDED.progress,
error_message = EXCLUDED.error_message,
spawned_task_ids = EXCLUDED.spawned_task_ids,
pending_questions = EXCLUDED.pending_questions,
last_activity = NOW(),
updated_at = NOW()
RETURNING *
"#,
)
.bind(contract_id)
.bind(task_id)
.bind(conversation_history)
.bind(pending_task_ids)
.bind(phase)
.bind(state)
.bind(current_activity)
.bind(progress)
.bind(error_message)
.bind(spawned_task_ids)
.bind(pending_questions)
.fetch_one(pool)
.await
}
/// Mark supervisor as restored from a crash/interruption.
pub async fn mark_supervisor_restored(
pool: &PgPool,
contract_id: Uuid,
restoration_source: &str,
) -> Result<SupervisorState, sqlx::Error> {
sqlx::query_as::<_, SupervisorState>(
r#"
UPDATE supervisor_states
SET restoration_count = restoration_count + 1,
last_restored_at = NOW(),
restoration_source = $1,
state = 'initializing',
error_message = NULL,
last_activity = NOW(),
updated_at = NOW()
WHERE contract_id = $2
RETURNING *
"#,
)
.bind(restoration_source)
.bind(contract_id)
.fetch_one(pool)
.await
}
/// Get supervisors with pending questions (for re-delivery after restoration).
pub async fn get_supervisors_with_pending_questions(
pool: &PgPool,
owner_id: Uuid,
) -> Result<Vec<SupervisorState>, sqlx::Error> {
sqlx::query_as::<_, SupervisorState>(
r#"
SELECT ss.*
FROM supervisor_states ss
JOIN contracts c ON c.id = ss.contract_id
WHERE c.owner_id = $1
AND ss.pending_questions != '[]'::jsonb
AND c.status = 'active'
ORDER BY ss.last_activity DESC
"#,
)
.bind(owner_id)
.fetch_all(pool)
.await
}
/// Get supervisor state with full details for restoration.
/// Includes validation info.
pub async fn get_supervisor_state_for_restoration(
pool: &PgPool,
contract_id: Uuid,
) -> Result<Option<SupervisorState>, sqlx::Error> {
sqlx::query_as::<_, SupervisorState>(
r#"
SELECT * FROM supervisor_states WHERE contract_id = $1
"#,
)
.bind(contract_id)
.fetch_optional(pool)
.await
}
/// Validate spawned tasks are in expected states.
/// Returns map of task_id -> (status, updated_at).
pub async fn validate_spawned_tasks(
pool: &PgPool,
task_ids: &[Uuid],
) -> Result<std::collections::HashMap<Uuid, (String, chrono::DateTime<Utc>)>, sqlx::Error> {
use sqlx::Row;
let rows = sqlx::query(
r#"
SELECT id, status, updated_at
FROM tasks
WHERE id = ANY($1)
"#,
)
.bind(task_ids)
.fetch_all(pool)
.await?;
let mut result = std::collections::HashMap::new();
for row in rows {
let id: Uuid = row.get("id");
let status: String = row.get("status");
let updated_at: chrono::DateTime<Utc> = row.get("updated_at");
result.insert(id, (status, updated_at));
}
Ok(result)
}
/// Update supervisor state when phase changes.
pub async fn update_supervisor_phase(
pool: &PgPool,
contract_id: Uuid,
new_phase: &str,
) -> Result<SupervisorState, sqlx::Error> {
sqlx::query_as::<_, SupervisorState>(
r#"
UPDATE supervisor_states
SET phase = $1,
state = 'working',
current_activity = 'Phase changed to ' || $1,
last_activity = NOW(),
updated_at = NOW()
WHERE contract_id = $2
RETURNING *
"#,
)
.bind(new_phase)
.bind(contract_id)
.fetch_one(pool)
.await
}
/// Update supervisor state on heartbeat (lightweight update).
pub async fn update_supervisor_heartbeat_state(
pool: &PgPool,
contract_id: Uuid,
state: &str,
current_activity: Option<&str>,
progress: i32,
pending_task_ids: &[Uuid],
) -> Result<(), sqlx::Error> {
sqlx::query(
r#"
UPDATE supervisor_states
SET state = $1,
current_activity = $2,
progress = $3,
pending_task_ids = $4,
last_activity = NOW(),
updated_at = NOW()
WHERE contract_id = $5
"#,
)
.bind(state)
.bind(current_activity)
.bind(progress)
.bind(pending_task_ids)
.bind(contract_id)
.execute(pool)
.await?;
Ok(())
}
// ============================================================================
// Supervisor Heartbeats
// ============================================================================
/// Record a supervisor heartbeat.
/// This creates a historical record for monitoring and dead supervisor detection.
pub async fn create_supervisor_heartbeat(
pool: &PgPool,
supervisor_task_id: Uuid,
contract_id: Uuid,
state: &str,
phase: &str,
current_activity: Option<&str>,
progress: i32,
pending_task_ids: &[Uuid],
) -> Result<SupervisorHeartbeatRecord, sqlx::Error> {
sqlx::query_as::<_, SupervisorHeartbeatRecord>(
r#"
INSERT INTO supervisor_heartbeats (
supervisor_task_id, contract_id, state, phase, current_activity, progress, pending_task_ids, timestamp
)
VALUES ($1, $2, $3, $4, $5, $6, $7, NOW())
RETURNING *
"#,
)
.bind(supervisor_task_id)
.bind(contract_id)
.bind(state)
.bind(phase)
.bind(current_activity)
.bind(progress)
.bind(pending_task_ids)
.fetch_one(pool)
.await
}
/// Get the latest heartbeat for a supervisor task.
pub async fn get_latest_supervisor_heartbeat(
pool: &PgPool,
supervisor_task_id: Uuid,
) -> Result<Option<SupervisorHeartbeatRecord>, sqlx::Error> {
sqlx::query_as::<_, SupervisorHeartbeatRecord>(
r#"
SELECT * FROM supervisor_heartbeats
WHERE supervisor_task_id = $1
ORDER BY timestamp DESC
LIMIT 1
"#,
)
.bind(supervisor_task_id)
.fetch_optional(pool)
.await
}
/// Get recent heartbeats for a supervisor task.
pub async fn get_supervisor_heartbeats(
pool: &PgPool,
supervisor_task_id: Uuid,
limit: i64,
) -> Result<Vec<SupervisorHeartbeatRecord>, sqlx::Error> {
sqlx::query_as::<_, SupervisorHeartbeatRecord>(
r#"
SELECT * FROM supervisor_heartbeats
WHERE supervisor_task_id = $1
ORDER BY timestamp DESC
LIMIT $2
"#,
)
.bind(supervisor_task_id)
.bind(limit)
.fetch_all(pool)
.await
}
/// Get recent heartbeats for a contract.
pub async fn get_contract_supervisor_heartbeats(
pool: &PgPool,
contract_id: Uuid,
limit: i64,
) -> Result<Vec<SupervisorHeartbeatRecord>, sqlx::Error> {
sqlx::query_as::<_, SupervisorHeartbeatRecord>(
r#"
SELECT * FROM supervisor_heartbeats
WHERE contract_id = $1
ORDER BY timestamp DESC
LIMIT $2
"#,
)
.bind(contract_id)
.bind(limit)
.fetch_all(pool)
.await
}
/// Delete old heartbeats beyond the TTL (24 hours by default).
/// Returns the number of deleted records.
pub async fn cleanup_old_heartbeats(
pool: &PgPool,
ttl_hours: i64,
) -> Result<u64, sqlx::Error> {
let result = sqlx::query(
r#"
DELETE FROM supervisor_heartbeats
WHERE timestamp < NOW() - ($1 || ' hours')::INTERVAL
"#,
)
.bind(ttl_hours.to_string())
.execute(pool)
.await?;
Ok(result.rows_affected())
}
/// Find supervisors that have not sent a heartbeat within the timeout period.
/// Returns list of (supervisor_task_id, contract_id, last_heartbeat_timestamp).
pub async fn find_stale_supervisors(
pool: &PgPool,
timeout_seconds: i64,
) -> Result<Vec<(Uuid, Uuid, chrono::DateTime<Utc>)>, sqlx::Error> {
let rows = sqlx::query(
r#"
WITH latest_heartbeats AS (
SELECT DISTINCT ON (supervisor_task_id)
supervisor_task_id,
contract_id,
timestamp
FROM supervisor_heartbeats
ORDER BY supervisor_task_id, timestamp DESC
)
SELECT
lh.supervisor_task_id,
lh.contract_id,
lh.timestamp
FROM latest_heartbeats lh
JOIN tasks t ON t.id = lh.supervisor_task_id
WHERE t.status = 'running'
AND lh.timestamp < NOW() - ($1 || ' seconds')::INTERVAL
"#,
)
.bind(timeout_seconds.to_string())
.fetch_all(pool)
.await?;
let mut result = Vec::new();
for row in rows {
use sqlx::Row;
let supervisor_task_id: Uuid = row.get("supervisor_task_id");
let contract_id: Uuid = row.get("contract_id");
let timestamp: chrono::DateTime<Utc> = row.get("timestamp");
result.push((supervisor_task_id, contract_id, timestamp));
}
Ok(result)
}
// ============================================================================
// Contract Supervisor
// ============================================================================
/// Update contract's supervisor task ID.
pub async fn update_contract_supervisor(
pool: &PgPool,
contract_id: Uuid,
supervisor_task_id: Uuid,
) -> Result<Contract, sqlx::Error> {
sqlx::query_as::<_, Contract>(
r#"
UPDATE contracts
SET supervisor_task_id = $1,
updated_at = NOW()
WHERE id = $2
RETURNING *
"#,
)
.bind(supervisor_task_id)
.bind(contract_id)
.fetch_one(pool)
.await
}
/// Mark a deliverable as complete for a specific phase.
/// Uses JSONB operations to append the deliverable_id to the phase's array.
pub async fn mark_deliverable_complete(
pool: &PgPool,
contract_id: Uuid,
phase: &str,
deliverable_id: &str,
) -> Result<Contract, sqlx::Error> {
// Use jsonb_set to add the deliverable to the phase's array
// If the phase key doesn't exist, create an empty array first
// COALESCE handles the case where the phase array doesn't exist yet
sqlx::query_as::<_, Contract>(
r#"
UPDATE contracts
SET completed_deliverables = jsonb_set(
completed_deliverables,
ARRAY[$2::text],
COALESCE(completed_deliverables->$2, '[]'::jsonb) || to_jsonb($3::text),
true
),
updated_at = NOW()
WHERE id = $1
AND NOT (COALESCE(completed_deliverables->$2, '[]'::jsonb) ? $3)
RETURNING *
"#,
)
.bind(contract_id)
.bind(phase)
.bind(deliverable_id)
.fetch_one(pool)
.await
}
/// Clear all completed deliverables for a specific phase.
/// Used when phase changes or deliverables need to be reset.
pub async fn clear_phase_deliverables(
pool: &PgPool,
contract_id: Uuid,
phase: &str,
) -> Result<Contract, sqlx::Error> {
sqlx::query_as::<_, Contract>(
r#"
UPDATE contracts
SET completed_deliverables = completed_deliverables - $2,
updated_at = NOW()
WHERE id = $1
RETURNING *
"#,
)
.bind(contract_id)
.bind(phase)
.fetch_one(pool)
.await
}
/// Get the supervisor task for a contract.
pub async fn get_contract_supervisor_task(
pool: &PgPool,
contract_id: Uuid,
) -> Result<Option<Task>, sqlx::Error> {
sqlx::query_as::<_, Task>(
r#"
SELECT t.* FROM tasks t
JOIN contracts c ON c.supervisor_task_id = t.id
WHERE c.id = $1
"#,
)
.bind(contract_id)
.fetch_optional(pool)
.await
}
// ============================================================================
// Task Tree Queries
// ============================================================================
/// Get full task tree for a contract.
pub async fn get_contract_task_tree(
pool: &PgPool,
contract_id: Uuid,
) -> Result<Vec<Task>, sqlx::Error> {
sqlx::query_as::<_, Task>(
r#"
WITH RECURSIVE task_tree AS (
-- Base case: root tasks (no parent)
SELECT * FROM tasks
WHERE contract_id = $1 AND parent_task_id IS NULL
UNION ALL
-- Recursive case: children of current level
SELECT t.* FROM tasks t
JOIN task_tree tt ON t.parent_task_id = tt.id
)
SELECT * FROM task_tree
ORDER BY depth, created_at
"#,
)
.bind(contract_id)
.fetch_all(pool)
.await
}
/// Get task tree from a specific root task.
pub async fn get_task_tree(pool: &PgPool, root_task_id: Uuid) -> Result<Vec<Task>, sqlx::Error> {
sqlx::query_as::<_, Task>(
r#"
WITH RECURSIVE task_tree AS (
-- Base case: the root task
SELECT * FROM tasks WHERE id = $1
UNION ALL
-- Recursive case: children of current level
SELECT t.* FROM tasks t
JOIN task_tree tt ON t.parent_task_id = tt.id
)
SELECT * FROM task_tree
ORDER BY depth, created_at
"#,
)
.bind(root_task_id)
.fetch_all(pool)
.await
}
// ============================================================================
// Daemon Selection
// ============================================================================
/// Get daemons with capacity info for selection.
pub async fn get_available_daemons(
pool: &PgPool,
owner_id: Uuid,
) -> Result<Vec<DaemonWithCapacity>, sqlx::Error> {
sqlx::query_as::<_, DaemonWithCapacity>(
r#"
SELECT id, owner_id, connection_id, hostname, machine_id,
max_concurrent_tasks, current_task_count,
capacity_score, task_queue_length, supports_migration,
status, last_heartbeat_at, connected_at
FROM daemons
WHERE owner_id = $1 AND status = 'connected'
ORDER BY
COALESCE(capacity_score, 100) DESC,
(max_concurrent_tasks - current_task_count) DESC,
COALESCE(task_queue_length, 0) ASC
"#,
)
.bind(owner_id)
.fetch_all(pool)
.await
}
/// Get daemons with capacity info for selection, excluding specified daemon IDs.
/// Used for task retry to avoid reassigning to daemons that have already failed.
pub async fn get_available_daemons_excluding(
pool: &PgPool,
owner_id: Uuid,
exclude_daemon_ids: &[Uuid],
) -> Result<Vec<DaemonWithCapacity>, sqlx::Error> {
sqlx::query_as::<_, DaemonWithCapacity>(
r#"
SELECT id, owner_id, connection_id, hostname, machine_id,
max_concurrent_tasks, current_task_count,
capacity_score, task_queue_length, supports_migration,
status, last_heartbeat_at, connected_at
FROM daemons
WHERE owner_id = $1
AND status = 'connected'
AND id != ALL($2)
ORDER BY
COALESCE(capacity_score, 100) DESC,
(max_concurrent_tasks - current_task_count) DESC,
COALESCE(task_queue_length, 0) ASC
"#,
)
.bind(owner_id)
.bind(exclude_daemon_ids)
.fetch_all(pool)
.await
}
/// Create a daemon task assignment.
pub async fn create_daemon_task_assignment(
pool: &PgPool,
daemon_id: Uuid,
task_id: Uuid,
) -> Result<DaemonTaskAssignment, sqlx::Error> {
sqlx::query_as::<_, DaemonTaskAssignment>(
r#"
INSERT INTO daemon_task_assignments (daemon_id, task_id)
VALUES ($1, $2)
RETURNING *
"#,
)
.bind(daemon_id)
.bind(task_id)
.fetch_one(pool)
.await
}
/// Update daemon task assignment status.
pub async fn update_daemon_task_assignment_status(
pool: &PgPool,
task_id: Uuid,
status: &str,
) -> Result<DaemonTaskAssignment, sqlx::Error> {
sqlx::query_as::<_, DaemonTaskAssignment>(
r#"
UPDATE daemon_task_assignments
SET status = $1
WHERE task_id = $2
RETURNING *
"#,
)
.bind(status)
.bind(task_id)
.fetch_one(pool)
.await
}
/// Get daemon task assignment for a task.
pub async fn get_daemon_task_assignment(
pool: &PgPool,
task_id: Uuid,
) -> Result<Option<DaemonTaskAssignment>, sqlx::Error> {
sqlx::query_as::<_, DaemonTaskAssignment>(
"SELECT * FROM daemon_task_assignments WHERE task_id = $1",
)
.bind(task_id)
.fetch_optional(pool)
.await
}
// ============================================================================
// Repository History Functions
// ============================================================================
use super::models::RepositoryHistoryEntry;
/// List all repository history entries for an owner, ordered by use_count DESC, last_used_at DESC.
pub async fn list_repository_history_for_owner(
pool: &PgPool,
owner_id: Uuid,
) -> Result<Vec<RepositoryHistoryEntry>, sqlx::Error> {
sqlx::query_as::<_, RepositoryHistoryEntry>(
r#"
SELECT id, owner_id, name, repository_url, local_path, source_type, use_count, last_used_at, created_at
FROM repository_history
WHERE owner_id = $1
ORDER BY use_count DESC, last_used_at DESC
"#,
)
.bind(owner_id)
.fetch_all(pool)
.await
}
/// Get repository suggestions for an owner, optionally filtered by source type and query.
pub async fn get_repository_suggestions(
pool: &PgPool,
owner_id: Uuid,
source_type: Option<&str>,
query: Option<&str>,
limit: i32,
) -> Result<Vec<RepositoryHistoryEntry>, sqlx::Error> {
// Build query dynamically based on filters
let mut sql = String::from(
r#"
SELECT id, owner_id, name, repository_url, local_path, source_type, use_count, last_used_at, created_at
FROM repository_history
WHERE owner_id = $1
"#,
);
let mut param_idx = 2;
if source_type.is_some() {
sql.push_str(&format!(" AND source_type = ${}", param_idx));
param_idx += 1;
}
if query.is_some() {
sql.push_str(&format!(
" AND (LOWER(name) LIKE ${} OR LOWER(COALESCE(repository_url, '')) LIKE ${} OR LOWER(COALESCE(local_path, '')) LIKE ${})",
param_idx, param_idx, param_idx
));
param_idx += 1;
}
sql.push_str(&format!(
" ORDER BY use_count DESC, last_used_at DESC LIMIT ${}",
param_idx
));
// Build and execute query with the appropriate bindings
let mut query_builder = sqlx::query_as::<_, RepositoryHistoryEntry>(&sql).bind(owner_id);
if let Some(st) = source_type {
query_builder = query_builder.bind(st);
}
if let Some(q) = query {
let search_pattern = format!("%{}%", q.to_lowercase());
query_builder = query_builder.bind(search_pattern);
}
query_builder = query_builder.bind(limit);
query_builder.fetch_all(pool).await
}
/// Add or update a repository history entry.
/// If an entry with the same URL (for remote) or path (for local) already exists,
/// increment use_count and update last_used_at and name.
/// Otherwise, create a new entry.
pub async fn add_or_update_repository_history(
pool: &PgPool,
owner_id: Uuid,
name: &str,
repository_url: Option<&str>,
local_path: Option<&str>,
source_type: &str,
) -> Result<RepositoryHistoryEntry, sqlx::Error> {
// Use UPSERT (INSERT ... ON CONFLICT)
if source_type == "remote" {
let url = repository_url.ok_or_else(|| {
sqlx::Error::Protocol("repository_url required for remote type".to_string())
})?;
sqlx::query_as::<_, RepositoryHistoryEntry>(
r#"
INSERT INTO repository_history (owner_id, name, repository_url, local_path, source_type, use_count, last_used_at)
VALUES ($1, $2, $3, NULL, $4, 1, NOW())
ON CONFLICT (owner_id, repository_url) WHERE source_type = 'remote' AND repository_url IS NOT NULL
DO UPDATE SET
name = EXCLUDED.name,
use_count = repository_history.use_count + 1,
last_used_at = NOW()
RETURNING id, owner_id, name, repository_url, local_path, source_type, use_count, last_used_at, created_at
"#,
)
.bind(owner_id)
.bind(name)
.bind(url)
.bind(source_type)
.fetch_one(pool)
.await
} else if source_type == "local" {
let path = local_path.ok_or_else(|| {
sqlx::Error::Protocol("local_path required for local type".to_string())
})?;
sqlx::query_as::<_, RepositoryHistoryEntry>(
r#"
INSERT INTO repository_history (owner_id, name, repository_url, local_path, source_type, use_count, last_used_at)
VALUES ($1, $2, NULL, $3, $4, 1, NOW())
ON CONFLICT (owner_id, local_path) WHERE source_type = 'local' AND local_path IS NOT NULL
DO UPDATE SET
name = EXCLUDED.name,
use_count = repository_history.use_count + 1,
last_used_at = NOW()
RETURNING id, owner_id, name, repository_url, local_path, source_type, use_count, last_used_at, created_at
"#,
)
.bind(owner_id)
.bind(name)
.bind(path)
.bind(source_type)
.fetch_one(pool)
.await
} else {
Err(sqlx::Error::Protocol(format!(
"Invalid source_type: {}",
source_type
)))
}
}
/// Delete a repository history entry.
/// Returns true if an entry was deleted, false if not found.
pub async fn delete_repository_history(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
) -> Result<bool, sqlx::Error> {
let result = sqlx::query(
r#"
DELETE FROM repository_history
WHERE id = $1 AND owner_id = $2
"#,
)
.bind(id)
.bind(owner_id)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
// ============================================================================
// Conversation Snapshots
// ============================================================================
/// Create a new conversation snapshot
pub async fn create_conversation_snapshot(
pool: &PgPool,
task_id: Uuid,
checkpoint_id: Option<Uuid>,
snapshot_type: &str,
message_count: i32,
conversation_state: serde_json::Value,
metadata: Option<serde_json::Value>,
) -> Result<ConversationSnapshot, sqlx::Error> {
sqlx::query_as::<_, ConversationSnapshot>(
r#"
INSERT INTO conversation_snapshots (task_id, checkpoint_id, snapshot_type, message_count, conversation_state, metadata)
VALUES ($1, $2, $3, $4, $5, $6)
RETURNING *
"#
)
.bind(task_id)
.bind(checkpoint_id)
.bind(snapshot_type)
.bind(message_count)
.bind(conversation_state)
.bind(metadata)
.fetch_one(pool)
.await
}
/// Get a conversation snapshot by ID
pub async fn get_conversation_snapshot(
pool: &PgPool,
id: Uuid,
) -> Result<Option<ConversationSnapshot>, sqlx::Error> {
sqlx::query_as::<_, ConversationSnapshot>(
"SELECT * FROM conversation_snapshots WHERE id = $1"
)
.bind(id)
.fetch_optional(pool)
.await
}
/// Get conversation snapshot at a specific checkpoint
pub async fn get_conversation_at_checkpoint(
pool: &PgPool,
checkpoint_id: Uuid,
) -> Result<Option<ConversationSnapshot>, sqlx::Error> {
sqlx::query_as::<_, ConversationSnapshot>(
"SELECT * FROM conversation_snapshots WHERE checkpoint_id = $1 ORDER BY created_at DESC LIMIT 1"
)
.bind(checkpoint_id)
.fetch_optional(pool)
.await
}
/// List conversation snapshots for a task
pub async fn list_conversation_snapshots(
pool: &PgPool,
task_id: Uuid,
limit: Option<i32>,
) -> Result<Vec<ConversationSnapshot>, sqlx::Error> {
let limit = limit.unwrap_or(100);
sqlx::query_as::<_, ConversationSnapshot>(
"SELECT * FROM conversation_snapshots WHERE task_id = $1 ORDER BY created_at DESC LIMIT $2"
)
.bind(task_id)
.bind(limit)
.fetch_all(pool)
.await
}
/// Delete conversation snapshots older than retention period
pub async fn cleanup_old_snapshots(
pool: &PgPool,
retention_days: i32,
) -> Result<u64, sqlx::Error> {
let result = sqlx::query(
"DELETE FROM conversation_snapshots WHERE created_at < NOW() - INTERVAL '1 day' * $1"
)
.bind(retention_days)
.execute(pool)
.await?;
Ok(result.rows_affected())
}
// ============================================================================
// History Events
// ============================================================================
/// Record a new history event
#[allow(clippy::too_many_arguments)]
pub async fn record_history_event(
pool: &PgPool,
owner_id: Uuid,
contract_id: Option<Uuid>,
task_id: Option<Uuid>,
event_type: &str,
event_subtype: Option<&str>,
phase: Option<&str>,
event_data: serde_json::Value,
) -> Result<HistoryEvent, sqlx::Error> {
sqlx::query_as::<_, HistoryEvent>(
r#"
INSERT INTO history_events (owner_id, contract_id, task_id, event_type, event_subtype, phase, event_data)
VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING *
"#
)
.bind(owner_id)
.bind(contract_id)
.bind(task_id)
.bind(event_type)
.bind(event_subtype)
.bind(phase)
.bind(event_data)
.fetch_one(pool)
.await
}
/// Get contract history timeline
pub async fn get_contract_history(
pool: &PgPool,
contract_id: Uuid,
owner_id: Uuid,
filters: &HistoryQueryFilters,
) -> Result<(Vec<HistoryEvent>, i64), sqlx::Error> {
let limit = filters.limit.unwrap_or(100);
let mut query = String::from(
"SELECT * FROM history_events WHERE contract_id = $1 AND owner_id = $2"
);
let mut count_query = String::from(
"SELECT COUNT(*) FROM history_events WHERE contract_id = $1 AND owner_id = $2"
);
let mut param_count = 2;
if filters.phase.is_some() {
param_count += 1;
query.push_str(&format!(" AND phase = ${}" , param_count));
count_query.push_str(&format!(" AND phase = ${}", param_count));
}
if filters.from.is_some() {
param_count += 1;
query.push_str(&format!(" AND created_at >= ${}", param_count));
count_query.push_str(&format!(" AND created_at >= ${}", param_count));
}
if filters.to.is_some() {
param_count += 1;
query.push_str(&format!(" AND created_at <= ${}", param_count));
count_query.push_str(&format!(" AND created_at <= ${}", param_count));
}
query.push_str(" ORDER BY created_at DESC");
query.push_str(&format!(" LIMIT {}", limit));
// Build and execute the query dynamically
let mut q = sqlx::query_as::<_, HistoryEvent>(&query)
.bind(contract_id)
.bind(owner_id);
if let Some(ref phase) = filters.phase {
q = q.bind(phase);
}
if let Some(ref from) = filters.from {
q = q.bind(from);
}
if let Some(ref to) = filters.to {
q = q.bind(to);
}
let events = q.fetch_all(pool).await?;
// Get total count
let mut cq = sqlx::query_scalar::<_, i64>(&count_query)
.bind(contract_id)
.bind(owner_id);
if let Some(ref phase) = filters.phase {
cq = cq.bind(phase);
}
if let Some(ref from) = filters.from {
cq = cq.bind(from);
}
if let Some(ref to) = filters.to {
cq = cq.bind(to);
}
let count = cq.fetch_one(pool).await?;
Ok((events, count))
}
/// Get task history
pub async fn get_task_history(
pool: &PgPool,
task_id: Uuid,
owner_id: Uuid,
filters: &HistoryQueryFilters,
) -> Result<(Vec<HistoryEvent>, i64), sqlx::Error> {
let limit = filters.limit.unwrap_or(100);
let events = sqlx::query_as::<_, HistoryEvent>(
r#"
SELECT * FROM history_events
WHERE task_id = $1 AND owner_id = $2
ORDER BY created_at DESC
LIMIT $3
"#
)
.bind(task_id)
.bind(owner_id)
.bind(limit)
.fetch_all(pool)
.await?;
let count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM history_events WHERE task_id = $1 AND owner_id = $2"
)
.bind(task_id)
.bind(owner_id)
.fetch_one(pool)
.await?;
Ok((events, count))
}
/// Get unified timeline for an owner
pub async fn get_timeline(
pool: &PgPool,
owner_id: Uuid,
filters: &HistoryQueryFilters,
) -> Result<(Vec<HistoryEvent>, i64), sqlx::Error> {
let limit = filters.limit.unwrap_or(100);
let events = sqlx::query_as::<_, HistoryEvent>(
r#"
SELECT * FROM history_events
WHERE owner_id = $1
ORDER BY created_at DESC
LIMIT $2
"#
)
.bind(owner_id)
.bind(limit)
.fetch_all(pool)
.await?;
let count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM history_events WHERE owner_id = $1"
)
.bind(owner_id)
.fetch_one(pool)
.await?;
Ok((events, count))
}
// ============================================================================
// Task Conversation Retrieval
// ============================================================================
// Helper struct for parsing task output events
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct TaskOutputEvent {
message_type: String,
content: Option<String>,
tool_name: Option<String>,
tool_input: Option<serde_json::Value>,
is_error: Option<bool>,
cost_usd: Option<f32>,
}
/// Get task conversation messages (reconstructed from task_events)
pub async fn get_task_conversation(
pool: &PgPool,
task_id: Uuid,
include_tool_calls: bool,
include_tool_results: bool,
limit: Option<i32>,
) -> Result<Vec<ConversationMessage>, sqlx::Error> {
let limit = limit.unwrap_or(1000);
// Get output events that represent conversation turns
let events = sqlx::query_as::<_, TaskEvent>(
r#"
SELECT * FROM task_events
WHERE task_id = $1 AND event_type = 'output'
ORDER BY created_at ASC
LIMIT $2
"#
)
.bind(task_id)
.bind(limit)
.fetch_all(pool)
.await?;
// Convert task events to conversation messages
let mut messages = Vec::new();
for event in events {
if let Some(data) = event.event_data {
// Parse the event data to extract message info
if let Ok(output) = serde_json::from_value::<TaskOutputEvent>(data.clone()) {
let should_include = match output.message_type.as_str() {
"tool_use" => include_tool_calls,
"tool_result" => include_tool_results,
_ => true,
};
if should_include {
messages.push(ConversationMessage {
id: event.id.to_string(),
role: match output.message_type.as_str() {
"assistant" => "assistant".to_string(),
"tool_use" => "assistant".to_string(),
"tool_result" => "tool".to_string(),
"system" => "system".to_string(),
"error" => "system".to_string(),
_ => "user".to_string(),
},
content: output.content.unwrap_or_default(),
timestamp: event.created_at,
tool_calls: None,
tool_name: output.tool_name,
tool_input: output.tool_input,
tool_result: None,
is_error: output.is_error,
token_count: None,
cost_usd: output.cost_usd.map(|c| c as f64),
});
}
}
}
}
Ok(messages)
}
/// Get supervisor conversation (from supervisor_states)
pub async fn get_supervisor_conversation_full(
pool: &PgPool,
contract_id: Uuid,
) -> Result<Option<SupervisorState>, sqlx::Error> {
get_supervisor_state(pool, contract_id).await
}
// =============================================================================
// Anonymous Task Cleanup Functions
// =============================================================================
/// Delete stale anonymous tasks (tasks with contract_id = NULL) that:
/// - Are in a terminal state (done, failed, merged)
/// - Are older than the specified number of days
///
/// Returns the number of deleted tasks.
pub async fn cleanup_stale_anonymous_tasks(
pool: &PgPool,
max_age_days: i32,
) -> Result<i64, sqlx::Error> {
let result = sqlx::query(
r#"
DELETE FROM tasks
WHERE contract_id IS NULL
AND status IN ('done', 'failed', 'merged')
AND created_at < NOW() - INTERVAL '1 day' * $1
"#,
)
.bind(max_age_days)
.execute(pool)
.await?;
Ok(result.rows_affected() as i64)
}
// ============================================================================
// Checkpoint Patches (for task recovery)
// ============================================================================
/// Create a checkpoint patch for task recovery.
pub async fn create_checkpoint_patch(
pool: &PgPool,
task_id: Uuid,
checkpoint_id: Option<Uuid>,
base_commit_sha: &str,
patch_data: &[u8],
files_count: i32,
ttl_hours: i64,
) -> Result<CheckpointPatch, sqlx::Error> {
sqlx::query_as::<_, CheckpointPatch>(
r#"
INSERT INTO checkpoint_patches (
task_id, checkpoint_id, base_commit_sha, patch_data,
patch_size_bytes, files_count, expires_at
)
VALUES ($1, $2, $3, $4, $5, $6, NOW() + INTERVAL '1 hour' * $7)
RETURNING *
"#,
)
.bind(task_id)
.bind(checkpoint_id)
.bind(base_commit_sha)
.bind(patch_data)
.bind(patch_data.len() as i32)
.bind(files_count)
.bind(ttl_hours)
.fetch_one(pool)
.await
}
/// Get the latest checkpoint patch for a task.
pub async fn get_latest_checkpoint_patch(
pool: &PgPool,
task_id: Uuid,
) -> Result<Option<CheckpointPatch>, sqlx::Error> {
sqlx::query_as::<_, CheckpointPatch>(
r#"
SELECT * FROM checkpoint_patches
WHERE task_id = $1 AND expires_at > NOW()
ORDER BY created_at DESC
LIMIT 1
"#,
)
.bind(task_id)
.fetch_optional(pool)
.await
}
/// Get a checkpoint patch by ID.
pub async fn get_checkpoint_patch(
pool: &PgPool,
id: Uuid,
) -> Result<Option<CheckpointPatch>, sqlx::Error> {
sqlx::query_as::<_, CheckpointPatch>(
"SELECT * FROM checkpoint_patches WHERE id = $1",
)
.bind(id)
.fetch_optional(pool)
.await
}
/// List all checkpoint patches for a task (without patch data for efficiency).
pub async fn list_checkpoint_patches(
pool: &PgPool,
task_id: Uuid,
) -> Result<Vec<CheckpointPatchInfo>, sqlx::Error> {
sqlx::query_as::<_, CheckpointPatchInfo>(
r#"
SELECT id, task_id, checkpoint_id, base_commit_sha,
patch_size_bytes, files_count, created_at, expires_at
FROM checkpoint_patches
WHERE task_id = $1
ORDER BY created_at DESC
"#,
)
.bind(task_id)
.fetch_all(pool)
.await
}
/// Delete expired checkpoint patches.
/// Returns the number of deleted patches.
pub async fn cleanup_expired_checkpoint_patches(
pool: &PgPool,
) -> Result<i64, sqlx::Error> {
let result = sqlx::query("DELETE FROM checkpoint_patches WHERE expires_at < NOW()")
.execute(pool)
.await?;
Ok(result.rows_affected() as i64)
}
/// Delete all checkpoint patches for a task.
pub async fn delete_checkpoint_patches_for_task(
pool: &PgPool,
task_id: Uuid,
) -> Result<i64, sqlx::Error> {
let result = sqlx::query("DELETE FROM checkpoint_patches WHERE task_id = $1")
.bind(task_id)
.execute(pool)
.await?;
Ok(result.rows_affected() as i64)
}
// =============================================================================
// Red Team Notifications
// =============================================================================
// =============================================================================
// Supervisor Status API Helpers
// =============================================================================
/// Get supervisor status for a contract.
/// Returns combined information from supervisor_states and tasks tables.
pub async fn get_supervisor_status(
pool: &PgPool,
contract_id: Uuid,
owner_id: Uuid,
) -> Result<Option<SupervisorStatusInfo>, sqlx::Error> {
// Query to get supervisor status by joining supervisor_states with tasks
sqlx::query_as::<_, SupervisorStatusInfo>(
r#"
SELECT
ss.task_id,
COALESCE(t.status, 'unknown') as supervisor_state,
ss.phase,
t.progress_summary as current_activity,
ss.pending_task_ids,
ss.last_activity as last_heartbeat,
t.status as task_status,
t.daemon_id IS NOT NULL as is_running
FROM supervisor_states ss
JOIN tasks t ON t.id = ss.task_id
WHERE ss.contract_id = $1
AND t.owner_id = $2
"#,
)
.bind(contract_id)
.bind(owner_id)
.fetch_optional(pool)
.await
}
/// Internal struct to hold supervisor status query result
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct SupervisorStatusInfo {
pub task_id: Uuid,
pub supervisor_state: String,
pub phase: String,
pub current_activity: Option<String>,
#[sqlx(try_from = "Vec<Uuid>")]
pub pending_task_ids: Vec<Uuid>,
pub last_heartbeat: chrono::DateTime<chrono::Utc>,
pub task_status: String,
pub is_running: bool,
}
/// Get supervisor activity history from history_events table.
/// This provides a timeline of supervisor activities that can serve as "heartbeats".
pub async fn get_supervisor_activity_history(
pool: &PgPool,
contract_id: Uuid,
limit: i32,
offset: i32,
) -> Result<Vec<SupervisorActivityEntry>, sqlx::Error> {
sqlx::query_as::<_, SupervisorActivityEntry>(
r#"
SELECT
created_at as timestamp,
COALESCE(event_subtype, 'activity') as state,
event_data->>'activity' as activity,
(event_data->>'progress')::INTEGER as progress,
COALESCE(phase, 'unknown') as phase,
CASE
WHEN event_data->'pending_task_ids' IS NOT NULL
THEN ARRAY(SELECT jsonb_array_elements_text(event_data->'pending_task_ids'))::UUID[]
ELSE ARRAY[]::UUID[]
END as pending_task_ids
FROM history_events
WHERE contract_id = $1
AND event_type IN ('supervisor', 'phase', 'task')
ORDER BY created_at DESC
LIMIT $2 OFFSET $3
"#,
)
.bind(contract_id)
.bind(limit)
.bind(offset)
.fetch_all(pool)
.await
}
/// Internal struct to hold supervisor activity entry
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct SupervisorActivityEntry {
pub timestamp: chrono::DateTime<chrono::Utc>,
pub state: String,
pub activity: Option<String>,
pub progress: Option<i32>,
pub phase: String,
#[sqlx(try_from = "Vec<Uuid>")]
pub pending_task_ids: Vec<Uuid>,
}
/// Count total supervisor activity history entries for a contract.
pub async fn count_supervisor_activity_history(
pool: &PgPool,
contract_id: Uuid,
) -> Result<i64, sqlx::Error> {
let result: (i64,) = sqlx::query_as(
r#"
SELECT COUNT(*)
FROM history_events
WHERE contract_id = $1
AND event_type IN ('supervisor', 'phase', 'task')
"#,
)
.bind(contract_id)
.fetch_one(pool)
.await?;
Ok(result.0)
}
/// Update supervisor state last_activity timestamp.
/// This acts as a "sync" operation to refresh the supervisor's heartbeat.
pub async fn sync_supervisor_state(
pool: &PgPool,
contract_id: Uuid,
) -> Result<Option<SupervisorState>, sqlx::Error> {
sqlx::query_as::<_, SupervisorState>(
r#"
UPDATE supervisor_states
SET last_activity = NOW(),
updated_at = NOW()
WHERE contract_id = $1
RETURNING *
"#,
)
.bind(contract_id)
.fetch_optional(pool)
.await
}
// =============================================================================
// Directive Operations (top-level orchestration entity)
// =============================================================================
// TODO: Implement directive CRUD functions
// - create_directive_for_owner
// - get_directive_for_owner
// - list_directives_for_owner
// - update_directive_for_owner
// - archive_directive_for_owner
// - update_directive_status
// =============================================================================
// Directive Chain Operations (generated execution plans)
// =============================================================================
// TODO: Implement chain CRUD functions
// - create_directive_chain
// - get_current_chain
// - supersede_chain
// =============================================================================
// Chain Step Operations (nodes in the DAG)
// =============================================================================
// TODO: Implement step CRUD functions
// - create_chain_step
// - update_chain_step
// - delete_chain_step
// - find_ready_steps
// - update_step_status
// - update_step_contract
// - update_step_confidence
// - increment_step_rework_count
// =============================================================================
// Directive Evaluation Operations
// =============================================================================
// TODO: Implement evaluation functions
// - create_directive_evaluation
// - list_step_evaluations
// - list_directive_evaluations
// =============================================================================
// Directive Event Operations (audit stream)
// =============================================================================
// TODO: Implement event functions
// - emit_directive_event
// - list_directive_events
// =============================================================================
// Directive Verifier Operations
// =============================================================================
// TODO: Implement verifier CRUD functions
// - create_directive_verifier
// - list_directive_verifiers
// - update_directive_verifier
// =============================================================================
// Directive Approval Operations (human-in-the-loop)
// =============================================================================
// TODO: Implement approval functions
// - create_approval_request
// - resolve_approval
// - list_pending_approvals
// NOTE: Old chain functions removed. See git history for reference.
// Old functions included: create_chain_for_owner, get_chain_for_owner,
// list_chains_for_owner, update_chain_for_owner, delete_chain_for_owner,
// add_contract_to_chain, remove_contract_from_chain, list_chain_contracts,
// get_chain_with_contracts, list_chain_repositories, add_chain_repository,
// delete_chain_repository, set_chain_repository_primary, get_chain_graph,
// record_chain_event, list_chain_events, increment_chain_loop, complete_chain,
// get_ready_chain_contracts, is_chain_complete, get_chain_editor_data,
// create_chain_contract_definition, list_chain_contract_definitions,
// update_chain_contract_definition, delete_chain_contract_definition,
// get_chain_definition_graph, update_chain_status, progress_chain,
// create_chain_directive, get_chain_directive, update_chain_directive,
// delete_chain_directive, create_contract_evaluation, get_contract_evaluation,
// list_chain_evaluations, update_chain_contract_evaluation_status,
// mark_chain_contract_original_completion, get_chain_contract_by_contract_id,
// init_chain_for_owner.
// =============================================================================
// Directive Operations
// =============================================================================
/// Create a new directive for an owner.
pub async fn create_directive_for_owner(
pool: &PgPool,
owner_id: Uuid,
req: CreateDirectiveRequest,
) -> Result<Directive, sqlx::Error> {
let title = req.title.unwrap_or_else(|| truncate_string(&req.goal, 100));
let autonomy_level = req.autonomy_level.unwrap_or_else(|| "guardrails".to_string());
let green_threshold = req.confidence_threshold_green.unwrap_or(0.85);
let yellow_threshold = req.confidence_threshold_yellow.unwrap_or(0.60);
let requirements = req.requirements.unwrap_or(serde_json::json!([]));
let acceptance_criteria = req.acceptance_criteria.unwrap_or(serde_json::json!([]));
sqlx::query_as::<_, Directive>(
r#"
INSERT INTO directives (
owner_id, title, goal, requirements, acceptance_criteria,
autonomy_level, confidence_threshold_green, confidence_threshold_yellow,
repository_url, local_path, base_branch,
max_total_cost_usd, max_wall_time_minutes
)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
RETURNING *
"#,
)
.bind(owner_id)
.bind(&title)
.bind(&req.goal)
.bind(&requirements)
.bind(&acceptance_criteria)
.bind(&autonomy_level)
.bind(green_threshold)
.bind(yellow_threshold)
.bind(&req.repository_url)
.bind(&req.local_path)
.bind(&req.base_branch)
.bind(req.max_total_cost_usd)
.bind(req.max_wall_time_minutes)
.fetch_one(pool)
.await
}
/// Get a directive by ID, scoped to owner.
pub async fn get_directive_for_owner(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
) -> Result<Option<Directive>, sqlx::Error> {
sqlx::query_as::<_, Directive>(
r#"
SELECT * FROM directives WHERE id = $1 AND owner_id = $2
"#,
)
.bind(id)
.bind(owner_id)
.fetch_optional(pool)
.await
}
/// Get a directive by ID (no owner check - for internal use).
pub async fn get_directive(pool: &PgPool, id: Uuid) -> Result<Option<Directive>, sqlx::Error> {
sqlx::query_as::<_, Directive>(
r#"SELECT * FROM directives WHERE id = $1"#,
)
.bind(id)
.fetch_optional(pool)
.await
}
/// List directives for an owner.
pub async fn list_directives_for_owner(
pool: &PgPool,
owner_id: Uuid,
status_filter: Option<&str>,
) -> Result<Vec<DirectiveSummary>, sqlx::Error> {
let query = if let Some(status) = status_filter {
sqlx::query_as::<_, DirectiveSummary>(
r#"
SELECT
d.id, d.title, d.goal, d.status, d.autonomy_level,
dc.current_confidence,
COALESCE(dc.completed_steps, 0) as completed_steps,
COALESCE(dc.total_steps, 0) as total_steps,
d.chain_generation_count, d.started_at, d.created_at
FROM directives d
LEFT JOIN directive_chains dc ON dc.id = d.current_chain_id
WHERE d.owner_id = $1 AND d.status = $2
ORDER BY d.created_at DESC
"#,
)
.bind(owner_id)
.bind(status)
} else {
sqlx::query_as::<_, DirectiveSummary>(
r#"
SELECT
d.id, d.title, d.goal, d.status, d.autonomy_level,
dc.current_confidence,
COALESCE(dc.completed_steps, 0) as completed_steps,
COALESCE(dc.total_steps, 0) as total_steps,
d.chain_generation_count, d.started_at, d.created_at
FROM directives d
LEFT JOIN directive_chains dc ON dc.id = d.current_chain_id
WHERE d.owner_id = $1
ORDER BY d.created_at DESC
"#,
)
.bind(owner_id)
};
query.fetch_all(pool).await
}
/// Update a directive with optimistic locking.
pub async fn update_directive_for_owner(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
req: UpdateDirectiveRequest,
) -> Result<Directive, RepositoryError> {
// First get current version
let current = sqlx::query_scalar::<_, i32>(
"SELECT version FROM directives WHERE id = $1 AND owner_id = $2"
)
.bind(id)
.bind(owner_id)
.fetch_optional(pool)
.await?
.ok_or_else(|| RepositoryError::Database(sqlx::Error::RowNotFound))?;
if current != req.version {
return Err(RepositoryError::VersionConflict {
expected: req.version,
actual: current,
});
}
let directive = sqlx::query_as::<_, Directive>(
r#"
UPDATE directives SET
title = COALESCE($3, title),
goal = COALESCE($4, goal),
requirements = COALESCE($5, requirements),
acceptance_criteria = COALESCE($6, acceptance_criteria),
constraints = COALESCE($7, constraints),
external_dependencies = COALESCE($8, external_dependencies),
autonomy_level = COALESCE($9, autonomy_level),
confidence_threshold_green = COALESCE($10, confidence_threshold_green),
confidence_threshold_yellow = COALESCE($11, confidence_threshold_yellow),
max_total_cost_usd = COALESCE($12, max_total_cost_usd),
max_wall_time_minutes = COALESCE($13, max_wall_time_minutes),
max_rework_cycles = COALESCE($14, max_rework_cycles),
max_chain_regenerations = COALESCE($15, max_chain_regenerations),
version = version + 1,
updated_at = NOW()
WHERE id = $1 AND owner_id = $2 AND version = $16
RETURNING *
"#,
)
.bind(id)
.bind(owner_id)
.bind(&req.title)
.bind(&req.goal)
.bind(&req.requirements)
.bind(&req.acceptance_criteria)
.bind(&req.constraints)
.bind(&req.external_dependencies)
.bind(&req.autonomy_level)
.bind(req.confidence_threshold_green)
.bind(req.confidence_threshold_yellow)
.bind(req.max_total_cost_usd)
.bind(req.max_wall_time_minutes)
.bind(req.max_rework_cycles)
.bind(req.max_chain_regenerations)
.bind(req.version)
.fetch_one(pool)
.await?;
Ok(directive)
}
/// Update directive status.
pub async fn update_directive_status(
pool: &PgPool,
id: Uuid,
status: &str,
) -> Result<Directive, sqlx::Error> {
sqlx::query_as::<_, Directive>(
r#"
UPDATE directives SET
status = $2,
started_at = CASE WHEN $2 = 'active' AND started_at IS NULL THEN NOW() ELSE started_at END,
completed_at = CASE WHEN $2 IN ('completed', 'failed', 'archived') THEN NOW() ELSE completed_at END,
updated_at = NOW()
WHERE id = $1
RETURNING *
"#,
)
.bind(id)
.bind(status)
.fetch_one(pool)
.await
}
/// Archive a directive (soft delete).
pub async fn archive_directive_for_owner(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
) -> Result<bool, sqlx::Error> {
let result = sqlx::query(
r#"
UPDATE directives SET status = 'archived', updated_at = NOW()
WHERE id = $1 AND owner_id = $2
"#,
)
.bind(id)
.bind(owner_id)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Get directive with full progress info.
pub async fn get_directive_with_progress(
pool: &PgPool,
id: Uuid,
owner_id: Uuid,
) -> Result<Option<DirectiveWithProgress>, sqlx::Error> {
let directive = match get_directive_for_owner(pool, id, owner_id).await? {
Some(d) => d,
None => return Ok(None),
};
let chain = if let Some(chain_id) = directive.current_chain_id {
get_directive_chain(pool, chain_id).await?
} else {
None
};
let steps = if let Some(ref c) = chain {
list_chain_steps(pool, c.id).await?
} else {
vec![]
};
let recent_events = list_directive_events(pool, id, Some(20)).await?;
let pending_approvals = list_pending_approvals(pool, id).await?;
Ok(Some(DirectiveWithProgress {
directive,
chain,
steps,
recent_events,
pending_approvals,
}))
}
// =============================================================================
// Directive Chain Operations
// =============================================================================
/// Create a new chain generation for a directive.
pub async fn create_directive_chain(
pool: &PgPool,
directive_id: Uuid,
name: &str,
description: Option<&str>,
rationale: Option<&str>,
planning_model: Option<&str>,
) -> Result<DirectiveChain, sqlx::Error> {
// Get next generation number
let generation = sqlx::query_scalar::<_, i32>(
"SELECT COALESCE(MAX(generation), 0) + 1 FROM directive_chains WHERE directive_id = $1"
)
.bind(directive_id)
.fetch_one(pool)
.await?;
let chain = sqlx::query_as::<_, DirectiveChain>(
r#"
INSERT INTO directive_chains (directive_id, generation, name, description, rationale, planning_model)
VALUES ($1, $2, $3, $4, $5, $6)
RETURNING *
"#,
)
.bind(directive_id)
.bind(generation)
.bind(name)
.bind(description)
.bind(rationale)
.bind(planning_model)
.fetch_one(pool)
.await?;
// Update directive to point to new chain and increment generation count
sqlx::query(
r#"
UPDATE directives SET
current_chain_id = $2,
chain_generation_count = chain_generation_count + 1,
updated_at = NOW()
WHERE id = $1
"#,
)
.bind(directive_id)
.bind(chain.id)
.execute(pool)
.await?;
Ok(chain)
}
/// Get a directive chain by ID.
pub async fn get_directive_chain(pool: &PgPool, id: Uuid) -> Result<Option<DirectiveChain>, sqlx::Error> {
sqlx::query_as::<_, DirectiveChain>(
"SELECT * FROM directive_chains WHERE id = $1"
)
.bind(id)
.fetch_optional(pool)
.await
}
/// Get the current chain for a directive.
pub async fn get_current_chain(pool: &PgPool, directive_id: Uuid) -> Result<Option<DirectiveChain>, sqlx::Error> {
sqlx::query_as::<_, DirectiveChain>(
r#"
SELECT dc.* FROM directive_chains dc
JOIN directives d ON d.current_chain_id = dc.id
WHERE d.id = $1
"#,
)
.bind(directive_id)
.fetch_optional(pool)
.await
}
/// Update chain status.
pub async fn update_chain_status(
pool: &PgPool,
chain_id: Uuid,
status: &str,
) -> Result<DirectiveChain, sqlx::Error> {
sqlx::query_as::<_, DirectiveChain>(
r#"
UPDATE directive_chains SET
status = $2,
started_at = CASE WHEN $2 = 'active' AND started_at IS NULL THEN NOW() ELSE started_at END,
completed_at = CASE WHEN $2 IN ('completed', 'failed', 'superseded') THEN NOW() ELSE completed_at END,
updated_at = NOW()
WHERE id = $1
RETURNING *
"#,
)
.bind(chain_id)
.bind(status)
.fetch_one(pool)
.await
}
/// Supersede a chain (mark as superseded and update directive).
pub async fn supersede_chain(pool: &PgPool, chain_id: Uuid) -> Result<(), sqlx::Error> {
sqlx::query(
r#"
UPDATE directive_chains SET status = 'superseded', completed_at = NOW(), updated_at = NOW()
WHERE id = $1
"#,
)
.bind(chain_id)
.execute(pool)
.await?;
Ok(())
}
// =============================================================================
// Chain Step Operations
// =============================================================================
/// Create a new step in a chain.
pub async fn create_chain_step(
pool: &PgPool,
chain_id: Uuid,
req: AddStepRequest,
) -> Result<ChainStep, sqlx::Error> {
let step_type = req.step_type.unwrap_or_else(|| "execute".to_string());
let contract_type = req.contract_type.unwrap_or_else(|| "simple".to_string());
let phases = req.phases.unwrap_or_default();
let depends_on = req.depends_on.unwrap_or_default();
let requirement_ids = req.requirement_ids.unwrap_or_default();
let acceptance_criteria_ids = req.acceptance_criteria_ids.unwrap_or_default();
let verifier_config = req.verifier_config.unwrap_or(serde_json::json!({}));
// Get next order index
let order_index = sqlx::query_scalar::<_, i32>(
"SELECT COALESCE(MAX(order_index), 0) + 1 FROM chain_steps WHERE chain_id = $1"
)
.bind(chain_id)
.fetch_one(pool)
.await?;
let step = sqlx::query_as::<_, ChainStep>(
r#"
INSERT INTO chain_steps (
chain_id, name, description, step_type, contract_type,
initial_phase, task_plan, phases, depends_on, parallel_group,
requirement_ids, acceptance_criteria_ids, verifier_config,
editor_x, editor_y, order_index
)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
RETURNING *
"#,
)
.bind(chain_id)
.bind(&req.name)
.bind(&req.description)
.bind(&step_type)
.bind(&contract_type)
.bind(&req.initial_phase)
.bind(&req.task_plan)
.bind(&phases)
.bind(&depends_on)
.bind(&req.parallel_group)
.bind(&requirement_ids)
.bind(&acceptance_criteria_ids)
.bind(&verifier_config)
.bind(req.editor_x.unwrap_or(0.0))
.bind(req.editor_y.unwrap_or(0.0))
.bind(order_index)
.fetch_one(pool)
.await?;
// Update chain total_steps count
sqlx::query(
"UPDATE directive_chains SET total_steps = total_steps + 1, updated_at = NOW() WHERE id = $1"
)
.bind(chain_id)
.execute(pool)
.await?;
Ok(step)
}
/// Get a chain step by ID.
pub async fn get_chain_step(pool: &PgPool, id: Uuid) -> Result<Option<ChainStep>, sqlx::Error> {
sqlx::query_as::<_, ChainStep>(
"SELECT * FROM chain_steps WHERE id = $1"
)
.bind(id)
.fetch_optional(pool)
.await
}
/// List all steps in a chain.
pub async fn list_chain_steps(pool: &PgPool, chain_id: Uuid) -> Result<Vec<ChainStep>, sqlx::Error> {
sqlx::query_as::<_, ChainStep>(
"SELECT * FROM chain_steps WHERE chain_id = $1 ORDER BY order_index"
)
.bind(chain_id)
.fetch_all(pool)
.await
}
/// Update a chain step.
pub async fn update_chain_step(
pool: &PgPool,
step_id: Uuid,
req: UpdateStepRequest,
) -> Result<ChainStep, sqlx::Error> {
sqlx::query_as::<_, ChainStep>(
r#"
UPDATE chain_steps SET
name = COALESCE($2, name),
description = COALESCE($3, description),
task_plan = COALESCE($4, task_plan),
depends_on = COALESCE($5, depends_on),
requirement_ids = COALESCE($6, requirement_ids),
acceptance_criteria_ids = COALESCE($7, acceptance_criteria_ids),
verifier_config = COALESCE($8, verifier_config),
editor_x = COALESCE($9, editor_x),
editor_y = COALESCE($10, editor_y)
WHERE id = $1
RETURNING *
"#,
)
.bind(step_id)
.bind(&req.name)
.bind(&req.description)
.bind(&req.task_plan)
.bind(&req.depends_on)
.bind(&req.requirement_ids)
.bind(&req.acceptance_criteria_ids)
.bind(&req.verifier_config)
.bind(req.editor_x)
.bind(req.editor_y)
.fetch_one(pool)
.await
}
/// Delete a chain step.
pub async fn delete_chain_step(pool: &PgPool, step_id: Uuid) -> Result<bool, sqlx::Error> {
// Get chain_id first for updating count
let chain_id = sqlx::query_scalar::<_, Uuid>(
"SELECT chain_id FROM chain_steps WHERE id = $1"
)
.bind(step_id)
.fetch_optional(pool)
.await?;
let result = sqlx::query("DELETE FROM chain_steps WHERE id = $1")
.bind(step_id)
.execute(pool)
.await?;
// Update chain total_steps count
if let Some(cid) = chain_id {
sqlx::query(
"UPDATE directive_chains SET total_steps = total_steps - 1, updated_at = NOW() WHERE id = $1"
)
.bind(cid)
.execute(pool)
.await?;
}
Ok(result.rows_affected() > 0)
}
/// Find steps that are ready to execute (all dependencies met, status=pending).
pub async fn find_ready_steps(pool: &PgPool, chain_id: Uuid) -> Result<Vec<ChainStep>, sqlx::Error> {
sqlx::query_as::<_, ChainStep>(
r#"
SELECT s.* FROM chain_steps s
WHERE s.chain_id = $1
AND s.status = 'pending'
AND NOT EXISTS (
SELECT 1 FROM chain_steps dep
WHERE dep.id = ANY(s.depends_on)
AND dep.status NOT IN ('passed', 'skipped')
)
ORDER BY s.order_index
"#,
)
.bind(chain_id)
.fetch_all(pool)
.await
}
/// Update step status.
pub async fn update_step_status(
pool: &PgPool,
step_id: Uuid,
status: &str,
) -> Result<ChainStep, sqlx::Error> {
let step = sqlx::query_as::<_, ChainStep>(
r#"
UPDATE chain_steps SET
status = $2,
started_at = CASE WHEN $2 = 'running' AND started_at IS NULL THEN NOW() ELSE started_at END,
completed_at = CASE WHEN $2 IN ('passed', 'failed', 'skipped') THEN NOW() ELSE completed_at END
WHERE id = $1
RETURNING *
"#,
)
.bind(step_id)
.bind(status)
.fetch_one(pool)
.await?;
// Update chain completed_steps and failed_steps counts
if status == "passed" || status == "skipped" {
sqlx::query(
"UPDATE directive_chains SET completed_steps = completed_steps + 1, updated_at = NOW() WHERE id = $1"
)
.bind(step.chain_id)
.execute(pool)
.await?;
} else if status == "failed" {
sqlx::query(
"UPDATE directive_chains SET failed_steps = failed_steps + 1, updated_at = NOW() WHERE id = $1"
)
.bind(step.chain_id)
.execute(pool)
.await?;
}
Ok(step)
}
/// Link a step to a contract.
pub async fn update_step_contract(
pool: &PgPool,
step_id: Uuid,
contract_id: Uuid,
supervisor_task_id: Option<Uuid>,
) -> Result<ChainStep, sqlx::Error> {
sqlx::query_as::<_, ChainStep>(
r#"
UPDATE chain_steps SET contract_id = $2, supervisor_task_id = $3
WHERE id = $1
RETURNING *
"#,
)
.bind(step_id)
.bind(contract_id)
.bind(supervisor_task_id)
.fetch_one(pool)
.await
}
/// Update step confidence score and level.
pub async fn update_step_confidence(
pool: &PgPool,
step_id: Uuid,
score: f64,
level: &str,
evaluation_id: Uuid,
) -> Result<ChainStep, sqlx::Error> {
sqlx::query_as::<_, ChainStep>(
r#"
UPDATE chain_steps SET
confidence_score = $2,
confidence_level = $3,
last_evaluation_id = $4,
evaluation_count = evaluation_count + 1
WHERE id = $1
RETURNING *
"#,
)
.bind(step_id)
.bind(score)
.bind(level)
.bind(evaluation_id)
.fetch_one(pool)
.await
}
/// Increment step rework count.
pub async fn increment_step_rework_count(pool: &PgPool, step_id: Uuid) -> Result<ChainStep, sqlx::Error> {
sqlx::query_as::<_, ChainStep>(
r#"
UPDATE chain_steps SET rework_count = rework_count + 1, status = 'rework'
WHERE id = $1
RETURNING *
"#,
)
.bind(step_id)
.fetch_one(pool)
.await
}
/// Get chain graph for visualization.
pub async fn get_chain_graph(
pool: &PgPool,
chain_id: Uuid,
) -> Result<DirectiveChainGraphResponse, sqlx::Error> {
let chain = get_directive_chain(pool, chain_id).await?
.ok_or_else(|| sqlx::Error::RowNotFound)?;
let steps = list_chain_steps(pool, chain_id).await?;
let nodes: Vec<DirectiveChainGraphNode> = steps.iter().map(|s| {
DirectiveChainGraphNode {
id: s.id,
name: s.name.clone(),
step_type: s.step_type.clone(),
status: s.status.clone(),
confidence_score: s.confidence_score,
confidence_level: s.confidence_level.clone(),
contract_id: s.contract_id,
editor_x: s.editor_x,
editor_y: s.editor_y,
}
}).collect();
let mut edges = Vec::new();
for step in &steps {
for dep_id in &step.depends_on {
edges.push(DirectiveChainGraphEdge {
source: *dep_id,
target: step.id,
});
}
}
Ok(DirectiveChainGraphResponse {
chain_id,
directive_id: chain.directive_id,
nodes,
edges,
})
}
// =============================================================================
// Directive Evaluation Operations
// =============================================================================
/// Create a directive evaluation.
pub async fn create_directive_evaluation(
pool: &PgPool,
directive_id: Uuid,
chain_id: Option<Uuid>,
step_id: Option<Uuid>,
contract_id: Option<Uuid>,
evaluation_type: &str,
evaluator: Option<&str>,
passed: bool,
overall_score: Option<f64>,
confidence_level: Option<&str>,
programmatic_results: serde_json::Value,
llm_results: serde_json::Value,
criteria_results: serde_json::Value,
summary_feedback: &str,
rework_instructions: Option<&str>,
) -> Result<DirectiveEvaluation, sqlx::Error> {
// Get next evaluation number for this step/directive
let evaluation_number = if let Some(sid) = step_id {
sqlx::query_scalar::<_, i32>(
"SELECT COALESCE(MAX(evaluation_number), 0) + 1 FROM directive_evaluations WHERE step_id = $1"
)
.bind(sid)
.fetch_one(pool)
.await?
} else {
sqlx::query_scalar::<_, i32>(
"SELECT COALESCE(MAX(evaluation_number), 0) + 1 FROM directive_evaluations WHERE directive_id = $1 AND step_id IS NULL"
)
.bind(directive_id)
.fetch_one(pool)
.await?
};
sqlx::query_as::<_, DirectiveEvaluation>(
r#"
INSERT INTO directive_evaluations (
directive_id, chain_id, step_id, contract_id,
evaluation_type, evaluation_number, evaluator,
passed, overall_score, confidence_level,
programmatic_results, llm_results, criteria_results,
summary_feedback, rework_instructions,
completed_at
)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, NOW())
RETURNING *
"#,
)
.bind(directive_id)
.bind(chain_id)
.bind(step_id)
.bind(contract_id)
.bind(evaluation_type)
.bind(evaluation_number)
.bind(evaluator)
.bind(passed)
.bind(overall_score)
.bind(confidence_level)
.bind(&programmatic_results)
.bind(&llm_results)
.bind(&criteria_results)
.bind(summary_feedback)
.bind(rework_instructions)
.fetch_one(pool)
.await
}
/// List evaluations for a step.
pub async fn list_step_evaluations(
pool: &PgPool,
step_id: Uuid,
) -> Result<Vec<DirectiveEvaluation>, sqlx::Error> {
sqlx::query_as::<_, DirectiveEvaluation>(
"SELECT * FROM directive_evaluations WHERE step_id = $1 ORDER BY evaluation_number DESC"
)
.bind(step_id)
.fetch_all(pool)
.await
}
/// List evaluations for a directive.
pub async fn list_directive_evaluations(
pool: &PgPool,
directive_id: Uuid,
limit: Option<i64>,
) -> Result<Vec<DirectiveEvaluation>, sqlx::Error> {
let limit = limit.unwrap_or(100);
sqlx::query_as::<_, DirectiveEvaluation>(
"SELECT * FROM directive_evaluations WHERE directive_id = $1 ORDER BY created_at DESC LIMIT $2"
)
.bind(directive_id)
.bind(limit)
.fetch_all(pool)
.await
}
// =============================================================================
// Directive Event Operations
// =============================================================================
/// Emit a directive event.
pub async fn emit_directive_event(
pool: &PgPool,
directive_id: Uuid,
chain_id: Option<Uuid>,
step_id: Option<Uuid>,
event_type: &str,
severity: &str,
event_data: Option<serde_json::Value>,
actor_type: &str,
actor_id: Option<Uuid>,
) -> Result<DirectiveEvent, sqlx::Error> {
sqlx::query_as::<_, DirectiveEvent>(
r#"
INSERT INTO directive_events (
directive_id, chain_id, step_id, event_type, severity, event_data, actor_type, actor_id
)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
RETURNING *
"#,
)
.bind(directive_id)
.bind(chain_id)
.bind(step_id)
.bind(event_type)
.bind(severity)
.bind(event_data)
.bind(actor_type)
.bind(actor_id)
.fetch_one(pool)
.await
}
/// List directive events.
pub async fn list_directive_events(
pool: &PgPool,
directive_id: Uuid,
limit: Option<i64>,
) -> Result<Vec<DirectiveEvent>, sqlx::Error> {
let limit = limit.unwrap_or(100);
sqlx::query_as::<_, DirectiveEvent>(
"SELECT * FROM directive_events WHERE directive_id = $1 ORDER BY created_at DESC LIMIT $2"
)
.bind(directive_id)
.bind(limit)
.fetch_all(pool)
.await
}
// =============================================================================
// Directive Verifier Operations
// =============================================================================
/// Create a directive verifier.
pub async fn create_directive_verifier(
pool: &PgPool,
directive_id: Uuid,
name: &str,
verifier_type: &str,
command: Option<&str>,
working_directory: Option<&str>,
auto_detect: bool,
detect_files: Vec<String>,
weight: f64,
required: bool,
) -> Result<DirectiveVerifier, sqlx::Error> {
sqlx::query_as::<_, DirectiveVerifier>(
r#"
INSERT INTO directive_verifiers (
directive_id, name, verifier_type, command, working_directory,
auto_detect, detect_files, weight, required
)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
RETURNING *
"#,
)
.bind(directive_id)
.bind(name)
.bind(verifier_type)
.bind(command)
.bind(working_directory)
.bind(auto_detect)
.bind(&detect_files)
.bind(weight)
.bind(required)
.fetch_one(pool)
.await
}
/// List verifiers for a directive.
pub async fn list_directive_verifiers(
pool: &PgPool,
directive_id: Uuid,
) -> Result<Vec<DirectiveVerifier>, sqlx::Error> {
sqlx::query_as::<_, DirectiveVerifier>(
"SELECT * FROM directive_verifiers WHERE directive_id = $1 ORDER BY name"
)
.bind(directive_id)
.fetch_all(pool)
.await
}
/// Update a directive verifier.
pub async fn update_directive_verifier(
pool: &PgPool,
verifier_id: Uuid,
enabled: Option<bool>,
command: Option<&str>,
weight: Option<f64>,
required: Option<bool>,
) -> Result<DirectiveVerifier, sqlx::Error> {
sqlx::query_as::<_, DirectiveVerifier>(
r#"
UPDATE directive_verifiers SET
enabled = COALESCE($2, enabled),
command = COALESCE($3, command),
weight = COALESCE($4, weight),
required = COALESCE($5, required),
updated_at = NOW()
WHERE id = $1
RETURNING *
"#,
)
.bind(verifier_id)
.bind(enabled)
.bind(command)
.bind(weight)
.bind(required)
.fetch_one(pool)
.await
}
/// Update verifier last run result.
pub async fn update_verifier_result(
pool: &PgPool,
verifier_id: Uuid,
result: serde_json::Value,
) -> Result<DirectiveVerifier, sqlx::Error> {
sqlx::query_as::<_, DirectiveVerifier>(
r#"
UPDATE directive_verifiers SET last_run_at = NOW(), last_result = $2, updated_at = NOW()
WHERE id = $1
RETURNING *
"#,
)
.bind(verifier_id)
.bind(result)
.fetch_one(pool)
.await
}
// =============================================================================
// Directive Approval Operations
// =============================================================================
/// Create an approval request.
pub async fn create_approval_request(
pool: &PgPool,
directive_id: Uuid,
step_id: Option<Uuid>,
approval_type: &str,
description: &str,
context: Option<serde_json::Value>,
urgency: &str,
expires_at: Option<chrono::DateTime<Utc>>,
) -> Result<DirectiveApproval, sqlx::Error> {
sqlx::query_as::<_, DirectiveApproval>(
r#"
INSERT INTO directive_approvals (
directive_id, step_id, approval_type, description, context, urgency, expires_at
)
VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING *
"#,
)
.bind(directive_id)
.bind(step_id)
.bind(approval_type)
.bind(description)
.bind(context)
.bind(urgency)
.bind(expires_at)
.fetch_one(pool)
.await
}
/// Resolve an approval request.
pub async fn resolve_approval(
pool: &PgPool,
approval_id: Uuid,
status: &str,
response: Option<&str>,
responded_by: Uuid,
) -> Result<DirectiveApproval, sqlx::Error> {
sqlx::query_as::<_, DirectiveApproval>(
r#"
UPDATE directive_approvals SET
status = $2,
response = $3,
responded_by = $4,
responded_at = NOW()
WHERE id = $1
RETURNING *
"#,
)
.bind(approval_id)
.bind(status)
.bind(response)
.bind(responded_by)
.fetch_one(pool)
.await
}
/// List pending approvals for a directive.
pub async fn list_pending_approvals(
pool: &PgPool,
directive_id: Uuid,
) -> Result<Vec<DirectiveApproval>, sqlx::Error> {
sqlx::query_as::<_, DirectiveApproval>(
r#"
SELECT * FROM directive_approvals
WHERE directive_id = $1 AND status = 'pending'
ORDER BY
CASE urgency
WHEN 'critical' THEN 1
WHEN 'high' THEN 2
WHEN 'normal' THEN 3
ELSE 4
END,
created_at
"#,
)
.bind(directive_id)
.fetch_all(pool)
.await
}
/// Get step by contract ID.
pub async fn get_step_by_contract_id(
pool: &PgPool,
contract_id: Uuid,
) -> Result<Option<ChainStep>, sqlx::Error> {
sqlx::query_as::<_, ChainStep>(
"SELECT * FROM chain_steps WHERE contract_id = $1"
)
.bind(contract_id)
.fetch_optional(pool)
.await
}
// =============================================================================
// Helper Functions
// =============================================================================
/// Helper to truncate string to max length
fn truncate_string(s: &str, max_len: usize) -> String {
if s.len() <= max_len {
s.to_string()
} else {
format!("{}...", &s[..max_len - 3])
}
}
