/**
 * DiskSessionStore — filesystem-backed implementation of
 * {@link SessionStore}.
 *
 * Every mutation is write-tmp-rename (Convention #8). Directory layout
 * matches session-hierarchy.md §7 / §13.4:
 *
 *   {rootDir}/projects/{projectId}/
 *     project.json
 *     sessions/{sessionId}/
 *       session.json
 *       summary.json
 *       subsessions/{subSessionId}/
 *         subsession.json
 *
 * It holds the session ENTITIES (status, actor, ownership version,
 * sub-session edges, summaries) and nothing of the conversation. A session's
 * messages are records in its session log (`SessionLog`,
 * `~/.namzu/projects/<slug>/<session-id>.jsonl`), read through
 * `foldSessionMessages` and written only by the turn recorder under the
 * session lease; listing turns and children across sessions is the
 * `SessionIndex`. This store no longer writes `messages.jsonl`, and one left
 * behind by an older build is never read.
 *
 * Tenant scoping is enforced through the JSON payload (`tenantId` field on
 * every record) rather than the path layout; cross-tenant reads reject with
 * {@link TenantIsolationError} (Convention #17, session-hierarchy.md §12.2).
 *
 * Constructor takes `rootDir`. This entity tree is NOT the session layout:
 * it is keyed by project id, so a `rootDir` equal to `NAMZU_HOME` puts
 * UUID-named directories under `projects/`, which `namzu state` reports as
 * legacy. Give it a directory of its own. Moving the entities onto the
 * session log and index needs record types for session status, ownership
 * and sub-session edges that the log does not have yet.
 */

import { createHash } from 'node:crypto'
import { mkdir, rm } from 'node:fs/promises'
import { join } from 'node:path'
import {
	ProjectRootPathTakenError,
	StaleProjectError,
	StaleSessionError,
	TenantIsolationError,
} from '../../session/errors.js'
import { SessionAlreadySummarizedError } from '../../session/summary/errors.js'
import type { SessionId, TenantId } from '../../types/ids/index.js'
import type { Project, ProjectStatus } from '../../types/project/entity.js'
import type { Session } from '../../types/session/entity.js'
import type { ProjectId, SubSessionId, SummaryId, TopicId } from '../../types/session/ids.js'
import type {
	CreateProjectParams,
	CreateSessionParams,
	CreateSubSessionParams,
	ProjectConfigInput,
	SessionStore,
	SessionView,
} from '../../types/session/store.js'
import type { SubSession } from '../../types/session/sub-session.js'
import type { DeliverableRef } from '../../types/summary/deliverable.js'
import type {
	SessionSummaryKeyDecision,
	SessionSummaryOutcome,
	SessionSummaryRef,
} from '../../types/summary/ref.js'
import {
	asProjectId,
	asSessionId,
	asSubSessionId,
	asTopicId,
	generateProjectId,
	generateSessionId,
	generateSubSessionId,
	isEntityId,
} from '../../utils/id.js'
import { DiskRecordStore } from '../kv/record-store.js'
import {
	DiskRevisionRecordStore,
	type RevisionedRecordLocation,
} from '../kv/revision-record-store.js'
import { defineSchema } from '../schema.js'
import { canonicalizePath, rootPathIndexKey } from './canonical-path.js'
import { getAncestry, getChildren, orderChildren } from './linkage.js'
import type { LinkageView } from './linkage.js'

/**
 * This store's on-disk format, versioned as a unit — which is how a
 * migration would actually be written and shipped, and it keeps every call
 * site free of schema plumbing.
 *
 * Bump `current` and add the migration for the step you are leaving when
 * the shape changes.
 */
const SCHEMA = defineSchema({
	kind: 'session-store',
	current: 4,
	migrations: {
		1: migrateSessionStoreThreadIdToTopicId,
		2: migrateSessionStoreTopicIdPrefix,
		// v3 → v4 tagged the lines of `messages.jsonl`, which this store no
		// longer reads; for every record it still reads the step is identity.
		3: (record) => record,
	},
})

/**
 * Read, write and list, through the one implementation.
 *
 * This file carried its own `readJson`, its own `atomicWriteJson` and
 * thirteen `readdir` scans — the same twenty lines four stores each kept a
 * private copy of. The properties are not obvious ones (a missing file is
 * an empty read, a record from a NEWER build is refused rather than read
 * partially and written back with the difference gone, a listing needs a
 * stable order), and every one fixed here had to be remembered into the
 * other three.

 */
const records = new DiskRecordStore<unknown>(SCHEMA)

interface RootPathBinding {
	readonly canonicalPath: string
	readonly tenantId: string
	readonly projectId: ProjectId
	readonly storageRevision: number
}

interface BoundProject {
	readonly projectId: ProjectId
	readonly source: 'immutable' | 'legacy'
}

const rootPathBindings = new DiskRevisionRecordStore<RootPathBinding>(
	SCHEMA,
	'project root-path binding',
	(record) => record.storageRevision,
)

/**
 * Validate topic UUIDs before accepting a stored record. Prefixed IDs are
 * rejected without rewriting stored references.
 * This migrator runs over every session-store record kind; records without
 * a topicId are intentionally untouched.
 */
export function migrateSessionStoreTopicIdPrefix(
	record: Record<string, unknown>,
): Record<string, unknown> {
	const topicId = record.topicId
	if (typeof topicId === 'string') {
		asTopicId(topicId)
	}
	return record
}

