Struct MemoryStore 

pub struct MemoryStore { /* private fields */ }

Store for generated memory state and memory extraction/consolidation jobs.

Implementations

impl MemoryStore

pub async fn clear_memory_data(&self) -> Result<()>

Deletes all persisted memory state in one transaction.

This removes every stage1_outputs row and all jobs rows for the stage-1 (memory_stage1) and phase-2 (memory_consolidate_global) memory pipelines.

pub async fn record_stage1_output_usage( &self, thread_ids: &[ThreadId], ) -> Result<usize>

Record usage for cited stage-1 outputs.

Each thread id increments usage_count by one and sets last_usage to the current Unix timestamp. Missing rows are ignored.

pub async fn claim_stage1_jobs_for_startup( &self, current_thread_id: ThreadId, params: Stage1StartupClaimParams<'_>, ) -> Result<Vec<Stage1JobClaim>>

Selects and claims stage-1 startup jobs for stale threads.

Query behavior:

• starts from threads filtered to active threads and allowed sources (push_thread_filters)
• excludes threads with memory_mode != 'enabled'
• excludes the current thread id
• keeps only threads whose millisecond updated_at is in the age window
• checks memory staleness against the memories DB
• orders by updated_at_ms DESC and applies scan_limit to bound state-DB work before probing the memories DB

For each selected thread, this function calls Self::try_claim_stage1_job with source_updated_at = thread.updated_at.timestamp() and returns up to max_claimed successful claims.

pub async fn list_stage1_outputs_for_global( &self, n: usize, ) -> Result<Vec<Stage1Output>>

Lists the most recent non-empty stage-1 outputs for global consolidation.

Query behavior:

• filters out rows where both raw_memory and rollout_summary are blank
• hydrates thread cwd, rollout_path, and git_branch from the state DB
• filters out missing or non-enabled threads
• orders by source_updated_at DESC, thread_id DESC
• returns the first n visible outputs

pub async fn prune_stage1_outputs_for_retention( &self, max_unused_days: i64, limit: usize, ) -> Result<usize>

Prunes stale stage-1 outputs while preserving the latest phase-2 baseline and stage-1 job watermarks.

Query behavior:

• considers only rows with selected_for_phase2 = 0
• keeps recency as COALESCE(last_usage, source_updated_at)
• removes rows older than max_unused_days
• prunes at most limit rows ordered from stalest to newest

pub async fn get_phase2_input_selection( &self, n: usize, max_unused_days: i64, ) -> Result<Vec<Stage1Output>>

Returns the current phase-2 input set.

Query behavior:

• current selection keeps only non-empty stage-1 outputs whose last_usage is within max_unused_days, or whose source_updated_at is within that window when the memory has never been used
• eligible rows are ranked by usage_count DESC, COALESCE(last_usage, source_updated_at) DESC, source_updated_at DESC, thread_id DESC
• the selected top-N rows are returned in stable thread_id ASC order

The returned rows are the complete Phase 2 filesystem input. Phase 2 syncs these rows directly; deletions are represented by the workspace diff against the previous successful memory baseline.

pub async fn mark_thread_memory_mode_polluted( &self, thread_id: ThreadId, ) -> Result<bool>

Marks a thread as polluted and enqueues phase-2 forgetting when the thread participated in the last successful phase-2 baseline.

pub async fn try_claim_stage1_job( &self, thread_id: ThreadId, worker_id: ThreadId, source_updated_at: i64, lease_seconds: i64, max_running_jobs: usize, ) -> Result<Stage1JobClaimOutcome>

Attempts to claim a stage-1 job for a thread at source_updated_at.

Claim semantics:

• skips as up-to-date when either:
  • stage1_outputs.source_updated_at >= source_updated_at, or
  • jobs.last_success_watermark >= source_updated_at
• inserts or updates a jobs row to running only when:
  • global running job count for memory_stage1 is below max_running_jobs
  • existing row is not actively running with a valid lease
  • retry backoff (if present) has elapsed, or source_updated_at advanced
  • retries remain, or source_updated_at advanced (which resets retries)

The update path refreshes ownership token, lease, and input_watermark. If claiming fails, a follow-up read maps current row state to a precise skip outcome (SkippedRunning, SkippedRetryBackoff, or SkippedRetryExhausted).

pub async fn mark_stage1_job_succeeded( &self, thread_id: ThreadId, ownership_token: &str, source_updated_at: i64, raw_memory: &str, rollout_summary: &str, rollout_slug: Option<&str>, ) -> Result<bool>

Marks a claimed stage-1 job successful and upserts generated output.

Transaction behavior:

• updates jobs only for the currently owned running row
• sets status='done' and last_success_watermark = input_watermark
• upserts stage1_outputs for the thread, replacing existing output only when source_updated_at is newer or equal
• preserves any existing selected_for_phase2 baseline until the next successful phase-2 run rewrites the baseline selection, including the snapshot timestamp chosen during that run
• persists optional rollout_slug for rollout summary artifact naming
• enqueues/advances the global phase-2 job watermark using source_updated_at

pub async fn mark_stage1_job_succeeded_no_output( &self, thread_id: ThreadId, ownership_token: &str, ) -> Result<bool>

Marks a claimed stage-1 job successful when extraction produced no output.

Transaction behavior:

• updates jobs only for the currently owned running row
• sets status='done' and last_success_watermark = input_watermark
• deletes any existing stage1_outputs row for the thread
• enqueues/advances the global phase-2 job watermark using the claimed input_watermark only when deleting an existing stage1_outputs row

pub async fn mark_stage1_job_failed( &self, thread_id: ThreadId, ownership_token: &str, failure_reason: &str, retry_delay_seconds: i64, ) -> Result<bool>

Marks a claimed stage-1 job as failed and schedules retry backoff.

Query behavior:

• updates only the owned running row for (kind='memory_stage1', job_key)
• sets status='error', clears lease, writes last_error
• decrements retry_remaining
• sets retry_at = now + retry_delay_seconds

pub async fn enqueue_global_consolidation( &self, input_watermark: i64, ) -> Result<()>

Enqueues or advances the global phase-2 consolidation job watermark.

The underlying upsert keeps the job running when already running, resets pending/error jobs to pending, and advances input_watermark as bookkeeping even when source_updated_at is older than prior maxima. Phase 2 does not use this watermark as a dirty check; git workspace diffing decides whether consolidation work exists after the lock is claimed.

pub async fn try_claim_global_phase2_job( &self, worker_id: ThreadId, lease_seconds: i64, ) -> Result<Phase2JobClaimOutcome>

Attempts to claim the global phase-2 consolidation lock.

Claim semantics:

• reads the singleton global job row (kind='memory_consolidate_global')
• creates and claims the singleton row when it does not exist yet
• does not use DB watermarks to decide whether Phase 2 has work; git workspace dirtiness is the source of truth after the caller materializes inputs
• returns SkippedRetryUnavailable when retry backoff is active
• returns SkippedRunning when an active running lease exists
• returns SkippedCooldown when the latest successful run finished within the phase-2 success cooldown
• otherwise updates the row to running, sets ownership + lease, and returns Claimed

pub async fn heartbeat_global_phase2_job( &self, ownership_token: &str, lease_seconds: i64, ) -> Result<bool>

Extends the lease for an owned running phase-2 global job.

Query behavior:

• UPDATE jobs SET lease_until = ? for the singleton global row
• requires status='running' and matching ownership_token

pub async fn mark_global_phase2_job_succeeded( &self, ownership_token: &str, completed_watermark: i64, selected_outputs: &[Stage1Output], ) -> Result<bool>

Marks the owned running global phase-2 job as succeeded.

Query behavior:

• updates only the owned running singleton global row
• sets status='done', clears lease/errors
• advances last_success_watermark to max(existing_last_success_watermark, completed_watermark)
• rewrites selected_for_phase2 so only the exact selected stage-1 snapshots remain marked as part of the latest successful phase-2 selection, and persists each selected snapshot’s source_updated_at

pub async fn mark_global_phase2_job_failed( &self, ownership_token: &str, failure_reason: &str, retry_delay_seconds: i64, ) -> Result<bool>

Marks the owned running global phase-2 job as failed and schedules retry.

Query behavior:

• updates only the owned running singleton global row
• sets status='error', clears lease
• writes failure reason and retry time
• decrements retry_remaining without going below zero

pub async fn mark_global_phase2_job_failed_if_unowned( &self, ownership_token: &str, failure_reason: &str, retry_delay_seconds: i64, ) -> Result<bool>

Fallback failure finalization when ownership may have been lost.

Query behavior:

• same state transition as Self::mark_global_phase2_job_failed
• matches rows where ownership_token = ? OR ownership_token IS NULL
• allows recovering a stuck unowned running row

Trait Implementations

impl Clone for MemoryStore

fn clone(&self) -> MemoryStore

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.