diff --git a/apps/desktop/src/main/__tests__/new-session-project.test.ts b/apps/desktop/src/main/__tests__/new-session-project.test.ts index bc5fd2fcf5..88b0507f4c 100644 --- a/apps/desktop/src/main/__tests__/new-session-project.test.ts +++ b/apps/desktop/src/main/__tests__/new-session-project.test.ts @@ -175,7 +175,7 @@ test('new sessions resolve a merged project alias to the surviving project id', const original = await catalog.register(originalPath); await rm(originalPath, { recursive: true, force: true }); const duplicate = await catalog.register(cwd); - await catalog.relink(original.id, cwd, async () => {}); + await catalog.relinkWithSessions(original.id, cwd); const resolved = await resolveNewSessionProjectInput( makeInput(cwd, { projectId: duplicate.id }), diff --git a/apps/desktop/src/main/__tests__/project-management-service.test.ts b/apps/desktop/src/main/__tests__/project-management-service.test.ts index e0471308a0..4a5153074d 100644 --- a/apps/desktop/src/main/__tests__/project-management-service.test.ts +++ b/apps/desktop/src/main/__tests__/project-management-service.test.ts @@ -1,9 +1,9 @@ import assert from 'node:assert/strict'; -import { mkdir, mkdtemp, realpath, rename, rm } from 'node:fs/promises'; +import { mkdir, mkdtemp, realpath, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { test } from 'node:test'; -import { createProjectCatalog, createSessionStore } from '@maka/storage'; +import { createProjectCatalog } from '@maka/storage'; import { createProjectManagementService } from '../project-management-service.js'; test('project management service owns selection and reversible lifecycle actions', async () => { @@ -20,12 +20,6 @@ test('project management service owns selection and reversible lifecycle actions }); const service = createProjectManagementService({ catalog, - sessions: { - listHeaders: async () => [], - updateHeader: async () => { - throw new Error('No sessions expected'); - }, - }, chooseDirectory: async () => nextDirectory, selection: { currentSelection: async () => ({ @@ -80,12 +74,6 @@ test('project management service rejects malformed IPC identities before catalog const base = await mkdtemp(join(tmpdir(), 'maka-project-service-input-')); const service = createProjectManagementService({ catalog: createProjectCatalog(join(base, 'storage')), - sessions: { - listHeaders: async () => [], - updateHeader: async () => { - throw new Error('No sessions expected'); - }, - }, chooseDirectory: async () => undefined, selection: { currentSelection: async () => ({ projectId: undefined, path: base }), @@ -112,12 +100,6 @@ test('project management service resolves a legacy path into one canonical selec const savedSelections: Array<{ projectId: string | null; projectPath: string }> = []; const service = createProjectManagementService({ catalog, - sessions: { - listHeaders: async () => [], - updateHeader: async () => { - throw new Error('No sessions expected'); - }, - }, chooseDirectory: async () => undefined, selection: { currentSelection: async () => ({ @@ -153,12 +135,6 @@ test('project management service persists an explicit no-project selection in ma const savedSelections: Array<{ projectId: string | null; projectPath: string }> = []; const service = createProjectManagementService({ catalog: createProjectCatalog(join(base, 'storage')), - sessions: { - listHeaders: async () => [], - updateHeader: async () => { - throw new Error('No sessions expected'); - }, - }, chooseDirectory: async () => undefined, selection: { currentSelection: async () => ({ @@ -208,12 +184,6 @@ test('archiving the current project resolves fallback or no-project inside main' }; const service = createProjectManagementService({ catalog, - sessions: { - listHeaders: async () => [], - updateHeader: async () => { - throw new Error('No sessions expected'); - }, - }, chooseDirectory: async () => undefined, selection: { currentSelection: async () => selection, @@ -251,128 +221,3 @@ test('archiving the current project resolves fallback or no-project inside main' await rm(base, { recursive: true, force: true }); } }); - -test('relinking merges a project that was accidentally added from its new path', async () => { - const base = await mkdtemp(join(tmpdir(), 'maka-project-service-merge-')); - const oldPath = join(base, 'old-location'); - const newPath = join(base, 'new-location'); - const secondPath = join(base, 'second-location'); - const storage = join(base, 'storage'); - await mkdir(oldPath); - let nextDirectory: string | undefined = oldPath; - let nextId = 0; - const catalog = createProjectCatalog(storage, { - now: () => 1_000, - createId: () => `project-${++nextId}`, - }); - const sessions = createSessionStore(storage); - let failUpdateNumber: number | undefined; - let updateCount = 0; - const service = createProjectManagementService({ - catalog, - sessions: { - listHeaders: () => sessions.listHeaders(), - updateHeader: async (sessionId, patch) => { - updateCount += 1; - if (updateCount === failUpdateNumber) { - throw new Error('injected session reassignment failure'); - } - return sessions.updateHeader(sessionId, patch); - }, - }, - chooseDirectory: async () => nextDirectory, - selection: { - currentSelection: async () => ({ - projectId: undefined, - path: nextDirectory ? await realpath(nextDirectory) : base, - }), - setSelection: () => {}, - }, - }); - - try { - const original = await service.add(); - assert.equal(original.ok, true); - if (!original.ok) throw new Error('Expected original project'); - await service.rename(original.project.id, 'Original name'); - await rename(oldPath, newPath); - - nextDirectory = newPath; - const duplicate = await service.add(); - assert.equal(duplicate.ok, true); - if (!duplicate.ok) throw new Error('Expected duplicate project'); - const oldSession = await sessions.create( - makeSessionInput(oldPath, original.project.id, 'Old history'), - ); - const newSession = await sessions.create( - makeSessionInput(newPath, duplicate.project.id, 'New history'), - ); - - failUpdateNumber = 2; - await assert.rejects( - () => service.relink(original.project.id), - /injected session reassignment failure/, - ); - assert.deepEqual( - (await catalog.list()).map((project) => project.id).sort(), - [original.project.id, duplicate.project.id].sort(), - ); - - failUpdateNumber = undefined; - updateCount = 0; - const merged = await service.relink(original.project.id); - - assert.equal(merged.ok, true); - if (!merged.ok) throw new Error('Expected merged project'); - assert.equal(merged.project.id, original.project.id); - assert.equal(merged.project.name, 'Original name'); - assert.equal(merged.project.preferredPath, await realpath(newPath)); - assert.deepEqual( - (await catalog.list()).map((project) => project.id), - [original.project.id], - ); - assert.equal( - (await sessions.readHeaderSnapshot(oldSession.id)).projectId, - original.project.id, - ); - assert.equal( - (await sessions.readHeaderSnapshot(oldSession.id)).cwd, - await realpath(newPath), - ); - assert.equal( - (await sessions.readHeaderSnapshot(newSession.id)).projectId, - original.project.id, - ); - - const lateAliasSession = await sessions.create( - makeSessionInput(newPath, duplicate.project.id, 'Late alias history'), - ); - await rename(newPath, secondPath); - nextDirectory = secondPath; - const relinkedAgain = await service.relink(original.project.id); - - assert.equal(relinkedAgain.ok, true); - const lateAliasHeader = await sessions.readHeaderSnapshot(lateAliasSession.id); - assert.equal(lateAliasHeader.projectId, original.project.id); - assert.equal( - lateAliasHeader.cwd, - await realpath(secondPath), - ); - } finally { - await sessions.close?.(); - await rm(base, { recursive: true, force: true }); - } -}); - -function makeSessionInput(cwd: string, projectId: string, name: string) { - return { - cwd, - projectId, - backend: 'fake' as const, - llmConnectionSlug: 'fake', - model: 'fake-model', - permissionMode: 'ask' as const, - name, - labels: [], - }; -} diff --git a/apps/desktop/src/main/__tests__/runtime-host-project-catalog.test.ts b/apps/desktop/src/main/__tests__/runtime-host-project-catalog.test.ts new file mode 100644 index 0000000000..1127f0fac7 --- /dev/null +++ b/apps/desktop/src/main/__tests__/runtime-host-project-catalog.test.ts @@ -0,0 +1,24 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; +import { RuntimeHostOperationError } from '@maka/runtime-host/client'; +import { isProjectPathMismatchError } from '@maka/storage'; +import { createRuntimeHostProjectCatalog } from '../runtime-host-project-catalog.js'; + +test('Host touch conflicts retain the Project path mismatch contract', async () => { + const catalog = createRuntimeHostProjectCatalog(() => + ({ + touchProject: async () => { + throw new RuntimeHostOperationError( + 'project.catalog.mutate', + 'operation_conflict', + 'Path does not belong to project project-1', + ); + }, + }) as never, + ); + + await assert.rejects( + () => catalog.touch('project-1', '/workspace/other'), + (error: unknown) => isProjectPathMismatchError(error), + ); +}); diff --git a/apps/desktop/src/main/__tests__/runtime-host-project-session-catalog.test.ts b/apps/desktop/src/main/__tests__/runtime-host-project-session-catalog.test.ts deleted file mode 100644 index af0d014838..0000000000 --- a/apps/desktop/src/main/__tests__/runtime-host-project-session-catalog.test.ts +++ /dev/null @@ -1,58 +0,0 @@ -import assert from "node:assert/strict"; -import { describe, test } from "node:test"; -import { createRuntimeHostProjectSessionCatalog } from "../runtime-host-project-session-catalog.js"; - -describe("Runtime Host project Session catalog", () => { - test("projects Host Session headers without taking local authority", async () => { - const catalog = createRuntimeHostProjectSessionCatalog({ - listSessions: async () => [ - { id: "session-1", cwd: "/workspace/one", projectId: "project-1" }, - ], - relocateSessionCwd: async () => assert.fail("unexpected relocation"), - updateSessionMetadata: async () => assert.fail("unexpected metadata update"), - } as never); - - assert.deepEqual(await catalog.listHeaders(), [ - { id: "session-1", cwd: "/workspace/one", projectId: "project-1" }, - ]); - }); - - test("relocates cwd and project association in one Host mutation", async () => { - const calls: unknown[][] = []; - const catalog = createRuntimeHostProjectSessionCatalog({ - listSessions: async () => [], - relocateSessionCwd: async (...args: unknown[]) => { - calls.push(args); - return {} as never; - }, - updateSessionMetadata: async () => assert.fail("must not split the mutation"), - } as never); - - await catalog.updateHeader("session-1", { - cwd: "/workspace/next", - projectId: "project-2", - }); - - assert.deepEqual(calls, [ - ["session-1", "/workspace/next", "project-2"], - ]); - }); - - test("updates a project association without relocating cwd", async () => { - const calls: unknown[][] = []; - const catalog = createRuntimeHostProjectSessionCatalog({ - listSessions: async () => [], - relocateSessionCwd: async () => assert.fail("unexpected relocation"), - updateSessionMetadata: async (...args: unknown[]) => { - calls.push(args); - return {} as never; - }, - } as never); - - await catalog.updateHeader("session-1", { projectId: "project-2" }); - - assert.deepEqual(calls, [ - ["session-1", { projectId: "project-2" }], - ]); - }); -}); diff --git a/apps/desktop/src/main/project-management-service.ts b/apps/desktop/src/main/project-management-service.ts index f870fdf97d..ee5b240972 100644 --- a/apps/desktop/src/main/project-management-service.ts +++ b/apps/desktop/src/main/project-management-service.ts @@ -1,4 +1,4 @@ -import type { ProjectCatalog, ProjectRecord } from '@maka/storage'; +import type { ProjectRecord } from '@maka/core'; import type { CurrentProjectSelection } from './project-root-controller.js'; type DirectoryActionResult = @@ -29,19 +29,18 @@ export interface ProjectManagementService { restore(projectId: unknown): Promise; } -export interface ProjectSessionCatalog { - listHeaders(): Promise< - Array<{ readonly id: string; readonly cwd: string; readonly projectId?: string | null }> - >; - updateHeader( - sessionId: string, - patch: { readonly cwd?: string; readonly projectId?: string | null }, - ): Promise; +export interface ProjectManagementCatalog { + list(): Promise; + register(path: string): Promise; + select(projectId: string): Promise<{ project: ProjectRecord; path: string }>; + relink(projectId: string, path: string): Promise; + rename(projectId: string, name: string): Promise; + archive(projectId: string): Promise; + restore(projectId: string): Promise; } export function createProjectManagementService(deps: { - catalog: ProjectCatalog; - sessions: ProjectSessionCatalog; + catalog: ProjectManagementCatalog; chooseDirectory(): Promise; selection: { currentSelection(): Promise; @@ -112,38 +111,17 @@ export function createProjectManagementService(deps: { const id = requireProjectId(projectId); const path = await deps.chooseDirectory(); if (!path) return { ok: false, reason: 'cancelled' }; - let selectedProjectWasRelinked = false; - const prepareSessions = async (context: { - projectId: string; - projectAliases: string[]; - destinationPath: string; - previousLocations: Array<{ path: string }>; - conflictingProjectId?: string; - conflictingProjectAliases?: string[]; - }) => { - const selectedPath = (await deps.selection.currentSelection()).path; - selectedProjectWasRelinked = context.previousLocations.some( - (location) => location.path === selectedPath, - ); - const survivingIds = new Set([context.projectId, ...context.projectAliases]); - const conflictingIds = new Set([ - ...(context.conflictingProjectId ? [context.conflictingProjectId] : []), - ...(context.conflictingProjectAliases ?? []), - ]); - for (const header of await deps.sessions.listHeaders()) { - if (header.projectId && survivingIds.has(header.projectId)) { - await deps.sessions.updateHeader(header.id, { - cwd: context.destinationPath, - ...(header.projectId !== context.projectId - ? { projectId: context.projectId } - : {}), - }); - } else if (header.projectId && conflictingIds.has(header.projectId)) { - await deps.sessions.updateHeader(header.id, { projectId: context.projectId }); - } - } - }; - const project = await deps.catalog.relink(id, path, prepareSessions); + const [selection, projects] = await Promise.all([ + deps.selection.currentSelection(), + deps.catalog.list(), + ]); + const previous = projects.find( + (project) => project.id === id || project.aliases?.includes(id), + ); + const selectedProjectWasRelinked = previous?.locations.some( + (location) => location.path === selection.path, + ); + const project = await deps.catalog.relink(id, path); if (selectedProjectWasRelinked && project.preferredPath) { deps.selection.setSelection(project.id, project.preferredPath); } diff --git a/apps/desktop/src/main/runtime-host-boot.ts b/apps/desktop/src/main/runtime-host-boot.ts index d48c9bfbcf..2d0a286fdd 100644 --- a/apps/desktop/src/main/runtime-host-boot.ts +++ b/apps/desktop/src/main/runtime-host-boot.ts @@ -23,7 +23,6 @@ import { McpClientManager } from "@maka/mcp"; import { createSettingsStore, createMcpConfigStore, - createProjectCatalog, createSqlitePlanReminderStore, } from "@maka/storage"; import { registerAppIpc } from "./app-ipc-main.js"; @@ -100,7 +99,7 @@ import { registerRuntimeHostOAuthIpc } from "./runtime-host-oauth-ipc-main.js"; import { RuntimeHostOAuthPresentation } from "./runtime-host-oauth-presentation.js"; import { registerRuntimeHostPermissionsIpc } from "./runtime-host-permissions-ipc-main.js"; import { registerRuntimeHostSearchIpc } from "./runtime-host-search-ipc-main.js"; -import { createRuntimeHostProjectSessionCatalog } from "./runtime-host-project-session-catalog.js"; +import { createRuntimeHostProjectCatalog } from "./runtime-host-project-catalog.js"; import { toDesktopHostSessionSummary } from "./runtime-host-session-catalog-ipc-main.js"; import { loadRuntimeHostSettings, @@ -157,10 +156,6 @@ if (e2eFixture) { } } const settingsStore = createSettingsStore(workspaceRoot); -const projectCatalog = createProjectCatalog(workspaceRoot, { - onLegacyImportFailure: (error) => - console.error("[projects] projects.json could not be imported:", error), -}); const mcpConfigStore = createMcpConfigStore(workspaceRoot); const mcpManager = new McpClientManager({ clientName: "maka-desktop", @@ -238,13 +233,12 @@ const oauthPresentation = new RuntimeHostOAuthPresentation( ); let owner: RuntimeHostDesktopOwner | undefined; let runtimePolicyClient: DesktopRuntimeHostClient | undefined; +const projectCatalog = createRuntimeHostProjectCatalog(() => { + if (!runtimePolicyClient) throw new Error("Runtime Host client is unavailable"); + return runtimePolicyClient; +}); const projectManagement: ProjectManagementService = createProjectManagementService({ catalog: projectCatalog, - sessions: { - listHeaders: () => runtimeHostProjectSessionCatalog().listHeaders(), - updateHeader: (sessionId, patch) => - runtimeHostProjectSessionCatalog().updateHeader(sessionId, patch), - }, chooseDirectory: async () => { const result = await mainWindowController.showOpenDialog({ title: "Add project", @@ -254,10 +248,6 @@ const projectManagement: ProjectManagementService = createProjectManagementServi }, selection: projectRoot, }); -function runtimeHostProjectSessionCatalog() { - if (!runtimePolicyClient) throw new Error("Runtime Host client is unavailable"); - return createRuntimeHostProjectSessionCatalog(runtimePolicyClient); -} const mcpCapabilityPublisher = createCapabilityRevisionPublisher(() => mcpManager.toolSnapshotRevision(), ); @@ -522,6 +512,9 @@ function registerHostClientIpc( const unsubscribeSessionCatalogChanges = client.subscribeSessionCatalogChanges( ({ sessionId }) => emitSessionsChanged("updated", sessionId), ); + const unsubscribeProjectCatalogChanges = client.subscribeProjectCatalogChanges(() => { + mainWindowController.send("projects:changed"); + }); const capabilityBinding = mcpCapabilityPublisher.bind( controls.refreshClientCapabilities, ); @@ -759,6 +752,7 @@ function registerHostClientIpc( return async () => { unsubscribeConfigurationChanges(); unsubscribeSessionCatalogChanges(); + unsubscribeProjectCatalogChanges(); candidateSettingsBotsIpc.dispose(); if (settingsBotsIpc === candidateSettingsBotsIpc) { settingsBotsIpc = undefined; diff --git a/apps/desktop/src/main/runtime-host-client.ts b/apps/desktop/src/main/runtime-host-client.ts index dd9320ae2e..e42da6e8d7 100644 --- a/apps/desktop/src/main/runtime-host-client.ts +++ b/apps/desktop/src/main/runtime-host-client.ts @@ -1,5 +1,6 @@ import { createHash, randomUUID } from "node:crypto"; import type { AttachmentRef, ShellRunUpdate } from "@maka/core/events"; +import type { ProjectRecord } from "@maka/core"; import type { PlanSessionState, PlanUserControlInput } from "@maka/core/plan"; import { decodeStoredMessageForRead, @@ -34,6 +35,7 @@ import { readRuntimeHostConnectionCatalog, readRuntimeHostInvocableSkills, readRuntimeHostResources, + readRuntimeHostProjects, readRuntimeHostSessions, readRuntimeHostSkillCatalog, } from "@maka/runtime-host/client"; @@ -64,6 +66,9 @@ import { type PlanQueryResult, type PricingMutation, type PricingQueryResult, + type ProjectCatalogMutateInput, + type ProjectCatalogMutateResult, + type ProjectCatalogProject, type QueueRetractInput, type QueueRetractResult, type SessionCatalogFilter, @@ -200,6 +205,11 @@ export class DesktopRuntimeHostClient { return this.connection.subscribeConfigurationChanges(listener); } + subscribeProjectCatalogChanges(listener: (revision: number) => void): () => void { + this.#assertOpen(); + return this.connection.subscribeProjectCatalogChanges(listener); + } + subscribeSessionCatalogChanges( listener: (frame: SessionCatalogChangedFrame) => void, ): () => void { @@ -482,6 +492,78 @@ export class DesktopRuntimeHostClient { } } + async listProjects(): Promise { + this.#assertOpen(); + try { + return (await readRuntimeHostProjects(this.connection)).map(toProjectRecord); + } catch (error) { + if (!(error instanceof RuntimeHostCatalogReadError)) throw error; + throw new DesktopRuntimeHostClientError( + "catalog_unstable", + "Project catalog kept changing while Desktop read it", + ); + } + } + + async registerProject(path: string): Promise { + const result = await this.#mutateProject({ kind: "register", path }); + return this.#projectForMutation(result); + } + + async selectProject(projectId: string): Promise<{ project: ProjectRecord; path: string }> { + const result = await this.#mutateProject({ kind: "select", projectId }); + if (result.kind !== "selection") throw invalidProjection("Project selection"); + return { project: await this.#projectById(result.projectId), path: result.path }; + } + + async touchProject(projectId: string, path?: string): Promise { + return this.#projectForMutation( + await this.#mutateProject({ kind: "touch", projectId, path: path ?? null }), + ); + } + + async relinkProject(projectId: string, path: string): Promise { + return this.#projectForMutation( + await this.#mutateProject({ kind: "relink", projectId, path }), + ); + } + + async renameProject(projectId: string, name: string): Promise { + return this.#projectForMutation( + await this.#mutateProject({ kind: "rename", projectId, name }), + ); + } + + async archiveProject(projectId: string): Promise { + return this.#projectForMutation( + await this.#mutateProject({ kind: "archive", projectId }), + ); + } + + async restoreProject(projectId: string): Promise { + return this.#projectForMutation( + await this.#mutateProject({ kind: "restore", projectId }), + ); + } + + #mutateProject(input: ProjectCatalogMutateInput) { + this.#assertOpen(); + return this.#request("project.catalog.mutate", input); + } + + async #projectForMutation(result: ProjectCatalogMutateResult): Promise { + if (result.kind !== "project") throw invalidProjection("Project mutation"); + return this.#projectById(result.projectId); + } + + async #projectById(projectId: string): Promise { + const project = (await this.listProjects()).find( + (candidate) => candidate.id === projectId || candidate.aliases?.includes(projectId), + ); + if (!project) throw invalidProjection("Project mutation"); + return project; + } + async listArtifacts(sessionId: string): Promise { for (let attempt = 0; attempt < MAX_OPTIMISTIC_ATTEMPTS; attempt += 1) { const first = await this.#request("artifact.query", { @@ -1456,6 +1538,18 @@ function requireSessionProjection( ); } +function toProjectRecord(project: ProjectCatalogProject): ProjectRecord { + return { + id: project.id, + ...(project.aliases.length === 0 ? {} : { aliases: [...project.aliases] }), + name: project.name, + locations: project.locations.map((location) => ({ ...location })), + ...(project.archivedAt === null ? {} : { archivedAt: project.archivedAt }), + available: project.available, + ...(project.preferredPath === null ? {} : { preferredPath: project.preferredPath }), + }; +} + function clientClosed(): DesktopRuntimeHostClientError { return new DesktopRuntimeHostClientError( "client_closed", diff --git a/apps/desktop/src/main/runtime-host-project-catalog.ts b/apps/desktop/src/main/runtime-host-project-catalog.ts new file mode 100644 index 0000000000..b3a644c777 --- /dev/null +++ b/apps/desktop/src/main/runtime-host-project-catalog.ts @@ -0,0 +1,45 @@ +import type { ProjectRecord } from "@maka/core"; +import { RuntimeHostOperationError } from "@maka/runtime-host/client"; +import { ProjectPathMismatchError } from "@maka/storage"; +import type { ProjectManagementCatalog } from "./project-management-service.js"; +import type { DesktopRuntimeHostClient } from "./runtime-host-client.js"; + +export interface DesktopProjectCatalog extends ProjectManagementCatalog { + touch(projectId: string, path?: string): Promise; +} + +type RuntimeHostProjectClient = Pick< + DesktopRuntimeHostClient, + | "archiveProject" + | "listProjects" + | "registerProject" + | "relinkProject" + | "renameProject" + | "restoreProject" + | "selectProject" + | "touchProject" +>; + +export function createRuntimeHostProjectCatalog( + resolveClient: () => RuntimeHostProjectClient, +): DesktopProjectCatalog { + return { + list: () => resolveClient().listProjects(), + register: (path) => resolveClient().registerProject(path), + select: (projectId) => resolveClient().selectProject(projectId), + async touch(projectId, path) { + try { + return await resolveClient().touchProject(projectId, path); + } catch (error) { + if (error instanceof RuntimeHostOperationError && error.code === "operation_conflict") { + throw new ProjectPathMismatchError(projectId, path ?? ""); + } + throw error; + } + }, + relink: (projectId, path) => resolveClient().relinkProject(projectId, path), + rename: (projectId, name) => resolveClient().renameProject(projectId, name), + archive: (projectId) => resolveClient().archiveProject(projectId), + restore: (projectId) => resolveClient().restoreProject(projectId), + }; +} diff --git a/apps/desktop/src/main/runtime-host-project-session-catalog.ts b/apps/desktop/src/main/runtime-host-project-session-catalog.ts deleted file mode 100644 index ee1108164f..0000000000 --- a/apps/desktop/src/main/runtime-host-project-session-catalog.ts +++ /dev/null @@ -1,33 +0,0 @@ -import type { ProjectSessionCatalog } from "./project-management-service.js"; -import type { DesktopRuntimeHostClient } from "./runtime-host-client.js"; - -type RuntimeHostProjectClient = Pick< - DesktopRuntimeHostClient, - "listSessions" | "relocateSessionCwd" | "updateSessionMetadata" ->; - -export function createRuntimeHostProjectSessionCatalog( - client: RuntimeHostProjectClient, -): ProjectSessionCatalog { - return { - async listHeaders() { - return (await client.listSessions()).map((session) => ({ - id: session.id, - cwd: session.cwd, - projectId: session.projectId, - })); - }, - - async updateHeader(sessionId, patch) { - if (patch.cwd !== undefined) { - return client.relocateSessionCwd(sessionId, patch.cwd, patch.projectId); - } - if (patch.projectId !== undefined) { - return client.updateSessionMetadata(sessionId, { - projectId: patch.projectId, - }); - } - return undefined; - }, - }; -} diff --git a/apps/desktop/src/preload/bridge-contract.d.ts b/apps/desktop/src/preload/bridge-contract.d.ts index 7292a92f23..fa7065e3a4 100644 --- a/apps/desktop/src/preload/bridge-contract.d.ts +++ b/apps/desktop/src/preload/bridge-contract.d.ts @@ -391,6 +391,7 @@ export interface MakaBridge { }; projects: { list(): Promise; + subscribeChanges(handler: () => void): () => void; add(): Promise< { ok: true; project: ProjectRecord; path: string } | { ok: false; reason: 'cancelled' } >; diff --git a/apps/desktop/src/preload/preload.ts b/apps/desktop/src/preload/preload.ts index e842d30bf0..c4503f4430 100644 --- a/apps/desktop/src/preload/preload.ts +++ b/apps/desktop/src/preload/preload.ts @@ -427,6 +427,11 @@ const makaBridge = { list(): Promise { return ipcRenderer.invoke('projects:list'); }, + subscribeChanges(handler: () => void): () => void { + const listener = () => handler(); + ipcRenderer.on('projects:changed', listener); + return () => ipcRenderer.off('projects:changed', listener); + }, add(): Promise< { ok: true; project: ProjectRecord; path: string } | { ok: false; reason: 'cancelled' } > { diff --git a/apps/desktop/src/renderer/settings/projects-settings-page.tsx b/apps/desktop/src/renderer/settings/projects-settings-page.tsx index f825d9cc95..f034d66ca0 100644 --- a/apps/desktop/src/renderer/settings/projects-settings-page.tsx +++ b/apps/desktop/src/renderer/settings/projects-settings-page.tsx @@ -1,4 +1,4 @@ -import { useCallback, useEffect, useState } from 'react'; +import { useCallback, useEffect, useRef, useState } from 'react'; import type { AppSettings, ProjectRecord, UpdateAppSettingsResult } from '@maka/core'; import { Badge, @@ -47,19 +47,23 @@ export function ProjectsSettingsPage(props: { const [homePath, setHomePath] = useState(undefined); const [renamingId, setRenamingId] = useState(null); const [draftName, setDraftName] = useState(''); + const reloadGeneration = useRef(0); const reload = useCallback(async () => { + const generation = ++reloadGeneration.current; const next = await window.maka.projects.list(); - if (mountedRef.current) setProjects(next); + if (mountedRef.current && generation === reloadGeneration.current) setProjects(next); }, [mountedRef]); useEffect(() => { void reload(); + const unsubscribe = window.maka.projects.subscribeChanges(() => void reload()); // Paths render unabbreviated until this lands, which is why // `collapseHomePath` treats an unknown home as a no-op rather than a bug. void window.maka.app.info().then((info) => { if (mountedRef.current) setHomePath(info.homePath); }); + return unsubscribe; }, [reload, mountedRef]); // Archived projects are removed-from-Maka, not deleted; they belong to the diff --git a/apps/desktop/src/renderer/use-project-context.ts b/apps/desktop/src/renderer/use-project-context.ts index ee338325bc..718c81d32a 100644 --- a/apps/desktop/src/renderer/use-project-context.ts +++ b/apps/desktop/src/renderer/use-project-context.ts @@ -63,23 +63,30 @@ export function useAppShellProjectContext(options: { useEffect(() => { let cancelled = false; - void Promise.all([window.maka.projects.list(), window.maka.app.info()]).then( - ([next, info]) => { - if (cancelled) return; - setProjects(next); - setAppInfo({ - projectId: info.projectId, - projectPath: info.projectPath, - projectGit: info.projectGit, - }); - setSelectedProjectId(info.projectId); - }, - () => { - // Project management failures surface at the next user action. - }, - ); + let refreshGeneration = 0; + const refresh = () => { + const generation = ++refreshGeneration; + return Promise.all([window.maka.projects.list(), window.maka.app.info()]).then( + ([next, info]) => { + if (cancelled || generation !== refreshGeneration) return; + setProjects(next); + setAppInfo({ + projectId: info.projectId, + projectPath: info.projectPath, + projectGit: info.projectGit, + }); + setSelectedProjectId(info.projectId); + }, + () => { + // Project management failures surface at the next user action. + }, + ); + }; + const unsubscribe = window.maka.projects.subscribeChanges(() => void refresh()); + void refresh(); return () => { cancelled = true; + unsubscribe(); }; }, []); diff --git a/packages/runtime-host/src/__tests__/authenticated-websocket.test.ts b/packages/runtime-host/src/__tests__/authenticated-websocket.test.ts index 314e6da091..9a3949e13f 100644 --- a/packages/runtime-host/src/__tests__/authenticated-websocket.test.ts +++ b/packages/runtime-host/src/__tests__/authenticated-websocket.test.ts @@ -45,6 +45,8 @@ test('one Local IPC owner and one authenticated WebSocket Client control the sam 'session.metadata.update', 'session.create', 'client.capability.replace', + 'project.catalog.query', + 'project.catalog.mutate', ], canPublishClientCapabilities: false, canUseHostPaths: false, @@ -92,6 +94,19 @@ test('one Local IPC owner and one authenticated WebSocket Client control the sam (error: unknown) => error instanceof RuntimeHostOperationError && error.code === 'unauthorized', ); + await assert.rejects( + remote.request('project.catalog.query', { kind: 'list_start' }), + (error: unknown) => + error instanceof RuntimeHostOperationError && error.code === 'unauthorized', + ); + await assert.rejects( + remote.request('project.catalog.mutate', { + kind: 'select', + projectId: 'project-1', + }), + (error: unknown) => + error instanceof RuntimeHostOperationError && error.code === 'unauthorized', + ); const created = await local.request('session.create', { sessionId: 'shared-session', diff --git a/packages/runtime-host/src/__tests__/catalog-reader.test.ts b/packages/runtime-host/src/__tests__/catalog-reader.test.ts index 09b544d885..a4dc06e0f7 100644 --- a/packages/runtime-host/src/__tests__/catalog-reader.test.ts +++ b/packages/runtime-host/src/__tests__/catalog-reader.test.ts @@ -4,6 +4,7 @@ import type { RuntimeHostConnection } from '../client/connection.js'; import { RuntimeHostCatalogReadError, readRuntimeHostConnectionCatalog, + readRuntimeHostProjects, readRuntimeHostSessions, readRuntimeHostSkillCatalog, } from '../client/catalog-reader.js'; @@ -101,6 +102,64 @@ test('reassembles per-item relay profiles into the connection profile table', as ]); }); +test('reassembles Project aliases and locations across bounded pages without truncation', async () => { + const aliases = Array.from({ length: 300 }, (_, index) => `project-alias-${index}`); + const root = process.platform === 'win32' ? 'C:\\workspace' : '/workspace'; + const locations = Array.from({ length: 70 }, (_, index) => ({ + path: `${root}${process.platform === 'win32' ? '\\' : '/'}location-${index}`, + isWorktree: index > 0, + })); + const items = [ + { + kind: 'project' as const, + projectIndex: 0, + id: 'project-1', + name: 'Project', + aliasCount: aliases.length, + locationCount: locations.length, + archivedAt: null, + available: true, + preferredPath: locations[0]!.path, + }, + ...aliases.map((alias, itemIndex) => ({ + kind: 'alias' as const, + projectIndex: 0, + itemIndex, + alias, + })), + ...locations.map((location, itemIndex) => ({ + kind: 'location' as const, + projectIndex: 0, + itemIndex, + location, + })), + ]; + const connection = fakeConnection(async (_operation, input) => { + const offset = input.kind === 'list_start' ? 0 : Number(input.cursor); + const page = items.slice(offset, offset + 64); + const nextOffset = offset + page.length; + return { + kind: 'page', + revision: `sha256:${'a'.repeat(64)}`, + projectCount: 1, + items: page, + nextCursor: nextOffset < items.length ? String(nextOffset) : null, + }; + }); + + assert.deepEqual(await readRuntimeHostProjects(connection), [ + { + id: 'project-1', + aliases, + name: 'Project', + locations, + archivedAt: null, + available: true, + preferredPath: locations[0]!.path, + }, + ]); +}); + test('rejects a Connection catalog with a missing index', async () => { const connection = fakeConnection(async () => ({ kind: 'page', diff --git a/packages/runtime-host/src/__tests__/project-catalog-coordinator.test.ts b/packages/runtime-host/src/__tests__/project-catalog-coordinator.test.ts new file mode 100644 index 0000000000..0e70991bb3 --- /dev/null +++ b/packages/runtime-host/src/__tests__/project-catalog-coordinator.test.ts @@ -0,0 +1,128 @@ +import assert from 'node:assert/strict'; +import { mkdir, mkdtemp, realpath, rename, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import { createProjectCatalog, createSessionStore } from '@maka/storage'; +import type { ConnectionContext } from '../server/operation-dispatcher.js'; +import { HostProjectCatalogChangeService } from '../server/project-catalog-change-service.js'; +import { HostProjectCatalogCoordinator } from '../server/project-catalog-coordinator.js'; +import { HostProjectMembershipGate } from '../server/project-membership-gate.js'; +import { HostSessionCatalogChangeService } from '../server/session-catalog-change-service.js'; + +test('Host Project Catalog relink merges identities and reassigns every affected Session', async () => { + const base = await mkdtemp(join(tmpdir(), 'maka-host-project-catalog-')); + const storageRoot = join(base, 'storage'); + const oldPath = join(base, 'old-location'); + const newPath = join(base, 'new-location'); + await mkdir(oldPath); + const catalog = createProjectCatalog(storageRoot, { + now: () => 1_000, + createId: (() => { + let id = 0; + return () => `project-${++id}`; + })(), + }); + const sessions = createSessionStore(storageRoot); + const projectChanges = new HostProjectCatalogChangeService(); + const sessionChanges = new HostSessionCatalogChangeService(); + const projectFrames: unknown[] = []; + const sessionFrames: unknown[] = []; + projectChanges.attachConnection('desktop', { + send: async (frame) => { + projectFrames.push(frame); + }, + }); + projectChanges.attachConnection('tui', { + send: async (frame) => { + projectFrames.push(frame); + }, + }); + sessionChanges.attachConnection('desktop', { + send: async (frame) => { + sessionFrames.push(frame); + }, + }); + const coordinator = new HostProjectCatalogCoordinator( + catalog, + projectChanges, + sessionChanges, + new HostProjectMembershipGate(), + () => assert.fail('ordinary project mutations must not drain the Host'), + ); + + try { + const original = await catalog.register(oldPath); + await rename(oldPath, newPath); + const destinationPath = await realpath(newPath); + const duplicate = await catalog.register(newPath); + const oldSession = await sessions.create(sessionInput(oldPath, original.id)); + const newSession = await sessions.create(sessionInput(destinationPath, duplicate.id)); + + const relinked = await coordinator.handlers['project.catalog.mutate']( + { kind: 'relink', projectId: original.id, path: newPath }, + connection(), + ); + assert.equal(relinked.ok, true); + if (!relinked.ok || relinked.result.kind !== 'project') return; + assert.equal(relinked.result.projectId, original.id); + const [project] = await catalog.list(); + assert.deepEqual(project?.aliases, [duplicate.id]); + assert.equal(project?.preferredPath, destinationPath); + assert.deepEqual( + (await catalog.list()).map(({ id }) => id), + [original.id], + ); + + for (const sessionId of [oldSession.id, newSession.id]) { + const header = await sessions.readHeaderSnapshot(sessionId); + assert.equal(header.projectId, original.id); + assert.equal(header.cwd, destinationPath); + } + assert.deepEqual(projectFrames, [ + { kind: 'project.catalog.changed', revision: 1 }, + { kind: 'project.catalog.changed', revision: 1 }, + ]); + assert.deepEqual( + sessionFrames.map((frame) => (frame as { revision: number }).revision), + [1, 2], + ); + assert.deepEqual( + sessionFrames.map((frame) => (frame as { sessionId: string }).sessionId).sort(), + [oldSession.id, newSession.id].sort(), + ); + + const listed = await coordinator.handlers['project.catalog.query']( + { kind: 'list_start' }, + connection(), + ); + assert.equal(listed.ok, true); + assert.equal(listed.ok && listed.result.kind, 'page'); + assert.equal(listed.ok && listed.result.kind === 'page' && listed.result.projectCount, 1); + } finally { + catalog.close(); + await sessions.close?.(); + await rm(base, { recursive: true, force: true }); + } +}); + +function sessionInput(cwd: string, projectId: string) { + return { + cwd, + projectId, + backend: 'fake' as const, + llmConnectionSlug: 'fake', + model: 'fake-model', + permissionMode: 'ask' as const, + }; +} + +function connection(): ConnectionContext { + return { + hostEpoch: 'host-1', + connectionId: 'desktop', + surface: 'desktop', + principal: 'local_os_user', + acquireResidency: () => ({ release: () => {} }), + }; +} diff --git a/packages/runtime-host/src/__tests__/project-catalog-protocol.test.ts b/packages/runtime-host/src/__tests__/project-catalog-protocol.test.ts new file mode 100644 index 0000000000..7b6dc2b3a3 --- /dev/null +++ b/packages/runtime-host/src/__tests__/project-catalog-protocol.test.ts @@ -0,0 +1,138 @@ +import assert from 'node:assert/strict'; +import { describe, test } from 'node:test'; +import { + decodeClientFrame, + decodeHostFrame, + decodeProjectCatalogQueryResult, + HOST_OPERATION_SPECS, + PROJECT_CATALOG_PAGE_MAX_ITEMS, + RuntimeHostProtocolError, +} from '../protocol/index.js'; + +const projectPath = process.platform === 'win32' ? 'C:\\workspace' : '/workspace'; +const revision = `sha256:${'a'.repeat(64)}` as const; + +describe('Project catalog protocol', () => { + test('declares bounded ready operations and exact invalidations', () => { + assert.deepEqual( + Object.fromEntries( + (['project.catalog.query', 'project.catalog.mutate'] as const).map((operation) => [ + operation, + { + mode: HOST_OPERATION_SPECS[operation].mode, + availability: HOST_OPERATION_SPECS[operation].availability, + }, + ]), + ), + { + 'project.catalog.query': { mode: 'query', availability: 'ready' }, + 'project.catalog.mutate': { mode: 'command', availability: 'ready' }, + }, + ); + const frame = { kind: 'project.catalog.changed' as const, revision: 1 }; + assert.deepEqual(decodeHostFrame(frame), frame); + assert.throws(() => decodeHostFrame({ ...frame, extra: true }), isProtocolError); + }); + + test('round-trips every closed mutation shape and correlates its result', () => { + for (const input of [ + { kind: 'register', path: projectPath }, + { kind: 'select', projectId: 'project-1' }, + { kind: 'touch', projectId: 'project-1', path: null }, + { kind: 'touch', projectId: 'project-1', path: projectPath }, + { kind: 'relink', projectId: 'project-1', path: projectPath }, + { kind: 'rename', projectId: 'project-1', name: 'Project' }, + { kind: 'archive', projectId: 'project-1' }, + { kind: 'restore', projectId: 'project-1' }, + ] as const) { + const request = { + requestId: `request-${input.kind}`, + operation: 'project.catalog.mutate' as const, + input, + }; + assert.deepEqual(decodeClientFrame(request), request); + } + assert.deepEqual( + decodeHostFrame({ + requestId: 'request-select', + operation: 'project.catalog.mutate', + ok: true, + result: { kind: 'selection', projectId: 'project-1', path: projectPath }, + }), + { + requestId: 'request-select', + operation: 'project.catalog.mutate', + ok: true, + result: { kind: 'selection', projectId: 'project-1', path: projectPath }, + }, + ); + assert.throws( + () => + HOST_OPERATION_SPECS['project.catalog.mutate'].assertOutputForInput?.( + { kind: 'select', projectId: 'project-1' }, + { kind: 'project', projectId: 'project-1' }, + ), + isProtocolError, + ); + }); + + test('rejects relative paths, open records, oversized pages, and stale shapes', () => { + assert.throws( + () => + decodeClientFrame({ + requestId: 'request-1', + operation: 'project.catalog.mutate', + input: { kind: 'register', path: 'relative/project' }, + }), + isProtocolError, + ); + assert.throws( + () => + decodeClientFrame({ + requestId: 'request-2', + operation: 'project.catalog.query', + input: { kind: 'list_start', includeArchived: true }, + }), + isProtocolError, + ); + assert.throws( + () => + decodeProjectCatalogQueryResult({ + kind: 'page', + revision, + projectCount: 1, + items: Array.from({ length: PROJECT_CATALOG_PAGE_MAX_ITEMS + 1 }, projectHeaderItem), + nextCursor: null, + }), + isProtocolError, + ); + assert.throws( + () => + decodeProjectCatalogQueryResult({ + kind: 'revision_changed', + expected: revision, + actual: revision, + items: [], + }), + isProtocolError, + ); + }); +}); + +function projectHeaderItem() { + return { + kind: 'project', + projectIndex: 0, + id: 'project-1', + name: 'Project', + aliasCount: 0, + locationCount: 1, + archivedAt: null, + available: true, + preferredPath: projectPath, + } as const; +} + +function isProtocolError(error: unknown): boolean { + return error instanceof RuntimeHostProtocolError; +} diff --git a/packages/runtime-host/src/__tests__/project-catalog-two-client-uds.test.ts b/packages/runtime-host/src/__tests__/project-catalog-two-client-uds.test.ts new file mode 100644 index 0000000000..f94eee62f7 --- /dev/null +++ b/packages/runtime-host/src/__tests__/project-catalog-two-client-uds.test.ts @@ -0,0 +1,178 @@ +import assert from 'node:assert/strict'; +import { mkdir, mkdtemp, realpath, rename, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import { openInteractiveProjectCatalogForWrite } from '@maka/storage'; +import { openInteractiveExecutionStoresForWrite } from '@maka/storage/execution-stores'; +import { + resolveRootControlNamespace, + resolveStorageRoot, + tryAcquireInteractiveRootOwner, +} from '@maka/storage/root-authority'; +import { + connectRuntimeHost, + readRuntimeHostProjects, + readRuntimeHostSessions, + type RuntimeHostConnection, +} from '../client/index.js'; +import { RUNTIME_HOST_PROTOCOL_VERSION, type ClientSurface } from '../protocol/index.js'; +import { createExecutionRuntimeHostComposition } from '../server/execution-composition.js'; +import { RuntimeHostKernel } from '../server/index.js'; + +const PROTOCOL = { + min: RUNTIME_HOST_PROTOCOL_VERSION, + max: RUNTIME_HOST_PROTOCOL_VERSION, +} as const; +const REQUEST_TIMEOUT_MS = 5_000; + +test('two UDS clients converge on one Host-owned Project Catalog', { + skip: process.platform === 'win32', + timeout: 20_000, +}, async () => { + const base = await mkdtemp(join(tmpdir(), 'maka-project-catalog-two-client-')); + const dataRoot = join(base, 'data-root'); + const oldPath = join(base, 'old-location'); + const newPath = join(base, 'new-location'); + await mkdir(oldPath); + const capability = await resolveStorageRoot({ path: dataRoot, kind: 'interactive' }); + const seeded = await seedCatalog(capability, oldPath, newPath); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + if (!owner) return; + let host: RuntimeHostKernel | undefined; + const connections: RuntimeHostConnection[] = []; + try { + host = await RuntimeHostKernel.start({ + owner, + idleGraceMs: 30_000, + compositionFactory: createExecutionRuntimeHostComposition, + }); + const [desktop, tui] = await Promise.all([ + connectClient(dataRoot, 'desktop'), + connectClient(dataRoot, 'tui'), + ]); + connections.push(desktop, tui); + assert.deepEqual(await readRuntimeHostProjects(desktop), await readRuntimeHostProjects(tui)); + + const desktopChanged = nextProjectChange(desktop); + const tuiChanged = nextProjectChange(tui); + const relinked = await desktop.request( + 'project.catalog.mutate', + { kind: 'relink', projectId: seeded.originalProjectId, path: newPath }, + REQUEST_TIMEOUT_MS, + ); + assert.equal(relinked.kind, 'project'); + assert.deepEqual(await Promise.all([desktopChanged, tuiChanged]), [1, 1]); + + const [desktopProjects, tuiProjects] = await Promise.all([ + readRuntimeHostProjects(desktop), + readRuntimeHostProjects(tui), + ]); + assert.deepEqual(tuiProjects, desktopProjects); + assert.equal(desktopProjects.length, 1); + assert.equal(desktopProjects[0]?.id, seeded.originalProjectId); + assert.deepEqual(desktopProjects[0]?.aliases, [seeded.duplicateProjectId]); + + const sessions = await readRuntimeHostSessions(tui); + for (const sessionId of seeded.sessionIds) { + const session = sessions.find(({ id }) => id === sessionId); + assert.ok(session); + assert.equal(session && 'kind' in session, false); + if (!session || 'kind' in session) continue; + assert.equal(session.projectId, seeded.originalProjectId); + assert.equal(session.cwd, seeded.destinationPath); + } + } finally { + const cleanupErrors: unknown[] = []; + for (const connection of connections) { + await connection.close().catch((error: unknown) => cleanupErrors.push(error)); + } + await host?.close().catch((error: unknown) => cleanupErrors.push(error)); + if (!host && !owner.closed) { + await owner.close().catch((error: unknown) => cleanupErrors.push(error)); + } + await rm(join(resolveRootControlNamespace(), capability.rootId), { + recursive: true, + force: true, + }).catch((error: unknown) => cleanupErrors.push(error)); + await rm(base, { recursive: true, force: true }).catch((error: unknown) => + cleanupErrors.push(error), + ); + if (cleanupErrors.length > 0) { + throw new AggregateError(cleanupErrors, 'Project Catalog UDS test cleanup failed'); + } + } +}); + +async function seedCatalog( + capability: Awaited>>, + oldPath: string, + newPath: string, +): Promise<{ + originalProjectId: string; + duplicateProjectId: string; + destinationPath: string; + sessionIds: readonly [string, string]; +}> { + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + if (!owner) throw new Error('Unable to acquire Project Catalog seed owner'); + const catalog = await openInteractiveProjectCatalogForWrite(owner.lease); + const stores = await openInteractiveExecutionStoresForWrite(owner.lease); + try { + const original = await catalog.register(oldPath); + await rename(oldPath, newPath); + const destinationPath = await realpath(newPath); + const duplicate = await catalog.register(newPath); + const first = await stores.sessionStore.create(sessionInput(oldPath, original.id)); + const second = await stores.sessionStore.create(sessionInput(destinationPath, duplicate.id)); + return { + originalProjectId: original.id, + duplicateProjectId: duplicate.id, + destinationPath, + sessionIds: [first.id, second.id], + }; + } finally { + catalog.close(); + await stores.sessionStore.close?.(); + await owner.close(); + } +} + +function sessionInput(cwd: string, projectId: string) { + return { + cwd, + projectId, + backend: 'fake' as const, + llmConnectionSlug: 'fake', + model: 'fake-model', + permissionMode: 'ask' as const, + }; +} + +async function connectClient( + rootPath: string, + surface: ClientSurface, +): Promise { + const result = await connectRuntimeHost({ + rootPath, + surface, + protocol: PROTOCOL, + connectTimeoutMs: REQUEST_TIMEOUT_MS, + handshakeTimeoutMs: REQUEST_TIMEOUT_MS, + }); + if (result.kind !== 'connected') { + throw new Error(`Runtime Host Client did not connect: ${result.kind}`); + } + return result.connection; +} + +function nextProjectChange(connection: RuntimeHostConnection): Promise { + return new Promise((resolve) => { + const unsubscribe = connection.subscribeProjectCatalogChanges((revision) => { + unsubscribe(); + resolve(revision); + }); + }); +} diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index 6a424f52b6..fe12b04776 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -47,7 +47,7 @@ describe('Runtime Host bootstrap protocol', () => { test('keeps the experimental protocol at v0 with the declared authority operations', () => { assert.equal(RUNTIME_HOST_PROTOCOL_VERSION, 0); - assert.equal(RUNTIME_HOST_COMPATIBILITY_EPOCH, 11); + assert.equal(RUNTIME_HOST_COMPATIBILITY_EPOCH, 12); assert.deepEqual(Object.keys(HOST_OPERATION_SPECS).sort(), [ 'access.credential.issue', 'access.credential.revoke', @@ -104,6 +104,8 @@ describe('Runtime Host bootstrap protocol', () => { 'plan.turn.start', 'pricing.mutate', 'pricing.query', + 'project.catalog.mutate', + 'project.catalog.query', 'queue.retract', 'runtime.policy.mutate', 'runtime.policy.query', diff --git a/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts index 6fcc96ef08..70a91bea8a 100644 --- a/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts @@ -21,6 +21,7 @@ import { type SessionConfigurationUpdateInput, } from '../protocol/index.js'; import type { ConnectionContext } from '../server/operation-dispatcher.js'; +import { HostProjectMembershipGate } from '../server/project-membership-gate.js'; import { HostSessionCatalogCoordinator, type HostSessionCatalogCoordinatorOptions, @@ -458,6 +459,62 @@ test('creation fingerprints and persists the canonical cwd behind a symlink', as } }); +test('creation resolves a stale Client project path from current Host membership', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-session-create-project-')); + const stalePath = join(root, 'stale'); + const currentPath = join(root, 'current'); + await mkdir(currentPath); + try { + let created: Parameters[0] | undefined; + const fixture = createFixture({ + projectCatalog: { + list: async () => [ + { + id: 'project-current', + aliases: ['project-stale'], + name: 'Project', + locations: [{ path: currentPath, isWorktree: false }], + available: true, + preferredPath: currentPath, + }, + ], + } as never, + stores: { + createStableSession: async (request) => { + created = request; + return { + kind: 'existing', + record: headerSnapshot( + { + ...sessionHeader(request.sessionId, []), + cwd: request.input.cwd, + projectId: request.input.projectId, + }, + 1, + ), + }; + }, + }, + }); + + const outcome = await fixture.coordinator.handlers['session.create']( + { + sessionId: fixture.sessionId, + cwd: stalePath, + projectId: 'project-stale', + modelTarget: { kind: 'default' }, + }, + context, + ); + + assert.equal(outcome.ok, true); + assert.equal(created?.input.cwd, currentPath); + assert.equal(created?.input.projectId, 'project-current'); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + test('cwd relocation canonicalizes once and commits through Runtime authority', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-session-relocate-cwd-')); const target = join(root, 'target'); @@ -605,6 +662,7 @@ function createFixture( readonly manager?: Partial; readonly continuity?: Partial; readonly connection?: FixtureConnection; + readonly projectCatalog?: HostSessionCatalogCoordinatorOptions['projectCatalog']; } = {}, ) { const sessionId = 'session-1'; @@ -670,6 +728,8 @@ function createFixture( manager, admission: new SessionAdmissionGate(), continuity, + projectCatalog: options.projectCatalog ?? ({ list: async () => [] } as never), + projectMembership: new HostProjectMembershipGate(), requestDrain: () => { drains += 1; }, diff --git a/packages/runtime-host/src/client/catalog-reader.ts b/packages/runtime-host/src/client/catalog-reader.ts index eca125c61f..9477f0c4f5 100644 --- a/packages/runtime-host/src/client/catalog-reader.ts +++ b/packages/runtime-host/src/client/catalog-reader.ts @@ -1,18 +1,22 @@ -import type { - ConnectionCatalogCursor, - ConnectionCatalogPageItem, - ConnectionCatalogQueryResult, - RelayModelProfile, - RelayModelProfiles, - SessionCatalogFilter, - SessionCatalogItem, - SkillCatalogLocalContext, - SkillCatalogInvocableItem, - SkillCatalogInvocableTarget, - SkillCatalogPageItem, - SkillCatalogRevision, - SkillCatalogView, - OperationOutput, +import { + decodeProjectCatalogProject, + type ConnectionCatalogCursor, + type ConnectionCatalogPageItem, + type ConnectionCatalogQueryResult, + type RelayModelProfile, + type RelayModelProfiles, + type SessionCatalogFilter, + type SessionCatalogItem, + type SkillCatalogLocalContext, + type SkillCatalogInvocableItem, + type SkillCatalogInvocableTarget, + type SkillCatalogPageItem, + type SkillCatalogRevision, + type SkillCatalogView, + type OperationOutput, + type ProjectCatalogPageItem, + type ProjectCatalogProject, + type ProjectCatalogQueryResult, } from '../protocol/index.js'; import type { RuntimeHostConnection } from './connection.js'; @@ -44,7 +48,7 @@ export interface RuntimeHostConnectionCatalogSnapshot { export class RuntimeHostCatalogReadError extends Error { constructor( - readonly catalog: 'connection' | 'session' | 'skill' | 'runtime_resource', + readonly catalog: 'connection' | 'project' | 'session' | 'skill' | 'runtime_resource', readonly reason: 'unstable' | 'invalid_projection' | 'repeated_cursor', ) { super(`Runtime Host ${catalog} catalog read failed: ${reason}`); @@ -157,6 +161,30 @@ export async function readRuntimeHostSessions( return pages.flatMap((page) => page.sessions); } +export async function readRuntimeHostProjects( + connection: RuntimeHostCatalogConnection, +): Promise { + const { first, pages } = await collectStablePages( + 'project', + async () => { + const result = await connection.request('project.catalog.query', { kind: 'list_start' }); + return result.kind === 'page' ? result : null; + }, + async (revision, cursor) => { + const result = await connection.request('project.catalog.query', { + kind: 'list_continue', + revision, + cursor, + }); + return result.kind === 'page' ? result : null; + }, + ); + return assembleProjectCatalog( + first, + pages.flatMap((page) => page.items), + ); +} + export async function readRuntimeHostResources( connection: RuntimeHostCatalogConnection, sessionId: string, @@ -239,6 +267,59 @@ function uniqueCursor( return cursor; } +function assembleProjectCatalog( + first: Extract, + items: readonly ProjectCatalogPageItem[], +): ProjectCatalogProject[] { + const projects = new Map< + number, + { + header: Extract; + aliases: Map; + locations: Map; + } + >(); + for (const item of items) { + if (item.kind !== 'project') continue; + if (projects.has(item.projectIndex)) { + throw new RuntimeHostCatalogReadError('project', 'invalid_projection'); + } + projects.set(item.projectIndex, { header: item, aliases: new Map(), locations: new Map() }); + } + for (const item of items) { + if (item.kind === 'project') continue; + const project = projects.get(item.projectIndex); + if (!project) throw new RuntimeHostCatalogReadError('project', 'invalid_projection'); + const values = item.kind === 'alias' ? project.aliases : project.locations; + const expectedCount = + item.kind === 'alias' ? project.header.aliasCount : project.header.locationCount; + if (item.itemIndex >= expectedCount || values.has(item.itemIndex)) { + throw new RuntimeHostCatalogReadError('project', 'invalid_projection'); + } + if (item.kind === 'alias') project.aliases.set(item.itemIndex, item.alias); + else project.locations.set(item.itemIndex, item.location); + } + if (projects.size !== first.projectCount) { + throw new RuntimeHostCatalogReadError('project', 'invalid_projection'); + } + return [...projects.entries()] + .sort(([left], [right]) => left - right) + .map(([, { header, aliases, locations }]) => { + if (aliases.size !== header.aliasCount || locations.size !== header.locationCount) { + throw new RuntimeHostCatalogReadError('project', 'invalid_projection'); + } + return decodeProjectCatalogProject({ + id: header.id, + aliases: orderedValues(aliases), + name: header.name, + locations: orderedValues(locations), + archivedAt: header.archivedAt, + available: header.available, + preferredPath: header.preferredPath, + }); + }); +} + function assembleConnectionCatalog( first: Extract, items: readonly ConnectionCatalogPageItem[], diff --git a/packages/runtime-host/src/client/connection.ts b/packages/runtime-host/src/client/connection.ts index 0dfb40f00e..224bab6cae 100644 --- a/packages/runtime-host/src/client/connection.ts +++ b/packages/runtime-host/src/client/connection.ts @@ -48,6 +48,7 @@ import { type PlanQueryResult, type PlanTurnStartInput, type PlanTurnStartResult, + type ProjectCatalogChangedFrame, type ProtocolRange, type RequestFrame, type ResponseFrame, @@ -241,6 +242,7 @@ export interface RuntimeHostConnection { ): Promise; unregisterClientCapabilities(timeoutMs?: number): Promise; subscribeConfigurationChanges(listener: (revision: number) => void): () => void; + subscribeProjectCatalogChanges(listener: (revision: number) => void): () => void; subscribeSessionCatalogChanges(listener: (frame: SessionCatalogChangedFrame) => void): () => void; } @@ -298,6 +300,7 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { readonly #retiredSubscriptionIds = new Set(); readonly #clientCapabilities: ClientCapabilityChannel; readonly #configurationChangeListeners = new Set<(revision: number) => void>(); + readonly #projectCatalogChangeListeners = new Set<(revision: number) => void>(); readonly #sessionCatalogChangeListeners = new Set<(frame: SessionCatalogChangedFrame) => void>(); #livenessTimer: NodeJS.Timeout | undefined; #livenessProbePending = false; @@ -607,6 +610,11 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { return () => this.#configurationChangeListeners.delete(listener); } + subscribeProjectCatalogChanges(listener: (revision: number) => void): () => void { + this.#projectCatalogChangeListeners.add(listener); + return () => this.#projectCatalogChangeListeners.delete(listener); + } + subscribeSessionCatalogChanges( listener: (frame: SessionCatalogChangedFrame) => void, ): () => void { @@ -628,6 +636,9 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { case 'configuration.changed': this.#acceptConfigurationChanged(frame); continue; + case 'project.catalog.changed': + this.#acceptProjectCatalogChanged(frame); + continue; case 'session.catalog.changed': this.#acceptSessionCatalogChanged(frame); continue; @@ -699,6 +710,16 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { } } + #acceptProjectCatalogChanged(frame: ProjectCatalogChangedFrame): void { + for (const listener of this.#projectCatalogChangeListeners) { + try { + listener(frame.revision); + } catch { + // A presentation listener cannot invalidate the Host connection. + } + } + } + #acceptSessionCatalogChanged(frame: SessionCatalogChangedFrame): void { for (const listener of this.#sessionCatalogChangeListeners) { try { diff --git a/packages/runtime-host/src/client/index.ts b/packages/runtime-host/src/client/index.ts index b2272d42f7..e4dcfbe0e7 100644 --- a/packages/runtime-host/src/client/index.ts +++ b/packages/runtime-host/src/client/index.ts @@ -21,6 +21,7 @@ export { readRuntimeHostConnectionCatalog, readRuntimeHostInvocableSkills, readRuntimeHostResources, + readRuntimeHostProjects, readRuntimeHostSessions, readRuntimeHostSkillCatalog, type RuntimeHostConnectionCatalogEntry, diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 837822141a..71065a9eae 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -22,6 +22,10 @@ import { decodeSessionCatalogChangedFrame, type SessionCatalogChangedFrame, } from './session-catalog-change.js'; +import { + decodeProjectCatalogChangedFrame, + type ProjectCatalogChangedFrame, +} from './project-catalog-change.js'; import { decodeRequestFrame, decodeResponseFrame, @@ -39,6 +43,8 @@ export * from './client-capability.js'; export * from './configuration-change.js'; export * from './goal.js'; export * from './plan.js'; +export * from './project-catalog.js'; +export * from './project-catalog-change.js'; export * from './execution-inspect.js'; export * from './external-session.js'; export * from './message.js'; @@ -55,8 +61,8 @@ export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // The wire version remains v0 before the first release. This independent epoch // lets a new Client retire a stale same-version Host whose closed schema is no // longer safe to use. -// 11: authenticated root identity and admission authorization changed the closed schema. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 11 as const; +// 12: Host-owned Project Catalog operations and invalidation changed the closed schema. +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 12 as const; // A legal sandbox-boundary expansion can consume 64 KiB before its Interaction // envelope and independently bounded justification are added. Keep transport // capacity large enough to represent that domain value; narrower surfaces such @@ -123,6 +129,7 @@ export type HostFrame = | SubscriptionFrame | ClientCapabilityHostFrame | ConfigurationChangedFrame + | ProjectCatalogChangedFrame | SessionCatalogChangedFrame; export interface HostRegistration { @@ -221,6 +228,7 @@ export function decodeHostFrame(value: unknown): HostFrame { return decodeClientCapabilityHostFrame(frame); } if (frame.kind === 'configuration.changed') return decodeConfigurationChangedFrame(frame); + if (frame.kind === 'project.catalog.changed') return decodeProjectCatalogChangedFrame(frame); if (frame.kind === 'session.catalog.changed') return decodeSessionCatalogChangedFrame(frame); return decodeResponseFrame(frame); } diff --git a/packages/runtime-host/src/protocol/operations.ts b/packages/runtime-host/src/protocol/operations.ts index 65f85600a3..a9fd67aeaa 100644 --- a/packages/runtime-host/src/protocol/operations.ts +++ b/packages/runtime-host/src/protocol/operations.ts @@ -20,6 +20,7 @@ import { MEMORY_OPERATION_SPECS } from './memory.js'; import { NETWORK_PROXY_OPERATION_SPECS } from './network-proxy.js'; import { OAUTH_OPERATION_SPECS } from './oauth.js'; import { PLAN_OPERATION_SPECS } from './plan.js'; +import { PROJECT_CATALOG_OPERATION_SPECS } from './project-catalog.js'; import { composeOperationSpecMaps, type HostOperationError, @@ -131,6 +132,7 @@ export * from './memory.js'; export * from './network-proxy.js'; export * from './oauth.js'; export * from './plan.js'; +export * from './project-catalog.js'; export * from './runtime-policy.js'; export * from './runtime-resource.js'; export * from './session-catalog.js'; @@ -158,6 +160,7 @@ export const HOST_OPERATION_SPECS = composeOperationSpecMaps( RUNTIME_RESOURCE_OPERATION_SPECS, AUTOMATION_OPERATION_SPECS, PLAN_OPERATION_SPECS, + PROJECT_CATALOG_OPERATION_SPECS, MESSAGE_OPERATION_SPECS, TASK_LEDGER_OPERATION_SPECS, INTERACTION_OPERATION_SPECS, diff --git a/packages/runtime-host/src/protocol/project-catalog-change.ts b/packages/runtime-host/src/protocol/project-catalog-change.ts new file mode 100644 index 0000000000..e81cc58762 --- /dev/null +++ b/packages/runtime-host/src/protocol/project-catalog-change.ts @@ -0,0 +1,14 @@ +import { requireCount, requireExactRecord } from './codec.js'; + +export interface ProjectCatalogChangedFrame { + readonly kind: 'project.catalog.changed'; + readonly revision: number; +} + +export function decodeProjectCatalogChangedFrame(value: unknown): ProjectCatalogChangedFrame { + const frame = requireExactRecord(value, 'project catalog changed frame', ['kind', 'revision']); + return { + kind: 'project.catalog.changed', + revision: requireCount(frame.revision, 'project catalog change revision'), + }; +} diff --git a/packages/runtime-host/src/protocol/project-catalog.ts b/packages/runtime-host/src/protocol/project-catalog.ts new file mode 100644 index 0000000000..6deb636b3b --- /dev/null +++ b/packages/runtime-host/src/protocol/project-catalog.ts @@ -0,0 +1,422 @@ +import { + requireCount, + requireEncodedByteLimit, + requireEntityId, + requireExactRecord, + requireRecord, + requireUtf8String, +} from './codec.js'; +import { invalidProtocolFrame } from './errors.js'; +import { defineOperation } from './operation-spec.js'; + +export const PROJECT_CATALOG_PAGE_MAX_ITEMS = 64; +export const PROJECT_CATALOG_PAGE_MAX_BYTES = 48 * 1024; +export const PROJECT_CATALOG_CURSOR_MAX_BYTES = 128; +export const PROJECT_CATALOG_NAME_MAX_BYTES = 16 * 1024; +export const PROJECT_CATALOG_PATH_MAX_BYTES = 4 * 1024; + +const QUERY_ERRORS = [ + 'host_not_ready', + 'host_draining', + 'operation_unavailable', + 'invalid_request', + 'persistence_failed', + 'internal_failure', +] as const; + +const MUTATE_ERRORS = [ + 'host_not_ready', + 'host_draining', + 'operation_unavailable', + 'invalid_request', + 'not_found', + 'operation_conflict', + 'persistence_failed', + 'commit_outcome_unknown', + 'internal_failure', +] as const; + +export type ProjectCatalogRevision = `sha256:${string}`; + +export interface ProjectCatalogLocation { + readonly path: string; + readonly isWorktree: boolean; +} + +export interface ProjectCatalogProject { + readonly id: string; + readonly aliases: readonly string[]; + readonly name: string; + readonly locations: readonly ProjectCatalogLocation[]; + readonly archivedAt: number | null; + readonly available: boolean; + readonly preferredPath: string | null; +} + +export type ProjectCatalogPageItem = + | { + readonly kind: 'project'; + readonly projectIndex: number; + readonly id: string; + readonly name: string; + readonly aliasCount: number; + readonly locationCount: number; + readonly archivedAt: number | null; + readonly available: boolean; + readonly preferredPath: string | null; + } + | { + readonly kind: 'alias'; + readonly projectIndex: number; + readonly itemIndex: number; + readonly alias: string; + } + | { + readonly kind: 'location'; + readonly projectIndex: number; + readonly itemIndex: number; + readonly location: ProjectCatalogLocation; + }; + +export type ProjectCatalogQueryInput = + | { readonly kind: 'list_start' } + | { + readonly kind: 'list_continue'; + readonly revision: ProjectCatalogRevision; + readonly cursor: string; + }; + +export type ProjectCatalogQueryResult = + | { + readonly kind: 'page'; + readonly revision: ProjectCatalogRevision; + readonly projectCount: number; + readonly items: readonly ProjectCatalogPageItem[]; + readonly nextCursor: string | null; + } + | { + readonly kind: 'revision_changed'; + readonly expected: ProjectCatalogRevision; + readonly actual: ProjectCatalogRevision; + }; + +export type ProjectCatalogMutateInput = + | { readonly kind: 'register'; readonly path: string } + | { readonly kind: 'select'; readonly projectId: string } + | { readonly kind: 'touch'; readonly projectId: string; readonly path: string | null } + | { readonly kind: 'relink'; readonly projectId: string; readonly path: string } + | { readonly kind: 'rename'; readonly projectId: string; readonly name: string } + | { readonly kind: 'archive'; readonly projectId: string } + | { readonly kind: 'restore'; readonly projectId: string }; + +export type ProjectCatalogMutateResult = + | { readonly kind: 'project'; readonly projectId: string } + | { + readonly kind: 'selection'; + readonly projectId: string; + readonly path: string; + }; + +export const PROJECT_CATALOG_OPERATION_SPECS = { + 'project.catalog.query': defineOperation< + ProjectCatalogQueryInput, + ProjectCatalogQueryResult, + (typeof QUERY_ERRORS)[number] + >({ + mode: 'query', + availability: 'ready', + errors: QUERY_ERRORS, + decodeInput: decodeProjectCatalogQueryInput, + decodeOutput: decodeProjectCatalogQueryResult, + }), + 'project.catalog.mutate': defineOperation< + ProjectCatalogMutateInput, + ProjectCatalogMutateResult, + (typeof MUTATE_ERRORS)[number] + >({ + mode: 'command', + availability: 'ready', + errors: MUTATE_ERRORS, + decodeInput: decodeProjectCatalogMutateInput, + decodeOutput: decodeProjectCatalogMutateResult, + assertOutputForInput: (input, output) => { + if ((input.kind === 'select') !== (output.kind === 'selection')) { + throw invalidProtocolFrame('Project catalog mutation result kind does not match input'); + } + }, + }), +} as const; + +export function decodeProjectCatalogQueryInput(value: unknown): ProjectCatalogQueryInput { + const record = requireRecord(value, 'project catalog query input'); + if (record.kind === 'list_start') { + requireExactRecord(record, 'project catalog list start input', ['kind']); + return { kind: 'list_start' }; + } + if (record.kind === 'list_continue') { + const input = requireExactRecord(record, 'project catalog list continuation input', [ + 'kind', + 'revision', + 'cursor', + ]); + return { + kind: 'list_continue', + revision: revision(input.revision, 'project catalog revision'), + cursor: requireUtf8String( + input.cursor, + 'project catalog cursor', + PROJECT_CATALOG_CURSOR_MAX_BYTES, + ), + }; + } + throw invalidProtocolFrame('Invalid project catalog query kind'); +} + +export function decodeProjectCatalogQueryResult(value: unknown): ProjectCatalogQueryResult { + const record = requireRecord(value, 'project catalog query result'); + if (record.kind === 'revision_changed') { + const result = requireExactRecord(record, 'project catalog revision changed result', [ + 'kind', + 'expected', + 'actual', + ]); + return { + kind: 'revision_changed', + expected: revision(result.expected, 'expected project catalog revision'), + actual: revision(result.actual, 'actual project catalog revision'), + }; + } + if (record.kind !== 'page') throw invalidProtocolFrame('Invalid project catalog query result'); + const page = requireExactRecord(record, 'project catalog page result', [ + 'kind', + 'revision', + 'projectCount', + 'items', + 'nextCursor', + ]); + if (!Array.isArray(page.items) || page.items.length > PROJECT_CATALOG_PAGE_MAX_ITEMS) { + throw invalidProtocolFrame('Invalid project catalog page items'); + } + const decoded: ProjectCatalogQueryResult = { + kind: 'page', + revision: revision(page.revision, 'project catalog revision'), + projectCount: requireCount(page.projectCount, 'project catalog projectCount'), + items: page.items.map(decodeProjectCatalogPageItem), + nextCursor: + page.nextCursor === null + ? null + : requireUtf8String( + page.nextCursor, + 'project catalog next cursor', + PROJECT_CATALOG_CURSOR_MAX_BYTES, + ), + }; + requireEncodedByteLimit(decoded, 'project catalog page result', PROJECT_CATALOG_PAGE_MAX_BYTES); + return decoded; +} + +export function decodeProjectCatalogMutateInput(value: unknown): ProjectCatalogMutateInput { + const record = requireRecord(value, 'project catalog mutation input'); + switch (record.kind) { + case 'register': { + const input = requireExactRecord(record, 'project register input', ['kind', 'path']); + return { kind: 'register', path: absolutePath(input.path, 'project path') }; + } + case 'select': + case 'archive': + case 'restore': { + const input = requireExactRecord(record, `project ${record.kind} input`, [ + 'kind', + 'projectId', + ]); + return { kind: record.kind, projectId: projectId(input.projectId) }; + } + case 'touch': { + const input = requireExactRecord(record, 'project touch input', [ + 'kind', + 'projectId', + 'path', + ]); + return { + kind: 'touch', + projectId: projectId(input.projectId), + path: input.path === null ? null : absolutePath(input.path, 'project path'), + }; + } + case 'relink': { + const input = requireExactRecord(record, 'project relink input', [ + 'kind', + 'projectId', + 'path', + ]); + return { + kind: 'relink', + projectId: projectId(input.projectId), + path: absolutePath(input.path, 'project path'), + }; + } + case 'rename': { + const input = requireExactRecord(record, 'project rename input', [ + 'kind', + 'projectId', + 'name', + ]); + return { + kind: 'rename', + projectId: projectId(input.projectId), + name: requireUtf8String(input.name, 'project name', PROJECT_CATALOG_NAME_MAX_BYTES), + }; + } + default: + throw invalidProtocolFrame('Invalid project catalog mutation kind'); + } +} + +export function decodeProjectCatalogMutateResult(value: unknown): ProjectCatalogMutateResult { + const record = requireRecord(value, 'project catalog mutation result'); + if (record.kind === 'project') { + const result = requireExactRecord(record, 'project mutation result', ['kind', 'projectId']); + return { kind: 'project', projectId: projectId(result.projectId) }; + } + if (record.kind === 'selection') { + const result = requireExactRecord(record, 'project selection result', [ + 'kind', + 'projectId', + 'path', + ]); + return { + kind: 'selection', + projectId: projectId(result.projectId), + path: absolutePath(result.path, 'selected project path'), + }; + } + throw invalidProtocolFrame('Invalid project catalog mutation result kind'); +} + +function decodeProjectCatalogPageItem(value: unknown): ProjectCatalogPageItem { + const record = requireRecord(value, 'project catalog page item'); + if (record.kind === 'project') { + const item = requireExactRecord(record, 'project catalog header item', [ + 'kind', + 'projectIndex', + 'id', + 'name', + 'aliasCount', + 'locationCount', + 'archivedAt', + 'available', + 'preferredPath', + ]); + return { + kind: 'project', + projectIndex: requireCount(item.projectIndex, 'project index'), + id: projectId(item.id), + name: requireUtf8String(item.name, 'project name', PROJECT_CATALOG_NAME_MAX_BYTES), + aliasCount: requireCount(item.aliasCount, 'project alias count'), + locationCount: requireCount(item.locationCount, 'project location count'), + archivedAt: + item.archivedAt === null ? null : requireCount(item.archivedAt, 'project archivedAt'), + available: boolean(item.available, 'project available'), + preferredPath: + item.preferredPath === null + ? null + : absolutePath(item.preferredPath, 'project preferred path'), + }; + } + if (record.kind === 'alias') { + const item = requireExactRecord(record, 'project catalog alias item', [ + 'kind', + 'projectIndex', + 'itemIndex', + 'alias', + ]); + return { + kind: 'alias', + projectIndex: requireCount(item.projectIndex, 'project index'), + itemIndex: requireCount(item.itemIndex, 'project alias index'), + alias: projectId(item.alias), + }; + } + if (record.kind === 'location') { + const item = requireExactRecord(record, 'project catalog location item', [ + 'kind', + 'projectIndex', + 'itemIndex', + 'location', + ]); + return { + kind: 'location', + projectIndex: requireCount(item.projectIndex, 'project index'), + itemIndex: requireCount(item.itemIndex, 'project location index'), + location: decodeProjectLocation(item.location), + }; + } + throw invalidProtocolFrame('Invalid project catalog page item kind'); +} + +export function decodeProjectCatalogProject(value: unknown): ProjectCatalogProject { + const record = requireExactRecord(value, 'project catalog project', [ + 'id', + 'aliases', + 'name', + 'locations', + 'archivedAt', + 'available', + 'preferredPath', + ]); + if (!Array.isArray(record.aliases)) throw invalidProtocolFrame('Invalid project aliases'); + if (!Array.isArray(record.locations)) throw invalidProtocolFrame('Invalid project locations'); + const aliases = record.aliases.map(projectId); + if (new Set(aliases).size !== aliases.length) { + throw invalidProtocolFrame('Duplicate project aliases'); + } + const project: ProjectCatalogProject = { + id: projectId(record.id), + aliases, + name: requireUtf8String(record.name, 'project name', PROJECT_CATALOG_NAME_MAX_BYTES), + locations: record.locations.map(decodeProjectLocation), + archivedAt: + record.archivedAt === null ? null : requireCount(record.archivedAt, 'project archivedAt'), + available: boolean(record.available, 'project available'), + preferredPath: + record.preferredPath === null + ? null + : absolutePath(record.preferredPath, 'project preferred path'), + }; + return project; +} + +function decodeProjectLocation(value: unknown): ProjectCatalogLocation { + const item = requireExactRecord(value, 'project location', ['path', 'isWorktree']); + return { + path: absolutePath(item.path, 'project location path'), + isWorktree: boolean(item.isWorktree, 'project location isWorktree'), + }; +} + +function projectId(value: unknown): string { + return requireEntityId(value, 'projectId'); +} + +function absolutePath(value: unknown, label: string): string { + const path = requireUtf8String(value, label, PROJECT_CATALOG_PATH_MAX_BYTES); + const absolute = + process.platform === 'win32' + ? /^[A-Za-z]:[\\/]/.test(path) || /^[\\/]{2}[^\\/]+[\\/][^\\/]+(?:[\\/]|$)/.test(path) + : path.startsWith('/'); + if (!absolute) throw invalidProtocolFrame(`${label} must be absolute`); + return path; +} + +function revision(value: unknown, label: string): ProjectCatalogRevision { + const candidate = requireUtf8String(value, label, 71); + if (!/^sha256:[a-f0-9]{64}$/.test(candidate)) { + throw invalidProtocolFrame(`Invalid ${label}`); + } + return candidate as ProjectCatalogRevision; +} + +function boolean(value: unknown, label: string): boolean { + if (typeof value !== 'boolean') throw invalidProtocolFrame(`Invalid ${label}`); + return value; +} diff --git a/packages/runtime-host/src/server/connection-authority.ts b/packages/runtime-host/src/server/connection-authority.ts index 7722e2cb6a..8d5e8a0aef 100644 --- a/packages/runtime-host/src/server/connection-authority.ts +++ b/packages/runtime-host/src/server/connection-authority.ts @@ -93,6 +93,8 @@ function operationUsesHostPath(frame: RequestFrame): boolean { case 'skill.catalog.query': case 'skill.catalog.mutate': case 'skill.catalog.preview-update': + case 'project.catalog.query': + case 'project.catalog.mutate': return true; case 'skill.catalog.invocable.query': return frame.input.target.kind === 'new_session'; diff --git a/packages/runtime-host/src/server/connection-session.ts b/packages/runtime-host/src/server/connection-session.ts index 2dde574c8d..6cc42904dc 100644 --- a/packages/runtime-host/src/server/connection-session.ts +++ b/packages/runtime-host/src/server/connection-session.ts @@ -32,6 +32,10 @@ import type { HostSessionCatalogChangeService, SessionCatalogChangeConnection, } from './session-catalog-change-service.js'; +import type { + HostProjectCatalogChangeService, + ProjectCatalogChangeConnection, +} from './project-catalog-change-service.js'; import type { RuntimeHostConnectionAuthority } from './connection-authority.js'; import { authorizeClientCapabilityFrame, @@ -57,6 +61,7 @@ export interface RuntimeHostConnectionSessionOptions { resolveContinuity(): SessionContinuityService | undefined; resolveClientCapabilities?(): ClientCapabilityService | undefined; resolveConfigurationChanges?(): HostConfigurationChangeService | undefined; + resolveProjectCatalogChanges?(): HostProjectCatalogChangeService | undefined; resolveSessionCatalogChanges?(): HostSessionCatalogChangeService | undefined; beginOperation(frame: RequestFrame): Promise; onTeardown(): void; @@ -72,6 +77,7 @@ export class RuntimeHostConnectionSession { #clientCapabilityService: ClientCapabilityService | undefined; #clientCapabilities: ClientCapabilityConnection | undefined; #configurationChanges: ConfigurationChangeConnection | undefined; + #projectCatalogChanges: ProjectCatalogChangeConnection | undefined; #sessionCatalogChanges: SessionCatalogChangeConnection | undefined; #inputClosed = false; #closed = false; @@ -104,6 +110,7 @@ export class RuntimeHostConnectionSession { this.#detachContinuity(); this.#detachClientCapabilities(); this.#detachConfigurationChanges(); + this.#detachProjectCatalogChanges(); this.#detachSessionCatalogChanges(); const outcome = await Promise.race([ Promise.allSettled([...this.#requests.values()]).then(() => 'drained' as const), @@ -300,6 +307,7 @@ export class RuntimeHostConnectionSession { attachGlobalChanges(): void { if (this.#closed || this.#inputClosed) return; this.#attachConfigurationChanges(); + this.#attachProjectCatalogChanges(); this.#attachSessionCatalogChanges(); } @@ -308,6 +316,31 @@ export class RuntimeHostConnectionSession { this.#configurationChanges = undefined; } + #attachProjectCatalogChanges(): void { + if ( + !this.#options.connection.authority.canUseHostPaths || + !hasRuntimeHostOperationGrant(this.#options.connection.authority, 'project.catalog.query') + ) { + return; + } + const service = this.#options.resolveProjectCatalogChanges?.(); + if (!service || this.#projectCatalogChanges) return; + this.#projectCatalogChanges = service.attachConnection(this.#options.connection.connectionId, { + send: (frame) => { + try { + return this.#writer.enqueue(frame).flushed; + } catch (error) { + return Promise.reject(error); + } + }, + }); + } + + #detachProjectCatalogChanges(): void { + this.#projectCatalogChanges?.close(); + this.#projectCatalogChanges = undefined; + } + #attachSessionCatalogChanges(): void { if ( !hasRuntimeHostOperationGrant(this.#options.connection.authority, 'session.catalog.query') @@ -339,6 +372,7 @@ export class RuntimeHostConnectionSession { this.#detachContinuity(); this.#detachClientCapabilities(); this.#detachConfigurationChanges(); + this.#detachProjectCatalogChanges(); this.#detachSessionCatalogChanges(); this.#writer.close(); this.#options.transport.abort(); diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index d11f84dcd5..5375efc8c5 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -32,6 +32,10 @@ import { type MakaTool, type RuntimeHostedRootAuthority, } from '@maka/runtime'; +import { + openInteractiveProjectCatalogForWrite, + type InteractiveProjectCatalogWriter, +} from '@maka/storage'; import { createAgentGraphControlStore } from '@maka/storage/agent-graph-control-store'; import { createArtifactAttachmentResourceReader, @@ -107,6 +111,9 @@ import { HostNetworkProxyCoordinator } from './network-proxy-coordinator.js'; import { HostOAuthExecutionAuthority } from './oauth-execution-authority.js'; import { HostOAuthCoordinator, type HostOAuthCoordinatorInput } from './oauth-coordinator.js'; import { HostPlanCoordinator } from './plan-coordinator.js'; +import { HostProjectCatalogChangeService } from './project-catalog-change-service.js'; +import { HostProjectCatalogCoordinator } from './project-catalog-coordinator.js'; +import { HostProjectMembershipGate } from './project-membership-gate.js'; import type { DomainOperationHandlerMap } from './operation-dispatcher.js'; import { RootAdmissionOwner } from './root-admission-owner.js'; import { RootTurnCoordinator } from './root-turn-coordinator.js'; @@ -169,6 +176,7 @@ export async function createExecutionRuntimeHostComposition( dependencies: ExecutionRuntimeHostCompositionDependencies = {}, ): Promise { const stores = await openInteractiveExecutionStoresForWrite(context.owner.lease); + await stores.sessionStore.ready(); let graphControlStore: ReturnType | undefined; let taskLedgerStore: | Awaited> @@ -193,7 +201,13 @@ export async function createExecutionRuntimeHostComposition( let unsubscribeTaskLedger: (() => void) | undefined; let managedWorkspaceOwner: ManagedWorkspaceOwner | undefined; let workspaceExecution: RuntimeHostWorkspaceExecutionComposition | undefined; + let projectCatalog: InteractiveProjectCatalogWriter | undefined; try { + const openedProjectCatalog = await openInteractiveProjectCatalogForWrite(context.owner.lease, { + onLegacyImportFailure: (error) => + console.error('[runtime-host] projects.json could not be imported:', error), + }); + projectCatalog = openedProjectCatalog; const runtimePolicyStores = await openInteractiveRuntimePolicyStoresForWrite( context.owner.lease, ); @@ -408,6 +422,15 @@ export async function createExecutionRuntimeHostComposition( ); const configurationChanges = new HostConfigurationChangeService(); const sessionCatalogChanges = new HostSessionCatalogChangeService(); + const projectCatalogChanges = new HostProjectCatalogChangeService(); + const projectMembership = new HostProjectMembershipGate(); + const projects = new HostProjectCatalogCoordinator( + openedProjectCatalog, + projectCatalogChanges, + sessionCatalogChanges, + projectMembership, + context.requestDrain, + ); let rootCoordinator: RootTurnCoordinator | undefined; let canonicalProjection: CanonicalSessionProjectionReader | undefined; let memory: HostMemoryCoordinator | undefined; @@ -1044,6 +1067,8 @@ export async function createExecutionRuntimeHostComposition( manager, admission: sessionAdmission, continuity: continuityCoordinator, + projectCatalog: openedProjectCatalog, + projectMembership, requestDrain: context.requestDrain, }); const externalSessions = new HostExternalSessionCoordinator({ @@ -1145,6 +1170,7 @@ export async function createExecutionRuntimeHostComposition( ...runtimeResources.handlers, ...automations.handlers, ...plans.handlers, + ...projects.handlers, ...requireDeepResearch(deepResearch).handlers, ...requireDailyReview(dailyReview).handlers, ...webSearch.handlers, @@ -1364,6 +1390,11 @@ export async function createExecutionRuntimeHostComposition( } catch (error) { errors.push(error); } + try { + openedProjectCatalog.close(); + } catch (error) { + errors.push(error); + } try { await stores.sessionStore.close?.(); } catch (error) { @@ -1382,6 +1413,7 @@ export async function createExecutionRuntimeHostComposition( continuity: continuityCoordinator, clientCapabilities, configurationChanges, + projectCatalogChanges, sessionCatalogChanges, releaseConnection: (connectionId: string) => { artifacts.releaseConnection(connectionId); @@ -1467,6 +1499,11 @@ export async function createExecutionRuntimeHostComposition( } catch (closeError) { errors.push(closeError); } + try { + projectCatalog?.close(); + } catch (closeError) { + errors.push(closeError); + } try { await stores.sessionStore.close?.(); } catch (closeError) { diff --git a/packages/runtime-host/src/server/host-kernel.ts b/packages/runtime-host/src/server/host-kernel.ts index f50241af5f..7ddc570360 100644 --- a/packages/runtime-host/src/server/host-kernel.ts +++ b/packages/runtime-host/src/server/host-kernel.ts @@ -42,6 +42,7 @@ import { import type { SessionContinuityService } from './session-continuity-service.js'; import type { ClientCapabilityService } from './client-capability-service.js'; import type { HostConfigurationChangeService } from './configuration-change-service.js'; +import type { HostProjectCatalogChangeService } from './project-catalog-change-service.js'; import { runtimeHostLogBuffer } from '../process-diagnostics.js'; import type { HostSessionCatalogChangeService } from './session-catalog-change-service.js'; import { @@ -86,6 +87,7 @@ export interface RuntimeHostComposition { readonly continuity?: SessionContinuityService; readonly clientCapabilities?: ClientCapabilityService; readonly configurationChanges?: HostConfigurationChangeService; + readonly projectCatalogChanges?: HostProjectCatalogChangeService; readonly sessionCatalogChanges?: HostSessionCatalogChangeService; releaseConnection?(connectionId: string): void; beginDrain(): void; @@ -324,6 +326,7 @@ export class RuntimeHostKernel { resolveContinuity: () => this.#composition?.continuity, resolveClientCapabilities: () => this.#composition?.clientCapabilities, resolveConfigurationChanges: () => this.#composition?.configurationChanges, + resolveProjectCatalogChanges: () => this.#composition?.projectCatalogChanges, resolveSessionCatalogChanges: () => this.#composition?.sessionCatalogChanges, beginOperation: (request) => this.#beginOperation(request), onTeardown: releaseTransport, diff --git a/packages/runtime-host/src/server/operation-dispatcher.ts b/packages/runtime-host/src/server/operation-dispatcher.ts index 36cd950384..a9e2baff36 100644 --- a/packages/runtime-host/src/server/operation-dispatcher.ts +++ b/packages/runtime-host/src/server/operation-dispatcher.ts @@ -102,6 +102,7 @@ export type RuntimeResourceOperationKey = Extract; export type AutomationOperationKey = Extract; export type PlanOperationKey = Extract; +export type ProjectCatalogOperationKey = Extract; export type DeepResearchOperationKey = Extract; export type DailyReviewOperationKey = Extract; export type WebSearchOperationKey = Extract; @@ -160,6 +161,10 @@ export type ClientCapabilityOperationHandlerMap = Pick< >; export type AutomationOperationHandlerMap = Pick; export type PlanOperationHandlerMap = Pick; +export type ProjectCatalogOperationHandlerMap = Pick< + OperationHandlerMap, + ProjectCatalogOperationKey +>; export type DeepResearchOperationHandlerMap = Pick; export type DailyReviewOperationHandlerMap = Pick; export type WebSearchOperationHandlerMap = Pick; diff --git a/packages/runtime-host/src/server/project-catalog-change-service.ts b/packages/runtime-host/src/server/project-catalog-change-service.ts new file mode 100644 index 0000000000..62f43e9ae3 --- /dev/null +++ b/packages/runtime-host/src/server/project-catalog-change-service.ts @@ -0,0 +1,39 @@ +import type { ProjectCatalogChangedFrame } from '../protocol/index.js'; + +export interface ProjectCatalogChangeConnection { + close(): void; +} + +interface ProjectCatalogChangeSink { + send(frame: ProjectCatalogChangedFrame): Promise; +} + +export class HostProjectCatalogChangeService { + readonly #connections = new Map(); + #revision = 0; + + attachConnection( + connectionId: string, + sink: ProjectCatalogChangeSink, + ): ProjectCatalogChangeConnection { + this.#connections.set(connectionId, sink); + return { + close: () => { + if (this.#connections.get(connectionId) === sink) this.#connections.delete(connectionId); + }, + }; + } + + publish(): void { + this.#revision += 1; + const frame: ProjectCatalogChangedFrame = { + kind: 'project.catalog.changed', + revision: this.#revision, + }; + for (const [connectionId, sink] of this.#connections) { + void sink.send(frame).catch(() => { + if (this.#connections.get(connectionId) === sink) this.#connections.delete(connectionId); + }); + } + } +} diff --git a/packages/runtime-host/src/server/project-catalog-coordinator.ts b/packages/runtime-host/src/server/project-catalog-coordinator.ts new file mode 100644 index 0000000000..becfeeb07e --- /dev/null +++ b/packages/runtime-host/src/server/project-catalog-coordinator.ts @@ -0,0 +1,259 @@ +import { createHash } from 'node:crypto'; +import type { ProjectRecord } from '@maka/core'; +import { + ProjectArchivedError, + type ProjectCatalog, + ProjectNotFoundError, + ProjectPathConflictError, + ProjectPathMismatchError, + ProjectUnavailableError, +} from '@maka/storage'; +import { + decodeProjectCatalogProject, + PROJECT_CATALOG_PAGE_MAX_BYTES, + PROJECT_CATALOG_PAGE_MAX_ITEMS, + type OperationOutcome, + type ProjectCatalogMutateInput, + type ProjectCatalogMutateResult, + type ProjectCatalogPageItem, + type ProjectCatalogProject, + type ProjectCatalogQueryInput, + type ProjectCatalogQueryResult, + type ProjectCatalogRevision, +} from '../protocol/index.js'; +import type { ProjectCatalogOperationHandlerMap } from './operation-dispatcher.js'; +import type { HostProjectCatalogChangeService } from './project-catalog-change-service.js'; +import type { HostProjectMembershipGate } from './project-membership-gate.js'; +import type { HostSessionCatalogChangeService } from './session-catalog-change-service.js'; + +export class HostProjectCatalogCoordinator { + readonly handlers: ProjectCatalogOperationHandlerMap = { + 'project.catalog.query': (input) => this.#query(input), + 'project.catalog.mutate': (input) => this.#mutate(input), + }; + + constructor( + private readonly catalog: ProjectCatalog, + private readonly projectChanges: HostProjectCatalogChangeService, + private readonly sessionChanges: HostSessionCatalogChangeService, + private readonly membership: HostProjectMembershipGate, + private readonly requestDrain: () => void, + ) {} + + async #query( + input: ProjectCatalogQueryInput, + ): Promise> { + try { + const projects = (await this.catalog.list()).map(projectProject); + const items = projectCatalogItems(projects); + const revision = catalogRevision(items); + if (input.kind === 'list_continue' && input.revision !== revision) { + return successQuery({ + kind: 'revision_changed', + expected: input.revision, + actual: revision, + }); + } + const offset = input.kind === 'list_start' ? 0 : decodeCursor(input.cursor); + if ( + offset === undefined || + offset > items.length || + (input.kind === 'list_continue' && offset === items.length) + ) { + return queryFailure('invalid_request', 'Project catalog cursor is invalid'); + } + return successQuery(createPage(revision, projects.length, items, offset)); + } catch { + return queryFailure('persistence_failed', 'Project catalog is unavailable'); + } + } + + async #mutate( + input: ProjectCatalogMutateInput, + ): Promise> { + try { + const result = await this.membership.run(() => this.#applyMutation(input)); + this.projectChanges.publish(); + return { ok: true, result }; + } catch (error) { + if (error instanceof ProjectNotFoundError) { + return mutationFailure('not_found', error.message); + } + if ( + error instanceof ProjectArchivedError || + error instanceof ProjectUnavailableError || + error instanceof ProjectPathConflictError || + error instanceof ProjectPathMismatchError + ) { + return mutationFailure('operation_conflict', error.message); + } + if (error instanceof TypeError || isInvalidPathError(error)) { + return mutationFailure('invalid_request', 'Project catalog input is invalid'); + } + this.requestDrain(); + return mutationFailure( + 'commit_outcome_unknown', + 'Project catalog mutation outcome is unknown', + ); + } + } + + async #applyMutation(input: ProjectCatalogMutateInput): Promise { + switch (input.kind) { + case 'register': + return projectResult((await this.catalog.register(input.path)).id); + case 'select': { + const selected = await this.catalog.select(input.projectId); + return { + kind: 'selection', + projectId: selected.project.id, + path: selected.path, + }; + } + case 'touch': + return projectResult( + (await this.catalog.touch(input.projectId, input.path === null ? undefined : input.path)) + .id, + ); + case 'relink': { + const result = await this.catalog.relinkWithSessions(input.projectId, input.path); + for (const sessionId of result.updatedSessionIds) this.sessionChanges.publish(sessionId); + return projectResult(result.project.id); + } + case 'rename': + return projectResult((await this.catalog.rename(input.projectId, input.name)).id); + case 'archive': + return projectResult((await this.catalog.archive(input.projectId)).id); + case 'restore': + return projectResult((await this.catalog.restore(input.projectId)).id); + } + } +} + +function projectResult(projectId: string): ProjectCatalogMutateResult { + return { kind: 'project', projectId }; +} + +function projectProject(project: ProjectRecord): ProjectCatalogProject { + return decodeProjectCatalogProject({ + id: project.id, + aliases: [...(project.aliases ?? [])], + name: project.name, + locations: project.locations.map((location) => ({ ...location })), + archivedAt: project.archivedAt ?? null, + available: project.available, + preferredPath: project.preferredPath ?? null, + }); +} + +function projectCatalogItems(projects: readonly ProjectCatalogProject[]): ProjectCatalogPageItem[] { + return projects.flatMap((project, projectIndex): ProjectCatalogPageItem[] => [ + { + kind: 'project', + projectIndex, + id: project.id, + name: project.name, + aliasCount: project.aliases.length, + locationCount: project.locations.length, + archivedAt: project.archivedAt, + available: project.available, + preferredPath: project.preferredPath, + }, + ...project.aliases.map((alias, itemIndex) => ({ + kind: 'alias' as const, + projectIndex, + itemIndex, + alias, + })), + ...project.locations.map((location, itemIndex) => ({ + kind: 'location' as const, + projectIndex, + itemIndex, + location, + })), + ]); +} + +function catalogRevision(items: readonly ProjectCatalogPageItem[]): ProjectCatalogRevision { + return `sha256:${createHash('sha256').update(JSON.stringify(items)).digest('hex')}`; +} + +function createPage( + revision: ProjectCatalogRevision, + projectCount: number, + items: readonly ProjectCatalogPageItem[], + offset: number, +): ProjectCatalogQueryResult { + const pageItems: ProjectCatalogPageItem[] = []; + for (let index = offset; index < items.length; index += 1) { + if (pageItems.length >= PROJECT_CATALOG_PAGE_MAX_ITEMS) break; + const item = items[index]; + if (!item) throw new Error('Project catalog projection index was out of bounds'); + const nextOffset = index + 1; + const candidate: ProjectCatalogQueryResult = { + kind: 'page', + revision, + projectCount, + items: [...pageItems, item], + nextCursor: nextOffset < items.length ? encodeCursor(nextOffset) : null, + }; + if (encodedBytes(candidate) > PROJECT_CATALOG_PAGE_MAX_BYTES) break; + pageItems.push(item); + } + if (pageItems.length === 0 && offset < items.length) { + throw new Error('A Project catalog item exceeds the page byte limit'); + } + const nextOffset = offset + pageItems.length; + return { + kind: 'page', + revision, + projectCount, + items: pageItems, + nextCursor: nextOffset < items.length ? encodeCursor(nextOffset) : null, + }; +} + +function encodeCursor(offset: number): string { + return String(offset); +} + +function decodeCursor(cursor: string): number | undefined { + if (!/^(?:0|[1-9]\d*)$/.test(cursor)) return undefined; + const offset = Number(cursor); + return Number.isSafeInteger(offset) ? offset : undefined; +} + +function encodedBytes(value: unknown): number { + return Buffer.byteLength(JSON.stringify(value), 'utf8'); +} + +function isInvalidPathError(error: unknown): boolean { + const code = (error as NodeJS.ErrnoException)?.code; + return ( + code === 'EACCES' || + code === 'ENOENT' || + code === 'ENOTDIR' || + code === 'EINVAL' || + code === 'EPERM' + ); +} + +function successQuery( + result: ProjectCatalogQueryResult, +): OperationOutcome<'project.catalog.query'> { + return { ok: true, result }; +} + +function queryFailure( + code: 'invalid_request' | 'persistence_failed', + message: string, +): OperationOutcome<'project.catalog.query'> { + return { ok: false, error: { code, message } }; +} + +function mutationFailure( + code: 'invalid_request' | 'not_found' | 'operation_conflict' | 'commit_outcome_unknown', + message: string, +): OperationOutcome<'project.catalog.mutate'> { + return { ok: false, error: { code, message } }; +} diff --git a/packages/runtime-host/src/server/project-membership-gate.ts b/packages/runtime-host/src/server/project-membership-gate.ts new file mode 100644 index 0000000000..df76fbb311 --- /dev/null +++ b/packages/runtime-host/src/server/project-membership-gate.ts @@ -0,0 +1,12 @@ +export class HostProjectMembershipGate { + #tail: Promise = Promise.resolve(); + + run(operation: () => Promise): Promise { + const task = this.#tail.then(operation, operation); + this.#tail = task.then( + () => undefined, + () => undefined, + ); + return task; + } +} diff --git a/packages/runtime-host/src/server/session-catalog-coordinator.ts b/packages/runtime-host/src/server/session-catalog-coordinator.ts index 9d55c2a18a..8044b3579c 100644 --- a/packages/runtime-host/src/server/session-catalog-coordinator.ts +++ b/packages/runtime-host/src/server/session-catalog-coordinator.ts @@ -9,6 +9,7 @@ import { type ExecutionBoundarySummary, } from '@maka/core/sandbox-boundary'; import type { CreateSessionInput } from '@maka/core/runtime-inputs'; +import type { ProjectCatalog } from '@maka/storage'; import { DEFAULT_SESSION_NAME, normalizeUserSessionName } from '@maka/core/session-name'; import { isSessionStartModeLabel as isExecutionSemanticLabel, @@ -56,6 +57,7 @@ import { type SessionUpdateResult, } from '../protocol/index.js'; import type { SessionCatalogOperationHandlerMap } from './operation-dispatcher.js'; +import type { HostProjectMembershipGate } from './project-membership-gate.js'; import { type SessionAdmissionLease, SessionAdmissionGate } from './session-admission-gate.js'; import type { SessionContinuityCoordinator } from './session-continuity-coordinator.js'; @@ -105,6 +107,8 @@ export interface HostSessionCatalogCoordinatorOptions { readonly manager: SessionConfigurationAuthority; readonly admission: SessionAdmissionGate; readonly continuity: SessionContinuity; + readonly projectCatalog: ProjectCatalog; + readonly projectMembership: HostProjectMembershipGate; readonly requestDrain: () => void; } @@ -130,6 +134,8 @@ export class HostSessionCatalogCoordinator { readonly #manager: SessionConfigurationAuthority; readonly #admission: SessionAdmissionGate; readonly #continuity: SessionContinuity; + readonly #projectCatalog: ProjectCatalog; + readonly #projectMembership: HostProjectMembershipGate; readonly #requestDrain: () => void; constructor(options: HostSessionCatalogCoordinatorOptions) { @@ -138,6 +144,8 @@ export class HostSessionCatalogCoordinator { this.#manager = options.manager; this.#admission = options.admission; this.#continuity = options.continuity; + this.#projectCatalog = options.projectCatalog; + this.#projectMembership = options.projectMembership; this.#requestDrain = options.requestDrain; } @@ -249,24 +257,31 @@ export class HostSessionCatalogCoordinator { this.#resolveModel(input.modelTarget, input.thinkingLevel), this.#readRuntimePolicy(), ]); - const createInput: CreateSessionInput = { - cwd: prepared.cwd, - ...(input.projectId !== undefined ? { projectId: input.projectId } : {}), - name: prepared.name, - labels: [...prepared.labels], - backend: 'ai-sdk', - llmConnectionSlug: model.connectionSlug, - model: model.model, - ...(input.thinkingLevel === undefined ? {} : { thinkingLevel: input.thinkingLevel }), - permissionMode: prepared.permissionMode ?? policy.policy.chatDefaults.permissionMode, - collaborationMode: input.collaborationMode ?? 'agent', - orchestrationMode: input.orchestrationMode ?? 'default', - }; - commitAttempted = true; - const result = await this.#stores.createStableSession({ - sessionId: input.sessionId, - requestFingerprint: prepared.requestFingerprint, - input: createInput, + const result = await this.#projectMembership.run(async () => { + const workspace = await resolveSessionProjectWorkspace( + this.#projectCatalog, + prepared.cwd, + input.projectId, + ); + const createInput: CreateSessionInput = { + cwd: workspace.cwd, + ...(workspace.projectId !== undefined ? { projectId: workspace.projectId } : {}), + name: prepared.name, + labels: [...prepared.labels], + backend: 'ai-sdk', + llmConnectionSlug: model.connectionSlug, + model: model.model, + ...(input.thinkingLevel === undefined ? {} : { thinkingLevel: input.thinkingLevel }), + permissionMode: prepared.permissionMode ?? policy.policy.chatDefaults.permissionMode, + collaborationMode: input.collaborationMode ?? 'agent', + orchestrationMode: input.orchestrationMode ?? 'default', + }; + commitAttempted = true; + return this.#stores.createStableSession({ + sessionId: input.sessionId, + requestFingerprint: prepared.requestFingerprint, + input: createInput, + }); }); if (result.kind === 'conflict') { return createFailure( @@ -699,6 +714,30 @@ interface PreparedSessionCreate { readonly requestFingerprint: string; } +async function resolveSessionProjectWorkspace( + catalog: Pick, + cwd: string, + projectId: string | null | undefined, +): Promise<{ readonly cwd: string; readonly projectId?: string | null }> { + if (typeof projectId !== 'string') { + return { cwd, ...(projectId === undefined ? {} : { projectId }) }; + } + const project = (await catalog.list()).find( + (candidate) => candidate.id === projectId || candidate.aliases?.includes(projectId), + ); + if (!project) { + throw new SessionOperationFailure('operation_conflict', `Project does not exist: ${projectId}`); + } + if (project.archivedAt !== undefined) { + throw new SessionOperationFailure('operation_conflict', `Project is archived: ${projectId}`); + } + const resolved = project.preferredPath; + if (!project.available || !resolved) { + throw new SessionOperationFailure('operation_conflict', `Project is unavailable: ${projectId}`); + } + return { cwd: resolved, projectId: project.id }; +} + async function prepareCreate(input: SessionCreateInput): Promise { if (!isAbsolute(input.cwd)) { throw new SessionOperationFailure('invalid_request', 'Session cwd must be absolute'); @@ -716,7 +755,10 @@ async function prepareCreate(input: SessionCreateInput): Promise { + const base = await mkdtemp(join(tmpdir(), 'maka-project-catalog-authority-')); + const dataRoot = join(base, 'data'); + const projectRoot = join(base, 'project'); + await mkdir(projectRoot); + const capability = await resolveStorageRoot({ path: dataRoot, kind: 'interactive' }); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + if (!owner) return; + let writer: Awaited> | undefined; + try { + const [first, second] = await Promise.all([ + openInteractiveProjectCatalogForWrite(owner.lease), + openInteractiveProjectCatalogForWrite(owner.lease), + ]); + writer = first; + assert.equal(first, second); + assert.equal(authenticateInteractiveProjectCatalogWriter(first), first); + const project = await first.register(projectRoot); + assert.equal((await second.list())[0]?.id, project.id); + + await owner.close(); + await assert.rejects( + () => first.rename(project.id, 'Renamed'), + (error: unknown) => + error instanceof StorageRootAuthorityError && error.code === 'invalid_lease', + ); + } finally { + writer?.close(); + if (!owner.closed) await owner.close(); + await rm(base, { recursive: true, force: true }); + } +}); diff --git a/packages/storage/src/__tests__/project-catalog.test.ts b/packages/storage/src/__tests__/project-catalog.test.ts index 8cc456f7db..49c0a1254e 100644 --- a/packages/storage/src/__tests__/project-catalog.test.ts +++ b/packages/storage/src/__tests__/project-catalog.test.ts @@ -12,6 +12,7 @@ import { type ResolvedProjectLocation, resolveProjectLocation, } from '../project-catalog.js'; +import { createSessionStore } from '../session-store.js'; import { createGitRepositoryWithWorktree } from './fixtures/git-repository.js'; const execFileAsync = promisify(execFile); @@ -49,6 +50,17 @@ async function rm(path: string, options?: Parameters[1]): Promise await remove(path, options); } +function sessionInput(cwd: string, projectId: string) { + return { + cwd, + projectId, + backend: 'fake' as const, + llmConnectionSlug: 'fake', + model: 'fake-model', + permissionMode: 'ask' as const, + }; +} + test('a plain folder resolves without requiring the Git executable', async () => { const base = await mkdtemp(join(tmpdir(), 'maka-project-folder-no-git-')); try { @@ -291,56 +303,6 @@ test('two catalogs changing one project at the same time keep both changes', asy } }); -test('a relink whose merge target changes mid-flight fails instead of half-committing', async () => { - const base = await mkdtemp(join(tmpdir(), 'maka-project-relink-race-')); - try { - const storage = join(base, 'storage'); - const home = join(base, 'home'); - const shared = join(base, 'shared'); - const elsewhere = join(base, 'elsewhere'); - await Promise.all([mkdir(home), mkdir(shared), mkdir(elsewhere)]); - const first = createProjectCatalog(storage); - const second = createProjectCatalog(storage); - const moving = await first.register(home); - const target = await first.register(shared); - await Promise.all([first.list(), second.list()]); - - let releaseCallback!: () => void; - let callbackStarted!: () => void; - const gate = new Promise((release) => { - releaseCallback = release; - }); - const started = new Promise((resolve) => { - callbackStarted = resolve; - }); - let observed: string | undefined; - const relink = first.relink(moving.id, shared, async (context) => { - observed = context.conflictingProjectId; - callbackStarted(); - await gate; - }); - await started; - - // The callback was told to move `target`'s sessions onto `moving`. While it - // is doing that, the other window moves `target` somewhere else entirely. - await second.relink(target.id, elsewhere); - releaseCallback(); - - assert.equal(observed, target.id, 'precondition: the callback planned a merge'); - await assert.rejects(() => relink, /retry/); - const projects = await first.list(); - assert.deepEqual( - projects.map((project) => project.id).sort(), - [moving.id, target.id].sort(), - 'neither project may be merged away after the plan went stale', - ); - first.close(); - second.close(); - } finally { - await rm(base, { recursive: true, force: true }); - } -}); - test('relinking an unavailable project preserves its id and adopts the new directory', async () => { const base = await mkdtemp(join(tmpdir(), 'maka-project-relink-')); try { @@ -371,49 +333,44 @@ test('relinking an unavailable project preserves its id and adopts the new direc } }); -test('conflicting relink waits for a retryable merge before removing the duplicate project', async () => { - const base = await mkdtemp(join(tmpdir(), 'maka-project-relink-merge-')); +test('Host relink rolls Project and Session membership back in one transaction', async () => { + const base = await mkdtemp(join(tmpdir(), 'maka-project-session-relink-')); + const storage = join(base, 'storage'); + const originalPath = join(base, 'original'); + const destinationPath = join(base, 'destination'); + await Promise.all([mkdir(originalPath), mkdir(destinationPath)]); + const injected = new Error('injected atomic relink failure'); + const catalog = createProjectCatalog(storage, { + createId: (() => { + let id = 0; + return () => `project-${++id}`; + })(), + relinkFailpoint: () => { + throw injected; + }, + }); + const sessions = createSessionStore(storage); try { - const relocated = join(base, 'relocated'); - await mkdir(relocated); - let id = 0; - const catalog = createProjectCatalog(join(base, 'storage'), { - now: () => 1_000, - createId: () => `project-${++id}`, - }); - const originalPath = join(base, 'original'); - await mkdir(originalPath); const original = await catalog.register(originalPath); - await rm(originalPath, { recursive: true, force: true }); - const duplicate = await catalog.register(relocated); - const interrupted = new Error('session reassignment interrupted'); + const duplicate = await catalog.register(destinationPath); + const originalSession = await sessions.create(sessionInput(originalPath, original.id)); + const duplicateSession = await sessions.create(sessionInput(destinationPath, duplicate.id)); await assert.rejects( - () => - catalog.relink(original.id, relocated, async () => { - throw interrupted; - }), - (error) => error === interrupted, - ); - assert.deepEqual( - (await catalog.list()).map((project) => project.id).sort(), - [original.id, duplicate.id].sort(), + () => catalog.relinkWithSessions(original.id, destinationPath), + (error) => error === injected, ); - let mergedProjectId: string | undefined; - const merged = await catalog.relink(original.id, relocated, async (context) => { - mergedProjectId = context.conflictingProjectId; - }); - - assert.equal(mergedProjectId, duplicate.id); - assert.equal(merged.id, original.id); - assert.equal(merged.name, original.name); - assert.deepEqual((merged as typeof merged & { aliases?: string[] }).aliases, [duplicate.id]); assert.deepEqual( - (await catalog.list()).map((project) => project.id), - [original.id], + (await catalog.list()).map(({ id }) => id).sort(), + [original.id, duplicate.id].sort(), ); + assert.equal((await sessions.readHeaderSnapshot(originalSession.id)).projectId, original.id); + assert.equal((await sessions.readHeaderSnapshot(originalSession.id)).cwd, originalPath); + assert.equal((await sessions.readHeaderSnapshot(duplicateSession.id)).projectId, duplicate.id); + assert.equal((await sessions.readHeaderSnapshot(duplicateSession.id)).cwd, destinationPath); } finally { + await sessions.close?.(); await rm(base, { recursive: true, force: true }); } }); @@ -436,7 +393,7 @@ test('conflicting relink preserves every available worktree location from the me await catalog.register(repository); await catalog.register(linkedWorktree); - const relinked = await catalog.relink(original.id, repository, async () => {}); + const { project: relinked } = await catalog.relinkWithSessions(original.id, repository); assert.deepEqual( relinked.locations.map((location) => location.path).sort(), diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index d23b379d47..6fab1afdbf 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -326,6 +326,7 @@ async function createExecutionStoresForWrite run(() => conversationOperationalStateStore.purge(sessionId)), sessionStore: { + ready: () => run(() => sessionStore.ready()), create: (input, initialBoundary) => run(() => sessionStore.create(input, initialBoundary)), createImportedSession: (input, messages) => run(() => sessionStore.createImportedSession(input, messages)), diff --git a/packages/storage/src/index.ts b/packages/storage/src/index.ts index 848fa4393e..274a28197c 100644 --- a/packages/storage/src/index.ts +++ b/packages/storage/src/index.ts @@ -132,6 +132,7 @@ export * from './workspace-identity.js'; export * from './memory-bundle-store.js'; export * from './long-term-memory-store.js'; export * from './project-catalog.js'; +export * from './project-catalog-authority.js'; export * from './project-session-backfill.js'; export * from './git-worktree-child-executor.js'; export * from './managed-workspace-owner.js'; diff --git a/packages/storage/src/project-catalog-authority.ts b/packages/storage/src/project-catalog-authority.ts new file mode 100644 index 0000000000..17af9671f4 --- /dev/null +++ b/packages/storage/src/project-catalog-authority.ts @@ -0,0 +1,107 @@ +import { + assertStorageRootLease, + runWithStorageRootLease, + StorageRootAuthorityError, + type StorageRootLease, +} from './root-authority.js'; +import { createProjectCatalog, type ProjectCatalog } from './project-catalog.js'; + +const writerBrand: unique symbol = Symbol('InteractiveProjectCatalogWriter'); +const writers = new WeakSet(); +const writerByLease = new WeakMap(); +const writerOpeningByLease = new WeakMap>(); + +export interface InteractiveProjectCatalogWriter extends ProjectCatalog { + readonly kind: 'interactive'; + readonly access: 'write'; + readonly [writerBrand]: true; +} + +export function authenticateInteractiveProjectCatalogWriter( + writer: InteractiveProjectCatalogWriter, +): InteractiveProjectCatalogWriter { + if (!writers.has(writer)) { + throw new StorageRootAuthorityError( + 'invalid_lease', + 'Expected an authentic interactive Project Catalog writer', + ); + } + return writer; +} + +export async function openInteractiveProjectCatalogForWrite( + lease: StorageRootLease<'interactive', 'write'>, + options: { readonly onLegacyImportFailure?: (error: unknown) => void } = {}, +): Promise { + await assertStorageRootLease(lease, 'interactive', 'write'); + const existing = writerByLease.get(lease); + if (existing) return existing; + const opening = writerOpeningByLease.get(lease); + if (opening) return opening; + + const pending = Promise.resolve().then(async () => { + let catalog: ProjectCatalog | undefined; + try { + catalog = await runWithStorageRootLease(lease, 'interactive', 'write', async (root) => + createProjectCatalog(root, options), + ); + await assertStorageRootLease(lease, 'interactive', 'write'); + const recoveredExisting = writerByLease.get(lease); + if (recoveredExisting) { + catalog.close(); + return recoveredExisting; + } + const writer = createWriterFacade(lease, catalog); + writers.add(writer); + writerByLease.set(lease, writer); + return writer; + } catch (error) { + catalog?.close(); + throw error; + } + }); + writerOpeningByLease.set(lease, pending); + try { + return await pending; + } finally { + if (writerOpeningByLease.get(lease) === pending) writerOpeningByLease.delete(lease); + } +} + +function createWriterFacade( + lease: StorageRootLease<'interactive', 'write'>, + catalog: ProjectCatalog, +): InteractiveProjectCatalogWriter { + let closed = false; + const run = (operation: () => Promise): Promise => { + if (closed) { + return Promise.reject( + new StorageRootAuthorityError('invalid_lease', 'Project Catalog writer is closed'), + ); + } + return runWithStorageRootLease(lease, 'interactive', 'write', operation); + }; + const writer: InteractiveProjectCatalogWriter = { + kind: 'interactive', + access: 'write', + [writerBrand]: true, + list: () => run(() => catalog.list()), + register: (path) => run(() => catalog.register(path)), + resolveHistoricalPath: (path, usedAt) => run(() => catalog.resolveHistoricalPath(path, usedAt)), + select: (projectId) => run(() => catalog.select(projectId)), + touch: (projectId, path) => run(() => catalog.touch(projectId, path)), + relink: (projectId, path) => run(() => catalog.relink(projectId, path)), + relinkWithSessions: (projectId, path) => run(() => catalog.relinkWithSessions(projectId, path)), + rename: (projectId, name) => run(() => catalog.rename(projectId, name)), + archive: (projectId) => run(() => catalog.archive(projectId)), + restore: (projectId) => run(() => catalog.restore(projectId)), + close: () => { + if (closed) return; + closed = true; + if (writerByLease.get(lease) === writer) writerByLease.delete(lease); + writers.delete(writer); + catalog.close(); + }, + }; + return Object.freeze(writer); +} diff --git a/packages/storage/src/project-catalog.ts b/packages/storage/src/project-catalog.ts index 7c5f7672a3..01550405a4 100644 --- a/packages/storage/src/project-catalog.ts +++ b/packages/storage/src/project-catalog.ts @@ -3,12 +3,13 @@ import { randomUUID } from 'node:crypto'; import { readFile, realpath, rename, stat } from 'node:fs/promises'; import { basename, dirname, join, normalize, resolve } from 'node:path'; import { promisify } from 'node:util'; -import type { ProjectLocation, ProjectRecord } from '@maka/core'; +import type { ProjectLocation, ProjectRecord, SessionHeader } from '@maka/core'; import { hasEnclosingGitEntry } from './git-entry.js'; import { acquireOperationalStateDatabase, type OperationalStateDatabaseLease, } from './operational-state-store.js'; +import { normalizeSessionHeader } from './session-store.js'; export type { ProjectLocation, ProjectRecord } from '@maka/core'; @@ -26,33 +27,44 @@ export class ProjectPathMismatchError extends Error { } } -export function isProjectPathMismatchError(error: unknown): error is ProjectPathMismatchError { - return error instanceof ProjectPathMismatchError; +export class ProjectNotFoundError extends Error { + readonly name = 'ProjectNotFoundError'; + readonly code = 'project_not_found'; + + constructor(readonly projectId: string) { + super(`No such project: ${projectId}`); + } } -/** - * The catalog changed underneath a relink while its `beforeCommit` callback was - * still reassigning sessions, so the merge the caller was told to prepare is no - * longer the merge that would be committed. Relink is already retryable — its - * callback throwing leaves the catalog untouched — so failing here hands the - * decision back rather than committing a half-true one. - */ -export class ProjectRelinkContentionError extends Error { - readonly name = 'ProjectRelinkContentionError'; - readonly code = 'project_relink_contention'; +export class ProjectArchivedError extends Error { + readonly name = 'ProjectArchivedError'; + readonly code = 'project_archived'; constructor(readonly projectId: string) { - super(`Project changed while relinking, retry: ${projectId}`); + super(`Project is archived: ${projectId}`); } } -export interface ProjectRelinkContext { - projectId: string; - projectAliases: string[]; - destinationPath: string; - previousLocations: ProjectLocation[]; - conflictingProjectId?: string; - conflictingProjectAliases?: string[]; +export class ProjectUnavailableError extends Error { + readonly name = 'ProjectUnavailableError'; + readonly code = 'project_unavailable'; + + constructor(readonly projectId: string) { + super(`Project is unavailable: ${projectId}`); + } +} + +export class ProjectPathConflictError extends Error { + readonly name = 'ProjectPathConflictError'; + readonly code = 'project_path_conflict'; + + constructor(readonly conflictingProjectId: string) { + super(`Project path already belongs to project: ${conflictingProjectId}`); + } +} + +export function isProjectPathMismatchError(error: unknown): error is ProjectPathMismatchError { + return error instanceof ProjectPathMismatchError; } export interface ProjectCatalog { @@ -67,11 +79,11 @@ export interface ProjectCatalog { resolveHistoricalPath(path: string, usedAt?: number): Promise; select(projectId: string): Promise<{ project: ProjectRecord; path: string }>; touch(projectId: string, path?: string): Promise; - relink( + relink(projectId: string, path: string): Promise; + relinkWithSessions( projectId: string, path: string, - beforeCommit?: (context: ProjectRelinkContext) => Promise, - ): Promise; + ): Promise<{ project: ProjectRecord; updatedSessionIds: readonly string[] }>; rename(projectId: string, name: string): Promise; archive(projectId: string): Promise; restore(projectId: string): Promise; @@ -105,6 +117,7 @@ export function createProjectCatalog( createId?: () => string; /** Report a `projects.json` that could not be imported; the catalog still opens. */ onLegacyImportFailure?: (error: unknown) => void; + relinkFailpoint?: (stage: 'after_session_updates') => void; } = {}, ): ProjectCatalog { return new SqliteProjectCatalog( @@ -113,6 +126,7 @@ export function createProjectCatalog( deps.now ?? Date.now, deps.createId ?? randomUUID, deps.onLegacyImportFailure ?? (() => {}), + deps.relinkFailpoint, ); } @@ -137,6 +151,7 @@ class SqliteProjectCatalog implements ProjectCatalog { private readonly now: () => number, private readonly createId: () => string, private readonly onLegacyImportFailure: (error: unknown) => void, + private readonly relinkFailpoint?: (stage: 'after_session_updates') => void, ) {} close(): void { @@ -231,7 +246,7 @@ class SqliteProjectCatalog implements ProjectCatalog { // Probing the filesystem cannot happen inside the write transaction, so // availability is decided first and the choice is re-validated under it. const existing = findProjectById((await this.read()).projects, projectId); - if (!existing) throw new Error(`No such project: ${projectId}`); + if (!existing) throw new ProjectNotFoundError(projectId); const availablePaths = new Set( ( await Promise.all( @@ -243,14 +258,14 @@ class SqliteProjectCatalog implements ProjectCatalog { ); [selected, selectedPath] = await this.mutate((file) => { const project = findProjectById(file.projects, projectId); - if (!project) throw new Error(`No such project: ${projectId}`); + if (!project) throw new ProjectNotFoundError(projectId); if (project.archivedAt !== undefined) { - throw new Error(`Project is archived: ${projectId}`); + throw new ProjectArchivedError(projectId); } const location = project.locations .filter((item) => availablePaths.has(item.path)) .sort((a, b) => b.lastUsedAt - a.lastUsedAt || a.path.localeCompare(b.path))[0]; - if (!location) throw new Error(`Project is unavailable: ${projectId}`); + if (!location) throw new ProjectUnavailableError(projectId); const timestamp = this.now(); location.lastUsedAt = timestamp; project.lastUsedAt = timestamp; @@ -270,7 +285,7 @@ class SqliteProjectCatalog implements ProjectCatalog { : undefined; const touched = await this.mutate((file) => { const project = findProjectById(file.projects, projectId); - if (!project) throw new Error(`No such project: ${projectId}`); + if (!project) throw new ProjectNotFoundError(projectId); const location = resolvedPath ? project.locations.find((item) => item.path === resolvedPath) : [...project.locations].sort( @@ -287,78 +302,59 @@ class SqliteProjectCatalog implements ProjectCatalog { return this.present(touched); } - async relink( + async relink(projectId: string, path: string): Promise { + const resolved = await resolveProjectLocation({ path }); + const timestamp = this.now(); + const locationPath = + resolved.kind === 'git' ? resolved.git!.worktreeRoot : resolved.canonicalPath; + const relinked = await this.mutate((file) => { + const project = findProjectById(file.projects, projectId); + if (!project) throw new ProjectNotFoundError(projectId); + const conflict = file.projects.find( + (item) => item.id !== project.id && item.identity === resolved.identity, + ); + if (conflict) throw new ProjectPathConflictError(conflict.id); + return applyRelink(file, project, undefined, resolved, locationPath, timestamp); + }); + return this.present(relinked); + } + + async relinkWithSessions( projectId: string, path: string, - beforeCommit?: (context: ProjectRelinkContext) => Promise, - ): Promise { + ): Promise<{ project: ProjectRecord; updatedSessionIds: readonly string[] }> { const resolved = await resolveProjectLocation({ path }); const timestamp = this.now(); - let relinked: PersistedProject | undefined; const locationPath = resolved.kind === 'git' ? resolved.git!.worktreeRoot : resolved.canonicalPath; + let committed: + | { readonly project: PersistedProject; readonly updatedSessionIds: readonly string[] } + | undefined; await this.withQueue(async () => { - // `beforeCommit` reassigns the sessions of the project being merged away, - // so it cannot run inside the write transaction. It is shown the state it - // will act on, and the commit below refuses to proceed if that state no - // longer holds: re-deriving instead would leave the catalog consistent - // while the sessions the callback already moved point at the wrong owner. - const preview = await this.read(); - const previewProject = findProjectById(preview.projects, projectId); - if (!previewProject) throw new Error(`No such project: ${projectId}`); - const previewConflict = preview.projects.find( - (item) => item.id !== previewProject.id && item.identity === resolved.identity, - ); - if (previewConflict && !beforeCommit) { - throw new Error(`Project path already belongs to project: ${previewConflict.id}`); - } - await beforeCommit?.({ - projectId: previewProject.id, - projectAliases: [...(previewProject.aliases ?? [])], - destinationPath: locationPath, - previousLocations: previewProject.locations.map((location) => ({ ...location })), - ...(previewConflict - ? { - conflictingProjectId: previewConflict.id, - conflictingProjectAliases: [...(previewConflict.aliases ?? [])], - } - : {}), - }); - relinked = await this.mutate((file) => { + await this.importLegacyCatalogOnce(); + committed = this.lease.transaction('write', () => { + const file = this.selectCatalog(); const project = findProjectById(file.projects, projectId); - if (!project) throw new Error(`No such project: ${projectId}`); + if (!project) throw new ProjectNotFoundError(projectId); const conflict = file.projects.find( (item) => item.id !== project.id && item.identity === resolved.identity, ); - if (conflict && !beforeCommit) { - throw new Error(`Project path already belongs to project: ${conflict.id}`); - } - if (conflict?.id !== previewConflict?.id) { - throw new ProjectRelinkContentionError(projectId); - } - if (conflict) { - project.aliases = [ - ...new Set([...(project.aliases ?? []), conflict.id, ...(conflict.aliases ?? [])]), - ]; - file.projects = file.projects.filter((item) => item.id !== conflict.id); - } - project.identity = resolved.identity; - project.locations = [ - { - path: locationPath, - isWorktree: resolved.git?.isWorktree ?? false, - lastUsedAt: timestamp, - }, - ...(conflict?.locations - .filter((location) => location.path !== locationPath) - .map((location) => ({ ...location })) ?? []), - ]; - project.lastUsedAt = Math.max(timestamp, conflict?.lastUsedAt ?? 0); - return project; + const context = relinkContext(project, conflict, locationPath); + const updatedSessionIds = reassignProjectSessions(this.lease, context, timestamp); + this.relinkFailpoint?.('after_session_updates'); + const relinked = applyRelink(file, project, conflict, resolved, locationPath, timestamp); + this.replaceCatalog(normalizeProjectCatalogFile(file)); + return { + project: relinked, + updatedSessionIds, + }; }); }); - if (!relinked) throw new Error(`Failed to relink project: ${projectId}`); - return this.present(relinked); + if (!committed) throw new Error(`Failed to relink project and Sessions: ${projectId}`); + return { + project: await this.present(committed.project), + updatedSessionIds: committed.updatedSessionIds, + }; } async rename(projectId: string, name: string): Promise { @@ -367,7 +363,7 @@ class SqliteProjectCatalog implements ProjectCatalog { return this.present( await this.mutate((file) => { const project = findProjectById(file.projects, projectId); - if (!project) throw new Error(`No such project: ${projectId}`); + if (!project) throw new ProjectNotFoundError(projectId); project.name = trimmed; return project; }), @@ -378,7 +374,7 @@ class SqliteProjectCatalog implements ProjectCatalog { return this.present( await this.mutate((file) => { const project = findProjectById(file.projects, projectId); - if (!project) throw new Error(`No such project: ${projectId}`); + if (!project) throw new ProjectNotFoundError(projectId); project.archivedAt = this.now(); return project; }), @@ -389,7 +385,7 @@ class SqliteProjectCatalog implements ProjectCatalog { return this.present( await this.mutate((file) => { const project = findProjectById(file.projects, projectId); - if (!project) throw new Error(`No such project: ${projectId}`); + if (!project) throw new ProjectNotFoundError(projectId); delete project.archivedAt; return project; }), @@ -668,6 +664,119 @@ function findProjectById( ); } +interface ProjectSessionReassignment { + readonly projectId: string; + readonly projectAliases: readonly string[]; + readonly destinationPath: string; + readonly previousLocations: readonly ProjectLocation[]; + readonly conflictingProjectId?: string; + readonly conflictingProjectAliases?: readonly string[]; +} + +function relinkContext( + project: PersistedProject, + conflict: PersistedProject | undefined, + destinationPath: string, +): ProjectSessionReassignment { + return { + projectId: project.id, + projectAliases: [...(project.aliases ?? [])], + destinationPath, + previousLocations: project.locations.map((location) => ({ ...location })), + ...(conflict + ? { + conflictingProjectId: conflict.id, + conflictingProjectAliases: [...(conflict.aliases ?? [])], + } + : {}), + }; +} + +function applyRelink( + file: ProjectCatalogFile, + project: PersistedProject, + conflict: PersistedProject | undefined, + resolved: ResolvedProjectLocation, + locationPath: string, + timestamp: number, +): PersistedProject { + if (conflict) { + project.aliases = [ + ...new Set([...(project.aliases ?? []), conflict.id, ...(conflict.aliases ?? [])]), + ]; + file.projects = file.projects.filter((item) => item.id !== conflict.id); + } + project.identity = resolved.identity; + project.locations = [ + { + path: locationPath, + isWorktree: resolved.git?.isWorktree ?? false, + lastUsedAt: timestamp, + }, + ...(conflict?.locations + .filter((location) => location.path !== locationPath) + .map((location) => ({ ...location })) ?? []), + ]; + project.lastUsedAt = Math.max(timestamp, conflict?.lastUsedAt ?? 0); + return project; +} + +function reassignProjectSessions( + lease: OperationalStateDatabaseLease, + context: ProjectSessionReassignment, + committedAt: number, +): readonly string[] { + const survivingIds = new Set([context.projectId, ...context.projectAliases]); + const conflictingIds = new Set([ + ...(context.conflictingProjectId ? [context.conflictingProjectId] : []), + ...(context.conflictingProjectAliases ?? []), + ]); + const rows = lease.database + .prepare( + `SELECT session_id, payload_json, metadata_version + FROM session_metadata + ORDER BY session_id`, + ) + .all() as Array<{ + session_id: string; + payload_json: string; + metadata_version: number; + }>; + const update = lease.database.prepare( + `UPDATE session_metadata + SET payload_json = ?, metadata_version = ?, committed_at = ? + WHERE session_id = ? AND metadata_version = ?`, + ); + const updatedSessionIds: string[] = []; + for (const row of rows) { + const header = normalizeSessionHeader( + JSON.parse(row.payload_json) as SessionHeader, + row.session_id, + ); + let patch: Pick | undefined; + if (header.projectId && survivingIds.has(header.projectId)) { + patch = { cwd: context.destinationPath, projectId: context.projectId }; + } else if (header.projectId && conflictingIds.has(header.projectId)) { + patch = { cwd: header.cwd, projectId: context.projectId }; + } + if (!patch) continue; + const next = normalizeSessionHeader({ ...header, ...patch }, row.session_id); + const nextVersion = row.metadata_version + 1; + const result = update.run( + JSON.stringify(next), + nextVersion, + committedAt, + row.session_id, + row.metadata_version, + ); + if (result.changes !== 1) { + throw new Error(`Session metadata compare-and-set failed: ${row.session_id}`); + } + updatedSessionIds.push(row.session_id); + } + return updatedSessionIds; +} + function normalizeProjectLocation(value: unknown): PersistedProjectLocation { if (!value || typeof value !== 'object' || Array.isArray(value)) { throw new TypeError('Invalid project catalog.'); diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index 608d70d72c..1c1ca95f13 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -182,6 +182,8 @@ export interface SessionStore { } export interface SessionAuthorityStore extends SessionStore { + /** Complete one-time storage migrations before direct cross-domain transactions. */ + ready(): Promise; /** Atomically create a Session from already-converted Maka raw messages. */ createImportedSession( input: CreateSessionInput, @@ -343,6 +345,10 @@ class SqliteSessionStore implements SessionAuthorityStore { } } + ready(): Promise { + return this.ensureReady(); + } + async create( input: CreateSessionInput, initialBoundary?: ExecutionBoundary,