/**
 * v1 → v2: the FK field `session.json` carries to its owning Topic was
 * spelled `threadId`. NZ-TOPIC-01 renamed the layer, not the field
 * (comment on `types/topic/store.ts`); NZ-TOPIC-03 is that rename landing,
 * with this as its data migration.
 *
 * One migration function runs over every kind this schema stamps —
 * project.json, session.json, subsession.json and summary.json — via the
 * single shared `readJson` / `migrate` call. Only `PersistedSession` ever
 * carried `threadId`; an unconditional rewrite here would stamp
 * `topicId: undefined` onto the other kinds. Exported, not module-private,
 * so that guarantee is unit-testable directly against the function rather
 * than only observable through whichever deserializer happens to forward
 * an extra field today (most of them don't — they map named fields, which
 * is exactly why a stray key here would otherwise go unnoticed). NOT part
 * of the package's public surface: `store/session/index.ts` re-exports
 * `DiskSessionStore` by explicit name only, no wildcard.
 */
export function migrateSessionStoreThreadIdToTopicId(
	record: Record<string, unknown>,
): Record<string, unknown> {
	if (!('threadId' in record)) return record
	const { threadId, ...rest } = record
	return { ...rest, topicId: threadId }
}

/**
 * Config for {@link DiskSessionStore}. `rootDir` is absolute; all files live
 * under it per the layout documented in the module header.
 */
export interface DiskSessionStoreConfig {
	rootDir: string
}

interface PersistedProject {
	id: ProjectId
	tenantId: TenantId
	name: string
	config: Project['config']
	/** Absent in files written before the project gained a status. */
	status?: ProjectStatus
	/** Absent in files written before the project gained a CAS counter. */
	ownerVersion?: number
	/** Canonical, `realpath`-resolved. Absent for a project not on disk. */
	rootPath?: string
	createdAt: string
	updatedAt: string
}

interface PersistedSession {
	id: SessionId
	topicId: TopicId
	projectId: ProjectId
	tenantId: TenantId
	status: Session['status']
	currentActor: Session['currentActor']
	previousActors: Session['previousActors']
	workspaceId: Session['workspaceId']
	ownerVersion: number
	createdAt: string
	updatedAt: string
}

interface PersistedSubSession {
	id: SubSessionId
	parentSessionId: SessionId
	childSessionId: SessionId
	tenantId: TenantId
	kind: SubSession['kind']
	status: SubSession['status']
	spawnedBy: SubSession['spawnedBy']
	spawnedAt: string
	failureMode: SubSession['failureMode']
	completionMode: SubSession['completionMode']
	workspaceId: SubSession['workspaceId']
	broadcastGroupId?: string
	summaryRef?: SubSession['summaryRef']
	archiveRef?: SubSession['archiveRef']
	archivedAt?: string
	updatedAt: string
}

interface PersistedSummary {
	id: SummaryId
	sessionRef: SessionId
	tenantId: TenantId
	outcome: SessionSummaryOutcome
	deliverables: readonly DeliverableRef[]
	agentSummary: string
	keyDecisions: ReadonlyArray<{ at: string; summary: string }>
	at: string
	materializedBy: 'kernel'
}

/**
 * Non-terminal statuses from which {@link DiskSessionStore.recordSummary}
 * flips the owning session to `'idle'` as part of the atomic materialize +
 * transition contract (session-hierarchy.md §8.1).
 */
const SUMMARY_TERMINAL_FLIP_STATUSES: ReadonlySet<Session['status']> = new Set([
	'active',
	'locked',
	'awaiting_merge',
])

/**
 * Index of projectId → its directory path. Built lazily on lookup via
 * {@link DiskSessionStore.resolveProjectDir}; populated by create / getProject.
 */
interface ProjectIndexEntry {
	projectId: ProjectId
	path: string
}

/**
 * Index of sessionId → (projectId, path). Populated lazily similarly.
 */
interface SessionIndexEntry {
	sessionId: SessionId
	projectId: ProjectId
	path: string
}

export class DiskSessionStore implements SessionStore {
	private readonly rootDir: string
	private readonly projectIndex = new Map<ProjectId, ProjectIndexEntry>()
	private readonly sessionIndex = new Map<SessionId, SessionIndexEntry>()
	private readonly subSessionIndex = new Map<
		SubSessionId,
		{ subSessionId: SubSessionId; sessionId: SessionId; projectId: ProjectId; path: string }
	>()

	constructor(config: DiskSessionStoreConfig) {
		this.rootDir = config.rootDir
	}

	// Project CRUD ------------------------------------------------------------

	async createProject(params: CreateProjectParams, tenantId: TenantId): Promise<Project> {
		if (params.tenantId !== tenantId) {
			throw new TenantIsolationError({
				requested: tenantId,
				resource: `project(name=${params.name})`,
			})
		}
		const rootPath =
			params.rootPath === undefined ? undefined : await canonicalizePath(params.rootPath)
		if (rootPath !== undefined) {
			const existing = await this.boundProjectId(rootPath, tenantId)
			if (existing) {
				throw new ProjectRootPathTakenError({ rootPath, existingProjectId: existing.projectId })
			}
		}

		const now = new Date()
		const project: Project = {
			id: generateProjectId(),
			tenantId,
			name: params.name,
			config: {
				maxDelegationDepth: params.config?.maxDelegationDepth ?? 4,
				maxDelegationWidth: params.config?.maxDelegationWidth ?? 8,
				maxInterventionDepth: 10,
			},
			status: 'open',
			ownerVersion: 0,
			...(rootPath !== undefined ? { rootPath } : {}),
			createdAt: now,
			updatedAt: now,
		}
		const dir = join(this.rootDir, 'projects', project.id)
		await mkdir(dir, { recursive: true })
		await records.write(join(dir, 'project.json'), serializeProject(project))
		if (rootPath !== undefined) {
			try {
				await this.publishRootPathBinding(rootPath, tenantId, project.id)
			} catch (error) {
				let cleanupError: unknown
				try {
					await rm(dir, { recursive: true })
				} catch (cause) {
					cleanupError = cause
				}
				const winner = await this.boundProjectId(rootPath, tenantId)
				if (cleanupError) {
					throw new AggregateError(
						[error, cleanupError],
						`Project root binding failed and its unpublished candidate ${project.id} could not be removed.`,
					)
				}
				if (winner) {
					throw new ProjectRootPathTakenError({
						rootPath,
						existingProjectId: winner.projectId,
					})
				}
				throw error
			}
		}
		this.projectIndex.set(project.id, { projectId: project.id, path: dir })
		return project
	}

