use std::fs;
use std::path::Path;
use std::process::ExitCode;
use std::time::{SystemTime, UNIX_EPOCH};
use anyhow::{Context, Result, bail};
use serde::{Deserialize, Serialize};
use crate::db;
use crate::extensions;
use crate::handoff;
use crate::output::CommandReport;
use crate::paths::state::{ResolvedPod, StateLayout};
use crate::profile::{self, ProfileName};
use crate::repo::marker as repo_marker;
use crate::state::machine_presence::{self, DEFAULT_IDLE_LEASE_SECS, MachineRuntimeHealth};
use crate::state::{
compiled as compiled_state, escalation as escalation_state, projection_metadata,
runtime as runtime_state, session_gates,
};
use crate::telemetry::cost as telemetry_cost;
const SESSION_SCHEMA_VERSION: u32 = 4;
const STALE_AFTER_SECS: u64 = 8 * 60 * 60;
const MAX_ACTIVITY_CHARS: usize = 280;
#[derive(Debug, Clone, Copy, Default, Deserialize, Serialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub(crate) enum SessionMode {
#[default]
General,
Research,
Implement,
}
impl SessionMode {
pub(crate) fn as_str(self) -> &'static str {
match self {
Self::General => "general",
Self::Research => "research",
Self::Implement => "implement",
}
}
pub(crate) fn from_str(value: &str) -> Option<Self> {
match value {
"general" => Some(Self::General),
"research" => Some(Self::Research),
"implement" => Some(Self::Implement),
_ => None,
}
}
}
#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub(crate) enum SessionLifecycle {
Interactive,
Autonomous,
}
#[derive(Debug, Clone, Copy, Default, Deserialize, Serialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub(crate) enum SessionOwnerKind {
#[default]
Interactive,
RuntimeSupervisor,
RuntimeWorker,
}
impl SessionOwnerKind {
pub(crate) fn as_str(self) -> &'static str {
match self {
Self::Interactive => "interactive",
Self::RuntimeSupervisor => "runtime_supervisor",
Self::RuntimeWorker => "runtime_worker",
}
}
pub(crate) fn lifecycle(self) -> SessionLifecycle {
match self {
Self::Interactive => SessionLifecycle::Interactive,
Self::RuntimeSupervisor | Self::RuntimeWorker => SessionLifecycle::Autonomous,
}
}
pub(crate) fn from_str(value: &str) -> Option<Self> {
match value {
"interactive" => Some(Self::Interactive),
"runtime_supervisor" => Some(Self::RuntimeSupervisor),
"runtime_worker" => Some(Self::RuntimeWorker),
_ => None,
}
}
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub(crate) struct SessionStateFile {
pub(crate) schema_version: u32,
pub(crate) started_at_epoch_s: u64,
pub(crate) last_started_at_epoch_s: u64,
pub(crate) start_count: u32,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) session_id: Option<String>,
#[serde(default)]
pub(crate) mode: SessionMode,
#[serde(default)]
pub(crate) owner_kind: SessionOwnerKind,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) owner_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) supervisor_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) lease_ttl_secs: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) last_heartbeat_at_epoch_s: Option<u64>,
#[serde(default)]
pub(crate) revision: u64,
}
impl SessionStateFile {
pub(crate) fn lifecycle(&self) -> SessionLifecycle {
self.owner_kind.lifecycle()
}
pub(crate) fn lease_expires_at_epoch_s(&self) -> Option<u64> {
self.last_heartbeat_at_epoch_s.zip(self.lease_ttl_secs).map(
|(last_heartbeat_at_epoch_s, lease_ttl_secs)| {
last_heartbeat_at_epoch_s.saturating_add(lease_ttl_secs)
},
)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct SessionActivityState {
pub(crate) session_id: String,
pub(crate) actor_id: String,
pub(crate) current_activity: String,
pub(crate) updated_at_epoch_s: u64,
pub(crate) session_revision: u64,
}
#[derive(Debug, Clone)]
pub(crate) struct SessionStartOptions {
pub(crate) mode: Option<SessionMode>,
pub(crate) lifecycle: SessionLifecycle,
pub(crate) owner_kind: Option<SessionOwnerKind>,
pub(crate) actor_id: Option<String>,
pub(crate) supervisor_id: Option<String>,
pub(crate) lease_ttl_secs: Option<u64>,
}
impl SessionStartOptions {
pub(crate) fn interactive(mode: Option<SessionMode>) -> Self {
Self {
mode,
lifecycle: SessionLifecycle::Interactive,
owner_kind: None,
actor_id: None,
supervisor_id: None,
lease_ttl_secs: None,
}
}
}
impl Default for SessionStartOptions {
fn default() -> Self {
Self::interactive(None)
}
}
#[derive(Debug, Clone, Default)]
pub(crate) struct SessionClearOptions {
pub(crate) actor_id: Option<String>,
pub(crate) reason: Option<String>,
}
#[derive(Debug, Clone)]
pub(crate) struct SessionHeartbeatOptions {
pub(crate) actor_id: String,
pub(crate) activity: Option<String>,
}
#[derive(Debug, Clone)]
pub(crate) struct SessionTakeoverOptions {
pub(crate) actor_id: String,
pub(crate) supervisor_id: Option<String>,
pub(crate) reason: String,
}
#[derive(Debug, Clone, Serialize)]
pub(crate) struct SessionLifecycleProjection {
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) lifecycle: Option<SessionLifecycle>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) owner_kind: Option<SessionOwnerKind>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) actor_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) supervisor_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) lease_ttl_secs: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) last_heartbeat_at_epoch_s: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) lease_expires_at_epoch_s: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) stale: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) revision: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) current_activity: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) activity_updated_at_epoch_s: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) ownership_conflict: Option<bool>,
}
impl SessionLifecycleProjection {
pub(crate) fn missing() -> Self {
Self {
lifecycle: None,
owner_kind: None,
actor_id: None,
supervisor_id: None,
lease_ttl_secs: None,
last_heartbeat_at_epoch_s: None,
lease_expires_at_epoch_s: None,
stale: None,
revision: None,
current_activity: None,
activity_updated_at_epoch_s: None,
ownership_conflict: None,
}
}
}
#[derive(Serialize)]
pub struct SessionStateStartReport {
command: &'static str,
ok: bool,
profile: String,
path: String,
started_at_epoch_s: u64,
last_started_at_epoch_s: u64,
start_count: u32,
#[serde(skip_serializing_if = "Option::is_none")]
session_id: Option<String>,
mode: SessionMode,
lifecycle: SessionLifecycle,
owner_kind: SessionOwnerKind,
#[serde(skip_serializing_if = "Option::is_none")]
actor_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
supervisor_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
lease_ttl_secs: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
last_heartbeat_at_epoch_s: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
lease_expires_at_epoch_s: Option<u64>,
revision: u64,
#[serde(skip_serializing_if = "Option::is_none")]
current_activity: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
activity_updated_at_epoch_s: Option<u64>,
reset: bool,
warning: Option<String>,
}
impl SessionStateStartReport {
pub(crate) fn summary_line(&self) -> String {
format!(
"mode={} start_count={} started_at={} reset={}",
self.mode.as_str(),
self.start_count,
self.started_at_epoch_s,
self.reset
)
}
pub(crate) fn warning_message(&self) -> Option<&str> {
self.warning.as_deref()
}
}
#[derive(Serialize)]
pub struct SessionStateClearReport {
command: &'static str,
ok: bool,
profile: String,
path: String,
cleared: bool,
#[serde(skip_serializing_if = "Option::is_none")]
reason: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
lifecycle: Option<SessionLifecycle>,
#[serde(skip_serializing_if = "Option::is_none")]
owner_kind: Option<SessionOwnerKind>,
#[serde(skip_serializing_if = "Option::is_none")]
actor_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
supervisor_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
cleared_revision: Option<u64>,
}
#[derive(Serialize)]
pub struct SessionStateHeartbeatReport {
command: &'static str,
ok: bool,
profile: String,
path: String,
session_id: String,
actor_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
activity: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
current_activity: Option<String>,
last_heartbeat_at_epoch_s: u64,
lease_expires_at_epoch_s: u64,
revision: u64,
#[serde(skip_serializing_if = "Option::is_none")]
activity_updated_at_epoch_s: Option<u64>,
}
#[derive(Serialize)]
pub struct SessionStateTakeoverReport {
command: &'static str,
ok: bool,
profile: String,
path: String,
prior_session_id: String,
prior_actor_id: String,
session_id: String,
actor_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
supervisor_id: Option<String>,
lease_ttl_secs: u64,
last_heartbeat_at_epoch_s: u64,
lease_expires_at_epoch_s: u64,
revision: u64,
#[serde(skip_serializing_if = "Option::is_none")]
current_activity: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
activity_updated_at_epoch_s: Option<u64>,
reason: String,
}
impl CommandReport for SessionStateStartReport {
fn exit_code(&self) -> ExitCode {
ExitCode::SUCCESS
}
fn render_text(&self) {
if let Some(warning) = &self.warning {
println!("Warning: {warning}");
}
println!("Session state: {}", self.summary_line());
}
}
impl CommandReport for SessionStateClearReport {
fn exit_code(&self) -> ExitCode {
ExitCode::SUCCESS
}
fn render_text(&self) {
if self.cleared {
println!("Cleared session state at {}", self.path);
} else {
println!("No session state found at {}", self.path);
}
}
}
impl CommandReport for SessionStateHeartbeatReport {
fn exit_code(&self) -> ExitCode {
ExitCode::SUCCESS
}
fn render_text(&self) {
println!(
"Heartbeat recorded for `{}` on session `{}` (revision={}, lease_expires_at={}).",
self.actor_id, self.session_id, self.revision, self.lease_expires_at_epoch_s
);
}
}
impl CommandReport for SessionStateTakeoverReport {
fn exit_code(&self) -> ExitCode {
ExitCode::SUCCESS
}
fn render_text(&self) {
println!(
"Session takeover recorded: `{}` replaced `{}` on `{}` (revision={}).",
self.actor_id, self.prior_actor_id, self.path, self.revision
);
}
}
pub fn start(
repo_root: &Path,
explicit_profile: Option<&str>,
locality_id: Option<&str>,
options: SessionStartOptions,
) -> Result<SessionStateStartReport> {
let profile = profile::resolve(explicit_profile)?;
let layout = required_layout(repo_root, &profile)?;
let db = open_session_db(&layout)?;
let path = layout.state_db_path();
let now = now_epoch_s()?;
let existing = db::session::read(db.conn())?;
let ownership = resolve_start_ownership(&options)?;
let resolved_pod: Option<ResolvedPod> = if let Some(locality_id) = locality_id {
layout.resolved_pod(locality_id)?
} else {
None
};
let next_state = build_next_state(existing.clone(), now, &options, &ownership)?;
let reset = next_state.start_count == 1;
let warning = existing.as_ref().and_then(|state| {
if is_stale(state, now) {
None
} else if state.session_id.is_some() {
match next_state.lifecycle() {
SessionLifecycle::Interactive => Some(format!(
"active session already exists for profile `{profile}` in this clone; incrementing start_count"
)),
SessionLifecycle::Autonomous => Some(format!(
"active autonomous session owned by `{}` already exists for profile `{profile}` in this workspace; refreshing lease and incrementing start_count",
next_state.owner_id.as_deref().unwrap_or("unknown"),
)),
}
} else {
Some(format!(
"legacy session state detected for profile `{profile}`; starting a fresh session and refreshing session telemetry"
))
}
});
if let (true, Some(locality_id)) = (reset, locality_id) {
if let Some(ref old_state) = existing {
if let Some(old_id) = &old_state.session_id {
if let Err(err) = telemetry_cost::record_session_cost(
db.conn(),
repo_root,
&layout,
locality_id,
old_id,
) {
warn_session_cost_persistence("before stale reset", &err);
}
let old_ctx = extensions::dispatch::SessionBoundaryContext {
layout: &layout,
repo_root,
locality_id,
session_id: old_id,
pod: resolved_pod
.as_ref()
.map(|p| extensions::dispatch::PodContext {
name: &p.name,
shared_root: &p.shared_root,
}),
};
extensions::dispatch::on_session_cleared(&old_ctx)
.context("failed to clean up stale focus entries; aborting session reset")?;
}
}
if let Some(ref session_id) = next_state.session_id {
let new_ctx = extensions::dispatch::SessionBoundaryContext {
layout: &layout,
repo_root,
locality_id,
session_id,
pod: resolved_pod
.as_ref()
.map(|p| extensions::dispatch::PodContext {
name: &p.name,
shared_root: &p.shared_root,
}),
};
extensions::dispatch::on_session_started(&new_ctx)
.context("failed to adopt pre-session focus entries; aborting session reset")?;
}
escalation_state::clear_non_blocking_for_stale_reset(&layout)
.context("failed to clear non-blocking escalations on stale reset")?;
session_gates::clear_for_session_boundary(&layout)
.context("failed to clear execution gates on stale reset")?;
}
db::session::write(db.conn(), &next_state)?;
let current_activity = if next_state.lifecycle() == SessionLifecycle::Autonomous && !reset {
load_session_activity_for_state(db.conn(), &next_state)?
} else {
db::session_activity::delete(db.conn())?;
None
};
drop(db);
record_machine_presence_best_effort(
&layout,
locality_id,
MachineRuntimeHealth::Ready,
next_state.lease_ttl_secs,
now,
"after session state update",
);
if let Some(locality_id) = locality_id {
if let Some(ref session_id) = next_state.session_id {
record_projection_baseline(repo_root, &layout, locality_id, session_id);
}
}
Ok(SessionStateStartReport {
command: "session-state-start",
ok: true,
profile: profile.to_string(),
path: path.display().to_string(),
started_at_epoch_s: next_state.started_at_epoch_s,
last_started_at_epoch_s: next_state.last_started_at_epoch_s,
start_count: next_state.start_count,
session_id: next_state.session_id.clone(),
mode: next_state.mode,
lifecycle: next_state.lifecycle(),
owner_kind: next_state.owner_kind,
actor_id: next_state.owner_id.clone(),
supervisor_id: next_state.supervisor_id.clone(),
lease_ttl_secs: next_state.lease_ttl_secs,
last_heartbeat_at_epoch_s: next_state.last_heartbeat_at_epoch_s,
lease_expires_at_epoch_s: next_state.lease_expires_at_epoch_s(),
revision: next_state.revision,
current_activity: current_activity
.as_ref()
.map(|value| value.current_activity.clone()),
activity_updated_at_epoch_s: current_activity
.as_ref()
.map(|value| value.updated_at_epoch_s),
reset,
warning,
})
}
fn mint_session_id() -> String {
format!("ses_{}", ulid::Ulid::new())
}
fn build_next_state(
existing: Option<SessionStateFile>,
now: u64,
options: &SessionStartOptions,
ownership: &ResolvedStartOwnership,
) -> Result<SessionStateFile> {
match existing {
Some(state) if !is_stale(&state, now) && state.session_id.is_some() => {
match (ownership.owner_kind.lifecycle(), state.lifecycle()) {
(SessionLifecycle::Interactive, SessionLifecycle::Interactive) => {
Ok(refreshed_session_state(
&state,
now,
options.mode.unwrap_or(state.mode),
ownership,
))
}
(SessionLifecycle::Autonomous, SessionLifecycle::Autonomous) => {
let existing_actor_id = state.owner_id.as_deref().unwrap_or("unknown");
if ownership.owner_id != existing_actor_id {
bail!(
"active autonomous session is owned by `{existing_actor_id}`; explicit takeover or clear is required before a different actor can start"
);
}
Ok(refreshed_session_state(
&state,
now,
options.mode.unwrap_or(state.mode),
ownership,
))
}
(SessionLifecycle::Interactive, SessionLifecycle::Autonomous) => bail!(
"active autonomous session is owned by `{}`; clear it or wait for it to become stale before switching back to interactive work",
state.owner_id.as_deref().unwrap_or("unknown"),
),
(SessionLifecycle::Autonomous, SessionLifecycle::Interactive) => bail!(
"active interactive session already exists in this workspace; clear it or wait for it to become stale before autonomous start"
),
}
}
Some(state) => {
if ownership.owner_kind.lifecycle() == SessionLifecycle::Autonomous
&& state.lifecycle() == SessionLifecycle::Autonomous
&& state.owner_id.as_deref() != Some(ownership.owner_id.as_str())
{
bail!(
"stale autonomous session is owned by `{}`; use `ccd session-state takeover` to adopt it",
state.owner_id.as_deref().unwrap_or("unknown"),
);
}
if ownership.owner_kind.lifecycle() == SessionLifecycle::Interactive
&& state.lifecycle() == SessionLifecycle::Autonomous
{
bail!(
"stale autonomous session is owned by `{}`; clear it or use explicit takeover before interactive start",
state.owner_id.as_deref().unwrap_or("unknown"),
);
}
Ok(fresh_session_state(
now,
options.mode.unwrap_or_default(),
ownership,
Some(next_revision(&state)),
options
.supervisor_id
.clone()
.or_else(|| state.supervisor_id.clone()),
))
}
None => Ok(fresh_session_state(
now,
options.mode.unwrap_or_default(),
ownership,
None,
options.supervisor_id.clone(),
)),
}
}
fn warn_session_cost_persistence(context: &str, err: &anyhow::Error) {
eprintln!("warning: failed to persist session cost {context}: {err:#}");
}
fn warn_machine_presence_persistence(context: &str, err: &anyhow::Error) {
eprintln!("warning: failed to record machine presence {context}: {err:#}");
}
fn record_machine_presence_best_effort(
layout: &StateLayout,
locality_id: Option<&str>,
runtime_health: MachineRuntimeHealth,
lease_ttl_secs: Option<u64>,
observed_at_epoch_s: u64,
context: &str,
) {
if let Err(err) = machine_presence::record_machine_presence(
layout,
locality_id,
runtime_health,
lease_ttl_secs,
observed_at_epoch_s,
) {
warn_machine_presence_persistence(context, &err);
}
}
pub(crate) fn load_session_id(layout: &StateLayout) -> Result<Option<String>> {
let now = now_epoch_s()?;
let Some(db) = try_open_session_db(layout)? else {
return Ok(None);
};
db::session::load_session_id(db.conn(), now)
}
pub fn clear(
repo_root: &Path,
explicit_profile: Option<&str>,
locality_id: Option<&str>,
options: SessionClearOptions,
) -> Result<SessionStateClearReport> {
let profile = profile::resolve(explicit_profile)?;
let layout = required_layout(repo_root, &profile)?;
let path = layout.state_db_path();
let resolved_pod: Option<ResolvedPod> = if let Some(locality_id) = locality_id {
layout.resolved_pod(locality_id)?
} else {
None
};
let mut cleared_lifecycle = None;
let mut cleared_owner_kind = None;
let mut cleared_actor_id = None;
let mut cleared_supervisor_id = None;
let mut cleared_revision = None;
let cleared = if let Some(db) = try_open_session_db(&layout)? {
let existing = db::session::read(db.conn())?;
authorize_clear(existing.as_ref(), options.actor_id.as_deref())?;
if let Some(state) = existing.as_ref() {
cleared_lifecycle = Some(state.lifecycle());
cleared_owner_kind = Some(state.owner_kind);
cleared_actor_id = state.owner_id.clone();
cleared_supervisor_id = state.supervisor_id.clone();
cleared_revision = Some(next_revision(state));
}
if let (Some(state), Some(locality_id)) = (&existing, locality_id) {
if let Some(session_id) = &state.session_id {
if let Err(err) = telemetry_cost::record_session_cost(
db.conn(),
repo_root,
&layout,
locality_id,
session_id,
) {
warn_session_cost_persistence("before session clear", &err);
}
let ctx = extensions::dispatch::SessionBoundaryContext {
layout: &layout,
repo_root,
locality_id,
session_id,
pod: resolved_pod
.as_ref()
.map(|p| extensions::dispatch::PodContext {
name: &p.name,
shared_root: &p.shared_root,
}),
};
extensions::dispatch::on_session_cleared(&ctx)
.context("failed to clean up focus entries; aborting session clear")?;
}
}
if existing.is_some() {
db::session_activity::delete(db.conn())?;
db::session::delete(db.conn())?;
true
} else {
false
}
} else {
false
};
escalation_state::clear_all_for_session_boundary(&layout)
.context("failed to clear escalation state on session close-out")?;
session_gates::clear_for_session_boundary(&layout)
.context("failed to clear execution gates on session close-out")?;
let cleared_at_epoch_s = now_epoch_s()?;
record_machine_presence_best_effort(
&layout,
locality_id,
MachineRuntimeHealth::Idle,
Some(DEFAULT_IDLE_LEASE_SECS),
cleared_at_epoch_s,
"after session clear",
);
Ok(SessionStateClearReport {
command: "session-state-clear",
ok: true,
profile: profile.to_string(),
path: path.display().to_string(),
cleared,
reason: options.reason,
lifecycle: cleared_lifecycle,
owner_kind: cleared_owner_kind,
actor_id: cleared_actor_id,
supervisor_id: cleared_supervisor_id,
cleared_revision,
})
}
pub fn heartbeat(
repo_root: &Path,
explicit_profile: Option<&str>,
options: SessionHeartbeatOptions,
) -> Result<SessionStateHeartbeatReport> {
let profile = profile::resolve(explicit_profile)?;
let layout = required_layout(repo_root, &profile)?;
let path = layout.state_db_path();
let db = open_session_db(&layout)?;
let now = now_epoch_s()?;
let locality_id = repo_marker::load(repo_root)?.map(|marker| marker.locality_id);
let Some(existing) = db::session::read(db.conn())? else {
bail!(
"no active session telemetry is available; run `ccd session-state start --path .` first"
);
};
if existing.lifecycle() != SessionLifecycle::Autonomous {
bail!("heartbeat requires an active autonomous session");
}
if existing.owner_id.as_deref() != Some(options.actor_id.as_str()) {
bail!(
"heartbeat requires actor `{}` to match the active autonomous owner `{}`",
options.actor_id,
existing.owner_id.as_deref().unwrap_or("unknown"),
);
}
let session_id = existing
.session_id
.clone()
.ok_or_else(|| anyhow::anyhow!("active autonomous session is missing session_id"))?;
if is_stale(&existing, now) {
bail!(
"autonomous session lease for `{}` is stale; restart with the same actor or use takeover deliberately",
options.actor_id,
);
}
let mut next = existing.clone();
next.last_heartbeat_at_epoch_s = Some(now);
next.revision = next_revision(&existing);
db::session::write(db.conn(), &next)?;
let activity = match normalize_activity(options.activity)? {
Some(current_activity) => {
let current = SessionActivityState {
session_id: session_id.clone(),
actor_id: options.actor_id.clone(),
current_activity,
updated_at_epoch_s: now,
session_revision: next.revision,
};
db::session_activity::write(db.conn(), ¤t)?;
Some(current)
}
None => load_session_activity_for_state(db.conn(), &next)?,
};
record_machine_presence_best_effort(
&layout,
locality_id.as_deref(),
MachineRuntimeHealth::Ready,
next.lease_ttl_secs,
now,
"after persisting session heartbeat",
);
Ok(SessionStateHeartbeatReport {
command: "session-state-heartbeat",
ok: true,
profile: profile.to_string(),
path: path.display().to_string(),
session_id,
actor_id: options.actor_id,
activity: activity
.as_ref()
.map(|value| value.current_activity.clone()),
last_heartbeat_at_epoch_s: now,
lease_expires_at_epoch_s: next.lease_expires_at_epoch_s().unwrap_or(now),
revision: next.revision,
current_activity: activity
.as_ref()
.map(|value| value.current_activity.clone()),
activity_updated_at_epoch_s: activity.as_ref().map(|value| value.updated_at_epoch_s),
})
}
pub fn takeover(
repo_root: &Path,
explicit_profile: Option<&str>,
locality_id: Option<&str>,
options: SessionTakeoverOptions,
) -> Result<SessionStateTakeoverReport> {
let profile = profile::resolve(explicit_profile)?;
let layout = required_layout(repo_root, &profile)?;
let db = open_session_db(&layout)?;
let path = layout.state_db_path();
let now = now_epoch_s()?;
let Some(existing) = db::session::read(db.conn())? else {
bail!("no stale autonomous session exists to take over");
};
if existing.lifecycle() != SessionLifecycle::Autonomous {
bail!("takeover only applies to stale autonomous sessions");
}
if !is_stale(&existing, now) {
bail!("takeover is only allowed after the active autonomous lease becomes stale");
}
let prior_actor_id = existing
.owner_id
.clone()
.unwrap_or_else(|| "unknown".to_owned());
if prior_actor_id == options.actor_id {
bail!(
"takeover requires a different actor than the stale owner; use `ccd session-state start` to recover the same actor lease"
);
}
let lease_ttl_secs = existing.lease_ttl_secs.ok_or_else(|| {
anyhow::anyhow!(
"stale autonomous session is missing lease_ttl_secs and cannot be taken over safely"
)
})?;
let prior_session_id = existing
.session_id
.clone()
.ok_or_else(|| anyhow::anyhow!("stale autonomous session is missing session_id"))?;
let resolved_pod: Option<ResolvedPod> = if let Some(locality_id) = locality_id {
layout.resolved_pod(locality_id)?
} else {
None
};
if let Some(locality_id) = locality_id {
if let Err(err) = telemetry_cost::record_session_cost(
db.conn(),
repo_root,
&layout,
locality_id,
&prior_session_id,
) {
warn_session_cost_persistence("before session takeover", &err);
}
let clear_ctx = extensions::dispatch::SessionBoundaryContext {
layout: &layout,
repo_root,
locality_id,
session_id: &prior_session_id,
pod: resolved_pod
.as_ref()
.map(|p| extensions::dispatch::PodContext {
name: &p.name,
shared_root: &p.shared_root,
}),
};
extensions::dispatch::on_session_cleared(&clear_ctx)
.context("failed to clean up focus entries; aborting session takeover")?;
}
let next_state = SessionStateFile {
schema_version: SESSION_SCHEMA_VERSION,
started_at_epoch_s: now,
last_started_at_epoch_s: now,
start_count: 1,
session_id: Some(mint_session_id()),
mode: existing.mode,
owner_kind: SessionOwnerKind::RuntimeWorker,
owner_id: Some(options.actor_id.clone()),
supervisor_id: options
.supervisor_id
.clone()
.or_else(|| existing.supervisor_id.clone()),
lease_ttl_secs: Some(lease_ttl_secs),
last_heartbeat_at_epoch_s: Some(now),
revision: next_revision(&existing),
};
if let Some(locality_id) = locality_id {
if let Some(session_id) = next_state.session_id.as_deref() {
let start_ctx = extensions::dispatch::SessionBoundaryContext {
layout: &layout,
repo_root,
locality_id,
session_id,
pod: resolved_pod
.as_ref()
.map(|p| extensions::dispatch::PodContext {
name: &p.name,
shared_root: &p.shared_root,
}),
};
extensions::dispatch::on_session_started(&start_ctx)
.context("failed to adopt focus entries; aborting session takeover")?;
}
}
db::session::write(db.conn(), &next_state)?;
db::session_activity::delete(db.conn())?;
record_machine_presence_best_effort(
&layout,
locality_id,
MachineRuntimeHealth::Ready,
next_state.lease_ttl_secs,
now,
"during session takeover",
);
Ok(SessionStateTakeoverReport {
command: "session-state-takeover",
ok: true,
profile: profile.to_string(),
path: path.display().to_string(),
prior_session_id,
prior_actor_id,
session_id: next_state.session_id.clone().unwrap_or_default(),
actor_id: options.actor_id,
supervisor_id: next_state.supervisor_id.clone(),
lease_ttl_secs,
last_heartbeat_at_epoch_s: now,
lease_expires_at_epoch_s: next_state.lease_expires_at_epoch_s().unwrap_or(now),
revision: next_state.revision,
current_activity: None,
activity_updated_at_epoch_s: None,
reason: options.reason,
})
}
fn resolve_start_ownership(options: &SessionStartOptions) -> Result<ResolvedStartOwnership> {
match options.lifecycle {
SessionLifecycle::Interactive => {
if options.owner_kind.is_some() {
bail!("`--owner-kind` is only valid with `--lifecycle autonomous`");
}
if options.actor_id.is_some() {
bail!("`--actor-id` is only valid with `--lifecycle autonomous`");
}
if options.supervisor_id.is_some() {
bail!("`--supervisor-id` is only valid with `--lifecycle autonomous`");
}
if options.lease_ttl_secs.is_some() {
bail!("`--lease-seconds` is only valid with `--lifecycle autonomous`");
}
Ok(ResolvedStartOwnership {
owner_kind: SessionOwnerKind::Interactive,
owner_id: "interactive".to_owned(),
supervisor_id: None,
lease_ttl_secs: None,
})
}
SessionLifecycle::Autonomous => {
let actor_id = options.actor_id.clone().ok_or_else(|| {
anyhow::anyhow!("`--actor-id` is required with `--lifecycle autonomous`")
})?;
let lease_ttl_secs = options.lease_ttl_secs.ok_or_else(|| {
anyhow::anyhow!("`--lease-seconds` is required with `--lifecycle autonomous`")
})?;
let owner_kind = match options
.owner_kind
.unwrap_or(SessionOwnerKind::RuntimeWorker)
{
SessionOwnerKind::RuntimeWorker => SessionOwnerKind::RuntimeWorker,
SessionOwnerKind::RuntimeSupervisor => SessionOwnerKind::RuntimeSupervisor,
SessionOwnerKind::Interactive => {
bail!("`interactive` is not a valid `--owner-kind` for autonomous sessions")
}
};
Ok(ResolvedStartOwnership {
owner_kind,
owner_id: actor_id,
supervisor_id: options.supervisor_id.clone(),
lease_ttl_secs: Some(lease_ttl_secs),
})
}
}
}
fn refreshed_session_state(
existing: &SessionStateFile,
now: u64,
mode: SessionMode,
ownership: &ResolvedStartOwnership,
) -> SessionStateFile {
SessionStateFile {
schema_version: SESSION_SCHEMA_VERSION,
started_at_epoch_s: existing.started_at_epoch_s,
last_started_at_epoch_s: now,
start_count: existing.start_count.saturating_add(1),
session_id: existing.session_id.clone(),
mode,
owner_kind: ownership.owner_kind,
owner_id: Some(ownership.owner_id.clone()),
supervisor_id: ownership
.supervisor_id
.clone()
.or_else(|| existing.supervisor_id.clone()),
lease_ttl_secs: ownership.lease_ttl_secs,
last_heartbeat_at_epoch_s: match ownership.owner_kind.lifecycle() {
SessionLifecycle::Interactive => None,
SessionLifecycle::Autonomous => Some(now),
},
revision: next_revision(existing),
}
}
fn fresh_session_state(
now: u64,
mode: SessionMode,
ownership: &ResolvedStartOwnership,
next_revision: Option<u64>,
supervisor_id: Option<String>,
) -> SessionStateFile {
SessionStateFile {
schema_version: SESSION_SCHEMA_VERSION,
started_at_epoch_s: now,
last_started_at_epoch_s: now,
start_count: 1,
session_id: Some(mint_session_id()),
mode,
owner_kind: ownership.owner_kind,
owner_id: Some(ownership.owner_id.clone()),
supervisor_id,
lease_ttl_secs: ownership.lease_ttl_secs,
last_heartbeat_at_epoch_s: match ownership.owner_kind.lifecycle() {
SessionLifecycle::Interactive => None,
SessionLifecycle::Autonomous => Some(now),
},
revision: next_revision.unwrap_or(1),
}
}
fn next_revision(existing: &SessionStateFile) -> u64 {
existing.revision.saturating_add(1).max(1)
}
fn authorize_clear(existing: Option<&SessionStateFile>, actor_id: Option<&str>) -> Result<()> {
let Some(existing) = existing else {
return Ok(());
};
if existing.lifecycle() == SessionLifecycle::Interactive {
return Ok(());
}
let Some(actor_id) = actor_id else {
bail!(
"clearing an autonomous session requires `--actor-id` matching the owner or supervisor"
);
};
if existing.owner_id.as_deref() == Some(actor_id)
|| existing.supervisor_id.as_deref() == Some(actor_id)
{
return Ok(());
}
bail!(
"actor `{actor_id}` is not authorized to clear the active autonomous session owned by `{}`",
existing.owner_id.as_deref().unwrap_or("unknown"),
)
}
#[derive(Debug, Clone)]
struct ResolvedStartOwnership {
owner_kind: SessionOwnerKind,
owner_id: String,
supervisor_id: Option<String>,
lease_ttl_secs: Option<u64>,
}
pub(crate) fn lifecycle_projection(
state: &SessionStateFile,
now_epoch_s: u64,
caller_actor_id: Option<&str>,
activity: Option<&SessionActivityState>,
) -> SessionLifecycleProjection {
let actor_id = state.owner_id.clone();
let stale = is_stale(state, now_epoch_s);
let ownership_conflict = caller_actor_id.map(|caller_actor_id| {
state.lifecycle() == SessionLifecycle::Autonomous
&& state.owner_id.as_deref() != Some(caller_actor_id)
&& !stale
});
let projected_activity = activity.filter(|activity| activity_matches_state(state, activity));
SessionLifecycleProjection {
lifecycle: Some(state.lifecycle()),
owner_kind: Some(state.owner_kind),
actor_id,
supervisor_id: state.supervisor_id.clone(),
lease_ttl_secs: state.lease_ttl_secs,
last_heartbeat_at_epoch_s: state.last_heartbeat_at_epoch_s,
lease_expires_at_epoch_s: state.lease_expires_at_epoch_s(),
stale: Some(stale),
revision: Some(state.revision),
current_activity: projected_activity.map(|value| value.current_activity.clone()),
activity_updated_at_epoch_s: projected_activity.map(|value| value.updated_at_epoch_s),
ownership_conflict,
}
}
pub(crate) fn load_for_layout(layout: &StateLayout) -> Result<Option<SessionStateFile>> {
let Some(db) = try_open_session_db(layout)? else {
return Ok(None);
};
db::session::read(db.conn())
}
pub(crate) fn load_activity_for_layout(
layout: &StateLayout,
) -> Result<Option<SessionActivityState>> {
let Some(db) = try_open_session_db(layout)? else {
return Ok(None);
};
db::session_activity::read(db.conn())
}
pub(crate) fn now_epoch_s() -> Result<u64> {
Ok(SystemTime::now()
.duration_since(UNIX_EPOCH)
.context("system clock is before UNIX_EPOCH")?
.as_secs())
}
pub(crate) fn session_minutes(state: &SessionStateFile, now_epoch_s: u64) -> u64 {
now_epoch_s.saturating_sub(state.started_at_epoch_s) / 60
}
pub(crate) fn is_stale(state: &SessionStateFile, now_epoch_s: u64) -> bool {
if state.session_id.is_none() {
return true;
}
match state.lifecycle() {
SessionLifecycle::Interactive => {
now_epoch_s.saturating_sub(state.last_started_at_epoch_s) > STALE_AFTER_SECS
}
SessionLifecycle::Autonomous => state
.lease_expires_at_epoch_s()
.is_none_or(|lease_expires_at_epoch_s| lease_expires_at_epoch_s <= now_epoch_s),
}
}
fn open_session_db(layout: &StateLayout) -> Result<db::StateDb> {
let db = db::StateDb::open(&layout.state_db_path())?;
migrate_session_json(&db, layout)?;
Ok(db)
}
fn try_open_session_db(layout: &StateLayout) -> Result<Option<db::StateDb>> {
if layout.state_db_path().exists() || layout.session_state_path().exists() {
open_session_db(layout).map(Some)
} else {
Ok(None)
}
}
fn migrate_session_json(db: &db::StateDb, layout: &StateLayout) -> Result<()> {
let path = layout.session_state_path();
let contents = match fs::read_to_string(&path) {
Ok(c) => c,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()),
Err(e) => return Err(e).with_context(|| format!("failed to read {}", path.display())),
};
if db::session::read(db.conn())?.is_none() {
let state: SessionStateFile = serde_json::from_str(&contents)
.with_context(|| format!("failed to parse {}", path.display()))?;
db::session::write(db.conn(), &state)?;
}
let mut migrated = path.as_os_str().to_owned();
migrated.push(".migrated");
fs::rename(&path, Path::new(&migrated))
.with_context(|| format!("failed to rename {} to .migrated", path.display()))?;
Ok(())
}
fn required_layout(repo_root: &Path, profile: &ProfileName) -> Result<StateLayout> {
let layout = StateLayout::resolve(repo_root, profile.clone())?;
ensure_profile_exists(&layout)?;
Ok(layout)
}
fn ensure_profile_exists(layout: &StateLayout) -> Result<()> {
let profile_root = layout.profile_root();
if profile_root.is_dir() {
return Ok(());
}
bail!(
"profile `{}` does not exist at {}; bootstrap it before using session-state",
layout.profile(),
profile_root.display()
)
}
fn record_projection_baseline(
repo_root: &Path,
layout: &StateLayout,
locality_id: &str,
session_id: &str,
) {
let Ok(runtime) = runtime_state::load_runtime_state(repo_root, layout, locality_id) else {
return;
};
let Ok(store) =
compiled_state::preview_for_target(&runtime, compiled_state::ProjectionTarget::Session)
else {
return;
};
let base_digests = compiled_state::compute_projection_digests(&store);
let execution_gates_json =
serialize_view_or_empty(|| serde_json::to_string(&runtime.execution_gates.view));
let escalation_json = serialize_view_or_empty(|| {
let entries = escalation_state::load_for_layout(layout).unwrap_or_default();
serde_json::to_string(&escalation_state::build_view(layout, &entries))
});
let recovery_json = serialize_view_or_empty(|| {
serde_json::to_string(&runtime_state::recovery_view(&runtime.recovery))
});
let git_state_json = if layout.resolved_substrate().is_git() {
handoff::read_git_state(repo_root, handoff::BranchMode::AllowDetachedHead)
.ok()
.and_then(|git| serde_json::to_string(&Some(git)).ok())
.unwrap_or_default()
} else {
serialize_view_or_empty(|| serde_json::to_string(&None::<handoff::GitState>))
};
let session_state_json = serialize_view_or_empty(|| {
serde_json::to_string(&build_session_state_view_for_baseline(layout))
});
let extended = compiled_state::compute_extended_digests(
&base_digests,
&execution_gates_json,
&escalation_json,
&recovery_json,
&git_state_json,
&session_state_json,
);
if let Err(error) = projection_metadata::record_baseline(layout, &store, session_id, &extended)
{
projection_metadata::warn_record_error(layout, &error);
}
}
fn serialize_view_or_empty<F: FnOnce() -> serde_json::Result<String>>(f: F) -> String {
f().unwrap_or_default()
}
fn build_session_state_view_for_baseline(layout: &StateLayout) -> serde_json::Value {
let tracked_session = load_for_layout(layout).ok().flatten();
let tracked_activity = load_activity_for_layout(layout).ok().flatten();
let path = layout.state_db_path().display().to_string();
match tracked_session {
Some(ref state) => {
let status = if state.session_id.is_some() {
"active"
} else {
"stale"
};
let mut map = serde_json::Map::new();
map.insert("path".into(), serde_json::Value::String(path));
map.insert("status".into(), serde_json::Value::String(status.into()));
if let Some(ref sid) = state.session_id {
map.insert("session_id".into(), serde_json::Value::String(sid.clone()));
}
map.insert(
"mode".into(),
serde_json::to_value(state.mode).unwrap_or_default(),
);
map.insert(
"start_count".into(),
serde_json::Value::Number(state.start_count.into()),
);
let projection = lifecycle_projection(
state,
now_epoch_s().unwrap_or_default(),
None,
tracked_activity.as_ref(),
);
if let Ok(serde_json::Value::Object(proj_map)) = serde_json::to_value(&projection) {
for (k, v) in proj_map {
map.insert(k, v);
}
}
serde_json::Value::Object(map)
}
None => {
let mut map = serde_json::Map::new();
map.insert("path".into(), serde_json::Value::String(path));
map.insert("status".into(), serde_json::Value::String("missing".into()));
let missing = SessionLifecycleProjection::missing();
if let Ok(serde_json::Value::Object(proj_map)) = serde_json::to_value(&missing) {
for (k, v) in proj_map {
map.insert(k, v);
}
}
serde_json::Value::Object(map)
}
}
}
fn normalize_activity(activity: Option<String>) -> Result<Option<String>> {
let Some(activity) = activity else {
return Ok(None);
};
let trimmed = activity.trim();
if trimmed.is_empty() {
return Ok(None);
}
if trimmed.chars().count() > MAX_ACTIVITY_CHARS {
bail!("activity text is too long (max {MAX_ACTIVITY_CHARS} characters)");
}
Ok(Some(trimmed.to_owned()))
}
fn load_session_activity_for_state(
conn: &rusqlite::Connection,
state: &SessionStateFile,
) -> Result<Option<SessionActivityState>> {
let Some(activity) = db::session_activity::read(conn)? else {
return Ok(None);
};
if !activity_matches_state(state, &activity) {
return Ok(None);
}
Ok(Some(activity))
}
fn activity_matches_state(state: &SessionStateFile, activity: &SessionActivityState) -> bool {
state.lifecycle() == SessionLifecycle::Autonomous
&& Some(activity.session_id.as_str()) == state.session_id.as_deref()
&& state.owner_id.as_deref() == Some(activity.actor_id.as_str())
}
#[cfg(test)]
mod tests {
use super::*;
fn interactive_state(
schema_version: u32,
started_at_epoch_s: u64,
last_started_at_epoch_s: u64,
start_count: u32,
session_id: Option<&str>,
mode: SessionMode,
) -> SessionStateFile {
SessionStateFile {
schema_version,
started_at_epoch_s,
last_started_at_epoch_s,
start_count,
session_id: session_id.map(str::to_owned),
mode,
owner_kind: SessionOwnerKind::Interactive,
owner_id: session_id.map(|_| "interactive".to_owned()),
supervisor_id: None,
lease_ttl_secs: None,
last_heartbeat_at_epoch_s: None,
revision: u64::from(session_id.is_some()),
}
}
fn interactive_start(
mode: Option<SessionMode>,
) -> (SessionStartOptions, ResolvedStartOwnership) {
let options = SessionStartOptions::interactive(mode);
let ownership = resolve_start_ownership(&options).unwrap();
(options, ownership)
}
#[test]
fn deserialize_v2_without_session_id() {
let json = r#"{
"schema_version": 2,
"started_at_epoch_s": 1000,
"last_started_at_epoch_s": 2000,
"start_count": 3
}"#;
let state: SessionStateFile = serde_json::from_str(json).unwrap();
assert_eq!(state.schema_version, 2);
assert_eq!(state.start_count, 3);
assert!(state.session_id.is_none());
}
#[test]
fn deserialize_v3_with_session_id() {
let json = r#"{
"schema_version": 3,
"started_at_epoch_s": 1000,
"last_started_at_epoch_s": 2000,
"start_count": 1,
"session_id": "ses_01ABC"
}"#;
let state: SessionStateFile = serde_json::from_str(json).unwrap();
assert_eq!(state.schema_version, 3);
assert_eq!(state.session_id.as_deref(), Some("ses_01ABC"));
}
#[test]
fn deserialize_corrupted_json_fails() {
let json = r#"{ not valid json }"#;
let result = serde_json::from_str::<SessionStateFile>(json);
assert!(result.is_err());
}
#[test]
fn serialize_v3_omits_none_session_id() {
let state = interactive_state(3, 1000, 2000, 1, None, SessionMode::General);
let json = serde_json::to_string(&state).unwrap();
assert!(!json.contains("session_id"));
}
#[test]
fn serialize_v3_includes_session_id_when_present() {
let state = interactive_state(3, 1000, 2000, 1, Some("ses_01ABC"), SessionMode::General);
let json = serde_json::to_string(&state).unwrap();
assert!(json.contains("ses_01ABC"));
}
#[test]
fn start_mints_session_id_on_fresh_start() {
let now = 1_000_000;
let (options, ownership) = interactive_start(None);
let state = build_next_state(None, now, &options, &ownership).unwrap();
assert_eq!(state.start_count, 1);
assert_eq!(state.started_at_epoch_s, now);
assert_eq!(state.mode, SessionMode::General);
let sid = state.session_id.expect("should mint session_id");
assert!(
sid.starts_with("ses_"),
"session_id should have ses_ prefix, got: {sid}"
);
assert_eq!(state.owner_kind, SessionOwnerKind::Interactive);
assert_eq!(state.owner_id.as_deref(), Some("interactive"));
assert_eq!(state.revision, 1);
}
#[test]
fn start_preserves_session_id_on_non_stale_increment() {
let now = 1_000_000;
let existing = interactive_state(
3,
now - 100,
now - 50,
2,
Some("ses_EXISTING"),
SessionMode::Research,
);
let (options, ownership) = interactive_start(None);
let state = build_next_state(Some(existing), now, &options, &ownership).unwrap();
assert_eq!(state.start_count, 3);
assert_eq!(state.session_id.as_deref(), Some("ses_EXISTING"));
assert_eq!(state.started_at_epoch_s, now - 100);
assert_eq!(state.mode, SessionMode::Research);
assert_eq!(state.revision, 2);
}
#[test]
fn start_mints_new_id_on_stale_reset() {
let now = 1_000_000;
let stale_time = now - STALE_AFTER_SECS - 1;
let existing = interactive_state(
3,
stale_time - 100,
stale_time,
5,
Some("ses_OLD"),
SessionMode::Research,
);
let (options, ownership) = interactive_start(None);
let state = build_next_state(Some(existing), now, &options, &ownership).unwrap();
assert_eq!(state.start_count, 1);
assert_eq!(state.mode, SessionMode::General);
let sid = state.session_id.expect("should mint new session_id");
assert!(sid.starts_with("ses_"));
assert_ne!(sid, "ses_OLD");
assert_eq!(state.revision, 2);
}
#[test]
fn start_mints_new_id_when_v2_file_has_no_session_id() {
let now = 1_000_000;
let existing = interactive_state(2, now - 100, now - 50, 3, None, SessionMode::General);
let (options, ownership) = interactive_start(None);
let state = build_next_state(Some(existing), now, &options, &ownership).unwrap();
assert_eq!(state.start_count, 1);
let sid = state
.session_id
.expect("should mint session_id for v2 upgrade");
assert!(sid.starts_with("ses_"));
assert_eq!(state.revision, 1);
}
#[test]
fn start_accepts_explicit_mode_override() {
let now = 1_000_000;
let existing = interactive_state(
3,
now - 100,
now - 50,
2,
Some("ses_EXISTING"),
SessionMode::Research,
);
let (options, ownership) = interactive_start(Some(SessionMode::Implement));
let state = build_next_state(Some(existing), now, &options, &ownership).unwrap();
assert_eq!(state.start_count, 3);
assert_eq!(state.mode, SessionMode::Implement);
}
#[test]
fn db_roundtrip_session_state() {
let conn = rusqlite::Connection::open_in_memory().unwrap();
crate::db::schema::initialize(&conn).unwrap();
let state = interactive_state(
3,
1_000_000,
1_000_050,
2,
Some("ses_DB_RT"),
SessionMode::Research,
);
db::session::write(&conn, &state).unwrap();
let loaded = db::session::read(&conn).unwrap().expect("should exist");
assert_eq!(loaded.session_id.as_deref(), Some("ses_DB_RT"));
assert_eq!(loaded.start_count, 2);
assert_eq!(loaded.mode, SessionMode::Research);
}
#[test]
fn db_load_session_id_returns_none_for_stale() {
let conn = rusqlite::Connection::open_in_memory().unwrap();
crate::db::schema::initialize(&conn).unwrap();
let now = 1_000_000u64;
let state = interactive_state(
3,
now - STALE_AFTER_SECS - 200,
now - STALE_AFTER_SECS - 100,
1,
Some("ses_STALE"),
SessionMode::General,
);
db::session::write(&conn, &state).unwrap();
assert!(db::session::load_session_id(&conn, now).unwrap().is_none());
}
#[test]
fn db_load_session_id_returns_id_for_active() {
let conn = rusqlite::Connection::open_in_memory().unwrap();
crate::db::schema::initialize(&conn).unwrap();
let now = 1_000_000u64;
let state = interactive_state(
3,
now - 100,
now - 50,
1,
Some("ses_ACTIVE"),
SessionMode::General,
);
db::session::write(&conn, &state).unwrap();
assert_eq!(
db::session::load_session_id(&conn, now).unwrap().as_deref(),
Some("ses_ACTIVE")
);
}
}