use std::collections::BTreeMap;
use std::fs::{self, File, OpenOptions};
use std::io::{self, Write};
use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
use std::path::{Path, PathBuf};
use std::process;
use std::thread;
use std::time::{Duration, Instant};
use anyhow::{Context, Result, bail};
use chrono::Utc;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use crate::config::TERMINAL_HOST_PROTOCOL_VERSION;
use crate::terminal_host::TerminalSessionInfo;
pub const REGISTRY_SCHEMA_VERSION: u64 = 1;
const LOCK_WAIT_TIMEOUT: Duration = Duration::from_secs(10);
const LOCK_RETRY_INTERVAL: Duration = Duration::from_millis(50);
#[derive(Debug, Clone)]
pub struct RegistryStore {
root: PathBuf,
registry_path: PathBuf,
lock_path: PathBuf,
audit_path: PathBuf,
messages_dir: PathBuf,
}
#[derive(Debug)]
pub struct RegistryLock {
path: PathBuf,
}
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct Registry {
pub schema_version: u64,
pub roles: BTreeMap<String, RegistryEntry>,
}
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct RegistryEntry {
pub agent: String,
pub workspace_id: String,
pub session_id: String,
pub created_at: String,
pub last_verified_at: String,
pub terminal_host_protocol_version: u64,
pub source: String,
pub owner: String,
}
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ForceSendAuditEntry {
pub timestamp: String,
pub role: Option<String>,
pub session_id: String,
pub workspace_id: String,
pub payload_hash: String,
pub byte_size: usize,
pub action: String,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CleanupSummary {
pub removed_roles: usize,
}
impl Default for Registry {
fn default() -> Self {
Self {
schema_version: REGISTRY_SCHEMA_VERSION,
roles: BTreeMap::new(),
}
}
}
impl RegistryStore {
pub fn new(root: impl Into<PathBuf>) -> Self {
let root = root.into();
Self {
registry_path: root.join("registry.json"),
lock_path: root.join("registry.lock"),
audit_path: root.join("audit.log"),
messages_dir: root.join("messages"),
root,
}
}
pub fn root(&self) -> &Path {
&self.root
}
pub fn registry_path(&self) -> &Path {
&self.registry_path
}
pub fn messages_dir(&self) -> &Path {
&self.messages_dir
}
pub fn acquire_lock(&self) -> Result<RegistryLock> {
ensure_private_dir(&self.root)?;
let start = Instant::now();
loop {
match OpenOptions::new()
.write(true)
.create_new(true)
.mode(0o600)
.open(&self.lock_path)
{
Ok(mut file) => {
writeln!(file, "pid={}", process::id())?;
file.sync_all()?;
return Ok(RegistryLock {
path: self.lock_path.clone(),
});
}
Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
if start.elapsed() >= LOCK_WAIT_TIMEOUT {
bail!(
"timed out waiting for registry lock; remove stale .ssctl/registry.lock if no ssctl process is running"
);
}
thread::sleep(LOCK_RETRY_INTERVAL);
}
Err(error) => {
return Err(error).context("failed to create registry lock");
}
}
}
}
pub fn load(&self) -> Result<Registry> {
match fs::read_to_string(&self.registry_path) {
Ok(contents) => {
let registry: Registry =
serde_json::from_str(&contents).context("failed to parse registry")?;
if registry.schema_version != REGISTRY_SCHEMA_VERSION {
bail!(
"unsupported registry schema version: {}",
registry.schema_version
);
}
Ok(registry)
}
Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(Registry::default()),
Err(error) => Err(error).context("failed to read registry"),
}
}
pub fn save(&self, registry: &Registry) -> Result<()> {
if registry.schema_version != REGISTRY_SCHEMA_VERSION {
bail!(
"unsupported registry schema version: {}",
registry.schema_version
);
}
ensure_private_dir(&self.root)?;
let tmp_path = self
.root
.join(format!("registry.json.tmp.{}", process::id()));
let mut tmp = OpenOptions::new()
.write(true)
.create(true)
.truncate(true)
.mode(0o600)
.open(&tmp_path)
.context("failed to create temporary registry file")?;
serde_json::to_writer_pretty(&mut tmp, registry)?;
tmp.write_all(b"\n")?;
tmp.sync_all()?;
drop(tmp);
fs::rename(&tmp_path, &self.registry_path).context("failed to replace registry")?;
sync_dir(&self.root)?;
Ok(())
}
pub fn append_force_send_audit(&self, entry: &ForceSendAuditEntry) -> Result<()> {
ensure_private_dir(&self.root)?;
let mut file = OpenOptions::new()
.create(true)
.append(true)
.mode(0o600)
.open(&self.audit_path)
.context("failed to open audit log")?;
serde_json::to_writer(&mut file, entry)?;
file.write_all(b"\n")?;
file.sync_all()?;
Ok(())
}
pub fn write_message_snapshot(&self, body: &str) -> Result<PathBuf> {
ensure_private_dir(&self.root)?;
ensure_private_dir(&self.messages_dir)?;
let filename = format!("{}.md", timestamp_for_filename());
let path = self.messages_dir.join(filename);
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.mode(0o600)
.open(&path)
.context("failed to create message snapshot")?;
file.write_all(body.as_bytes())?;
file.sync_all()?;
Ok(path)
}
}
impl Drop for RegistryLock {
fn drop(&mut self) {
let _ = fs::remove_file(&self.path);
}
}
impl RegistryEntry {
pub fn new(agent: String, workspace_id: String, session_id: String, source: String) -> Self {
let now = timestamp_now();
Self {
agent,
workspace_id,
session_id,
created_at: now.clone(),
last_verified_at: now,
terminal_host_protocol_version: TERMINAL_HOST_PROTOCOL_VERSION,
source,
owner: "ssctl".to_owned(),
}
}
pub fn touch_verified(&mut self) {
self.last_verified_at = timestamp_now();
self.terminal_host_protocol_version = TERMINAL_HOST_PROTOCOL_VERSION;
}
}
impl ForceSendAuditEntry {
pub fn new(
role: Option<String>,
session_id: String,
workspace_id: String,
payload_hash: String,
byte_size: usize,
) -> Self {
Self {
timestamp: timestamp_now(),
role,
session_id,
workspace_id,
payload_hash,
byte_size,
action: "force-unregistered-send".to_owned(),
}
}
#[cfg(test)]
pub fn from_payload(
role: Option<String>,
session_id: String,
workspace_id: String,
payload: &str,
) -> Self {
Self::new(
role,
session_id,
workspace_id,
sha256_hex(payload.as_bytes()),
payload.len(),
)
}
}
pub fn cleanup_stale_sessions(
registry: &mut Registry,
sessions: &[TerminalSessionInfo],
) -> CleanupSummary {
let before = registry.roles.len();
registry.roles.retain(|_, entry| {
sessions.iter().any(|session| {
session.is_alive
&& session.session_id == entry.session_id
&& session.workspace_id.as_deref() == Some(entry.workspace_id.as_str())
})
});
CleanupSummary {
removed_roles: before.saturating_sub(registry.roles.len()),
}
}
pub fn role_for_session<'a>(
registry: &'a Registry,
session_id: &str,
) -> Option<(&'a str, &'a RegistryEntry)> {
registry
.roles
.iter()
.find(|(_, entry)| entry.session_id == session_id)
.map(|(role, entry)| (role.as_str(), entry))
}
pub fn sha256_hex(bytes: &[u8]) -> String {
let mut hasher = Sha256::new();
hasher.update(bytes);
hex::encode(hasher.finalize())
}
pub fn timestamp_now() -> String {
Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
}
pub fn timestamp_for_filename() -> String {
Utc::now().format("%Y%m%d-%H%M%S%.3fZ").to_string()
}
fn ensure_private_dir(path: &Path) -> Result<()> {
fs::create_dir_all(path).with_context(|| format!("failed to create {}", path.display()))?;
fs::set_permissions(path, fs::Permissions::from_mode(0o700))
.with_context(|| format!("failed to set permissions on {}", path.display()))?;
Ok(())
}
fn sync_dir(path: &Path) -> Result<()> {
File::open(path)
.with_context(|| format!("failed to open directory {}", path.display()))?
.sync_all()
.with_context(|| format!("failed to sync directory {}", path.display()))
}
#[cfg(test)]
mod tests {
use std::fs;
use super::{
ForceSendAuditEntry, Registry, RegistryEntry, RegistryStore, cleanup_stale_sessions,
};
use crate::terminal_host::TerminalSessionInfo;
#[test]
fn cleanup_removes_missing_or_dead_sessions() {
let mut registry = Registry::default();
registry.roles.insert(
"alive".to_owned(),
RegistryEntry::new(
"codex".to_owned(),
"workspace-1".to_owned(),
"session-1".to_owned(),
"superset agents create".to_owned(),
),
);
registry.roles.insert(
"dead".to_owned(),
RegistryEntry::new(
"codex".to_owned(),
"workspace-1".to_owned(),
"session-2".to_owned(),
"superset agents create".to_owned(),
),
);
let summary = cleanup_stale_sessions(
&mut registry,
&[TerminalSessionInfo {
session_id: "session-1".to_owned(),
workspace_id: Some("workspace-1".to_owned()),
pane_id: None,
is_alive: true,
attached_clients: 0,
pid: None,
created_at: None,
last_attached_at: None,
shell: None,
}],
);
assert_eq!(summary.removed_roles, 1);
assert!(registry.roles.contains_key("alive"));
assert!(!registry.roles.contains_key("dead"));
}
#[test]
fn store_round_trips_registry_and_audit_without_payload() {
let root = std::env::temp_dir().join(format!("ssctl-registry-test-{}", std::process::id()));
let _ = fs::remove_dir_all(&root);
let store = RegistryStore::new(&root);
let _lock = store.acquire_lock().unwrap();
let mut registry = Registry::default();
registry.roles.insert(
"worker".to_owned(),
RegistryEntry::new(
"codex".to_owned(),
"workspace-1".to_owned(),
"session-1".to_owned(),
"superset agents create".to_owned(),
),
);
store.save(®istry).unwrap();
assert_eq!(store.load().unwrap().roles.len(), 1);
let audit = ForceSendAuditEntry::from_payload(
None,
"session-1".to_owned(),
"workspace-1".to_owned(),
"secret payload",
);
store.append_force_send_audit(&audit).unwrap();
let audit_contents = fs::read_to_string(root.join("audit.log")).unwrap();
assert!(audit_contents.contains("payloadHash"));
assert!(!audit_contents.contains("secret payload"));
drop(_lock);
let _ = fs::remove_dir_all(&root);
}
}