	/**
	 * Reads ONE index file. Not a scan of `projects/*` — that opens every
	 * `project.json` on the machine to answer a question about one
	 * directory, and gets slower with every project a host has ever made.
	 */
	async findProjectByRootPath(rootPath: string, tenantId: TenantId): Promise<Project | null> {
		const canonical = await canonicalizePath(rootPath)
		const bound = await this.boundProjectId(canonical, tenantId)
		if (!bound) return null
		const project = await this.getProject(bound.projectId, tenantId)
		if (!project) {
			throw new Error(
				`Project root-path binding for ${canonical} points to missing Project ${bound.projectId}.`,
			)
		}
		if (
			(bound.source === 'immutable' && project.rootPath !== canonical) ||
			(bound.source === 'legacy' &&
				project.rootPath !== undefined &&
				project.rootPath !== canonical)
		) {
			throw new Error(
				`Project root-path binding for ${canonical} points to ${bound.projectId}, but that Project declares ${project.rootPath ?? 'no rootPath'}.`,
			)
		}
		return project
	}

	private async boundProjectId(rootPath: string, tenantId: TenantId): Promise<BoundProject | null> {
		const binding = await rootPathBindings.read(this.rootPathBindingLocation(rootPath, tenantId))
		if (binding) {
			if (binding.canonicalPath !== rootPath || binding.tenantId !== tenantId) {
				throw new Error(
					`Project root-path binding hash collision or corruption for ${rootPath}; stored identity does not match the lookup.`,
				)
			}
		}

		// Read compatibility for stores written before per-root immutable bindings.
		// New writes never touch this shared JSON object: its read-modify-write
		// publication loses entries when two processes create different Projects.
		const index = await records.read<Record<string, ProjectId>>(this.rootPathIndexPath())
		const legacyProjectId = index?.[rootPathIndexKey(rootPath, tenantId)]
		if (binding && legacyProjectId && binding.projectId !== legacyProjectId) {
			throw new Error(
				`Conflicting Project root-path bindings for ${rootPath}: immutable binding names ${binding.projectId}, legacy index names ${legacyProjectId}.`,
			)
		}
		if (binding) return { projectId: binding.projectId, source: 'immutable' }
		return legacyProjectId ? { projectId: legacyProjectId, source: 'legacy' } : null
	}

	private rootPathIndexPath(): string {
		return join(this.rootDir, 'projects', 'root-path-index.json')
	}

	private rootPathBindingLocation(rootPath: string, tenantId: TenantId): RevisionedRecordLocation {
		const key = rootPathIndexKey(rootPath, tenantId)
		const digest = createHash('sha256').update(key).digest('hex')
		const root = join(this.rootDir, 'projects', '.root-path-index', digest.slice(0, 2))
		return {
			legacyPath: join(root, `${digest.slice(2)}.json`),
			revisionsDir: join(root, `${digest.slice(2)}.revisions`),
		}
	}

	private async publishRootPathBinding(
		rootPath: string,
		tenantId: TenantId,
		projectId: ProjectId,
	): Promise<void> {
		await rootPathBindings.transact(this.rootPathBindingLocation(rootPath, tenantId), (current) => {
			if (current) {
				throw new ProjectRootPathTakenError({
					rootPath,
					existingProjectId: current.projectId,
				})
			}
			return {
				record: {
					canonicalPath: rootPath,
					tenantId,
					projectId,
					storageRevision: 1,
				},
				result: undefined,
			}
		})
	}

	async getProject(projectId: ProjectId, tenantId: TenantId): Promise<Project | null> {
		const dir = this.projectDir(projectId)
		const raw = await records.read<PersistedProject>(join(dir, 'project.json'))
		if (!raw) return null
		this.assertTenant(raw.tenantId, tenantId, `project(${projectId})`)
		return deserializeProject(raw)
	}

	async updateProject(
		projectId: ProjectId,
		config: ProjectConfigInput,
		tenantId: TenantId,
	): Promise<Project | null> {
		const existing = await this.getProject(projectId, tenantId)
		if (!existing) return null
		// Per field, like the in-memory store: an omitted limit is left alone
		// rather than reset.
		const project: Project = {
			...existing,
			config: {
				...existing.config,
				...(config.maxDelegationDepth !== undefined
					? { maxDelegationDepth: config.maxDelegationDepth }
					: {}),
				...(config.maxDelegationWidth !== undefined
					? { maxDelegationWidth: config.maxDelegationWidth }
					: {}),
			},
			updatedAt: new Date(),
		}
		await records.write(join(this.projectDir(projectId), 'project.json'), serializeProject(project))
		return project
	}

