diff --git a/docs/development.md b/docs/development.md index 91acc48..ba62db5 100644 --- a/docs/development.md +++ b/docs/development.md @@ -1,6 +1,6 @@ # Development -Flow is an ESM TypeScript package built and tested with Git and the versions +Flow is an ESM TypeScript package built and tested with Git and versions pinned in `package.json`. ## Local setup @@ -24,8 +24,8 @@ bun run package:smoke ``` The opt-in real-host check launches the pinned `opencode-ai` package through -`bunx`. It therefore requires registry access or an already populated Bun -cache; it does not use a separately installed OpenCode binary: +`bunx`. It requires registry access or a populated Bun cache rather than a +separately installed OpenCode binary: ```bash bun run smoke:live @@ -33,27 +33,30 @@ bun run smoke:live ### Bumping the pinned host -`@opencode-ai/plugin` and `zod` are on the ignore list in -`.github/dependabot.yml`, so they are raised by hand. The plugin pin is also the -host version `smoke:live` launches, so bumping it is what puts a new host under -test: +`@opencode-ai/plugin` and `zod` are on `.github/dependabot.yml`'s ignore list, +so they are raised by hand. The plugin pin is also the host version +`smoke:live` launches, so bumping it puts a new host under test: 1. Move the `devDependencies` pin, and `peerDependencies` too if the new version falls outside the declared range. 2. Run `bun run check`, then `bun run smoke:live` for the real host. -Widen the peer range only for a host that has been smoke-tested; a range that -admits versions no check has launched is a compatibility claim with nothing -behind it. +Widen the peer range only for a smoke-tested host; a wider range makes a +compatibility claim no check has run. ## Source layout - `src/domain/` owns Session v5 values, invariants, and transitions that use only JavaScript/Node standard-library primitives. - `src/application/` owns use cases and repository ports. -- `src/infrastructure/` owns filesystem persistence and source fingerprinting. +- `src/infrastructure/` owns filesystem persistence and source fingerprinting: + `fs/workspace-paths.ts` (workspace root validation, `.flow` layout), + `fs/managed-fs.ts` (managed filesystem primitives), `fs/session-lock.ts` + (cross-process session lock), `fs/workspace.ts` (session file protocol). - `src/platform/opencode/` owns OpenCode hooks, host schemas, commands, tools, - validation capture, and the duplicate-runtime guard. + validation capture, and the duplicate-runtime guard: `command-hook.ts` + (slash-command hook), `tool-guard.ts` (leadership and auto-drive guard + around the tools), `plugin.ts` (wiring only). - `src/guidance/`, `skills/`, and prompt surfaces own concise workflow judgment. - `tests/` prove state-machine, persistence, platform, package, and host contracts. @@ -62,57 +65,55 @@ Dependencies point inward. Domain code does not import filesystem or host APIs; application code depends on domain; infrastructure implements application ports; the OpenCode platform composes the outer layers. -There is no distribution/activation subsystem, cache inventory, repair journal, -or Flow-owned installer. OpenCode installs and loads the npm package from its -native plugin command and normal plugin configuration. +There is no distribution/activation subsystem, cache inventory, repair +journal, or Flow-owned installer: OpenCode installs and loads the npm package +from its native plugin command and normal configuration. ## Change discipline -- Keep Session v5 as one canonical run aggregate. Derive status and progress - instead of adding parallel ledgers or cached counters. +- Keep Session v5 as one canonical run aggregate: derive status and progress + instead of parallel ledgers or cached counters. - Every mutation needs a revision guard and stable operation ID. Exact replay is safe; conflicting reuse fails. -- Only the reserved reviewer may create a new completion. While the Session v5 - workflow remains active, every caller receives an exact accepted completion - replay through a read-only path that does not cancel validation or write +- Only the reserved reviewer may create a new completion; while the Session v5 + workflow remains active, every other caller gets an exact accepted-completion + replay through a read-only path that neither cancels validation nor writes session state. -- Keep validation host-observed and session-native. Do not add caller-authored +- Keep validation host-observed and session-native: no caller-authored success, detached receipt stores, or clock requirements. -- Treat validation scope as a coverage claim. `broad` means the canonical - repository gate, byte for byte. Do not promote a narrow command by - relabeling it. -- Validation commands are persisted. Never inline secrets. Raw output is - intentionally reduced to completeness and a digest rather than stored or - projected. +- Treat validation scope as a coverage claim: `broad` means the canonical + repository gate, byte for byte, not a narrow command relabeled. +- Validation commands are persisted, never with inline secrets. Raw output is + reduced to completeness and a digest rather than stored or projected. - Keep one review per run. A final review requires broad validation and is not a second pass. The reviewer submits through `flow_feature_complete`; the manager never proxies its verdict. -- Prefer deletion when a test or document exists only for a removed concept. - Do not preserve a dual stack for pre-v6 active state. -- Use table-driven lifecycle and persistence tests. Avoid registries that test +- Prefer deletion when a test or document exists only for a removed concept, + not a dual stack for pre-v6 active state. +- Use table-driven lifecycle and persistence tests, not registries that test the presence of other tests. ## Documentation Update the README, maintainer contract, ADR, and changelog when a public -lifecycle or installation contract changes. Documentation must describe only -the current product; Git history owns superseded plans and experiments. +lifecycle or installation contract changes; documentation describes only the +current product, and Git history owns superseded plans and experiments. ## Model-driven wave evidence Deterministic CI validates schemas, permissions, prompts, and host integration -without provider credentials. It does not claim that a model actually overlaps -workers. Changes to wave behavior should therefore be exercised manually with a -real provider when available and accompanied by sanitized evidence of: +without provider credentials. It does not claim a model overlaps workers, so +changes to wave behavior should be exercised manually with a real provider +when available, with sanitized evidence of: - worker start/end times with a positive common overlap; - assigned versus changed paths and any scope drift; - permission prompts or denials and worker Bash calls; and - reviewer-owned `flow_feature_complete` submission. -Every wave-behavior change must include this evidence when marked verified, but -it is not a deterministic release gate. When a provider is unavailable, mark -the behavior unverified, record the review risk, and avoid performance or +Every wave-behavior change needs this evidence when marked verified, but it is +not a deterministic release gate. Without a provider, mark the behavior +unverified, record the review risk, and avoid performance or reliability claims. Do not persist prompts, secrets, raw provider payloads, or a wave ledger, and do not add provider credentials, a scheduler, or telemetry to CI. @@ -120,9 +121,9 @@ CI. ## Model-driven auto-continuation evidence Deterministic tests exercise the coordinator through the real plugin hooks and -the `promptAsync` client boundary. They do not prove how a configured model +the `promptAsync` client boundary, but do not prove how a configured model behaves after delivery. When auto-continuation behavior changes and a provider -is available, run one packed-plugin canary that records sanitized evidence of: +is available, run one packed-plugin canary recording sanitized evidence of: - idle `ready` delivery with the Flow token and compact revision; - recommendation or clarification at a checkpoint remaining waiting, followed @@ -144,27 +145,28 @@ deterministic hook and lifecycle gates. ## Release -Follow the [frozen-candidate sequence](release-qualification.md#running-it): finish -fixes and dependency updates, pass deterministic checks, then approve paid evals. -`bun run qualify -- --campaign-dir --canary ` seals the complete -two-provider campaign, exact-artifact canary and grader evidence. Commit that bundle -before tagging; never substitute interrupted results for qualification. +Follow the [frozen-candidate sequence](release-qualification.md#running-it): +finish fixes and dependency updates, pass deterministic checks, approve paid +evals. +`bun run qualify -- --campaign-dir --canary ` seals the +two-provider campaign, exact-artifact canary and grader evidence. Commit that +bundle before tagging; never substitute interrupted results for qualification. -Release tags use `v`. Blocking release checks include the -normal repository gate, package smoke, packed live OpenCode smoke, package -integrity generation, npm publication, and GitHub release assets. There is no -cross-version active-session gate because v6 is an explicit hard cutover. +Release tags use `v`. Blocking release checks: the normal +repository gate, package smoke, packed live OpenCode smoke, package integrity +generation, npm publication, and GitHub release assets. There is no +cross-version active-session gate; v6 is an explicit hard cutover. Publication accepts both annotated and lightweight tags, but the freshly fetched tag, workflow event, checkout, and current remote `main` tip must identify the -same commit immediately before npm publication. Network calls have explicit -deadlines. npm publication reconciles the immutable package integrity after every -result, including timeouts. GitHub publication first builds an exact draft under -the same ref proof. That draft is the recovery marker if npm succeeds and `main` -then advances. Finalization rechecks the remote tag, refuses conflicting metadata +same commit before npm publication. Network calls have explicit deadlines. npm +publication reconciles the immutable package integrity after every result, +including timeouts. GitHub publication first builds an exact draft under the +same ref proof. That draft recovers if npm succeeds and `main` then advances. +Finalization rechecks the remote tag, refuses conflicting metadata or assets, and publishes only after every asset digest matches. Reruns converge after partial success without replacing published bytes. Preparing an already-published release is read-only and requires exact assets. -Missing or pending assets fail preparation; use the explicit `github-publish` -recovery path with the original inputs and tag proof to restore a missing asset. +Missing or pending assets fail preparation; use the `github-publish` recovery +path with the original inputs and tag proof to restore a missing asset. diff --git a/src/infrastructure/fs/managed-fs.ts b/src/infrastructure/fs/managed-fs.ts new file mode 100644 index 0000000..ccdae16 --- /dev/null +++ b/src/infrastructure/fs/managed-fs.ts @@ -0,0 +1,162 @@ +import { randomUUID } from "node:crypto"; +import { constants } from "node:fs"; +import { lstat, mkdir, open, rename, rm, writeFile } from "node:fs/promises"; +import { dirname, join } from "node:path"; +import { setTimeout as sleep } from "node:timers/promises"; +import { UnreadableFlowSessionError } from "../../application/errors.js"; +import { + MAX_SESSION_BYTES, + SESSION_CLOSE_RESERVE_BYTES, +} from "../../domain/limits.js"; +import { flowDir, historyDir } from "./workspace-paths.js"; + +export class UnsafeFlowWorkspaceLayoutError extends Error { + readonly code = "UNSAFE_FLOW_WORKSPACE_LAYOUT"; +} + +export async function pathKind( + path: string, + expected: "file" | "directory", + description: string, +): Promise<"missing" | "present"> { + try { + const info = await lstat(path); + if (info.isSymbolicLink()) { + throw new UnsafeFlowWorkspaceLayoutError( + `Flow refuses a symbolic link for ${description}: ${path}.`, + ); + } + if ( + (expected === "file" && !info.isFile()) || + (expected === "directory" && !info.isDirectory()) + ) { + throw new UnsafeFlowWorkspaceLayoutError( + `Flow requires ${description} to be a ${expected}: ${path}.`, + ); + } + return "present"; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return "missing"; + throw error; + } +} + +async function ensureDirectory( + path: string, + description: string, +): Promise { + if ((await pathKind(path, "directory", description)) === "present") return; + try { + await mkdir(path, { mode: 0o700 }); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; + } + await pathKind(path, "directory", description); +} + +export async function ensureFlowDirectory(workspace: string): Promise { + const root = flowDir(workspace); + await ensureDirectory(root, "the Flow state directory"); + const ignore = join(root, ".gitignore"); + if ((await pathKind(ignore, "file", "the Flow ignore file")) === "missing") { + try { + await writeFile(ignore, "*\n", { + encoding: "utf8", + flag: "wx", + mode: 0o600, + }); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; + } + } +} + +export async function ensureHistoryDirectory(workspace: string): Promise { + await ensureFlowDirectory(workspace); + await ensureDirectory(historyDir(workspace), "the Flow history directory"); +} + +export async function readManaged( + path: string, + description: string, + synchronizeFile = false, +): Promise { + await pathKind(path, "file", description); + const noFollow = process.platform === "win32" ? 0 : constants.O_NOFOLLOW; + const access = synchronizeFile ? constants.O_RDWR : constants.O_RDONLY; + const handle = await open(path, access | noFollow); + try { + const stat = await handle.stat(); + if ( + !stat.isFile() || + stat.size > MAX_SESSION_BYTES + SESSION_CLOSE_RESERVE_BYTES + ) { + throw new UnreadableFlowSessionError( + `${description} is not a bounded regular file.`, + "state exceeds the supported session size", + ); + } + const contents = await handle.readFile("utf8"); + if (synchronizeFile) await handle.sync(); + return contents; + } finally { + await handle.close(); + } +} + +export async function syncDirectory(path: string): Promise { + if (process.platform === "win32") return; + const handle = await open(path, "r"); + try { + await handle.sync(); + } finally { + await handle.close(); + } +} + +async function renameReplacing(temporary: string, path: string): Promise { + let retry = 0; + while (true) { + try { + await rename(temporary, path); + return; + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + const transientWindowsError = + process.platform === "win32" && + (code === "EACCES" || code === "EBUSY" || code === "EPERM"); + if (!transientWindowsError || retry >= 20) throw error; + retry += 1; + // Preserve atomic replacement: wait for short-lived readers instead of + // unlinking the destination and exposing missing or partial state. + await sleep(retry * 5); + } + } +} + +export async function writeAtomically( + path: string, + contents: string, +): Promise { + const temporary = join( + dirname(path), + `.flow-write-${process.pid}-${randomUUID()}.tmp`, + ); + const handle = await open(temporary, "wx", 0o600); + try { + await handle.writeFile(contents, "utf8"); + await handle.sync(); + } catch (error) { + await handle.close(); + await rm(temporary, { force: true }); + throw error; + } + await handle.close(); + try { + await renameReplacing(temporary, path); + await syncDirectory(dirname(path)); + } catch (error) { + await rm(temporary, { force: true }); + throw error; + } +} diff --git a/src/infrastructure/fs/session-lock.ts b/src/infrastructure/fs/session-lock.ts new file mode 100644 index 0000000..6874f55 --- /dev/null +++ b/src/infrastructure/fs/session-lock.ts @@ -0,0 +1,132 @@ +import { randomUUID } from "node:crypto"; +import { mkdir, readFile, rm, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { setTimeout as sleep } from "node:timers/promises"; +import { ensureFlowDirectory, pathKind } from "./managed-fs.js"; +import { assertMutableWorkspaceRoot, flowDir } from "./workspace-paths.js"; + +const inProcessLocks = new Map>(); +const LOCK_TIMEOUT_MS = 30_000; + +async function orphanOwnerToken(lock: string): Promise { + try { + const owner = JSON.parse( + await readFile(join(lock, "owner.json"), "utf8"), + ) as { + token?: unknown; + pid?: unknown; + }; + if (typeof owner.token !== "string" || owner.token.length === 0) + return null; + const pid = owner.pid; + if (typeof pid !== "number" || !Number.isInteger(pid) || pid < 1) + return null; + try { + process.kill(pid, 0); + return null; + } catch (error) { + return (error as NodeJS.ErrnoException).code === "ESRCH" + ? owner.token + : null; + } + } catch { + return null; + } +} + +/** + * wx-create `claim` inside the lock. That binds the claim to this directory + * inode, so a live replacement is never moved off the canonical path. + * Re-check the owner token before deleting; a mismatch drops the claim file. + */ +export async function reclaimOrphanedLock(lock: string): Promise { + const token = await orphanOwnerToken(lock); + if (token === null) return false; + const claim = join(lock, "claim"); + try { + await writeFile(claim, "", { encoding: "utf8", flag: "wx", mode: 0o600 }); + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code === "EEXIST" || code === "ENOENT") return false; + throw error; + } + if ((await orphanOwnerToken(lock)) !== token) { + try { + await rm(claim); + } catch { + // Directory was replaced; the claim went with it. + } + return false; + } + await rm(lock, { recursive: true, force: true }); + return true; +} + +async function acquireLock(workspace: string): Promise<() => Promise> { + await ensureFlowDirectory(workspace); + const lock = join(flowDir(workspace), "session.lock"); + const started = Date.now(); + while (true) { + try { + await mkdir(lock, { mode: 0o700 }); + const token = randomUUID(); + try { + await writeFile( + join(lock, "owner.json"), + JSON.stringify({ token, pid: process.pid }), + { encoding: "utf8", flag: "wx", mode: 0o600 }, + ); + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code !== "EEXIST" && code !== "ENOENT") { + await rm(lock, { recursive: true, force: true }); + } + throw error; + } + return async () => { + try { + const owner = JSON.parse( + await readFile(join(lock, "owner.json"), "utf8"), + ) as { token?: unknown }; + if (owner.token === token) await rm(lock, { recursive: true }); + } catch { + // A replaced or damaged lock is not ours to remove. + } + }; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; + await pathKind(lock, "directory", "the Flow session lock"); + if (await reclaimOrphanedLock(lock)) continue; + if (Date.now() - started >= LOCK_TIMEOUT_MS) { + throw new Error( + `Timed out waiting for Flow session lock at ${lock}; inspect it before manual removal.`, + ); + } + await sleep(25); + } + } +} + +export async function withSessionLock( + workspace: string, + task: () => Promise, +): Promise { + const root = assertMutableWorkspaceRoot(workspace); + const previous = inProcessLocks.get(root) ?? Promise.resolve(); + let releaseQueue = () => {}; + const current = new Promise((resolveQueue) => { + releaseQueue = resolveQueue; + }); + const queued = previous.catch(() => undefined).then(() => current); + inProcessLocks.set(root, queued); + let releaseFile: (() => Promise) | null = null; + try { + await previous.catch(() => undefined); + releaseFile = await acquireLock(root); + return await task(); + } finally { + await releaseFile?.(); + releaseQueue(); + if (inProcessLocks.get(root) === queued) inProcessLocks.delete(root); + } +} diff --git a/src/infrastructure/fs/session-repository.ts b/src/infrastructure/fs/session-repository.ts index 3af4b0c..c3abeb3 100644 --- a/src/infrastructure/fs/session-repository.ts +++ b/src/infrastructure/fs/session-repository.ts @@ -1,15 +1,15 @@ import type { SessionRepository } from "../../application/ports/session-repository.js"; +import { withSessionLock } from "./session-lock.js"; import { createFileSourceIdentityProvider } from "./source-identity.js"; import { archiveAndClearSession, - assertMutableWorkspaceRoot, confirmActiveSessionDurability, loadArchivedSession, loadSession, quarantineUnreadableSession, saveSession, - withSessionLock, } from "./workspace.js"; +import { assertMutableWorkspaceRoot } from "./workspace-paths.js"; export function createFileSessionRepository( workspace: string, diff --git a/src/infrastructure/fs/source-identity.ts b/src/infrastructure/fs/source-identity.ts index 59993d2..3f02369 100644 --- a/src/infrastructure/fs/source-identity.ts +++ b/src/infrastructure/fs/source-identity.ts @@ -10,7 +10,7 @@ import { MAX_SOURCE_TOTAL_BYTES, } from "../../domain/limits.js"; import type { SourceDigest } from "../../domain/session.js"; -import { assertMutableWorkspaceRoot } from "./workspace.js"; +import { assertMutableWorkspaceRoot } from "./workspace-paths.js"; export class SourceIdentityError extends Error { readonly code = "FLOW_SOURCE_IDENTITY"; diff --git a/src/infrastructure/fs/workspace-paths.ts b/src/infrastructure/fs/workspace-paths.ts new file mode 100644 index 0000000..1c1bb94 --- /dev/null +++ b/src/infrastructure/fs/workspace-paths.ts @@ -0,0 +1,95 @@ +import { createHash } from "node:crypto"; +import { lstatSync, realpathSync } from "node:fs"; +import { homedir } from "node:os"; +import { join, parse, resolve } from "node:path"; +import { MAX_SESSION_ID_LENGTH } from "../../domain/limits.js"; + +class InvalidFlowWorkspaceRootError extends Error { + readonly code = "INVALID_FLOW_WORKSPACE_ROOT"; +} + +function normalizeWorkspaceRoot(rawPath: string | undefined): string | null { + const value = rawPath?.trim(); + if (!value) return null; + const normalized = resolve(value); + return parse(normalized).root === normalized ? null : normalized; +} + +export function assertMutableWorkspaceRoot(rawPath: string): string { + const candidate = normalizeWorkspaceRoot(rawPath); + if (!candidate) { + throw new InvalidFlowWorkspaceRootError( + "Flow requires a non-root workspace path.", + ); + } + let root: string; + try { + root = realpathSync(candidate); + } catch (error) { + throw new InvalidFlowWorkspaceRootError( + `Flow requires an existing workspace directory: ${candidate}.`, + { cause: error }, + ); + } + if (parse(root).root === root || !lstatSync(root).isDirectory()) { + throw new InvalidFlowWorkspaceRootError( + "Flow requires an existing non-root workspace directory.", + ); + } + const homes = [process.env.HOME, homedir()] + .filter((value): value is string => Boolean(value?.trim())) + .map((value) => { + try { + return realpathSync(resolve(value)); + } catch { + return resolve(value); + } + }); + if (homes.includes(root)) { + throw new InvalidFlowWorkspaceRootError( + "Flow refuses to use the home directory itself as mutable state.", + ); + } + return root; +} + +export function resolveWorkspaceRoot(context: { + worktree?: string | undefined; + directory?: string | undefined; +}): string { + const candidate = + normalizeWorkspaceRoot(context.worktree) ?? + normalizeWorkspaceRoot(context.directory); + if (!candidate) { + throw new InvalidFlowWorkspaceRootError( + "Flow could not resolve a workspace root from tool context.", + ); + } + return assertMutableWorkspaceRoot(candidate); +} + +export function flowDir(workspace: string): string { + return join(workspace, ".flow"); +} + +export function sessionPath(workspace: string): string { + return join(flowDir(workspace), "session.json"); +} + +export function historyDir(workspace: string): string { + return join(flowDir(workspace), "history"); +} + +function archivedSessionFilename(sessionId: string): string { + if (sessionId.length < 1 || sessionId.length > MAX_SESSION_ID_LENGTH) { + throw new Error("Invalid session id."); + } + return `${createHash("sha256").update(sessionId).digest("hex")}.json`; +} + +export function archivedSessionPath( + workspace: string, + sessionId: string, +): string { + return join(historyDir(workspace), archivedSessionFilename(sessionId)); +} diff --git a/src/infrastructure/fs/workspace.ts b/src/infrastructure/fs/workspace.ts index 920aa94..e88bb0e 100644 --- a/src/infrastructure/fs/workspace.ts +++ b/src/infrastructure/fs/workspace.ts @@ -1,19 +1,6 @@ -import { createHash, randomUUID } from "node:crypto"; -import { constants, lstatSync, realpathSync } from "node:fs"; -import { - link, - lstat, - mkdir, - open, - readFile, - rename, - rm, - unlink, - writeFile, -} from "node:fs/promises"; -import { homedir } from "node:os"; -import { dirname, join, parse, resolve } from "node:path"; -import { setTimeout as sleep } from "node:timers/promises"; +import { randomUUID } from "node:crypto"; +import { link, open, rename, rm, unlink } from "node:fs/promises"; +import { join } from "node:path"; import { ArchiveCollisionError, UnreadableFlowSessionError, @@ -22,23 +9,30 @@ import { import { SessionSchema } from "../../application/schema.js"; import { MAX_SESSION_BYTES, - MAX_SESSION_ID_LENGTH, SESSION_CLOSE_RESERVE_BYTES, } from "../../domain/limits.js"; import { sameSession } from "../../domain/operation.js"; import type { Session } from "../../domain/session.js"; +import { + ensureFlowDirectory, + ensureHistoryDirectory, + pathKind, + readManaged, + syncDirectory, + UnsafeFlowWorkspaceLayoutError, + writeAtomically, +} from "./managed-fs.js"; import { parseStrictJsonObject } from "./strict-json-object.js"; +import { + archivedSessionPath, + assertMutableWorkspaceRoot, + flowDir, + historyDir, + sessionPath, +} from "./workspace-paths.js"; export { ArchiveCollisionError } from "../../application/errors.js"; -class InvalidFlowWorkspaceRootError extends Error { - readonly code = "INVALID_FLOW_WORKSPACE_ROOT"; -} - -export class UnsafeFlowWorkspaceLayoutError extends Error { - readonly code = "UNSAFE_FLOW_WORKSPACE_LAYOUT"; -} - function provesManagedStateCollision(error: unknown): boolean { return ( error instanceof ArchiveCollisionError || @@ -48,240 +42,10 @@ function provesManagedStateCollision(error: unknown): boolean { ); } -function normalizeWorkspaceRoot(rawPath: string | undefined): string | null { - const value = rawPath?.trim(); - if (!value) return null; - const normalized = resolve(value); - return parse(normalized).root === normalized ? null : normalized; -} - -export function assertMutableWorkspaceRoot(rawPath: string): string { - const candidate = normalizeWorkspaceRoot(rawPath); - if (!candidate) { - throw new InvalidFlowWorkspaceRootError( - "Flow requires a non-root workspace path.", - ); - } - let root: string; - try { - root = realpathSync(candidate); - } catch (error) { - throw new InvalidFlowWorkspaceRootError( - `Flow requires an existing workspace directory: ${candidate}.`, - { cause: error }, - ); - } - if (parse(root).root === root || !lstatSync(root).isDirectory()) { - throw new InvalidFlowWorkspaceRootError( - "Flow requires an existing non-root workspace directory.", - ); - } - const homes = [process.env.HOME, homedir()] - .filter((value): value is string => Boolean(value?.trim())) - .map((value) => { - try { - return realpathSync(resolve(value)); - } catch { - return resolve(value); - } - }); - if (homes.includes(root)) { - throw new InvalidFlowWorkspaceRootError( - "Flow refuses to use the home directory itself as mutable state.", - ); - } - return root; -} - -export function resolveWorkspaceRoot(context: { - worktree?: string | undefined; - directory?: string | undefined; -}): string { - const candidate = - normalizeWorkspaceRoot(context.worktree) ?? - normalizeWorkspaceRoot(context.directory); - if (!candidate) { - throw new InvalidFlowWorkspaceRootError( - "Flow could not resolve a workspace root from tool context.", - ); - } - return assertMutableWorkspaceRoot(candidate); -} - -export function flowDir(workspace: string): string { - return join(workspace, ".flow"); -} - -export function sessionPath(workspace: string): string { - return join(flowDir(workspace), "session.json"); -} - -export function historyDir(workspace: string): string { - return join(flowDir(workspace), "history"); -} - -function archivedSessionFilename(sessionId: string): string { - if (sessionId.length < 1 || sessionId.length > MAX_SESSION_ID_LENGTH) { - throw new Error("Invalid session id."); - } - return `${createHash("sha256").update(sessionId).digest("hex")}.json`; -} - -export function archivedSessionPath( - workspace: string, - sessionId: string, -): string { - return join(historyDir(workspace), archivedSessionFilename(sessionId)); -} - -async function pathKind( - path: string, - expected: "file" | "directory", - description: string, -): Promise<"missing" | "present"> { - try { - const info = await lstat(path); - if (info.isSymbolicLink()) { - throw new UnsafeFlowWorkspaceLayoutError( - `Flow refuses a symbolic link for ${description}: ${path}.`, - ); - } - if ( - (expected === "file" && !info.isFile()) || - (expected === "directory" && !info.isDirectory()) - ) { - throw new UnsafeFlowWorkspaceLayoutError( - `Flow requires ${description} to be a ${expected}: ${path}.`, - ); - } - return "present"; - } catch (error) { - if ((error as NodeJS.ErrnoException).code === "ENOENT") return "missing"; - throw error; - } -} - -async function ensureDirectory( - path: string, - description: string, -): Promise { - if ((await pathKind(path, "directory", description)) === "present") return; - try { - await mkdir(path, { mode: 0o700 }); - } catch (error) { - if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; - } - await pathKind(path, "directory", description); -} - -async function ensureFlowDirectory(workspace: string): Promise { - const root = flowDir(workspace); - await ensureDirectory(root, "the Flow state directory"); - const ignore = join(root, ".gitignore"); - if ((await pathKind(ignore, "file", "the Flow ignore file")) === "missing") { - try { - await writeFile(ignore, "*\n", { - encoding: "utf8", - flag: "wx", - mode: 0o600, - }); - } catch (error) { - if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; - } - } -} - -async function ensureHistoryDirectory(workspace: string): Promise { - await ensureFlowDirectory(workspace); - await ensureDirectory(historyDir(workspace), "the Flow history directory"); -} - -async function readManaged( - path: string, - description: string, - synchronizeFile = false, -): Promise { - await pathKind(path, "file", description); - const noFollow = process.platform === "win32" ? 0 : constants.O_NOFOLLOW; - const access = synchronizeFile ? constants.O_RDWR : constants.O_RDONLY; - const handle = await open(path, access | noFollow); - try { - const stat = await handle.stat(); - if ( - !stat.isFile() || - stat.size > MAX_SESSION_BYTES + SESSION_CLOSE_RESERVE_BYTES - ) { - throw new UnreadableFlowSessionError( - `${description} is not a bounded regular file.`, - "state exceeds the supported session size", - ); - } - const contents = await handle.readFile("utf8"); - if (synchronizeFile) await handle.sync(); - return contents; - } finally { - await handle.close(); - } -} - -async function syncDirectory(path: string): Promise { - if (process.platform === "win32") return; - const handle = await open(path, "r"); - try { - await handle.sync(); - } finally { - await handle.close(); - } -} - type WorkspacePersistenceOptions = Readonly<{ synchronizeDirectory?: (path: string) => Promise; }>; -async function renameReplacing(temporary: string, path: string): Promise { - let retry = 0; - while (true) { - try { - await rename(temporary, path); - return; - } catch (error) { - const code = (error as NodeJS.ErrnoException).code; - const transientWindowsError = - process.platform === "win32" && - (code === "EACCES" || code === "EBUSY" || code === "EPERM"); - if (!transientWindowsError || retry >= 20) throw error; - retry += 1; - // Preserve atomic replacement: wait for short-lived readers instead of - // unlinking the destination and exposing missing or partial state. - await sleep(retry * 5); - } - } -} - -async function writeAtomically(path: string, contents: string): Promise { - const temporary = join( - dirname(path), - `.flow-write-${process.pid}-${randomUUID()}.tmp`, - ); - const handle = await open(temporary, "wx", 0o600); - try { - await handle.writeFile(contents, "utf8"); - await handle.sync(); - } catch (error) { - await handle.close(); - await rm(temporary, { force: true }); - throw error; - } - await handle.close(); - try { - await renameReplacing(temporary, path); - await syncDirectory(dirname(path)); - } catch (error) { - await rm(temporary, { force: true }); - throw error; - } -} - function parseSession(raw: string, description: string): Session { const parsed = parseStrictJsonObject(raw, description); if (!parsed.ok) { @@ -533,129 +297,3 @@ export async function quarantineUnreadableSession( await syncDirectory(historyDir(root)); return target; } - -const inProcessLocks = new Map>(); -const LOCK_TIMEOUT_MS = 30_000; - -async function orphanOwnerToken(lock: string): Promise { - try { - const owner = JSON.parse( - await readFile(join(lock, "owner.json"), "utf8"), - ) as { - token?: unknown; - pid?: unknown; - }; - if (typeof owner.token !== "string" || owner.token.length === 0) - return null; - const pid = owner.pid; - if (typeof pid !== "number" || !Number.isInteger(pid) || pid < 1) - return null; - try { - process.kill(pid, 0); - return null; - } catch (error) { - return (error as NodeJS.ErrnoException).code === "ESRCH" - ? owner.token - : null; - } - } catch { - return null; - } -} - -/** - * wx-create `claim` inside the lock. That binds the claim to this directory - * inode, so a live replacement is never moved off the canonical path. - * Re-check the owner token before deleting; a mismatch drops the claim file. - */ -export async function reclaimOrphanedLock(lock: string): Promise { - const token = await orphanOwnerToken(lock); - if (token === null) return false; - const claim = join(lock, "claim"); - try { - await writeFile(claim, "", { encoding: "utf8", flag: "wx", mode: 0o600 }); - } catch (error) { - const code = (error as NodeJS.ErrnoException).code; - if (code === "EEXIST" || code === "ENOENT") return false; - throw error; - } - if ((await orphanOwnerToken(lock)) !== token) { - try { - await rm(claim); - } catch { - // Directory was replaced; the claim went with it. - } - return false; - } - await rm(lock, { recursive: true, force: true }); - return true; -} - -async function acquireLock(workspace: string): Promise<() => Promise> { - await ensureFlowDirectory(workspace); - const lock = join(flowDir(workspace), "session.lock"); - const started = Date.now(); - while (true) { - try { - await mkdir(lock, { mode: 0o700 }); - const token = randomUUID(); - try { - await writeFile( - join(lock, "owner.json"), - JSON.stringify({ token, pid: process.pid }), - { encoding: "utf8", flag: "wx", mode: 0o600 }, - ); - } catch (error) { - const code = (error as NodeJS.ErrnoException).code; - if (code !== "EEXIST" && code !== "ENOENT") { - await rm(lock, { recursive: true, force: true }); - } - throw error; - } - return async () => { - try { - const owner = JSON.parse( - await readFile(join(lock, "owner.json"), "utf8"), - ) as { token?: unknown }; - if (owner.token === token) await rm(lock, { recursive: true }); - } catch { - // A replaced or damaged lock is not ours to remove. - } - }; - } catch (error) { - if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; - await pathKind(lock, "directory", "the Flow session lock"); - if (await reclaimOrphanedLock(lock)) continue; - if (Date.now() - started >= LOCK_TIMEOUT_MS) { - throw new Error( - `Timed out waiting for Flow session lock at ${lock}; inspect it before manual removal.`, - ); - } - await sleep(25); - } - } -} - -export async function withSessionLock( - workspace: string, - task: () => Promise, -): Promise { - const root = assertMutableWorkspaceRoot(workspace); - const previous = inProcessLocks.get(root) ?? Promise.resolve(); - let releaseQueue = () => {}; - const current = new Promise((resolveQueue) => { - releaseQueue = resolveQueue; - }); - const queued = previous.catch(() => undefined).then(() => current); - inProcessLocks.set(root, queued); - let releaseFile: (() => Promise) | null = null; - try { - await previous.catch(() => undefined); - releaseFile = await acquireLock(root); - return await task(); - } finally { - await releaseFile?.(); - releaseQueue(); - if (inProcessLocks.get(root) === queued) inProcessLocks.delete(root); - } -} diff --git a/src/platform/opencode/command-hook.ts b/src/platform/opencode/command-hook.ts new file mode 100644 index 0000000..e884eda --- /dev/null +++ b/src/platform/opencode/command-hook.ts @@ -0,0 +1,134 @@ +import type { FlowService } from "../../application/flow-service.js"; +import { FLOW_CORE_COMMANDS } from "../../config-shared.js"; +import { requestEvidenceAnchor } from "../../domain/request-evidence.js"; +import type { AutoDriveCoordinator } from "./auto-drive.js"; +import type { Hooks } from "./sdk.js"; + +type FlowCommandName = keyof typeof FLOW_CORE_COMMANDS; +type CommandHook = NonNullable; +type CommandOutput = Parameters[1]; +type Part = CommandOutput["parts"][number]; +type TextPart = Extract; +// Host assigns id, sessionID and messageID after the command hook returns. +type DraftTextPart = Omit; +const AUTO_STOPPED = "Flow auto stopped."; +function isFlowCommand(command: string): command is FlowCommandName { + return Object.hasOwn(FLOW_CORE_COMMANDS, command); +} +export function textPart( + text: string, + synthetic = false, + metadata?: Readonly>, +): DraftTextPart { + return { + type: "text", + text, + ...(synthetic ? { synthetic: true } : {}), + ...(metadata ? { metadata } : {}), + }; +} +/** + * The command hook's parts are typed with the identity the host assigns after + * the hook returns, so a part written here is a draft at runtime. This is the + * one place a draft crosses into the host's array. + */ +function asHostTextPart(part: DraftTextPart): TextPart { + return part as TextPart; +} +function rewriteCommand( + command: FlowCommandName, + args: string, + output: CommandOutput, +): void { + const config = FLOW_CORE_COMMANDS[command]; + const promptArgs = config.subtask + ? args + : "the preceding non-synthetic Flow request"; + const prompt = config.template.split("$ARGUMENTS").join(promptArgs); + if (!config.subtask) { + if (output.parts.some((part) => part.type === "subtask")) + throw new Error("Flow manager commands cannot contain subtask parts."); + const preserved = output.parts.filter((part) => part.type !== "text"); + output.parts.splice( + 0, + output.parts.length, + asHostTextPart( + textPart(args.trim() ? `Flow ${command}: ${args}` : `Flow ${command}`), + ), + asHostTextPart(textPart(prompt, true)), + ...preserved, + ); + return; + } + const part = output.parts[0]; + if (output.parts.length !== 1 || part?.type !== "subtask") + throw new Error(`/${command} requires exactly one reviewer subtask.`); + if (part.agent !== config.agent) + throw new Error(`/${command} must dispatch to '${config.agent}'.`); + // The host's subtask type does not declare `command`, but a command-dispatched + // subtask carries it at runtime, so its presence is checked, not asserted. + const declared = "command" in part ? part.command : undefined; + if (typeof declared !== "string" || declared.replace(/^\/+/, "") !== command) + throw new Error(`/${command} subtask identity did not match.`); + part.prompt = prompt; +} +export function createCommandHook( + options: Readonly<{ + assertOperational: (action: string) => void; + autoDrive: AutoDriveCoordinator; + flow: FlowService; + }>, +): CommandHook { + const { assertOperational, autoDrive, flow } = options; + return async (input, output) => { + const command = input.command.replace(/^\/+/, ""); + if (!isFlowCommand(command)) return; + const action = input.arguments.trim(); + if (command === "flow-auto" && /^(?:stop|cancel)$/i.test(action)) { + const confirmed = output.parts.some( + (part) => part.type === "text" && part.text === AUTO_STOPPED, + ); + const response = + autoDrive.deactivate(input.sessionID) || confirmed + ? AUTO_STOPPED + : "No Flow auto lease was active in this OpenCode session."; + output.parts[0] = asHostTextPart(textPart(response)); + output.parts.length = 1; + return; + } + assertOperational(`execute /${command}`); + if (command === "flow-auto" || command === "flow-plan") { + const evidence = requestEvidenceAnchor(input.arguments, input.sessionID); + if (evidence) { + await flow.status({ request: { view: "compact" } }); + await flow.requestAnchor({ goal: input.arguments, evidence }); + } + } + rewriteCommand(command, input.arguments, output); + if (command !== "flow-auto") + return void autoDrive.deactivate(input.sessionID); + const metadata = await autoDrive.activate(input.sessionID); + // Preflight, not a gate. The lifecycle works either way; what changes is + // whether the user is told up front that this host cannot carry the + // continuation, instead of watching Flow stop after every feature and + // guessing which of the two it is. + if (autoDrive.continuationSupport() === "unsupported") { + output.parts.unshift( + asHostTextPart( + textPart( + "Note: this OpenCode host does not report assistant message parentage, so Flow cannot continue automatically between features here. Each feature still runs normally; drive the next one with /flow-run.", + ), + ), + ); + } + const instruction = output.parts.find( + (part): part is TextPart => + part.type === "text" && part.synthetic === true, + ); + if (!instruction) { + autoDrive.deactivate(input.sessionID); + throw new Error("/flow-auto is missing its synthetic instruction."); + } + instruction.metadata = { ...instruction.metadata, ...metadata }; + }; +} diff --git a/src/platform/opencode/plugin.ts b/src/platform/opencode/plugin.ts index 2913efd..4c40a04 100644 --- a/src/platform/opencode/plugin.ts +++ b/src/platform/opencode/plugin.ts @@ -1,14 +1,12 @@ import { createHash } from "node:crypto"; import { readFile } from "node:fs/promises"; import { fileURLToPath } from "node:url"; -import { dataNote } from "../../application/flow-response.js"; import { - FLOW_CORE_COMMANDS, type FlowCodingModel, resolveFlowReviewerConfiguration, } from "../../config-shared.js"; -import { requestEvidenceAnchor } from "../../domain/request-evidence.js"; import { createWorkspaceFlowService } from "../../infrastructure/fs/workspace-flow-service.js"; +import { resolveWorkspaceRoot } from "../../infrastructure/fs/workspace-paths.js"; import { persistWorkspaceValidation, prepareWorkspaceValidation, @@ -16,253 +14,19 @@ import { } from "../../infrastructure/fs/workspace-validation.js"; import { resolveFlowPluginVersion } from "../../version.js"; import { AutoDriveCoordinator, autoDriveDelivery } from "./auto-drive.js"; +import { createCommandHook, textPart } from "./command-hook.js"; import { createConfigHook } from "./config.js"; import { createFlowPluginInstanceId, FLOW_LEADERSHIP_PROTOCOL_VERSION, - type FlowLeadershipHandle, - type FlowLeadershipReason, - type FlowLeadershipStatus, registerFlowPluginInstance, } from "./leadership.js"; import { createFlowLog } from "./logging.js"; import type { Hooks, Plugin } from "./sdk.js"; +import { guardTools } from "./tool-guard.js"; import { createTools } from "./tools.js"; import { ValidationCaptureCoordinator } from "./validation-capture.js"; -type FlowCommandName = keyof typeof FLOW_CORE_COMMANDS; -type CommandHook = NonNullable; -type CommandOutput = Parameters[1]; -type Part = CommandOutput["parts"][number]; -type TextPart = Extract; -// Host assigns id, sessionID and messageID after the command hook returns. -type DraftTextPart = Omit; -const MUTATION = - /^flow_(?:plan_save|plan_approve|run_start|review_start|feature_complete|feature_reset|session_close)$/; -const AUTO_STOPPED = "Flow auto stopped."; -function isFlowCommand(command: string): command is FlowCommandName { - return Object.hasOwn(FLOW_CORE_COMMANDS, command); -} -function acceptedMutation(tool: string, output: string) { - if (!MUTATION.test(tool)) return null; - try { - const response = JSON.parse(output); - const data = response.workflowData; - const closeAccepted = - tool === "flow_session_close" && - response.status === "error" && - data?.closeState?.durableAccepted === true; - const revision = data?.projection?.revision; - if ( - data?.operation?.replayed !== false || - (response.status !== "ok" && !closeAccepted) || - typeof revision !== "number" || - !Number.isSafeInteger(revision) - ) - return null; - const sessionId = data.projection?.sessionId; - return { - revision, - sessionId: typeof sessionId === "string" ? sessionId : undefined, - }; - } catch { - return null; - } -} -function textPart( - text: string, - synthetic = false, - metadata?: Readonly>, -): DraftTextPart { - return { - type: "text", - text, - ...(synthetic ? { synthetic: true } : {}), - ...(metadata ? { metadata } : {}), - }; -} -/** - * The command hook's parts are typed with the identity the host assigns after - * the hook returns, so a part written here is a draft at runtime. This is the - * one place a draft crosses into the host's array. - */ -function asHostTextPart(part: DraftTextPart): TextPart { - return part as TextPart; -} -function rewriteCommand( - command: FlowCommandName, - args: string, - output: CommandOutput, -): void { - const config = FLOW_CORE_COMMANDS[command]; - const promptArgs = config.subtask - ? args - : "the preceding non-synthetic Flow request"; - const prompt = config.template.split("$ARGUMENTS").join(promptArgs); - if (!config.subtask) { - if (output.parts.some((part) => part.type === "subtask")) - throw new Error("Flow manager commands cannot contain subtask parts."); - const preserved = output.parts.filter((part) => part.type !== "text"); - output.parts.splice( - 0, - output.parts.length, - asHostTextPart( - textPart(args.trim() ? `Flow ${command}: ${args}` : `Flow ${command}`), - ), - asHostTextPart(textPart(prompt, true)), - ...preserved, - ); - return; - } - const part = output.parts[0]; - if (output.parts.length !== 1 || part?.type !== "subtask") - throw new Error(`/${command} requires exactly one reviewer subtask.`); - if (part.agent !== config.agent) - throw new Error(`/${command} must dispatch to '${config.agent}'.`); - // The host's subtask type does not declare `command`, but a command-dispatched - // subtask carries it at runtime, so its presence is checked, not asserted. - const declared = "command" in part ? part.command : undefined; - if (typeof declared !== "string" || declared.replace(/^\/+/, "") !== command) - throw new Error(`/${command} subtask identity did not match.`); - part.prompt = prompt; -} -function createCommandHook( - assertOperational: (action: string) => void, - autoDrive: AutoDriveCoordinator, - workspace: string, -): CommandHook { - return async (input, output) => { - const command = input.command.replace(/^\/+/, ""); - if (!isFlowCommand(command)) return; - const action = input.arguments.trim(); - if (command === "flow-auto" && /^(?:stop|cancel)$/i.test(action)) { - const confirmed = output.parts.some( - (part) => part.type === "text" && part.text === AUTO_STOPPED, - ); - const response = - autoDrive.deactivate(input.sessionID) || confirmed - ? AUTO_STOPPED - : "No Flow auto lease was active in this OpenCode session."; - output.parts[0] = asHostTextPart(textPart(response)); - output.parts.length = 1; - return; - } - assertOperational(`execute /${command}`); - if (command === "flow-auto" || command === "flow-plan") { - const evidence = requestEvidenceAnchor(input.arguments, input.sessionID); - if (evidence) { - const flow = createWorkspaceFlowService(workspace); - await flow.status({ request: { view: "compact" } }); - await flow.requestAnchor({ goal: input.arguments, evidence }); - } - } - rewriteCommand(command, input.arguments, output); - if (command !== "flow-auto") - return void autoDrive.deactivate(input.sessionID); - const metadata = await autoDrive.activate(input.sessionID); - // Preflight, not a gate. The lifecycle works either way; what changes is - // whether the user is told up front that this host cannot carry the - // continuation, instead of watching Flow stop after every feature and - // guessing which of the two it is. - if (autoDrive.continuationSupport() === "unsupported") { - output.parts.unshift( - asHostTextPart( - textPart( - "Note: this OpenCode host does not report assistant message parentage, so Flow cannot continue automatically between features here. Each feature still runs normally; drive the next one with /flow-run.", - ), - ), - ); - } - const instruction = output.parts.find( - (part): part is TextPart => - part.type === "text" && part.synthetic === true, - ); - if (!instruction) { - autoDrive.deactivate(input.sessionID); - throw new Error("/flow-auto is missing its synthetic instruction."); - } - instruction.metadata = { ...instruction.metadata, ...metadata }; - }; -} -type FlowTools = NonNullable; - -/** - * Tools whose successful output is markdown prose rather than a Flow response - * envelope. A guard rejection must stay in the same shape the caller is reading, - * so these get a markdown failure instead of a JSON blob. - */ -const MARKDOWN_TOOLS = new Set(["flow_guidance"]); - -/** Actionable recovery for each non-operational leadership reason. */ -function guardRecovery(reason: FlowLeadershipReason): string { - switch (reason) { - case "duplicate-instances": - return "Two Flow plugin instances are registered for this project. Remove the duplicate installation so exactly one remains, then restart OpenCode."; - case "incompatible-registry": - return "Another Flow build owns an incompatible runtime registry. Align the installed Flow versions, then restart OpenCode."; - default: - return "Flow is not registered for this project. Restart OpenCode to re-register, then retry."; - } -} - -function guardRejection(name: string, status: FlowLeadershipStatus): string { - const recovery = guardRecovery(status.reason); - if (MARKDOWN_TOOLS.has(name)) { - return `${status.message}\n\nRecovery: ${recovery}`; - } - // The same envelope every other Flow failure uses, so a caller told to read - // `workflowData.failure.recovery` finds it here too. - return JSON.stringify({ - status: "error", - summary: status.message, - workflowData: { - dataNote: dataNote(), - failure: { summary: status.message, recovery }, - runtimeGuard: status, - }, - }); -} - -function guardTools( - tools: FlowTools, - runtimeGuard: FlowLeadershipHandle, - autoDrive: AutoDriveCoordinator, -): FlowTools { - return Object.fromEntries( - Object.entries(tools).map(([name, definition]) => [ - name, - { - ...definition, - execute: async (...args: Parameters) => { - if (args[1].agent === "flow-planner") - return JSON.stringify({ - status: "error", - summary: - "The planning specialist supplies advice only; the manager owns all Flow tools.", - workflowData: {}, - }); - const status = runtimeGuard.query(); - if (!status.operational) return guardRejection(name, status); - const output = await definition.execute(...args); - const mutation = acceptedMutation(name, String(output)); - const context = args[1]; - if (mutation) - autoDrive.observeMutation( - context.sessionID, - mutation.revision, - name === "flow_plan_save" && mutation.revision === 1 - ? mutation.sessionId - : undefined, - context.messageID, - name === "flow_review_start", - ); - return output; - }, - }, - ]), - ) as FlowTools; -} - const FlowPlugin: Plugin = async (ctx, pluginOptions) => { const log = createFlowLog(ctx); let reviewerConfiguration = resolveFlowReviewerConfiguration({ @@ -275,24 +39,27 @@ const FlowPlugin: Plugin = async (ctx, pluginOptions) => { const pluginEntrySha256 = `sha256:${createHash("sha256") .update(await readFile(fileURLToPath(import.meta.url))) .digest("hex")}`; - const runtimeGuard = registerFlowPluginInstance( - ctx.worktree ?? ctx.directory, - { - packageName: "opencode-plugin-flow", - version, - protocolVersion: FLOW_LEADERSHIP_PROTOCOL_VERSION, - instanceId: createFlowPluginInstanceId(), - }, - ); + let workspace: string; + try { + workspace = resolveWorkspaceRoot(ctx); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + log("error", `Flow ${version} cannot start here: ${message}`); + throw error; + } + const runtimeGuard = registerFlowPluginInstance(workspace, { + packageName: "opencode-plugin-flow", + version, + protocolVersion: FLOW_LEADERSHIP_PROTOCOL_VERSION, + instanceId: createFlowPluginInstanceId(), + }); const initial = runtimeGuard.query(); const level = initial.operational ? "info" : "error"; log(level, `Flow ${version}: ${initial.message}`); - const workspace = ctx.worktree ?? ctx.directory; + const flow = createWorkspaceFlowService(workspace); const autoDrive = new AutoDriveCoordinator({ readProjection: async () => { - const response = await createWorkspaceFlowService(workspace).status({ - request: { view: "compact" }, - }); + const response = await flow.status({ request: { view: "compact" } }); if (response.status !== "ok") throw new Error(response.summary); const projection = response.workflowData.projection; if (projection.view !== "compact") @@ -347,11 +114,11 @@ const FlowPlugin: Plugin = async (ctx, pluginOptions) => { }, }), tool: guardTools(tools, runtimeGuard, autoDrive), - "command.execute.before": createCommandHook( - (action) => runtimeGuard.assertOperational(action), + "command.execute.before": createCommandHook({ + assertOperational: (action) => runtimeGuard.assertOperational(action), autoDrive, - workspace, - ), + flow, + }), "chat.message": async (input, output) => { if ( !["flow-planner", "flow-reviewer", "flow-worker"].includes( @@ -385,8 +152,6 @@ const FlowPlugin: Plugin = async (ctx, pluginOptions) => { }, event: async (input) => { const event = input.event; - if (event.type === "session.deleted") - codingModels.delete(event.properties.info.id); if (event.type === "message.updated") return autoDrive.observeHostMessage( event.properties.info.sessionID, @@ -402,6 +167,8 @@ const FlowPlugin: Plugin = async (ctx, pluginOptions) => { event.type === "session.deleted" ? event.properties.info.id : event.properties.sessionID; + if (event.type === "session.deleted") + codingModels.delete(event.properties.info.id); if (!sessionID) return autoDrive.clear(); validation.cancel(sessionID); return void autoDrive.deactivate(sessionID); diff --git a/src/platform/opencode/tool-guard.ts b/src/platform/opencode/tool-guard.ts new file mode 100644 index 0000000..5ce1353 --- /dev/null +++ b/src/platform/opencode/tool-guard.ts @@ -0,0 +1,115 @@ +import { dataNote } from "../../application/flow-response.js"; +import type { AutoDriveCoordinator } from "./auto-drive.js"; +import type { + FlowLeadershipHandle, + FlowLeadershipReason, + FlowLeadershipStatus, +} from "./leadership.js"; +import type { Hooks } from "./sdk.js"; + +const MUTATION = + /^flow_(?:plan_save|plan_approve|run_start|review_start|feature_complete|feature_reset|session_close)$/; +function acceptedMutation(tool: string, output: string) { + if (!MUTATION.test(tool)) return null; + try { + const response = JSON.parse(output); + const data = response.workflowData; + const closeAccepted = + tool === "flow_session_close" && + response.status === "error" && + data?.closeState?.durableAccepted === true; + const revision = data?.projection?.revision; + if ( + data?.operation?.replayed !== false || + (response.status !== "ok" && !closeAccepted) || + typeof revision !== "number" || + !Number.isSafeInteger(revision) + ) + return null; + const sessionId = data.projection?.sessionId; + return { + revision, + sessionId: typeof sessionId === "string" ? sessionId : undefined, + }; + } catch { + return null; + } +} +type FlowTools = NonNullable; + +/** + * Tools whose successful output is markdown prose rather than a Flow response + * envelope. A guard rejection must stay in the same shape the caller is reading, + * so these get a markdown failure instead of a JSON blob. + */ +const MARKDOWN_TOOLS = new Set(["flow_guidance"]); + +/** Actionable recovery for each non-operational leadership reason. */ +function guardRecovery(reason: FlowLeadershipReason): string { + switch (reason) { + case "duplicate-instances": + return "Two Flow plugin instances are registered for this project. Remove the duplicate installation so exactly one remains, then restart OpenCode."; + case "incompatible-registry": + return "Another Flow build owns an incompatible runtime registry. Align the installed Flow versions, then restart OpenCode."; + default: + return "Flow is not registered for this project. Restart OpenCode to re-register, then retry."; + } +} + +function guardRejection(name: string, status: FlowLeadershipStatus): string { + const recovery = guardRecovery(status.reason); + if (MARKDOWN_TOOLS.has(name)) { + return `${status.message}\n\nRecovery: ${recovery}`; + } + // The same envelope every other Flow failure uses, so a caller told to read + // `workflowData.failure.recovery` finds it here too. + return JSON.stringify({ + status: "error", + summary: status.message, + workflowData: { + dataNote: dataNote(), + failure: { summary: status.message, recovery }, + runtimeGuard: status, + }, + }); +} + +export function guardTools( + tools: FlowTools, + runtimeGuard: FlowLeadershipHandle, + autoDrive: AutoDriveCoordinator, +): FlowTools { + return Object.fromEntries( + Object.entries(tools).map(([name, definition]) => [ + name, + { + ...definition, + execute: async (...args: Parameters) => { + if (args[1].agent === "flow-planner") + return JSON.stringify({ + status: "error", + summary: + "The planning specialist supplies advice only; the manager owns all Flow tools.", + workflowData: {}, + }); + const status = runtimeGuard.query(); + if (!status.operational) return guardRejection(name, status); + const output = await definition.execute(...args); + const mutation = acceptedMutation(name, String(output)); + const context = args[1]; + if (mutation) + autoDrive.observeMutation( + context.sessionID, + mutation.revision, + name === "flow_plan_save" && mutation.revision === 1 + ? mutation.sessionId + : undefined, + context.messageID, + name === "flow_review_start", + ); + return output; + }, + }, + ]), + ) as FlowTools; +} diff --git a/src/platform/opencode/tools.ts b/src/platform/opencode/tools.ts index 9784b2e..70d1cad 100644 --- a/src/platform/opencode/tools.ts +++ b/src/platform/opencode/tools.ts @@ -21,8 +21,8 @@ import { import { requestAuthority } from "../../domain/request-evidence.js"; import type { EvidencePlatform } from "../../domain/session.js"; import { FLOW_GUIDANCE_IDS, getFlowGuidance } from "../../guidance/catalog.js"; -import { resolveWorkspaceRoot } from "../../infrastructure/fs/workspace.js"; import { createWorkspaceFlowService } from "../../infrastructure/fs/workspace-flow-service.js"; +import { resolveWorkspaceRoot } from "../../infrastructure/fs/workspace-paths.js"; import type { AutoTimingSnapshot, ProcessLocalAutoContinuationSupport, diff --git a/tests/distribution-and-surface.test.ts b/tests/distribution-and-surface.test.ts index d815092..a9d265b 100644 --- a/tests/distribution-and-surface.test.ts +++ b/tests/distribution-and-surface.test.ts @@ -8,7 +8,7 @@ import { unlink, writeFile, } from "node:fs/promises"; -import { tmpdir } from "node:os"; +import { homedir, tmpdir } from "node:os"; import { join } from "node:path"; import type { ToolContext } from "@opencode-ai/plugin"; import { createFlowCoreConfigEntries } from "../src/config-shared.js"; @@ -56,10 +56,15 @@ function pluginContext( workspace: string, directory = workspace, promptCalls?: unknown[], + logCalls?: unknown[], ) { return { client: { - app: { log() {} }, + app: { + log(input: unknown) { + logCalls?.push(input); + }, + }, session: { promptAsync(input: unknown) { promptCalls?.push(input); @@ -387,6 +392,20 @@ describe("Flow distribution surface", () => { ), ).toBe(false); }); + + test("refuses to load from the home directory and says so", async () => { + const home = homedir(); + const logCalls: unknown[] = []; + await expect( + FlowPlugin(pluginContext(home, home, undefined, logCalls)), + ).rejects.toThrow( + "Flow refuses to use the home directory itself as mutable state.", + ); + expect(logCalls).toHaveLength(1); + expect( + String((logCalls[0] as { body: { message: string } }).body.message), + ).toContain("cannot start here"); + }); }); describe("command preflight", () => { diff --git a/tests/runtime-validation-capacity.test.ts b/tests/runtime-validation-capacity.test.ts index cef2680..361560a 100644 --- a/tests/runtime-validation-capacity.test.ts +++ b/tests/runtime-validation-capacity.test.ts @@ -18,8 +18,8 @@ import { recordValidation } from "../src/domain/transitions.js"; import { loadSession, saveSession, - sessionPath, } from "../src/infrastructure/fs/workspace.js"; +import { sessionPath } from "../src/infrastructure/fs/workspace-paths.js"; import { FEATURE, MemorySessionRepository, diff --git a/tests/session-capacity.test.ts b/tests/session-capacity.test.ts index baa01a8..bad4d11 100644 --- a/tests/session-capacity.test.ts +++ b/tests/session-capacity.test.ts @@ -20,8 +20,8 @@ import { loadArchivedSession, loadSession, saveSession, - sessionPath, } from "../src/infrastructure/fs/workspace.js"; +import { sessionPath } from "../src/infrastructure/fs/workspace-paths.js"; import { deterministicEnvironment, FEATURE, diff --git a/tests/workspace-lifecycle-integration.test.ts b/tests/workspace-lifecycle-integration.test.ts index a1c829f..695885c 100644 --- a/tests/workspace-lifecycle-integration.test.ts +++ b/tests/workspace-lifecycle-integration.test.ts @@ -8,12 +8,14 @@ import type { ToolContext } from "@opencode-ai/plugin"; import type { FlowService } from "../src/application/flow-service.js"; import type { SourceDigest } from "../src/domain/session.js"; import { - archivedSessionPath, loadArchivedSession, loadSession, - sessionPath, } from "../src/infrastructure/fs/workspace.js"; import { createWorkspaceFlowService } from "../src/infrastructure/fs/workspace-flow-service.js"; +import { + archivedSessionPath, + sessionPath, +} from "../src/infrastructure/fs/workspace-paths.js"; import { persistWorkspaceValidation, prepareWorkspaceValidation, diff --git a/tests/workspace-persistence.test.ts b/tests/workspace-persistence.test.ts index 562a16c..f69b433 100644 --- a/tests/workspace-persistence.test.ts +++ b/tests/workspace-persistence.test.ts @@ -22,27 +22,31 @@ import { } from "../src/application/errors.js"; import type { Session } from "../src/domain/session.js"; import { closeSession } from "../src/domain/transitions.js"; +import { UnsafeFlowWorkspaceLayoutError } from "../src/infrastructure/fs/managed-fs.js"; +import { + reclaimOrphanedLock, + withSessionLock, +} from "../src/infrastructure/fs/session-lock.js"; import { ArchiveCollisionError, archiveAndClearSession, - archivedSessionPath, - assertMutableWorkspaceRoot, confirmActiveSessionDurability, - flowDir, - historyDir, loadArchivedSession, loadSession, quarantineUnreadableSession, - reclaimOrphanedLock, saveSession, - sessionPath, - UnsafeFlowWorkspaceLayoutError, - withSessionLock, } from "../src/infrastructure/fs/workspace.js"; +import { + archivedSessionPath, + assertMutableWorkspaceRoot, + flowDir, + historyDir, + sessionPath, +} from "../src/infrastructure/fs/workspace-paths.js"; const temporaryRoots: string[] = []; -const workspaceModuleUrl = new URL( - "../src/infrastructure/fs/workspace.ts", +const sessionLockModuleUrl = new URL( + "../src/infrastructure/fs/session-lock.ts", import.meta.url, ).href; @@ -566,7 +570,7 @@ describe("session locks", () => { const script = ` import { writeFile } from "node:fs/promises"; - import { withSessionLock } from ${JSON.stringify(workspaceModuleUrl)}; + import { withSessionLock } from ${JSON.stringify(sessionLockModuleUrl)}; await withSessionLock(process.env.FLOW_TEST_WORKSPACE, async () => { await writeFile(process.env.FLOW_TEST_CHILD_ENTERED, "yes"); });