feat: add durable transcript import

package.json

@@ -82,7 +82,8 @@
 		"typecheck:turbo": "turbo run typecheck",
 		"bench": "bun scripts/bench-memory.ts",
 		"bench:ingest": "bun scripts/bench-memory.ts ingest",
-		"bench:evaluate": "bun scripts/bench-memory.ts --resume run --from-phase search"
+		"bench:evaluate": "bun scripts/bench-memory.ts --resume run --from-phase search",
+		"eval:transcript-import": "bun scripts/evals/transcript-import-real-daemon.ts"
 	},
 	"devDependencies": {
 		"@biomejs/biome": "^2.5.5",

platform/core/src/migrations/146-source-transcript-import.ts

@@ -0,0 +1,67 @@
+import type { MigrationDb } from "./contract";
+
+/** Migration 146: durable, agent-scoped transcript import ledgers. */
+export function up(db: MigrationDb): void {
+	db.exec(`
+		CREATE TABLE IF NOT EXISTS source_import_jobs (
+			id TEXT PRIMARY KEY, kind TEXT NOT NULL CHECK (kind = 'import'), agent_id TEXT NOT NULL,
+			schema_id TEXT NOT NULL, adapter_version INTEGER NOT NULL CHECK (adapter_version = 1),
+			state TEXT NOT NULL CHECK (state IN ('staging','inventorying','queued','running','paused','completed','completed_with_rejections','cancelled','failed')),
+			control_request TEXT CHECK (control_request IN ('pause','resume','retry','cancel') OR control_request IS NULL),
+			generation INTEGER NOT NULL DEFAULT 0, total INTEGER NOT NULL DEFAULT 0, imported INTEGER NOT NULL DEFAULT 0,
+			duplicate INTEGER NOT NULL DEFAULT 0, rejected INTEGER NOT NULL DEFAULT 0, pending INTEGER NOT NULL DEFAULT 0,
+			lease_token TEXT, lease_expires_at TEXT, error TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')),
+			updated_at TEXT NOT NULL DEFAULT (datetime('now')), started_at TEXT, completed_at TEXT, reconciled_at TEXT
+		);
+		CREATE INDEX IF NOT EXISTS idx_source_import_jobs_agent_state ON source_import_jobs(agent_id, state);
+
+		CREATE TABLE IF NOT EXISTS source_import_files (
+			id TEXT PRIMARY KEY, job_id TEXT NOT NULL REFERENCES source_import_jobs(id) ON DELETE CASCADE,
+			source_id TEXT NOT NULL, agent_id TEXT NOT NULL, ordinal INTEGER NOT NULL, name TEXT NOT NULL,
+			managed_path TEXT NOT NULL, size_bytes INTEGER NOT NULL DEFAULT 0, content_hash TEXT,
+			state TEXT NOT NULL CHECK (state IN ('staging','ready','inventorying','completed','failed')),
+			record_count INTEGER NOT NULL DEFAULT 0, blank_count INTEGER NOT NULL DEFAULT 0, malformed_count INTEGER NOT NULL DEFAULT 0,
+			inventory_version INTEGER NOT NULL DEFAULT 1, checkpoint_ordinal INTEGER NOT NULL DEFAULT 0,
+			checkpoint_byte_offset INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (datetime('now')),
+			UNIQUE(job_id, ordinal)
+		);
+		CREATE INDEX IF NOT EXISTS idx_source_import_files_job_state ON source_import_files(job_id, state);
+
+		CREATE TABLE IF NOT EXISTS source_import_records (
+			id TEXT PRIMARY KEY, job_id TEXT NOT NULL REFERENCES source_import_jobs(id) ON DELETE CASCADE,
+			file_id TEXT NOT NULL REFERENCES source_import_files(id) ON DELETE CASCADE, source_id TEXT NOT NULL, agent_id TEXT NOT NULL,
+			ordinal INTEGER NOT NULL, line_number INTEGER NOT NULL, byte_offset INTEGER NOT NULL, byte_length INTEGER NOT NULL,
+			raw_hash TEXT NOT NULL, external_identity TEXT, conversation_fingerprint TEXT, status TEXT NOT NULL CHECK (status IN ('pending','imported','duplicate','rejected','cancelled')),
+			canonical_id TEXT, canonical_key TEXT, rejection_code TEXT, attempt_count INTEGER NOT NULL DEFAULT 0,
+			created_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (datetime('now')),
+			UNIQUE(file_id, ordinal), UNIQUE(file_id, byte_offset)
+		);
+		CREATE INDEX IF NOT EXISTS idx_source_import_records_job_status ON source_import_records(job_id, status);
+		CREATE INDEX IF NOT EXISTS idx_source_import_records_source_status ON source_import_records(agent_id, source_id, status);
+
+		CREATE TABLE IF NOT EXISTS transcript_import_conversations (
+			agent_id TEXT NOT NULL, external_identity TEXT NOT NULL, canonical_key TEXT NOT NULL,
+			conversation_fingerprint TEXT NOT NULL, canonical_id TEXT NOT NULL, owner_source_id TEXT NOT NULL,
+			owner_record_id TEXT NOT NULL, state TEXT NOT NULL CHECK (state IN ('committing','committed','removed')),
+			content_hash TEXT NOT NULL, harness TEXT NOT NULL, timestamp TEXT NOT NULL,
+			created_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (datetime('now')),
+			PRIMARY KEY (agent_id, external_identity), UNIQUE(agent_id, canonical_key)
+		);
+		CREATE INDEX IF NOT EXISTS idx_transcript_import_conversations_source ON transcript_import_conversations(agent_id, owner_source_id);
+
+		CREATE TABLE IF NOT EXISTS source_import_record_attempts (
+			id INTEGER PRIMARY KEY AUTOINCREMENT, agent_id TEXT NOT NULL, job_id TEXT NOT NULL, file_id TEXT NOT NULL,
+			record_id TEXT NOT NULL, generation INTEGER NOT NULL, outcome TEXT NOT NULL, error_code TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now'))
+		);
+		CREATE INDEX IF NOT EXISTS idx_source_import_attempts_record ON source_import_record_attempts(agent_id, record_id, created_at);
+	`);
+	for (const column of ["source_id", "source_record_id", "source_meta_json"] as const) {
+		const exists = db
+			.prepare("SELECT 1 AS found FROM pragma_table_info('session_transcripts') WHERE name = ?")
+			.get(column);
+		if (exists == null) db.exec(`ALTER TABLE session_transcripts ADD COLUMN ${column} TEXT`);
+	}
+	db.exec(
+		"CREATE INDEX IF NOT EXISTS idx_session_transcripts_agent_source ON session_transcripts(agent_id, source_id)",
+	);
+}

platform/core/src/migrations/147-source-import-replay-file-slots.ts

@@ -0,0 +1,27 @@
+import type { MigrationDb } from "./contract";
+
+/** Migration 147: one configured Source may own multiple auditable replay file slots. */
+export function up(db: MigrationDb): void {
+	db.exec(`
+		CREATE TABLE source_import_files_147 (
+			id TEXT PRIMARY KEY, job_id TEXT NOT NULL REFERENCES source_import_jobs(id) ON DELETE CASCADE,
+			source_id TEXT NOT NULL, agent_id TEXT NOT NULL, ordinal INTEGER NOT NULL, name TEXT NOT NULL,
+			managed_path TEXT NOT NULL, size_bytes INTEGER NOT NULL DEFAULT 0, content_hash TEXT,
+			state TEXT NOT NULL CHECK (state IN ('staging','ready','inventorying','completed','failed')),
+			record_count INTEGER NOT NULL DEFAULT 0, blank_count INTEGER NOT NULL DEFAULT 0, malformed_count INTEGER NOT NULL DEFAULT 0,
+			inventory_version INTEGER NOT NULL DEFAULT 1, checkpoint_ordinal INTEGER NOT NULL DEFAULT 0,
+			checkpoint_byte_offset INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (datetime('now')),
+			UNIQUE(job_id, ordinal)
+		);
+		INSERT INTO source_import_files_147 (
+			id, job_id, source_id, agent_id, ordinal, name, managed_path, size_bytes, content_hash, state,
+			record_count, blank_count, malformed_count, inventory_version, checkpoint_ordinal, checkpoint_byte_offset, created_at, updated_at
+		)
+		SELECT id, job_id, source_id, agent_id, ordinal, name, managed_path, size_bytes, content_hash, state,
+			record_count, blank_count, malformed_count, inventory_version, checkpoint_ordinal, checkpoint_byte_offset, created_at, updated_at
+		FROM source_import_files;
+		DROP TABLE source_import_files;
+		ALTER TABLE source_import_files_147 RENAME TO source_import_files;
+		CREATE INDEX idx_source_import_files_job_state ON source_import_files(job_id, state);
+	`);
+}

platform/core/src/migrations/148-source-import-attempt-provenance.ts

@@ -0,0 +1,9 @@
+import type { MigrationDb } from "./contract";
+
+/** Migration 148: retain source identity on append-only import attempt tombstones. */
+export function up(db: MigrationDb): void {
+	const columns = db.prepare("PRAGMA table_info(source_import_record_attempts)").all() as Array<{ name?: unknown }>;
+	if (!columns.some((column) => column.name === "source_id")) {
+		db.exec("ALTER TABLE source_import_record_attempts ADD COLUMN source_id TEXT");
+	}
+}

platform/core/src/migrations/index.ts

@@ -153,6 +153,9 @@ import { up as sourceSyncFrontier } from "./142-source-sync-frontier";
 import { up as embeddingIndexProgress } from "./143-embedding-index-progress";
 import { up as memoryJobLeaseToken } from "./144-memory-job-lease-token";
 import { up as dreamingEvidenceReviews } from "./145-dreaming-evidence-reviews";
+import { up as sourceTranscriptImport } from "./146-source-transcript-import";
+import { up as sourceImportReplayFileSlots } from "./147-source-import-replay-file-slots";
+import { up as sourceImportAttemptProvenance } from "./148-source-import-attempt-provenance";
 
 export type { Migration, MigrationArtifacts, MigrationDb } from "./contract";
 
@@ -1313,6 +1316,37 @@ export const MIGRATIONS: readonly Migration[] = [
 		up: dreamingEvidenceReviews,
 		artifacts: { tables: ["dreaming_evidence_reviews"] },
 	},
+	{
+		version: 146,
+		name: "source-transcript-import",
+		up: sourceTranscriptImport,
+		artifacts: {
+			tables: [
+				"source_import_jobs",
+				"source_import_files",
+				"source_import_records",
+				"transcript_import_conversations",
+				"source_import_record_attempts",
+			],
+			columns: [
+				{ table: "session_transcripts", column: "source_id" },
+				{ table: "session_transcripts", column: "source_record_id" },
+				{ table: "session_transcripts", column: "source_meta_json" },
+			],
+		},
+	},
+	{
+		version: 147,
+		name: "source-import-replay-file-slots",
+		up: sourceImportReplayFileSlots,
+		artifacts: { tables: ["source_import_files"] },
+	},
+	{
+		version: 148,
+		name: "source-import-attempt-provenance",
+		up: sourceImportAttemptProvenance,
+		artifacts: { columns: [{ table: "source_import_record_attempts", column: "source_id" }] },
+	},
 ];
 
 /** Simple checksum for audit trail (hash of migration name + version). */