	async setProjectStatus(
		projectId: ProjectId,
		status: ProjectStatus,
		tenantId: TenantId,
		expectedOwnerVersion: number,
	): Promise<Project | null> {
		const existing = await this.getProject(projectId, tenantId)
		if (!existing) return null
		// Against the version on disk, not the caller's copy of it.
		if (existing.ownerVersion !== expectedOwnerVersion) {
			throw new StaleProjectError({
				projectId,
				expectedOwnerVersion,
				actualOwnerVersion: existing.ownerVersion,
			})
		}
		const project: Project = {
			...existing,
			status,
			ownerVersion: existing.ownerVersion + 1,
			updatedAt: new Date(),
		}
		await records.write(join(this.projectDir(projectId), 'project.json'), serializeProject(project))
		return project
	}

	async listProjects(tenantId: TenantId): Promise<readonly Project[]> {
		// Read from the directory rather than the lazily-built index: the index
		// only knows about projects this instance has already touched, so a
		// listing built from it would omit everything written by a previous
		// process — which for a store whose whole point is durability is the
		// wrong answer.
		const projectsRoot = join(this.rootDir, 'projects')
		const entries = await records.scanNames(projectsRoot, '')

		const found: Project[] = []
		for (const entry of entries) {
			if (!isEntityId(entry, 'project')) continue
			const raw = await records.read<PersistedProject>(join(projectsRoot, entry, 'project.json'))
			if (!raw) continue
			// Another tenant's project is absent, not an error — a listing is a
			// question about what you own, and refusing would leak that
			// somebody else's project is there.
			if (raw.tenantId !== tenantId) continue
			found.push(deserializeProject(raw))
		}
		// Tie-broken by id — see the in-memory store. On a fast filesystem two
		// projects share a millisecond routinely, and without this the order
		// came from readdir.
		found.sort((a, b) => a.createdAt.getTime() - b.createdAt.getTime() || a.id.localeCompare(b.id))
		return found
	}

	// Session CRUD ------------------------------------------------------------

	async createSession(params: CreateSessionParams, tenantId: TenantId): Promise<Session> {
		const id = params.id === undefined ? generateSessionId() : asSessionId(params.id)
		const project = await this.getProject(params.projectId, tenantId)
		if (!project) {
			throw new Error(`Project ${params.projectId} not found`)
		}
		const now = new Date()
		if (params.id !== undefined && (await this.getSession(id, tenantId))) {
			throw new Error(`Session ${params.id} already exists`)
		}
		const session: Session = {
			id,
			topicId: params.topicId,
			projectId: params.projectId,
			tenantId,
			status: 'idle',
			currentActor: params.currentActor,
			previousActors: [],
			workspaceId: null,
			ownerVersion: 0,
			createdAt: now,
			updatedAt: now,
		}
		const dir = join(this.projectDir(params.projectId), 'sessions', session.id)
		await mkdir(dir, { recursive: true })
		await records.write(join(dir, 'session.json'), serializeSession(session))
		this.sessionIndex.set(session.id, {
			sessionId: session.id,
			projectId: params.projectId,
			path: dir,
		})
		return session
	}

	async getSession(sessionId: SessionId, tenantId: TenantId): Promise<Session | null> {
		const located = await this.locateSession(sessionId)
		if (!located) return null
		const raw = await records.read<PersistedSession>(join(located.path, 'session.json'))
		if (!raw) return null
		this.assertTenant(raw.tenantId, tenantId, `session(${sessionId})`)
		return deserializeSession(raw)
	}

	async listSessionsByTopic(topicId: TopicId, tenantId: TenantId): Promise<readonly Session[]> {
		// Walk projects/*/sessions/* and filter on the persisted record. Sessions
		// don't live under a topic-scoped path in the current layout — the
		// denormalized `topicId` on every session.json is the authority.
		//
		// This used to say "matches DiskThreadStore.listThreads in scan
		// semantics" — that class is gone (NZ-TOPIC-02, ses_020: zero production
		// callers, never public, no tests). This function's own scan doesn't
		// change; the comparison just isn't holding anything up anymore, so it's
		// removed rather than left pointing at a deleted file. Renamed from
		// `listSessions` alongside the field it reads (NZ-TOPIC-03).
		//
		// Cost: O(all sessions across all projects in the root) per call. The
		// MVP disk store prioritizes simplicity over index freshness, matching
		// `buildLinkageView` / `locateSession` which use the same pattern. A
		// production driver would maintain a topicId → sessionIds secondary
		// index populated on createSession / deleteSession. Acceptable for
		// TopicManager archive/delete today because those operations are
		// admin-initiated and infrequent.
		const projectsDir = join(this.rootDir, 'projects')
		const projectDirs = await records.scanNames(projectsDir, '')

		const results: Session[] = []
		for (const rawProject of projectDirs) {
			if (!isEntityId(rawProject, 'project')) continue
			const sessionsRoot = join(projectsDir, rawProject, 'sessions')
			const sessionDirs = await records.scanNames(sessionsRoot, '')
			for (const rawSessionId of sessionDirs) {
				if (!isEntityId(rawSessionId, 'session')) continue
				const path = join(sessionsRoot, rawSessionId)
				const raw = await records.read<PersistedSession>(join(path, 'session.json'))
				if (!raw) continue
				if (raw.tenantId !== tenantId) continue
				if (raw.topicId !== topicId) continue
				results.push(deserializeSession(raw))
				this.sessionIndex.set(raw.id, {
					sessionId: raw.id,
					projectId: rawProject as ProjectId,
					path,
				})
			}
		}
		results.sort((a, b) => a.createdAt.getTime() - b.createdAt.getTime())
		return results
	}

