feat: distill long-horizon agent transcripts into durable memory

packages/daemon/src/app.ts

@@ -59,6 +59,7 @@ import { SpaceAgentRepository } from './storage/repositories/space-agent-reposit
 import { WorkflowHookRuntimeService } from './lib/space/workflow-hook-runtime-service';
 import { WorkflowHookStateRepository } from './storage/repositories/workflow-hook-state-repository';
 import { SpaceLongHorizonAgentRepository } from './storage/repositories/space-long-horizon-agent-repository';
+import { MemoryDistillationService } from './lib/space/memory-distillation-service';
 import { SpaceAgentManager } from './lib/space/managers/space-agent-manager';
 import { SpaceManager } from './lib/space/managers/space-manager';
 import type { SpaceRuntimeService } from './lib/space/runtime/space-runtime-service';
@@ -71,10 +72,15 @@ import {
   createMemoryConsolidationHandler,
   enqueueMemoryConsolidationIfMissing,
 } from './lib/job-handlers/memory-consolidation.handler';
+import {
+  createMemoryDistillationHandler,
+  enqueueMemoryDistillationIfMissing,
+} from './lib/job-handlers/memory-distillation.handler';
 import { createSkillValidateHandler } from './lib/job-handlers/skill-validate.handler';
 import {
   JOB_QUEUE_CLEANUP,
   MEMORY_CONSOLIDATION,
+  MEMORY_DISTILLATION,
   SKILL_VALIDATE,
   TASK_SCHEDULE_FIRE,
 } from './lib/job-queue-constants';
@@ -913,6 +919,21 @@ export async function createDaemonApp(options: CreateDaemonAppOptions): Promise<
       MEMORY_CONSOLIDATION,
       createMemoryConsolidationHandler(db.agentMemory, jobQueue)
     );
+    // Distill each active long-horizon agent's recent transcript into durable
+    // memory. Self-schedules; idle runs (no new messages past the cursor) make
+    // no LLM call, so it can run on a short cadence without burning tokens.
+    const memoryDistillationService = new MemoryDistillationService(
+      new SpaceLongHorizonAgentRepository(db.getDatabase()),
+      db.sdkMessages,
+      db.agentMemory,
+      db.distillationCursor,
+      earlySpaceRepo,
+      {}
+    );
+    jobProcessor.register(
+      MEMORY_DISTILLATION,
+      createMemoryDistillationHandler(memoryDistillationService, jobQueue)
+    );
 
     // Register task-schedule.fire handler.
     const taskScheduleRepo = new TaskScheduleRepository(db.getDatabase());
@@ -1026,6 +1047,8 @@ export async function createDaemonApp(options: CreateDaemonAppOptions): Promise<
     }
     enqueueMemoryConsolidationIfMissing(jobQueue, Date.now());
     logInfo('[Daemon] Ensured initial memory_consolidation job');
+    enqueueMemoryDistillationIfMissing(jobQueue, Date.now());
+    logInfo('[Daemon] Ensured initial memory_distillation job');
 
     // Start job queue processor last (after all handler registrations)
     jobProcessor.start();

packages/daemon/src/lib/agent/rewind-handler.ts