platform/core/src/migrations/migrations.test.ts

@@ -139,7 +139,7 @@ describe("migration framework", () => {
 			runMigrations(db);
 
 			const applied = db.query("SELECT MAX(version) AS version FROM schema_migrations").get() as { version: number };
-			expect(applied.version).toBe(145);
+			expect(applied.version).toBe(148);
 			expect(
 				(db.query("PRAGMA table_info(memory_jobs)").all() as Array<{ name: string }>).some(
 					(column) => column.name === "lease_token",

platform/daemon/src/daemon.ts

@@ -141,7 +141,9 @@ import {
 	startPipeline,
 	stopPipeline,
 } from "./pipeline";
+import { randomUUID } from "node:crypto";
 import { recordDreamingPassTelemetry } from "./pipeline/dreaming";
+import { dbOwnerTransaction } from "./db-owner-runtime";
 import { type DreamingWorkerHandle, startDreamingWorker } from "./pipeline/dreaming-worker";
 import { retireLegacyExtractionJobsAsync } from "./pipeline/extraction-fallback";
 import { invalidateTraversalCache } from "./pipeline/graph-traversal";
@@ -238,6 +240,8 @@ import {
 } from "./telemetry";
 import { type TranscriptCaptureWorkerHandle, startTranscriptCaptureWorker } from "./transcript-capture-worker";
 import { type TranscriptRecoveryWorkerHandle, startTranscriptRecoveryWorker } from "./transcript-recovery-worker";
+import { type TranscriptImportWorkerHandle, startTranscriptImportWorker } from "./transcript-import-worker";
+import { createOwnerTranscriptImportStore } from "./transcript-import-store";
 
 import { resolveDaemonRestartMode } from "./daemon-restart";
 import {
@@ -268,6 +272,7 @@ import { registerGraphiqRoutes } from "./routes/graphiq-routes.js";
 import { mountHealthRoutes } from "./routes/health.js";
 import { registerHooksRoutes } from "./routes/hooks-routes.js";
 import { registerImportRoutes } from "./routes/import-routes.js";
+import { registerTranscriptImportRoutes } from "./routes/transcript-import-routes.js";
 import { mountInferenceRoutes } from "./routes/inference.js";
 import { registerKnowledgeRoutes } from "./routes/knowledge-routes.js";
 import { mountMarketplaceReviewsRoutes } from "./routes/marketplace-reviews.js";
@@ -318,6 +323,7 @@ let embeddingPromotionRestart: Promise<void> | null = null;
 let skillReconcilerHandle: ReturnType<typeof startReconciler> | null = null;
 let transcriptCaptureWorkerHandle: TranscriptCaptureWorkerHandle | null = null;
 let transcriptRecoveryWorkerHandle: TranscriptRecoveryWorkerHandle | null = null;
+let transcriptImportWorkerHandle: TranscriptImportWorkerHandle | null = null;
 // These are mirrored into state.ts via setters for read access by
 // route modules. Only daemon.ts should assign or clear them.
 let telemetryRef: TelemetryCollector | undefined;
@@ -524,6 +530,7 @@ registerSecretRoutes(app);
 registerSessionRoutes(app, { gitConfig, stopGitSyncTimer, startGitSyncTimer, getGitStatus, gitPull, gitPush, gitSync });
 registerSourcesRoutes(app);
 registerImportRoutes(app);
+registerTranscriptImportRoutes(app);
 registerPipelineRoutes(app);
 registerReflectionRoutes(app);
 registerTelemetryRoutes(app);
@@ -1584,6 +1591,12 @@ async function stopPipelineRuntime(): Promise<void> {
 		reflectionWorkerHandle = null;
 	}
 
+	if (transcriptImportWorkerHandle) {
+		try {
+			await transcriptImportWorkerHandle.stop();
+		} catch {}
+		transcriptImportWorkerHandle = null;
+	}
 	if (transcriptRecoveryWorkerHandle) {
 		try {
 			await transcriptRecoveryWorkerHandle.stop();
@@ -2069,7 +2082,7 @@ async function cleanup() {
 	const renderWorker = getSynthesisRenderWorker();
 	if (renderWorker !== null) {
 		setSynthesisRenderWorker(null);
-		renderWorker.terminate().catch((e) => {
+		await renderWorker.terminate().catch((e) => {
 			logger.debug("daemon", "render worker terminate failed", {
 				error: e instanceof Error ? e.message : String(e),
 			});
@@ -2799,6 +2812,36 @@ async function main() {
 				resolveDaemonAgentId(),
 			);
 		}
+		if (!transcriptImportWorkerHandle) {
+			transcriptImportWorkerHandle = startTranscriptImportWorker({
+				store: createOwnerTranscriptImportStore(),
+				agentId: resolveDaemonAgentId(),
+				workspaceRoot: AGENTS_DIR,
+				onBatch: async (_jobId, sourceId) => {
+					const agentId = resolveDaemonAgentId();
+					const subjectRef = `source:${sourceId}`;
+					const details = JSON.stringify({ sourceId, reason: "transcript-import-committed" });
+					await dbOwnerTransaction(
+						[
+							{
+								sql: `INSERT INTO dreaming_attention
+									(id, agent_id, kind, subject_ref, details_json, priority)
+									VALUES (?, ?, 'evidence_requeue', ?, ?, 50)
+									ON CONFLICT(agent_id, kind, subject_ref) DO UPDATE SET
+									  details_json = excluded.details_json,
+									  priority = MAX(dreaming_attention.priority, excluded.priority),
+									  generation = dreaming_attention.generation + 1,
+									  resolved_at = NULL,
+									  resolved_by_pass_id = NULL`,
+								params: [randomUUID(), agentId, subjectRef, details],
+								result: "run",
+							},
+						],
+						{ operation: "sources.import.dreaming-attention", lane: "write" },
+					);
+				},
+			});
+		}
 
 		checkpointPruneTimer = setInterval(() => {
 			try {
@@ -2898,22 +2941,17 @@ async function main() {
 	const createBoundedServer = (...args: Parameters<typeof nodeCreateServer>) => {
 		const server = nodeCreateServer(...args);
 		server.on("request", (req, res) => {
-			let bytes = 0;
-			let aborted = false;
-			req.on("data", (chunk: Buffer) => {
-				if (aborted) return;
-				bytes += chunk.length;
-				if (bytes > REQUEST_BODY_LIMIT) {
-					aborted = true;
-					logger.warn("http", "Request body exceeded limit", { bytes, limit: REQUEST_BODY_LIMIT });
-					if (!res.headersSent) {
-						res.writeHead(413, { "Content-Type": "application/json" });
-						res.end(JSON.stringify({ error: "payload too large" }), () => {
-							req.socket?.destroy();
-						});
-					}
-				}
-			});
+			// Never attach a data listener here. Node switches the request into
+			// flowing mode and drains the one-shot stream before Hono can consume
+			// it. Streamed routes (notably transcript imports) enforce their own
+			// limits while reading the body. Content-Length is the only safe
+			// pre-read check at this layer; chunked bodies are checked downstream.
+			const declared = Number(req.headers["content-length"] ?? 0);
+			if (declared > REQUEST_BODY_LIMIT && !res.headersSent) {
+				logger.warn("http", "Request body exceeded declared limit", { bytes: declared, limit: REQUEST_BODY_LIMIT });
+				res.writeHead(413, { "Content-Type": "application/json" });
+				res.end(JSON.stringify({ error: "payload too large" }), () => req.socket?.destroy());
+			}
 		});
 		return server;
 	};

platform/daemon/src/db-owner-client.test.ts

@@ -675,6 +675,36 @@ describe("DB owner client", () => {
 		expect(rows).toEqual([{ id: "m1" }]);
 	});
 
+	test("rolls back a transaction when a required precondition changes zero rows", async () => {
+		const database = makeDb();
+		directory = database.directory;
+		client = createDbOwnerClient({ dbPath: database.path });
+		const transaction = client.submit(
+			{
+				kind: "transaction",
+				transaction: {
+					statements: [
+						{
+							sql: "INSERT INTO memories (id, content) VALUES (?, ?)",
+							params: ["m2", "rolled back"],
+							result: "run",
+						},
+						{
+							sql: "UPDATE memories SET content = ? WHERE id = ?",
+							params: ["must not persist", "missing"],
+							result: "run",
+							requireChanges: true,
+						},
+					],
+				},
+			},
+			{ operation: "memory.transaction-precondition", lane: "write", deadlineMs: 1_000 },
+		);
+		await expect(transaction.result).rejects.toThrow("DB owner transaction precondition changed zero rows");
+		const rows = await recallThroughDbOwner<{ id: string }>(client, "SELECT id FROM memories ORDER BY id");
+		expect(rows).toEqual([{ id: "m1" }]);
+	});
+
 	test("retires a broken stdin without starving the queue in a retry loop", async () => {
 		const database = makeDb();
 		directory = database.directory;

platform/daemon/src/db-owner-protocol.ts

@@ -213,6 +213,31 @@ export interface DbOwnerSourceEvidenceEligibility {
 	readonly legacyObsidianRoot?: string;
 }
 
+import type { CompletedTranscriptCommit } from "./transcript-import-commit";
+
+export interface DbOwnerTranscriptBulkCommit {
+	readonly agentId: string;
+	readonly jobId: string;
+	readonly generation: number;
+	readonly leaseToken: string;
+	readonly sourceId: string;
+	readonly harness: string;
+	readonly commits: readonly CompletedTranscriptCommit[];
+}
+export interface DbOwnerTranscriptImportControl {
+	readonly agentId: string;
+	readonly jobId: string;
+	readonly control: "pause" | "resume" | "retry" | "cancel";
+}
+export interface DbOwnerTranscriptImportReconcile {
+	readonly agentId: string;
+	readonly jobId: string;
+}
+export interface DbOwnerTranscriptImportPurge {
+	readonly agentId: string;
+	readonly sourceId: string;
+}
+
 export type DbOwnerRequest =
 	| { readonly kind: "initialize"; readonly agentsDir?: string }
 	| { readonly kind: "query"; readonly statement: DbOwnerStatement }
@@ -243,6 +268,10 @@ export type DbOwnerRequest =
 	| { readonly kind: "source_artifact_upsert"; readonly input: DbOwnerSourceArtifactUpsert }
 	| { readonly kind: "source_artifact_upsert_batch"; readonly input: readonly DbOwnerSourceArtifactUpsert[] }
 	| { readonly kind: "source_evidence_eligibility"; readonly input: DbOwnerSourceEvidenceEligibility }
+	| { readonly kind: "transcript_bulk_commit"; readonly input: DbOwnerTranscriptBulkCommit }
+	| { readonly kind: "transcript_import_control"; readonly input: DbOwnerTranscriptImportControl }
+	| { readonly kind: "transcript_import_reconcile"; readonly input: DbOwnerTranscriptImportReconcile }
+	| { readonly kind: "transcript_import_purge"; readonly input: DbOwnerTranscriptImportPurge }
 	| { readonly kind: "dreaming_hygiene_attention"; readonly input: DbOwnerDreamingHygieneAttention }
 	| { readonly kind: "dreaming_surprisal_attention"; readonly input: DbOwnerDreamingSurprisalAttention }
 	| { readonly kind: "dreaming_episodic_backlog"; readonly input: DbOwnerDreamingEpisodicBacklog }