	async listSessionsByProject(
		projectId: ProjectId,
		tenantId: TenantId,
	): Promise<readonly Session[]> {
		// One directory, not the whole root: sessions live under their project,
		// so this is the cheap direction. `listSessionsByTopic` has to scan
		// every project precisely because `topicId` is denormalised onto the
		// record rather than expressed in the layout.
		const sessionsRoot = join(this.projectDir(projectId), 'sessions')
		const sessionDirs = await records.scanNames(sessionsRoot, '')

		const results: Session[] = []
		for (const rawSessionId of sessionDirs) {
			if (!isEntityId(rawSessionId, 'session')) continue
			const path = join(sessionsRoot, rawSessionId)
			const raw = await records.read<PersistedSession>(join(path, 'session.json'))
			if (!raw) continue
			if (raw.tenantId !== tenantId) continue
			results.push(deserializeSession(raw))
			this.sessionIndex.set(raw.id, { sessionId: raw.id, projectId, path })
		}
		results.sort(
			(a, b) => a.createdAt.getTime() - b.createdAt.getTime() || a.id.localeCompare(b.id),
		)
		return results
	}

	async updateSession(
		session: Session,
		tenantId: TenantId,
		expectedOwnerVersion?: number,
	): Promise<void> {
		const located = await this.locateSession(session.id)
		if (!located) {
			throw new Error(`Session ${session.id} not found`)
		}
		if (session.tenantId !== tenantId) {
			throw new TenantIsolationError({
				requested: tenantId,
				resource: `session(${session.id}) payload`,
			})
		}
		const existing = await records.read<PersistedSession>(join(located.path, 'session.json'))
		if (existing) {
			this.assertTenant(existing.tenantId, tenantId, `session(${session.id})`)
		}
		// Against what is on disk, not against the payload. The write itself is
		// atomic; this read-compare-write is NOT a critical section, so two
		// processes can still both pass — see the contract note on
		// `SessionStore.updateSession`. In one process it is the real lock.
		if (
			expectedOwnerVersion !== undefined &&
			existing !== null &&
			existing.ownerVersion !== expectedOwnerVersion
		) {
			throw new StaleSessionError({
				sessionId: session.id,
				expectedVersion: expectedOwnerVersion,
				actualVersion: existing.ownerVersion,
			})
		}
		const updated: Session = { ...session, updatedAt: new Date() }
		await records.write(join(located.path, 'session.json'), serializeSession(updated))
	}

	async deleteSession(sessionId: SessionId, tenantId: TenantId): Promise<void> {
		const located = await this.locateSession(sessionId)
		if (!located) return // Idempotent: missing = no-op.
		const existing = await records.read<PersistedSession>(join(located.path, 'session.json'))
		if (!existing) return
		this.assertTenant(existing.tenantId, tenantId, `session(${sessionId})`)

		// Policy: reject if sub-sessions are attached. Callers must delete
		// children first — Convention #5 deny-by-default; no implicit cascade.
		// We check BOTH directions (this session as parent, or as child) to
		// match the in-memory semantics.
		const subsDir = join(located.path, 'subsessions')
		const subEntries = await records.scanNames(subsDir, '')
		if (subEntries.some((entry) => isEntityId(entry, 'subSession'))) {
			throw new Error(
				`Session ${sessionId} has attached sub-sessions; delete them before deleting the session`,
			)
		}

		// Also scan tree for sub-session records that reference this session as
		// `childSessionId`. We need to walk siblings; acceptable cost for the
		// MVP disk store since the broadcast rollback path always pairs a
		// deleteSubSession + deleteSession call on the child (no orphans at
		// steady state).
		const projectsDir = join(this.rootDir, 'projects')
		const projectDirs = await records.scanNames(projectsDir, '')
		for (const rawProject of projectDirs) {
			if (!isEntityId(rawProject, 'project')) continue
			const sessionsRoot = join(projectsDir, rawProject, 'sessions')
			const siblingSessions = await records.scanNames(sessionsRoot, '')
			for (const rawSib of siblingSessions) {
				if (!isEntityId(rawSib, 'session')) continue
				const sibSubsDir = join(sessionsRoot, rawSib, 'subsessions')
				const sibSubs = await records.scanNames(sibSubsDir, '')
				for (const rawSub of sibSubs) {
					if (!isEntityId(rawSub, 'subSession')) continue
					const subRaw = await records.read<PersistedSubSession>(
						join(sibSubsDir, rawSub, 'subsession.json'),
					)
					if (!subRaw) continue
					if (subRaw.childSessionId === sessionId || subRaw.parentSessionId === sessionId) {
						throw new Error(
							`Session ${sessionId} has attached sub-sessions; delete them before deleting the session`,
						)
					}
				}
			}
		}

		// Recursive removal — `fs.rm` with `recursive: true` is the atomic
		// primitive for bulk delete. No write-tmp-rename applies here (we're
		// destroying state, not creating it).
		await rm(located.path, { recursive: true, force: true })
		this.sessionIndex.delete(sessionId)
	}

	// SubSession CRUD ---------------------------------------------------------