@@ -309,6 +309,15 @@ export class RewindHandler {
 
     // Step 1: Delete the user message itself AND all messages after it from DB
     const messagesDeleted = db.deleteMessagesAtAndAfter(session.id, rewindPoint.timestamp);
+    // Clamp any long-horizon-agent distillation cursor for this session: after
+    // the high-rowid tail is removed, new messages get rowids at MAX(remaining)+1
+    // which can be <= the stale cursor and get silently skipped. Best-effort —
+    // never let distillation bookkeeping block a rewind.
+    try {
+      db.distillationCursor.clampCursorToRemainingMessages(session.id);
+    } catch {
+      // cursor table may not exist on partially-migrated DBs; rewind still succeeds
+    }
 
     // Step 2: Truncate the SDK JSONL file at this message
     // Use worktree path when available — SDK creates session files based on CWD,
@@ -746,6 +755,12 @@ export class RewindHandler {
       if (mode === 'conversation' || mode === 'both') {
         // Delete messages from DB at and after the earliest timestamp (inclusive)
         messagesDeleted = db.deleteMessagesAtAndAfter(session.id, earliestTimestamp);
+        // Clamp the distillation cursor for this session (see rewindToCheckpoint).
+        try {
+          db.distillationCursor.clampCursorToRemainingMessages(session.id);
+        } catch {
+          // cursor table may not exist on partially-migrated DBs; rewind still succeeds
+        }
 
         const rewindSdkPath = session.worktree
           ? session.worktree.worktreePath

packages/daemon/src/lib/db-query/scope-config.ts

@@ -560,6 +560,8 @@ const EXCLUDED_TABLE_NAMES: string[] = [
   'space_long_horizon_agent_forge_scopes',
   'space_long_horizon_agent_reminders',
   'space_long_horizon_agent_event_subscriptions',
+  // Per-agent distillation cursor — internal bookkeeping for the memory_distillation job.
+  'space_agent_memory_distillation',
   // Space agent management tables — agent goal/scope assignments and reminders;
   // internal MCP tool state, not useful for ad-hoc agent queries.
   'space_agent_goal_assignments',

packages/daemon/src/lib/job-handlers/memory-distillation.handler.ts

@@ -0,0 +1,86 @@
+import type { Job, JobQueueRepository } from '../../storage/repositories/job-queue-repository';
+import type { MemoryDistillationService } from '../space/memory-distillation-service';
+import { MEMORY_DISTILLATION } from '../job-queue-constants';
+
+/**
+ * Distillation cadence. Shorter than memory_consolidation's 24h because the
+ * per-agent cursor makes idle runs cheap (no LLM call when nothing is new), so
+ * active long-horizon agents get knowledge into memory within minutes of
+ * producing it instead of waiting up to a day.
+ */
+const NEXT_RUN_DELAY_MS = 30 * 60 * 1000;
+
+export interface MemoryDistillationJobPayload {
+  /** Present (and non-empty) on per-agent jobs; absent on the coordinator tick. */
+  agentId?: string;
+}
+
+export function createMemoryDistillationHandler(
+  service: MemoryDistillationService,
+  jobQueue?: JobQueueRepository
+) {
+  return async (job: Job): Promise<Record<string, unknown>> => {
+    const agentId = readOptionalString(job.payload.agentId);
+
+    // Per-agent job: distill a single agent. These are short (one bounded LLM
+    // call, or none when idle/in-backoff), so they finish well under the job
+    // queue's 5-min stale-reclaim threshold — and even if one were reclaimed
+    // and re-run, the per-agent cursor makes it idempotent.
+    if (agentId) {
+      const result = await service.distillAgentById(agentId);
+      return {
+        agentId,
+        distilled: result?.distilled ?? false,
+        messagesRead: result?.messagesRead ?? 0,
+        memoriesWritten: result?.memoriesWritten ?? 0,
+        skipped: result?.skipped,
+      };
+    }
+
+    // Coordinator tick: self-schedule the next tick, then fan out one short
+    // per-agent job per active LH agent. Splitting work this way (instead of
+    // one long distillAll) keeps every job short enough to never be reclaimed
+    // as stale and re-executed concurrently — which would duplicate paid LLM
+    // calls and race on cursor/memory writes.
+    const nextRunAt = Date.now() + NEXT_RUN_DELAY_MS;
+    if (jobQueue) enqueueMemoryDistillationIfMissing(jobQueue, nextRunAt);
+
+    const agentIds = service.listActiveAgentIds();
+    let dispatched = 0;
+    if (jobQueue) {
+      for (const id of agentIds) {
+        jobQueue.enqueue({
+          queue: MEMORY_DISTILLATION,
+          payload: { agentId: id },
+          runAt: Date.now(),
+        });
+        dispatched++;
+      }
+    }
+
+    return { coordinator: true, agentsDispatched: dispatched, nextRunAt };
+  };
+}
+
+/**
+ * Enqueue the next coordinator tick (payload `{}`, i.e. no agentId) unless one
+ * is already pending. Per-agent jobs (payload `{ agentId }`) are NOT counted —
+ * they're transient work, not the cadence.
+ *
+ * Uses a targeted payload query rather than `listJobs`+filter: `listJobs` is
+ * newest-first and bounded, and the coordinator is enqueued before the per-agent
+ * fan-out, so with many pending per-agent jobs the coordinator is the oldest row
+ * and would fall outside a `LIMIT` window — producing duplicate coordinators.
+ */
+export function enqueueMemoryDistillationIfMissing(
+  jobQueue: JobQueueRepository,
+  runAt = Date.now()
+): void {
+  if (!jobQueue.hasPendingJobWithoutPayloadField(MEMORY_DISTILLATION, 'agentId')) {
+    jobQueue.enqueue({ queue: MEMORY_DISTILLATION, payload: {}, runAt });
+  }
+}
+
+function readOptionalString(value: unknown): string | undefined {
+  return typeof value === 'string' && value.trim() ? value.trim() : undefined;
+}

packages/daemon/src/lib/job-queue-constants.ts

@@ -9,6 +9,7 @@ export const ROOM_TICK = 'room.tick';
 export const JOB_QUEUE_CLEANUP = 'job_queue.cleanup';
 export const SKILL_VALIDATE = 'skill.validate';
 export const MEMORY_CONSOLIDATION = 'memory_consolidation';
+export const MEMORY_DISTILLATION = 'memory_distillation';
 export const SPACE_CONVERSATION_FRICTION_ANALYZE = 'space.conversationFriction.analyze';
 export const GOAL_AUTOMATION_EXECUTE = 'goalAutomation.execute';