	async createSubSession(params: CreateSubSessionParams, tenantId: TenantId): Promise<SubSession> {
		const parent = await this.getSession(params.parentSessionId, tenantId)
		if (!parent) throw new Error(`Parent session ${params.parentSessionId} not found`)

		const child = await this.getSession(params.childSessionId, tenantId)
		if (!child) throw new Error(`Child session ${params.childSessionId} not found`)

		const parentLoc = this.sessionIndex.get(params.parentSessionId)
		if (!parentLoc) throw new Error(`Parent session ${params.parentSessionId} missing from index`)

		const now = new Date()
		const subSession: SubSession = {
			id: generateSubSessionId(),
			parentSessionId: params.parentSessionId,
			childSessionId: params.childSessionId,
			kind: params.kind,
			status: 'pending',
			spawnedBy: params.spawnedBy,
			spawnedAt: now,
			failureMode: params.failureMode ?? 'delegate',
			completionMode: params.completionMode ?? 'summary_ref',
			workspaceId: null,
			updatedAt: now,
		}
		const dir = join(parentLoc.path, 'subsessions', subSession.id)
		await mkdir(dir, { recursive: true })
		await records.write(join(dir, 'subsession.json'), serializeSubSession(subSession, tenantId))
		this.subSessionIndex.set(subSession.id, {
			subSessionId: subSession.id,
			sessionId: params.parentSessionId,
			projectId: parentLoc.projectId,
			path: dir,
		})
		return subSession
	}

	async getSubSession(subSessionId: SubSessionId, tenantId: TenantId): Promise<SubSession | null> {
		const located = await this.locateSubSession(subSessionId)
		if (!located) return null
		const raw = await records.read<PersistedSubSession>(join(located.path, 'subsession.json'))
		if (!raw) return null
		this.assertTenant(raw.tenantId, tenantId, `sub-session(${subSessionId})`)
		return deserializeSubSession(raw)
	}

	async updateSubSession(subSession: SubSession, tenantId: TenantId): Promise<void> {
		const located = await this.locateSubSession(subSession.id)
		if (!located) {
			throw new Error(`SubSession ${subSession.id} not found`)
		}
		const existing = await records.read<PersistedSubSession>(join(located.path, 'subsession.json'))
		if (existing) {
			this.assertTenant(existing.tenantId, tenantId, `sub-session(${subSession.id})`)
		}
		const updated: SubSession = { ...subSession, updatedAt: new Date() }
		await records.write(
			join(located.path, 'subsession.json'),
			serializeSubSession(updated, tenantId),
		)
	}

	async deleteSubSession(subSessionId: SubSessionId, tenantId: TenantId): Promise<void> {
		const located = await this.locateSubSession(subSessionId)
		if (!located) return // Idempotent: missing = no-op.
		const existing = await records.read<PersistedSubSession>(join(located.path, 'subsession.json'))
		if (!existing) {
			// Record vanished between locate + read — treat as already deleted.
			this.subSessionIndex.delete(subSessionId)
			return
		}
		this.assertTenant(existing.tenantId, tenantId, `sub-session(${subSessionId})`)

		await rm(located.path, { recursive: true, force: true })
		this.subSessionIndex.delete(subSessionId)
	}

	// Linkage -----------------------------------------------------------------

	async getChildren(sessionId: SessionId, tenantId: TenantId): Promise<readonly SubSession[]> {
		const session = await this.getSession(sessionId, tenantId)
		if (!session) return []
		const view = await this.buildLinkageView(tenantId)
		return orderChildren(getChildren(view, sessionId))
	}

	async getAncestry(sessionId: SessionId, tenantId: TenantId): Promise<readonly SessionId[]> {
		const session = await this.getSession(sessionId, tenantId)
		if (!session) return []
		const view = await this.buildLinkageView(tenantId)
		return getAncestry(view, sessionId)
	}

	async drill(sessionId: SessionId, tenantId: TenantId): Promise<SessionView | null> {
		const session = await this.getSession(sessionId, tenantId)
		if (!session) return null
		const view = await this.buildLinkageView(tenantId)
		return {
			session,
			children: orderChildren(getChildren(view, sessionId)),
			ancestry: getAncestry(view, sessionId),
		}
	}

	// Summary (§4.7 / §8.1) ---------------------------------------------------

	/**
	 * Atomic materialize-with-terminal-transition (§8.1). Two write-tmp-renames:
	 *
	 *   1. Persist `summary.json` under the session directory.
	 *   2. Flip `session.json#status` to `'idle'` if it's in a non-terminal
	 *      state (`'active' | 'locked' | 'awaiting_merge'`).
	 *
	 * Each rename is atomic individually. A crash between step 1 and step 2
	 * leaves summary present + session still non-terminal — recovery replays
	 * the flip via {@link SessionSummaryMaterializer.recover}. Idempotent when
	 * the same summary is re-presented (recovery path); rejects a *different*
	 * summary for the same session as {@link SessionAlreadySummarizedError}.
	 */
	async recordSummary(
		summary: SessionSummaryRef & { materializedBy: 'kernel' },
		tenantId: TenantId,
	): Promise<void> {
		if (summary.tenantId !== tenantId) {
			throw new TenantIsolationError({
				requested: tenantId,
				resource: `summary(${summary.id}) payload`,
			})
		}

		const located = await this.locateSession(summary.sessionRef)
		if (!located) {
			throw new Error(`Session ${summary.sessionRef} not found`)
		}
		const sessionRaw = await records.read<PersistedSession>(join(located.path, 'session.json'))
		if (!sessionRaw) {
			throw new Error(`Session ${summary.sessionRef} not found on disk`)
		}
		this.assertTenant(sessionRaw.tenantId, tenantId, `session(${summary.sessionRef})`)

		const summaryPath = join(located.path, 'summary.json')
		const existingRaw = await records.read<PersistedSummary>(summaryPath)
		if (existingRaw) {
			this.assertTenant(existingRaw.tenantId, tenantId, `summary(${existingRaw.id})`)
			if (existingRaw.id !== summary.id) {
				throw new SessionAlreadySummarizedError({
					sessionId: summary.sessionRef,
					existingSummaryId: existingRaw.id,
				})
			}
			// Same summary id — recovery replay. No duplicate write; fall through
			// to the status flip so crash-between-writes is recovered.
		} else {
			// Step 1: persist summary.
			await records.write(summaryPath, serializeSummary(summary))
		}

		// Step 2: flip session status atomically if still non-terminal.
		if (SUMMARY_TERMINAL_FLIP_STATUSES.has(sessionRaw.status)) {
			const flipped: PersistedSession = {
				...sessionRaw,
				status: 'idle',
				updatedAt: new Date().toISOString(),
			}
			await records.write(join(located.path, 'session.json'), flipped)
		}
	}

	async getSummary(sessionId: SessionId, tenantId: TenantId): Promise<SessionSummaryRef | null> {
		const located = await this.locateSession(sessionId)
		if (!located) return null
		const raw = await records.read<PersistedSummary>(join(located.path, 'summary.json'))
		if (!raw) return null
		this.assertTenant(raw.tenantId, tenantId, `summary(${raw.id})`)
		return deserializeSummary(raw)
	}

	// Helpers -----------------------------------------------------------------

	private assertTenant(actual: TenantId, requested: TenantId, resource: string): void {
		if (actual !== requested) {
			throw new TenantIsolationError({ requested, resource })
		}
	}

	private projectDir(projectId: ProjectId): string {
		asProjectId(projectId)
		const cached = this.projectIndex.get(projectId)
		if (cached) return cached.path
		const path = join(this.rootDir, 'projects', projectId)
		this.projectIndex.set(projectId, { projectId, path })
		return path
	}

	private async locateSession(sessionId: SessionId): Promise<SessionIndexEntry | null> {
		asSessionId(sessionId)
		const cached = this.sessionIndex.get(sessionId)
		if (cached) return cached

		const projectsDir = join(this.rootDir, 'projects')
		// ENOENT lists as empty; the loop below then falls through to the
		// same `return null` the old catch took directly.
		const projectDirs = await records.scanNames(projectsDir, '')
		for (const rawId of projectDirs) {
			if (!isEntityId(rawId, 'project')) continue
			const projectId = rawId as ProjectId
			const sessionsRoot = join(projectsDir, projectId, 'sessions')
			const sessionDirs = await records.scanNames(sessionsRoot, '')
			for (const rawSessionId of sessionDirs) {
				if (!isEntityId(rawSessionId, 'session')) continue
				if (rawSessionId === sessionId) {
					const entry: SessionIndexEntry = {
						sessionId,
						projectId,
						path: join(sessionsRoot, rawSessionId),
					}
					this.sessionIndex.set(sessionId, entry)
					return entry
				}
			}
		}
		return null
	}

	private async locateSubSession(subSessionId: SubSessionId): Promise<{
		subSessionId: SubSessionId
		sessionId: SessionId
		projectId: ProjectId
		path: string
	} | null> {
		asSubSessionId(subSessionId)
		const cached = this.subSessionIndex.get(subSessionId)
		if (cached) return cached

		const projectsDir = join(this.rootDir, 'projects')
		// ENOENT lists as empty; the loop below then falls through to the
		// same `return null` the old catch took directly.
		const projectDirs = await records.scanNames(projectsDir, '')
		for (const rawProject of projectDirs) {
			if (!isEntityId(rawProject, 'project')) continue
			const projectId = rawProject as ProjectId
			const sessionsRoot = join(projectsDir, projectId, 'sessions')
			const sessionDirs = await records.scanNames(sessionsRoot, '')
			for (const rawSession of sessionDirs) {
				if (!isEntityId(rawSession, 'session')) continue
				const sessionId = rawSession as SessionId
				const subsDir = join(sessionsRoot, sessionId, 'subsessions')
				const subDirs = await records.scanNames(subsDir, '')
				for (const rawSub of subDirs) {
					if (rawSub === subSessionId) {
						const entry = {
							subSessionId,
							sessionId,
							projectId,
							path: join(subsDir, rawSub),
						}
						this.subSessionIndex.set(subSessionId, entry)
						return entry
					}
				}
			}
		}
		return null
	}

	private async buildLinkageView(tenantId: TenantId): Promise<LinkageView> {
		// Walk the full projects → sessions → subsessions tree once per call.
		// Acceptable for an MVP disk store; a production impl would cache.
		const allSubs: SubSession[] = []
		const projectsDir = join(this.rootDir, 'projects')
		// ENOENT lists as empty, and a view over no sub-sessions IS the empty
		// view the old catch returned directly.
		const projectDirs = await records.scanNames(projectsDir, '')

		for (const rawProject of projectDirs) {
			if (!isEntityId(rawProject, 'project')) continue
			const sessionsRoot = join(projectsDir, rawProject, 'sessions')
			const sessionDirs = await records.scanNames(sessionsRoot, '')
			for (const rawSession of sessionDirs) {
				if (!isEntityId(rawSession, 'session')) continue
				const subsRoot = join(sessionsRoot, rawSession, 'subsessions')
				const subDirs = await records.scanNames(subsRoot, '')
				for (const rawSub of subDirs) {
					if (!isEntityId(rawSub, 'subSession')) continue
					const raw = await records.read<PersistedSubSession>(
						join(subsRoot, rawSub, 'subsession.json'),
					)
					if (!raw) continue
					if (raw.tenantId !== tenantId) continue
					allSubs.push(deserializeSubSession(raw))
				}
			}
		}

		return {
			findChildSubSessions: (parentSessionId) =>
				allSubs.filter((s) => s.parentSessionId === parentSessionId),
			findParentSubSession: (childSessionId) =>
				allSubs.find((s) => s.childSessionId === childSessionId) ?? null,
		}
	}
}

// Serialization helpers -----------------------------------------------------

function serializeProject(p: Project): PersistedProject {
	return {
		id: p.id,
		tenantId: p.tenantId,
		name: p.name,
		config: p.config,
		status: p.status,
		ownerVersion: p.ownerVersion,
		...(p.rootPath !== undefined ? { rootPath: p.rootPath } : {}),
		createdAt: p.createdAt.toISOString(),
		updatedAt: p.updatedAt.toISOString(),
	}
}

function deserializeProject(p: PersistedProject): Project {
	return {
		id: p.id,
		tenantId: p.tenantId,
		name: p.name,
		config: p.config,
		// A project.json written before these fields existed reads as an open
		// project at version 0, which is what it was. Leaving `ownerVersion`
		// undefined would be worse than a wrong default: every compare-and-set
		// against it would fail, so an existing store could never be closed.
		status: p.status ?? 'open',
		ownerVersion: p.ownerVersion ?? 0,
		...(p.rootPath !== undefined ? { rootPath: p.rootPath } : {}),
		createdAt: new Date(p.createdAt),
		updatedAt: new Date(p.updatedAt),
	}
}

function serializeSession(s: Session): PersistedSession {
	return {
		id: s.id,
		topicId: s.topicId,
		projectId: s.projectId,
		tenantId: s.tenantId,
		status: s.status,
		currentActor: s.currentActor,
		previousActors: s.previousActors,
		workspaceId: s.workspaceId,
		ownerVersion: s.ownerVersion,
		createdAt: s.createdAt.toISOString(),
		updatedAt: s.updatedAt.toISOString(),
	}
}

function deserializeSession(s: PersistedSession): Session {
	return {
		id: s.id,
		topicId: s.topicId,
		projectId: s.projectId,
		tenantId: s.tenantId,
		status: s.status,
		currentActor: s.currentActor,
		previousActors: s.previousActors,
		workspaceId: s.workspaceId,
		ownerVersion: s.ownerVersion,
		createdAt: new Date(s.createdAt),
		updatedAt: new Date(s.updatedAt),
	}
}

function serializeSubSession(s: SubSession, tenantId: TenantId): PersistedSubSession {
	return {
		id: s.id,
		parentSessionId: s.parentSessionId,
		childSessionId: s.childSessionId,
		tenantId,
		kind: s.kind,
		status: s.status,
		spawnedBy: s.spawnedBy,
		spawnedAt: s.spawnedAt.toISOString(),
		failureMode: s.failureMode,
		completionMode: s.completionMode,
		workspaceId: s.workspaceId,
		...(s.broadcastGroupId !== undefined && { broadcastGroupId: s.broadcastGroupId }),
		...(s.summaryRef !== undefined && { summaryRef: s.summaryRef }),
		...(s.archiveRef !== undefined && { archiveRef: s.archiveRef }),
		...(s.archivedAt !== undefined && { archivedAt: s.archivedAt.toISOString() }),
		updatedAt: s.updatedAt.toISOString(),
	}
}

function deserializeSubSession(s: PersistedSubSession): SubSession {
	return {
		id: s.id,
		parentSessionId: s.parentSessionId,
		childSessionId: s.childSessionId,
		kind: s.kind,
		status: s.status,
		spawnedBy: s.spawnedBy,
		spawnedAt: new Date(s.spawnedAt),
		failureMode: s.failureMode,
		completionMode: s.completionMode,
		workspaceId: s.workspaceId,
		...(s.broadcastGroupId !== undefined && { broadcastGroupId: s.broadcastGroupId }),
		...(s.summaryRef !== undefined && { summaryRef: s.summaryRef }),
		...(s.archiveRef !== undefined && { archiveRef: s.archiveRef }),
		...(s.archivedAt !== undefined && { archivedAt: new Date(s.archivedAt) }),
		updatedAt: new Date(s.updatedAt),
	}
}

function serializeSummary(s: SessionSummaryRef): PersistedSummary {
	return {
		id: s.id,
		sessionRef: s.sessionRef,
		tenantId: s.tenantId,
		outcome: s.outcome,
		deliverables: s.deliverables,
		agentSummary: s.agentSummary,
		keyDecisions: s.keyDecisions.map((k) => ({
			at: k.at.toISOString(),
			summary: k.summary,
		})),
		at: s.at.toISOString(),
		materializedBy: 'kernel',
	}
}

function deserializeSummary(s: PersistedSummary): SessionSummaryRef {
	const decisions: SessionSummaryKeyDecision[] = s.keyDecisions.map((k) => ({
		at: new Date(k.at),
		summary: k.summary,
	}))
	return {
		id: s.id,
		sessionRef: s.sessionRef,
		tenantId: s.tenantId,
		outcome: s.outcome,
		deliverables: s.deliverables,
		agentSummary: s.agentSummary,
		keyDecisions: decisions,
		at: new Date(s.at),
		materializedBy: 'kernel',
	}
}