diff --git a/src/kernel/flow.ts b/src/kernel/flow.ts
new file mode 100644
index 0000000..5236d2a
--- /dev/null
+++ b/src/kernel/flow.ts
@@ -0,0 +1,563 @@
+/**
+ * Flow 生命周期 kernel 逻辑（spec §2.3 flow_control）。
+ * 纯函数与 gh/git 执行分离；gh/git 调用走可注入 executor（测试用）。
+ */
+import { exec } from "node:child_process"
+import { promisify } from "node:util"
+import path from "node:path"
+import { gh as ghCli } from "../util/gh.js"
+import { escapeShellArg } from "../util/shell.js"
+import { pathExists } from "../util/fs.js"
+import { validateTitle } from "./slug.js"
+import { readProjectProfile } from "./profile.js"
+import { STAGE_LABEL_PREFIX, createFlowRecord, readFlowRecord, updateFlowRecord, setStageLabel } from "./records.js"
+import { worktreeStart } from "./worktree.js"
+import {
+  readIndex,
+  getFlowSession,
+  bindSession,
+  takeoverSession,
+  updateFlowSession,
+} from "./session-index.js"
+import { detectLegacyFlowRun } from "./legacy.js"
+
+const execAsync = promisify(exec)
+
+// ─── 可替换的 gh/git executor（测试用，仿 records.ts/setup.ts） ───
+
+export type CmdResult = { stdout: string; stderr: string }
+export type CmdFn = (args: string) => Promise<CmdResult>
+
+let flowGhExecutor: CmdFn | null = null
+let flowGitExecutor: CmdFn | null = null
+
+export function setFlowGhExecutor(fn: CmdFn | null): void {
+  flowGhExecutor = fn
+}
+
+export function setFlowGitExecutor(fn: CmdFn | null): void {
+  flowGitExecutor = fn
+}
+
+export async function flowGh(args: string): Promise<CmdResult> {
+  if (flowGhExecutor) return flowGhExecutor(args)
+  return ghCli(args)
+}
+
+export async function flowGit(args: string): Promise<CmdResult> {
+  if (flowGitExecutor) return flowGitExecutor(args)
+  return execAsync(`git ${args}`)
+}
+
+// ─── Flow 阶段 ───
+
+export type FlowStage = "requirements" | "design" | "tasks" | "code" | "review"
+
+export const FLOW_STAGES: FlowStage[] = ["requirements", "design", "tasks", "code", "review"]
+
+const STAGE_ORDER: Record<FlowStage, number> = {
+  requirements: 0,
+  design: 1,
+  tasks: 2,
+  code: 3,
+  review: 4,
+}
+
+/** 高风险 Flow 标签：design 与 tasks 之间需用户确认（PRD R11） */
+export const RISK_LABEL = "cabbage:risk:high"
+
+function isFlowStage(value: string): value is FlowStage {
+  return (FLOW_STAGES as string[]).includes(value)
+}
+
+// ─── Flow Record body 模板与 stage checklist ───
+
+/** 新建 Flow Record 的 body：目标/验收/阶段 checklist（spec §2.3 create-flow） */
+export function buildFlowBody(slug: string): string {
+  const stages = FLOW_STAGES.map(stage => `- [ ] ${stage}`).join("\n")
+  return [
+    `# Flow: ${slug}`,
+    "",
+    "## Goal",
+    "",
+    "<!-- 目标：由模型在 requirements 阶段填充 -->",
+    "",
+    "## Acceptance Criteria",
+    "",
+    "- [ ] TBD",
+    "",
+    "## Stages",
+    "",
+    stages,
+    "",
+  ].join("\n")
+}
+
+/** 从 body checklist 提取已完成阶段（按流程顺序） */
+export function getCompletedStages(body: string): FlowStage[] {
+  const completed: FlowStage[] = []
+  for (const line of body.split("\n")) {
+    const match = /^-\s+\[x\]\s+(\w+)/i.exec(line.trim())
+    if (match && isFlowStage(match[1])) completed.push(match[1])
+  }
+  return FLOW_STAGES.filter(stage => completed.includes(stage))
+}
+
+/** 把 stage 的 checkbox 标记为完成；已标记或缺失时幂等返回原 body */
+export function markStageComplete(body: string, stage: FlowStage): string {
+  return body.replace(new RegExp(`^- \\[ \\] ${stage}$`, "m"), `- [x] ${stage}`)
+}
+
+/** 从 issue labels 提取当前阶段（阶段序号最大的 cabbage:stage:* 标签） */
+export function getCurrentStage(labels: string[]): FlowStage | null {
+  let current: FlowStage | null = null
+  for (const label of labels) {
+    if (!label.startsWith(STAGE_LABEL_PREFIX)) continue
+    const stage = label.slice(STAGE_LABEL_PREFIX.length)
+    if (!isFlowStage(stage)) continue
+    if (current === null || STAGE_ORDER[stage] > STAGE_ORDER[current]) current = stage
+  }
+  return current
+}
+
+// ─── stage 门禁（spec §2.3：requirements 基线确认 / design 后才有 task / 高风险暂停） ───
+
+export type StageGateResult = { ok: true } | { ok: false; code: string; message: string }
+
+export type StageOp = "stage-start" | "stage-complete"
+
+/**
+ * 前置门禁判定：
+ * - stage-start X：前驱阶段须已 completed（requirements 无前置）
+ * - stage-complete X：X 未完成时前驱须已 completed；requirements 完成需 user_confirmed 一次
+ * - stage-start tasks：高风险 Flow（cabbage:risk:high）需 user_confirmed 确认
+ * - 已完成阶段重复 start/complete → 幂等通过
+ */
+export function checkStageGate(
+  op: StageOp,
+  stage: FlowStage,
+  completed: FlowStage[],
+  highRisk: boolean,
+  userConfirmed: boolean,
+): StageGateResult {
+  const done = completed.includes(stage)
+
+  if (op === "stage-complete") {
+    if (done) return { ok: true }
+    if (stage === "requirements") {
+      return userConfirmed ? { ok: true } : requirementConfirmationError()
+    }
+    if (!completed.includes(FLOW_STAGES[STAGE_ORDER[stage] - 1])) {
+      return previousStageError(stage)
+    }
+    return { ok: true }
+  }
+
+  // stage-start
+  if (done) return { ok: true }
+  if (stage !== "requirements" && !completed.includes(FLOW_STAGES[STAGE_ORDER[stage] - 1])) {
+    return previousStageError(stage)
+  }
+  if (stage === "tasks" && highRisk && !userConfirmed) {
+    return {
+      ok: false,
+      code: "RISK_CONFIRMATION_REQUIRED",
+      message:
+        "This flow is marked as high-risk; the design-to-tasks handoff requires user confirmation (user_confirmed: true)",
+    }
+  }
+  return { ok: true }
+}
+
+function requirementConfirmationError(): StageGateResult {
+  return {
+    ok: false,
+    code: "REQUIREMENTS_CONFIRMATION_REQUIRED",
+    message: "The requirements baseline must be confirmed once by the user (user_confirmed: true)",
+  }
+}
+
+function previousStageError(stage: FlowStage): StageGateResult {
+  const prev = FLOW_STAGES[STAGE_ORDER[stage] - 1]
+  const code = `${prev.toUpperCase()}_NOT_COMPLETE`
+  return { ok: false, code, message: `Stage "${prev}" must be completed before stage "${stage}"` }
+}
+
+// ─── status：Flow Record + 子任务 + 关联 PR checks 聚合（spec §2.3） ───
+
+export interface FlowRecordSummary {
+  number: number
+  title: string
+  state: string
+  labels: string[]
+}
+
+export interface SubtaskSummary {
+  number: number
+  title: string
+  state: string
+}
+
+export interface CheckSummary {
+  name: string
+  state: string
+}
+
+export interface RelatedPrSummary {
+  number: number
+  title: string
+  state: string
+  headRefName: string
+  checks: CheckSummary[]
+}
+
+export interface FlowStatusReport {
+  flow: FlowRecordSummary
+  stages: { completed: FlowStage[]; current: FlowStage | null }
+  subtasks: SubtaskSummary[]
+  pullRequests: RelatedPrSummary[]
+}
+
+export type FlowStatusResult =
+  | { ok: true; status: FlowStatusReport }
+  | { ok: false; code: "NOT_FOUND" | "GITHUB_ERROR"; message: string }
+
+/** 关联 PR：PR body 引用了 Flow 或任一子任务 Issue 号（保留 PR 原始字段） */
+export function filterRelatedPrs<T extends { number: number; body?: string }>(
+  prs: T[],
+  parentIssueNumber: number,
+  subtaskNumbers: number[],
+): T[] {
+  const refs = new Set([String(parentIssueNumber), ...subtaskNumbers.map(String)])
+  return prs.filter(pr => {
+    const mentioned = (pr.body ?? "").match(/#(\d+)/g)?.map(s => s.slice(1)) ?? []
+    return mentioned.some(n => refs.has(n))
+  })
+}
+
+/** 把 statusCheckRollup 精简为 {name, state} 列表（name 兼容 .name/.context） */
+export function summarizeChecks(rollup: Array<{ name?: string; context?: string; state: string }>): CheckSummary[] {
+  return rollup.map(check => ({ name: check.name ?? check.context ?? "unknown", state: check.state }))
+}
+
+/** 读 Flow Record + 子任务 + 关联 PR checks 汇总（spec §2.3 status） */
+export async function readFlowStatus(parentIssueNumber: number): Promise<FlowStatusResult> {
+  try {
+    const { stdout: flowOut } = await flowGh(
+      `issue view ${parentIssueNumber} --json number,title,state,labels,body --jq '{number, title, state, labels: [.labels[].name], body}'`,
+    )
+    const flow = JSON.parse(flowOut) as {
+      number: number
+      title: string
+      state: string
+      labels: string[]
+      body: string
+    }
+
+    const { stdout: subtasksOut } = await flowGh(
+      `issue list --parent ${parentIssueNumber} --json number,title,state --jq '[.[] | {number, title, state}]'`,
+    )
+    const subtasks = JSON.parse(subtasksOut) as SubtaskSummary[]
+
+    const { stdout: prsOut } = await flowGh(
+      `pr list --state all --json number,title,state,headRefName,body --jq '[.[] | {number, title, state, headRefName, body}]'`,
+    )
+    const related = filterRelatedPrs(
+      JSON.parse(prsOut) as Array<{ number: number; title: string; state: string; headRefName: string; body?: string }>,
+      parentIssueNumber,
+      [...subtasks.map(s => s.number)],
+    )
+
+    const pullRequests: RelatedPrSummary[] = []
+    for (const pr of related) {
+      const { stdout: rollupOut } = await flowGh(
+        `pr view ${pr.number} --json statusCheckRollup --jq '[.[] | {name, context, state}]'`,
+      )
+      const rollup = JSON.parse(rollupOut) as Array<{ name?: string; context?: string; state: string }>
+      pullRequests.push({ ...pr, checks: summarizeChecks(rollup) })
+    }
+
+    return {
+      ok: true,
+      status: {
+        flow: { number: flow.number, title: flow.title, state: flow.state, labels: flow.labels },
+        stages: { completed: getCompletedStages(flow.body), current: getCurrentStage(flow.labels) },
+        subtasks,
+        pullRequests,
+      },
+    }
+  } catch (err) {
+    const message = String(err)
+    return message.includes("Could not resolve") || message.includes("not found")
+      ? { ok: false, code: "NOT_FOUND", message }
+      : { ok: false, code: "GITHUB_ERROR", message }
+  }
+}
+
+// ─── planning worktree（spec §2.3 planning-start / planning-pr） ───
+
+export type PlanningResult =
+  | { ok: true; worktreePath: string; branch: string; prNumber?: number; reused?: boolean }
+  | { ok: false; code: string; message: string }
+
+/** 读 Flow Record 标题（即 flow slug） */
+async function readFlowSlug(parentIssueNumber: number): Promise<string> {
+  const { stdout } = await flowGh(`issue view ${parentIssueNumber} --json title --jq .title`)
+  return stdout.trim()
+}
+
+/** 读仓库默认分支 */
+async function readDefaultBranch(): Promise<string> {
+  const { stdout } = await flowGh(`repo view --json defaultBranchRef --jq .defaultBranchRef.name`)
+  return stdout.trim()
+}
+
+function planningBranch(slug: string): string {
+  return `docs/planning-${slug}`
+}
+
+function planningWorktreePath(projectDir: string, slug: string): string {
+  return path.join(projectDir, ".worktree", `planning-${slug}`)
+}
+
+/** 创建 planning worktree `.worktree/planning-<slug>`（architect 在此写文档） */
+export async function planningStart(projectDir: string, parentIssueNumber: number): Promise<PlanningResult> {
+  try {
+    const slug = await readFlowSlug(parentIssueNumber)
+    const base = await readDefaultBranch()
+    const result = await worktreeStart({
+      projectDir,
+      slug: `planning-${slug}`,
+      base,
+      branchType: "docs",
+    })
+    if (!result.ok) {
+      return { ok: false, code: result.code, message: result.message }
+    }
+    return { ok: true, worktreePath: result.path, branch: result.branch, reused: result.reused }
+  } catch (err) {
+    const message = String(err)
+    return message.includes("Could not resolve") || message.includes("not found")
+      ? { ok: false, code: "NOT_FOUND", message }
+      : { ok: false, code: "GITHUB_ERROR", message }
+  }
+}
+
+/** 对 planning worktree 的变更创建 Planning PR（Planning Baseline） */
+export async function planningPr(projectDir: string, parentIssueNumber: number): Promise<PlanningResult> {
+  try {
+    const slug = await readFlowSlug(parentIssueNumber)
+    const branch = planningBranch(slug)
+    const worktreeDir = planningWorktreePath(projectDir, slug)
+
+    if (!(await pathExists(worktreeDir))) {
+      return {
+        ok: false,
+        code: "PLANNING_WORKTREE_NOT_FOUND",
+        message: `Planning worktree ${worktreeDir} does not exist; run planning-start first`,
+      }
+    }
+
+    const { stdout: statusOut } = await flowGit(`-C '${escapeShellArg(worktreeDir)}' status --porcelain`)
+    if (statusOut.trim() === "") {
+      return { ok: false, code: "NOTHING_TO_COMMIT", message: "Planning worktree has no changes to commit" }
+    }
+
+    await flowGit(`-C '${escapeShellArg(worktreeDir)}' add -A`)
+    await flowGit(`-C '${escapeShellArg(worktreeDir)}' commit -m 'docs: planning baseline for ${slug}'`)
+    await flowGit(`-C '${escapeShellArg(worktreeDir)}' push -u origin ${branch}`)
+
+    const base = await readDefaultBranch()
+    const prTitle = `docs: planning baseline for ${slug}`
+    const prBody = `Planning Baseline for Flow #${parentIssueNumber}: CONTEXT.md + PRD + Design + ADR (spec PRD R11).`
+    const { stdout } = await flowGh(
+      `pr create --title '${escapeShellArg(prTitle)}' --body '${escapeShellArg(prBody)}' --base ${base} --head ${branch} --json number --jq .number`,
+    )
+    const prNumber = Number(stdout.trim())
+    return { ok: true, worktreePath: worktreeDir, branch, prNumber }
+  } catch (err) {
+    const message = String(err)
+    return message.includes("Could not resolve") || message.includes("not found")
+      ? { ok: false, code: "NOT_FOUND", message }
+      : { ok: false, code: "GITHUB_ERROR", message }
+  }
+}
+
+// ─── Flow 生命周期（spec §2.3 create-flow / complete-flow / cancel-flow / takeover） ───
+
+export interface CreateFlowInput {
+  projectDir: string
+  title: string
+  sessionID: string
+  /** 提供时对已有 Issue 建立 Flow 绑定（不新建）——legacy 检测 + 双 session 检查 */
+  parentIssueNumber?: number
+}
+
+export type CreateFlowResult =
+  | { ok: true; ref: { parentIssueNumber: number; slug: string } }
+  | { ok: false; code: string; message: string }
+
+/**
+ * create-flow：
+ * - 新建模式：validateTitle → Profile 门禁 → Draft Parent Issue（body 含目标/验收/阶段 checklist）→ bindSession
+ * - 绑定模式（提供 parentIssueNumber）：legacy 检测（§11.2）→ bindSession 双 session 检查（§10.3）
+ */
+export async function createFlow(input: CreateFlowInput): Promise<CreateFlowResult> {
+  if (input.parentIssueNumber === undefined) {
+    const validated = validateTitle(input.title)
+    if (!validated.ok) {
+      return { ok: false, code: validated.code, message: validated.message }
+    }
+    const profile = await readProjectProfile(input.projectDir)
+    if (profile.testCommand === null) {
+      return {
+        ok: false,
+        code: "SETUP_REQUIRED",
+        message: "Project Profile is not confirmed; run setup_control confirm-profile first (spec §9.3)",
+      }
+    }
+
+    const created = await createFlowRecord({ slug: validated.slug, body: buildFlowBody(validated.slug) })
+    if (!created.ok) {
+      return { ok: false, code: "ISSUE_CREATE_FAILED", message: created.error }
+    }
+    const bound = await bindSession(input.projectDir, created.ref.parentIssueNumber, input.sessionID)
+    if (!bound.ok) {
+      return { ok: false, code: bound.code, message: bound.message }
+    }
+    return { ok: true, ref: created.ref }
+  }
+
+  // 绑定已有 Issue：legacy 检测 → 双 session 检查
+  const n = input.parentIssueNumber
+  const legacy = await detectLegacyFlowRun(n, flowGh)
+  if (legacy.legacy) {
+    return {
+      ok: false,
+      code: "LEGACY_FLOW_DETECTED",
+      message:
+        "This issue is a legacy-architecture FlowRun; complete it with the old plugin version (0.x). New versions do not auto-migrate.",
+    }
+  }
+  const bound = await bindSession(input.projectDir, n, input.sessionID)
+  if (!bound.ok) {
+    return { ok: false, code: bound.code, message: bound.message }
+  }
+  let slug = input.title
+  try {
+    slug = await readFlowSlug(n)
+  } catch {
+    // 读不到标题时退化为传入 title
+  }
+  return { ok: true, ref: { parentIssueNumber: n, slug } }
+}
+
+export type FlowLifecycleResult =
+  | { ok: true; parentIssueNumber: number }
+  | { ok: false; code: string; message: string }
+
+/** complete-flow（仅 goal-verify 调用，goal 已验证）：关闭 Parent Issue + 标记 completed */
+export async function completeFlow(projectDir: string, parentIssueNumber: number): Promise<FlowLifecycleResult> {
+  try {
+    await flowGh(`issue close ${parentIssueNumber}`)
+    await updateFlowSession(projectDir, parentIssueNumber, { status: "completed" })
+    return { ok: true, parentIssueNumber }
+  } catch (err) {
+    return { ok: false, code: "GITHUB_ERROR", message: String(err) }
+  }
+}
+
+/** cancel-flow：受控取消（user_confirmed）→ 关闭 Issue + 标记 cancelled */
+export async function cancelFlow(
+  projectDir: string,
+  parentIssueNumber: number,
+  userConfirmed: boolean,
+): Promise<FlowLifecycleResult> {
+  if (!userConfirmed) {
+    return {
+      ok: false,
+      code: "USER_CONFIRMATION_REQUIRED",
+      message: "Cancelling a flow requires user confirmation (user_confirmed: true)",
+    }
+  }
+  try {
+    await flowGh(`issue close ${parentIssueNumber}`)
+    await updateFlowSession(projectDir, parentIssueNumber, { status: "cancelled" })
+    return { ok: true, parentIssueNumber }
+  } catch (err) {
+    return { ok: false, code: "GITHUB_ERROR", message: String(err) }
+  }
+}
+
+export type TakeoverFlowResult =
+  | { ok: true; parentIssueNumber: number; oldSessionID: string | null }
+  | { ok: false; code: string; message: string }
+
+/** takeover：双 session 接管（user_confirmed）→ 旧 session 置 paused + 新 session 绑定（§10.3） */
+export async function takeoverFlow(
+  projectDir: string,
+  parentIssueNumber: number,
+  newSessionID: string,
+  userConfirmed: boolean,
+): Promise<TakeoverFlowResult> {
+  const index = await readIndex(projectDir)
+  const old = getFlowSession(index, parentIssueNumber)
+  const result = await takeoverSession(projectDir, parentIssueNumber, newSessionID, userConfirmed)
+  if (!result.ok) {
+    return { ok: false, code: result.code, message: result.message }
+  }
+  return { ok: true, parentIssueNumber, oldSessionID: old?.sessionID ?? null }
+}
+
+// ─── stage-start / stage-complete（spec §2.3：门禁 + checklist + label） ───
+
+export type ApplyStageResult =
+  | { ok: true; stage: FlowStage; completedStages: FlowStage[]; highRisk: boolean }
+  | { ok: false; code: string; message: string }
+
+/** 读 Flow Record 的 labels（字符串数组） */
+async function readIssueLabels(parentIssueNumber: number): Promise<string[]> {
+  const { stdout } = await flowGh(`issue view ${parentIssueNumber} --json labels --jq '[.labels[].name]'`)
+  return JSON.parse(stdout) as string[]
+}
+
+/**
+ * stage-start / stage-complete：
+ * 门禁（checkStageGate）→ stage-complete 更新 body checklist（乐观锁）→ setStageLabel。
+ */
+export async function applyStageOp(
+  projectDir: string,
+  parentIssueNumber: number,
+  op: StageOp,
+  stage: FlowStage,
+  userConfirmed: boolean,
+): Promise<ApplyStageResult> {
+  const record = await readFlowRecord(parentIssueNumber)
+  if (!record.ok) {
+    return { ok: false, code: "NOT_FOUND", message: record.error }
+  }
+
+  const completed = getCompletedStages(record.body)
+  const labels = await readIssueLabels(parentIssueNumber)
+  const highRisk = labels.includes(RISK_LABEL)
+
+  const gate = checkStageGate(op, stage, completed, highRisk, userConfirmed)
+  if (!gate.ok) {
+    return { ok: false, code: gate.code, message: gate.message }
+  }
+
+  if (op === "stage-complete") {
+    const updated = await updateFlowRecord(parentIssueNumber, record.body, body => markStageComplete(body, stage))
+    if (!updated.ok) {
+      return { ok: false, code: "BODY_UPDATE_FAILED", message: updated.error }
+    }
+  }
+
+  const labelResult = await setStageLabel(parentIssueNumber, stage)
+  if (!labelResult.ok) {
+    return { ok: false, code: "LABEL_UPDATE_FAILED", message: labelResult.error ?? "label update failed" }
+  }
+
+  const newCompleted = op === "stage-complete" ? [...completed, stage] : completed
+  return { ok: true, stage, completedStages: newCompleted, highRisk }
+}
diff --git a/src/kernel/legacy.ts b/src/kernel/legacy.ts
new file mode 100644
index 0000000..63ce669
--- /dev/null
+++ b/src/kernel/legacy.ts
@@ -0,0 +1,40 @@
+/**
+ * 旧 FlowRun 检测（spec §11.2，自 flowrun/github.ts 只读部分迁移）。
+ * 不迁移、不读取旧状态——只判断 Issue body 是否仍是旧架构的 FlowRun cabinet。
+ */
+import { gh as ghCli } from "../util/gh.js"
+
+/** 旧 FlowRun cabinet 标记（flowrun/types.ts 同款） */
+export const CABINET_START_MARKER = "<!-- cabbage-flow-run:start -->"
+export const CABINET_END_MARKER = "<!-- cabbage-flow-run:end -->"
+
+export type LegacyFlowDetection = { legacy: true; flowRunId?: string } | { legacy: false }
+
+/** Issue body 同时含 start/end 标记 → 判定为旧架构 FlowRun（不自动迁移） */
+export function containsLegacyMarker(body: string): boolean {
+  return body.includes(CABINET_START_MARKER) && body.includes(CABINET_END_MARKER)
+}
+
+/** 从 cabinet JSON 块提取 flowRunId；缺失返回 null */
+function extractFlowRunId(body: string): string | null {
+  const match = /"flowRunId"\s*:\s*"([^"]+)"/.exec(body)
+  return match ? match[1] : null
+}
+
+/**
+ * 检测指定 Issue 是否为旧架构 FlowRun。
+ * @param ghFn gh 执行器（测试可注入；默认宿主 gh CLI）
+ */
+export async function detectLegacyFlowRun(
+  parentIssueNumber: number,
+  ghFn: (args: string) => Promise<{ stdout: string; stderr: string }> = ghCli,
+): Promise<LegacyFlowDetection> {
+  try {
+    const { stdout } = await ghFn(`issue view ${parentIssueNumber} --json body --jq .body`)
+    if (!containsLegacyMarker(stdout)) return { legacy: false }
+    const flowRunId = extractFlowRunId(stdout)
+    return flowRunId ? { legacy: true, flowRunId } : { legacy: true }
+  } catch {
+    return { legacy: false }
+  }
+}
diff --git a/src/plugin/flow-control.ts b/src/plugin/flow-control.ts
new file mode 100644
index 0000000..389f88f
--- /dev/null
+++ b/src/plugin/flow-control.ts
@@ -0,0 +1,333 @@
+/**
+ * flow_control 工具（spec §2.3）：Flow Record 生命周期（替代旧 FlowRun 状态机）。
+ * 九 op：create-flow / status / planning-start / planning-pr / stage-start / stage-complete /
+ *       complete-flow / cancel-flow / takeover。
+ * caller：primary 全量（除 complete-flow）；complete-flow 仅 goal-verify。
+ * server.ts 接线由后续批次统一处理（本批不注册，避免文件冲突）。
+ */
+import { tool } from "@opencode-ai/plugin/tool"
+import { requireToolCaller, CALLER_NOT_AUTHORIZED } from "../kernel/caller.js"
+import {
+  FLOW_STAGES,
+  createFlow,
+  readFlowStatus,
+  planningStart,
+  planningPr,
+  applyStageOp,
+  completeFlow,
+  cancelFlow,
+  takeoverFlow,
+  type FlowStage,
+} from "../kernel/flow.js"
+import { readIndex, getFlowSession } from "../kernel/session-index.js"
+import { detectLegacyFlowRun, type LegacyFlowDetection } from "../kernel/legacy.js"
+import { flowGh } from "../kernel/flow.js"
+
+/** session client 最小可见面：get（caller 判定 + goal 读取）+ update（goal 写入） */
+export interface FlowControlSessionClient {
+  session: {
+    get(input: { sessionID: string }): Promise<{ data?: { parentID?: string | null; metadata?: Record<string, unknown> } }>
+    update(input: { sessionID: string; metadata: Record<string, unknown> }): Promise<unknown>
+  }
+}
+
+/** flow_control 工具依赖 */
+export interface FlowControlDeps {
+  projectDir: string
+  sessionClient: FlowControlSessionClient
+}
+
+export type FlowControlResponse = {
+  ok: boolean
+  flow?: { parentIssueNumber: number; slug?: string }
+  status?: unknown
+  stage?: FlowStage
+  completedStages?: FlowStage[]
+  prNumber?: number
+  worktreePath?: string
+  branch?: string
+  oldSessionID?: string | null
+  error?: { code: string; message: string }
+}
+
+/**
+ * flow_control 工具工厂（spec §2.3）。
+ * caller 门禁：primary（除 complete-flow 外）；complete-flow 仅 goal-verify。
+ */
+export function createFlowControlTool(deps: FlowControlDeps) {
+  return tool({
+    description: `Control the Flow Record lifecycle (a Flow = one GitHub Parent Issue).
+
+Operations:
+- create-flow: Validate the functional title, derive the slug, create a Draft Parent Issue (body contains goal/acceptance/stage checklist, label cabbage:flow), bind the session, and write the minimal goal. When parent_issue_number is given, bind an existing issue instead (legacy FlowRun detection + double-session check).
+- status: Aggregate the Flow Record body, subtasks (issue list --parent), and related PR checks into one report.
+- planning-start: Create the planning worktree .worktree/planning-<slug> for the architect.
+- planning-pr: Commit/push the planning worktree changes and open the Planning PR (Planning Baseline).
+- stage-start / stage-complete: Enforce stage gates (requirements baseline confirmed once; design must complete before tasks; high-risk flows pause between design and tasks) and update the Parent Issue checklist + cabbage:stage:* label.
+- complete-flow: goal-verify only. After independent verification, close the Parent Issue.
+- cancel-flow: Controlled cancellation (requires user_confirmed).
+- takeover: Take over a flow bound to another active session (requires user_confirmed; old session goal is paused).
+
+Caller: primary for all ops except complete-flow (goal-verify only).`,
+    args: {
+      op: tool.schema
+        .enum(["create-flow", "status", "planning-start", "planning-pr", "stage-start", "stage-complete", "complete-flow", "cancel-flow", "takeover"])
+        .describe("Flow control operation"),
+      title: tool.schema.string().optional().describe("Functional title (kebab-case slug) for create-flow"),
+      parent_issue_number: tool.schema.number().optional().describe("Parent GitHub Issue number of the Flow Record"),
+      stage: tool.schema
+        .enum(["requirements", "design", "tasks", "code", "review"])
+        .optional()
+        .describe("Stage name (for stage-start/stage-complete)"),
+      user_confirmed: tool.schema.boolean().optional().describe("User confirmation (required for cancel-flow/takeover and high-risk gates)"),
+    },
+    async execute(args, ctx) {
+      const op = args.op as string
+
+      // caller 门禁（矩阵单一来源，§2.2）：complete-flow 仅 goal-verify；status 另开放 architect；其余仅 primary
+      const denied = await requireToolCaller(ctx, "flow_control", op, deps.sessionClient)
+      if (denied) return errorResponse(CALLER_NOT_AUTHORIZED, denied)
+
+      try {
+        switch (op) {
+          case "create-flow":
+            return handleCreateFlow(deps, args, ctx.sessionID)
+          case "status":
+            return handleStatus(deps, args)
+          case "planning-start":
+            return handlePlanningStart(deps, args)
+          case "planning-pr":
+            return handlePlanningPr(deps, args)
+          case "stage-start":
+          case "stage-complete":
+            return handleStage(deps, args, op)
+          case "complete-flow":
+            return handleCompleteFlow(deps, args, ctx)
+          case "cancel-flow":
+            return handleCancelFlow(deps, args)
+          case "takeover":
+            return handleTakeover(deps, args, ctx.sessionID)
+          default:
+            return errorResponse("UNKNOWN_OP", `Unknown op: "${op}"`)
+        }
+      } catch (err) {
+        return errorResponse("INTERNAL_ERROR", String(err))
+      }
+    },
+  })
+}
+
+/** 独立注册辅助：把 flow_control 挂到工具注册表（后续接线批次使用） */
+export function registerFlowControl(registry: Record<string, unknown>, deps: FlowControlDeps): void {
+  registry.flow_control = createFlowControlTool(deps)
+}
+
+// ─── op handlers ───
+
+async function handleCreateFlow(
+  deps: FlowControlDeps,
+  args: Record<string, unknown>,
+  sessionID: string,
+): Promise<string> {
+  const parent = args.parent_issue_number !== undefined ? Number(args.parent_issue_number) : undefined
+  const title = (args.title as string | undefined) ?? ""
+  if (parent === undefined && title.trim() === "") {
+    return errorResponse("POLICY_INVALID", "title is required for create-flow (or pass parent_issue_number to bind an existing issue)")
+  }
+
+  const result = await createFlow({
+    projectDir: deps.projectDir,
+    title,
+    sessionID,
+    parentIssueNumber: parent,
+  })
+  if (!result.ok) {
+    return errorResponse(result.code, result.message)
+  }
+
+  const goalWritten = await writeMinimalGoal(deps.sessionClient, sessionID, result.ref.parentIssueNumber)
+  if (!goalWritten) {
+    return errorResponse("GOAL_WRITE_FAILED", "flow created but failed to write the minimal goal to the session")
+  }
+
+  return okResponse({ flow: result.ref })
+}
+
+async function handleStatus(deps: FlowControlDeps, args: Record<string, unknown>): Promise<string> {
+  const n = requireParentIssueNumber(args)
+  if (typeof n === "string") return n
+
+  const legacy = await detectLegacyFlowRun(n, flowGh)
+  if (legacy.legacy) return legacyDetectedResponse(legacy)
+
+  const result = await readFlowStatus(n)
+  if (!result.ok) {
+    return errorResponse(result.code, result.message)
+  }
+  return okResponse({ status: result.status })
+}
+
+async function handlePlanningStart(deps: FlowControlDeps, args: Record<string, unknown>): Promise<string> {
+  const n = requireParentIssueNumber(args)
+  if (typeof n === "string") return n
+
+  const result = await planningStart(deps.projectDir, n)
+  if (!result.ok) {
+    return errorResponse(result.code, result.message)
+  }
+  return okResponse({ worktreePath: result.worktreePath, branch: result.branch })
+}
+
+async function handlePlanningPr(deps: FlowControlDeps, args: Record<string, unknown>): Promise<string> {
+  const n = requireParentIssueNumber(args)
+  if (typeof n === "string") return n
+
+  const result = await planningPr(deps.projectDir, n)
+  if (!result.ok) {
+    return errorResponse(result.code, result.message)
+  }
+  return okResponse({ prNumber: result.prNumber, worktreePath: result.worktreePath, branch: result.branch })
+}
+
+async function handleStage(
+  deps: FlowControlDeps,
+  args: Record<string, unknown>,
+  op: string,
+): Promise<string> {
+  const n = requireParentIssueNumber(args)
+  if (typeof n === "string") return n
+
+  const stage = args.stage as FlowStage | undefined
+  if (!stage || !FLOW_STAGES.includes(stage)) {
+    return errorResponse("POLICY_INVALID", "stage is required for stage ops: requirements|design|tasks|code|review")
+  }
+
+  const result = await applyStageOp(deps.projectDir, n, op as "stage-start" | "stage-complete", stage, args.user_confirmed === true)
+  if (!result.ok) {
+    return errorResponse(result.code, result.message)
+  }
+  return okResponse({ stage: result.stage, completedStages: result.completedStages })
+}
+
+async function handleCompleteFlow(
+  deps: FlowControlDeps,
+  args: Record<string, unknown>,
+  ctx: { sessionID: string },
+): Promise<string> {
+  const n = requireParentIssueNumber(args)
+  if (typeof n === "string") return n
+
+  // 独立 goal-verify 验证通过：绑定 session（或本会话）的 goal 必须已是 complete
+  const index = await readIndex(deps.projectDir)
+  const entry = getFlowSession(index, n)
+  const targetSessionID = entry?.sessionID ?? ctx.sessionID
+  const goalStatus = await readGoalStatus(deps.sessionClient, targetSessionID)
+  if (goalStatus !== "complete") {
+    return errorResponse(
+      "GOAL_NOT_COMPLETE",
+      "complete-flow requires the goal to be verified complete first (goal({op:'complete'}) by goal-verify)",
+    )
+  }
+
+  const result = await completeFlow(deps.projectDir, n)
+  if (!result.ok) {
+    return errorResponse(result.code, result.message)
+  }
+  return okResponse({ flow: { parentIssueNumber: result.parentIssueNumber } })
+}
+
+async function handleCancelFlow(deps: FlowControlDeps, args: Record<string, unknown>): Promise<string> {
+  const n = requireParentIssueNumber(args)
+  if (typeof n === "string") return n
+
+  const result = await cancelFlow(deps.projectDir, n, args.user_confirmed === true)
+  if (!result.ok) {
+    return errorResponse(result.code, result.message)
+  }
+  return okResponse({ flow: { parentIssueNumber: result.parentIssueNumber } })
+}
+
+async function handleTakeover(
+  deps: FlowControlDeps,
+  args: Record<string, unknown>,
+  sessionID: string,
+): Promise<string> {
+  const n = requireParentIssueNumber(args)
+  if (typeof n === "string") return n
+
+  const result = await takeoverFlow(deps.projectDir, n, sessionID, args.user_confirmed === true)
+  if (!result.ok) {
+    return errorResponse(result.code, result.message)
+  }
+  if (result.oldSessionID && result.oldSessionID !== sessionID) {
+    await pauseOldSessionGoal(deps.sessionClient, result.oldSessionID)
+  }
+  return okResponse({ flow: { parentIssueNumber: result.parentIssueNumber }, oldSessionID: result.oldSessionID })
+}
+
+// ─── helpers ───
+
+function requireParentIssueNumber(args: Record<string, unknown>): number | string {
+  const n = Number(args.parent_issue_number)
+  if (!Number.isInteger(n) || n <= 0) {
+    return errorResponse("POLICY_INVALID", "parent_issue_number is required")
+  }
+  return n
+}
+
+/** 写最小 goal（§10.2）：{parentIssueNumber, status:"active", continuationCount:0} */
+async function writeMinimalGoal(client: FlowControlSessionClient, sessionID: string, parentIssueNumber: number): Promise<boolean> {
+  try {
+    const { data } = await client.session.get({ sessionID })
+    const existing = data?.metadata ?? {}
+    await client.session.update({
+      sessionID,
+      metadata: { ...existing, goal: { parentIssueNumber, status: "active", continuationCount: 0 } },
+    })
+    return true
+  } catch {
+    return false
+  }
+}
+
+/** 读指定 session 的 goal 状态；无 goal 返回 null */
+async function readGoalStatus(client: FlowControlSessionClient, sessionID: string): Promise<string | null> {
+  try {
+    const { data } = await client.session.get({ sessionID })
+    const goal = data?.metadata?.goal as { status?: string } | undefined
+    return goal?.status ?? null
+  } catch {
+    return null
+  }
+}
+
+/** takeover 后把旧 session 的 goal 置 paused（§10.3） */
+async function pauseOldSessionGoal(client: FlowControlSessionClient, sessionID: string): Promise<void> {
+  try {
+    const { data } = await client.session.get({ sessionID })
+    const metadata = data?.metadata as Record<string, unknown> | undefined
+    const goal = metadata?.goal as { status?: string } | undefined
+    if (!goal) return
+    await client.session.update({ sessionID, metadata: { ...metadata, goal: { ...goal, status: "paused" } } })
+  } catch {
+    // 旧 session goal 置 paused 失败不阻断 takeover
+  }
+}
+
+function legacyDetectedResponse(legacy: Extract<LegacyFlowDetection, { legacy: true }>): string {
+  const detail = legacy.flowRunId ? ` (flowRunId: ${legacy.flowRunId})` : ""
+  return errorResponse(
+    "LEGACY_FLOW_DETECTED",
+    `This issue is a legacy-architecture FlowRun${detail}; complete it with the old plugin version (0.x). New versions do not auto-migrate.`,
+  )
+}
+
+function okResponse(overrides: Partial<FlowControlResponse> = {}): string {
+  const resp: FlowControlResponse = { ok: true, ...overrides }
+  return JSON.stringify(resp, null, 2)
+}
+
+function errorResponse(code: string, message: string): string {
+  const resp: FlowControlResponse = { ok: false, error: { code, message } }
+  return JSON.stringify(resp, null, 2)
+}
diff --git a/test/plugin/flow-control.test.ts b/test/plugin/flow-control.test.ts
new file mode 100644
index 0000000..304a94c
--- /dev/null
+++ b/test/plugin/flow-control.test.ts
@@ -0,0 +1,820 @@
+import { describe, it, expect, afterEach } from "vitest"
+import { mkdtemp, writeFile, mkdir, rm } from "node:fs/promises"
+import { tmpdir } from "node:os"
+import { join } from "node:path"
+import { containsLegacyMarker, detectLegacyFlowRun } from "../../src/kernel/legacy.js"
+import {
+  buildFlowBody,
+  getCompletedStages,
+  markStageComplete,
+  getCurrentStage,
+  checkStageGate,
+  filterRelatedPrs,
+  summarizeChecks,
+  readFlowStatus,
+  planningStart,
+  planningPr,
+  setFlowGhExecutor,
+  setFlowGitExecutor,
+  createFlow,
+  completeFlow,
+  cancelFlow,
+  takeoverFlow,
+} from "../../src/kernel/flow.js"
+import { setWorktreeGitExecutor } from "../../src/kernel/worktree.js"
+import { setRecordsGhExecutor } from "../../src/kernel/records.js"
+import { readIndex, writeIndex } from "../../src/kernel/session-index.js"
+import { createFlowControlTool, registerFlowControl } from "../../src/plugin/flow-control.js"
+
+describe("legacy FlowRun detection (spec §11.2)", () => {
+  it("containsLegacyMarker requires both cabinet markers", () => {
+    const legacyBody = "<!-- cabbage-flow-run:start -->\n```json\n{}\n```\n<!-- cabbage-flow-run:end -->"
+    expect(containsLegacyMarker(legacyBody)).toBe(true)
+    expect(containsLegacyMarker("no markers here")).toBe(false)
+    expect(containsLegacyMarker("<!-- cabbage-flow-run:start --> only")).toBe(false)
+    expect(containsLegacyMarker("<!-- cabbage-flow-run:end --> only")).toBe(false)
+  })
+
+  it("detectLegacyFlowRun reports legacy with the flowRunId", async () => {
+    const gh = async () => ({
+      stdout:
+        "<!-- cabbage-flow-run:start -->\n```json\n{\"flowRunId\": \"fr-123\"}\n```\n<!-- cabbage-flow-run:end -->",
+      stderr: "",
+    })
+    const result = await detectLegacyFlowRun(12, gh)
+    expect(result).toEqual({ legacy: true, flowRunId: "fr-123" })
+  })
+
+  it("detectLegacyFlowRun reports legacy without an id when the id is missing", async () => {
+    const gh = async () => ({ stdout: "<!-- cabbage-flow-run:start -->\n<!-- cabbage-flow-run:end -->", stderr: "" })
+    const result = await detectLegacyFlowRun(12, gh)
+    expect(result).toEqual({ legacy: true })
+  })
+
+  it("detectLegacyFlowRun returns legacy false for a new-style issue body", async () => {
+    const gh = async () => ({ stdout: "## Stages\n\n- [ ] requirements\n- [ ] design", stderr: "" })
+    const result = await detectLegacyFlowRun(12, gh)
+    expect(result).toEqual({ legacy: false })
+  })
+
+  it("detectLegacyFlowRun returns legacy false when the issue cannot be read", async () => {
+    const gh = async () => {
+      throw new Error("gh failed")
+    }
+    const result = await detectLegacyFlowRun(12, gh)
+    expect(result).toEqual({ legacy: false })
+  })
+})
+
+describe("flow kernel pure functions", () => {
+  describe("buildFlowBody", () => {
+    it("includes the slug, goal, acceptance, and the five stage checkboxes", () => {
+      const body = buildFlowBody("planning-baseline")
+      expect(body).toContain("planning-baseline")
+      for (const stage of ["requirements", "design", "tasks", "code", "review"]) {
+        expect(body).toContain(`- [ ] ${stage}`)
+      }
+    })
+  })
+
+  describe("getCompletedStages / markStageComplete", () => {
+    it("extracts completed stages from the checklist", () => {
+      const body = buildFlowBody("x").replace("- [ ] requirements", "- [x] requirements")
+      expect(getCompletedStages(body)).toEqual(["requirements"])
+    })
+
+    it("marks a stage complete and is idempotent", () => {
+      const body = buildFlowBody("x")
+      const marked = markStageComplete(body, "design")
+      expect(marked).toContain("- [x] design")
+      expect(getCompletedStages(marked)).toEqual(["design"])
+      expect(markStageComplete(marked, "design")).toBe(marked)
+    })
+
+    it("ignores non-stage checkboxes", () => {
+      const body = "## Acceptance Criteria\n\n- [x] TBD\n\n## Stages\n\n- [ ] code"
+      expect(getCompletedStages(body)).toEqual([])
+    })
+  })
+
+  describe("getCurrentStage", () => {
+    it("returns the furthest cabbage:stage:* label", () => {
+      expect(getCurrentStage(["cabbage:flow", "cabbage:stage:requirements"])).toBe("requirements")
+      expect(getCurrentStage(["cabbage:stage:requirements", "cabbage:stage:design"])).toBe("design")
+    })
+
+    it("returns null when no stage label exists", () => {
+      expect(getCurrentStage(["cabbage:flow"])).toBeNull()
+    })
+  })
+
+  describe("checkStageGate (spec §2.3 前置门禁)", () => {
+    it("stage-start requirements has no prerequisite", () => {
+      expect(checkStageGate("stage-start", "requirements", [], false, false).ok).toBe(true)
+    })
+
+    it("stage-start design requires requirements completed", () => {
+      expect(checkStageGate("stage-start", "design", [], false, false)).toMatchObject({
+        ok: false,
+        code: "REQUIREMENTS_NOT_COMPLETE",
+      })
+      expect(checkStageGate("stage-start", "design", ["requirements"], false, false).ok).toBe(true)
+    })
+
+    it("stage-start tasks requires design and pauses on high-risk without confirmation", () => {
+      expect(checkStageGate("stage-start", "tasks", ["requirements"], false, false)).toMatchObject({
+        ok: false,
+        code: "DESIGN_NOT_COMPLETE",
+      })
+      expect(checkStageGate("stage-start", "tasks", ["requirements", "design"], true, false)).toMatchObject({
+        ok: false,
+        code: "RISK_CONFIRMATION_REQUIRED",
+      })
+      expect(checkStageGate("stage-start", "tasks", ["requirements", "design"], true, true).ok).toBe(true)
+      expect(checkStageGate("stage-start", "tasks", ["requirements", "design"], false, false).ok).toBe(true)
+    })
+
+    it("stage-start code/review require the previous stage", () => {
+      expect(checkStageGate("stage-start", "code", ["requirements", "design"], false, false)).toMatchObject({
+        ok: false,
+        code: "TASKS_NOT_COMPLETE",
+      })
+      expect(checkStageGate("stage-start", "code", ["requirements", "design", "tasks"], false, false).ok).toBe(true)
+      expect(checkStageGate("stage-start", "review", ["requirements", "design", "tasks"], false, false)).toMatchObject({
+        ok: false,
+        code: "CODE_NOT_COMPLETE",
+      })
+    })
+
+    it("stage-complete requirements requires user confirmation once", () => {
+      expect(checkStageGate("stage-complete", "requirements", [], false, false)).toMatchObject({
+        ok: false,
+        code: "REQUIREMENTS_CONFIRMATION_REQUIRED",
+      })
+      expect(checkStageGate("stage-complete", "requirements", [], false, true).ok).toBe(true)
+    })
+
+    it("stage-complete is idempotent once the stage is already completed", () => {
+      expect(checkStageGate("stage-complete", "requirements", ["requirements"], false, false).ok).toBe(true)
+    })
+
+    it("stage-complete design requires requirements completed", () => {
+      expect(checkStageGate("stage-complete", "design", [], false, true)).toMatchObject({
+        ok: false,
+        code: "REQUIREMENTS_NOT_COMPLETE",
+      })
+      expect(checkStageGate("stage-complete", "design", ["requirements"], false, true).ok).toBe(true)
+    })
+  })
+})
+
+describe("status aggregation (spec §2.3 status)", () => {
+  describe("filterRelatedPrs / summarizeChecks", () => {
+    it("filters PRs referencing the flow or its subtasks", () => {
+      const prs = [
+        { number: 1, body: "Closes #13" },
+        { number: 2, body: "Refs #12" },
+        { number: 3, body: "unrelated work" },
+      ]
+      expect(filterRelatedPrs(prs, 12, [13])).toEqual([
+        { number: 1, body: "Closes #13" },
+        { number: 2, body: "Refs #12" },
+      ])
+    })
+
+    it("summarizes the check rollup into name/state pairs", () => {
+      expect(
+        summarizeChecks([{ name: "CI", state: "SUCCESS" }, { context: "lint", state: "FAILURE" }]),
+      ).toEqual([
+        { name: "CI", state: "SUCCESS" },
+        { name: "lint", state: "FAILURE" },
+      ])
+    })
+  })
+
+  describe("readFlowStatus", () => {
+    afterEach(() => {
+      setFlowGhExecutor(null)
+    })
+
+    it("aggregates flow record, subtasks, and related PR checks", async () => {
+      const calls: string[] = []
+      setFlowGhExecutor(async args => {
+        calls.push(args)
+        if (args.includes("issue view")) {
+          return {
+            stdout: JSON.stringify({
+              number: 12,
+              title: "planning-baseline",
+              state: "OPEN",
+              labels: ["cabbage:flow", "cabbage:stage:requirements"],
+              body: "## Stages\n\n- [x] requirements\n- [ ] design",
+            }),
+            stderr: "",
+          }
+        }
+        if (args.includes("issue list")) {
+          return { stdout: JSON.stringify([{ number: 13, title: "flow-record-layer", state: "OPEN" }]), stderr: "" }
+        }
+        if (args.includes("pr list")) {
+          return {
+            stdout: JSON.stringify([
+              { number: 99, title: "docs: planning baseline", state: "OPEN", headRefName: "docs/planning-x", body: "Closes #13" },
+            ]),
+            stderr: "",
+          }
+        }
+        if (args.includes("pr view")) {
+          return { stdout: JSON.stringify([{ name: "CI", state: "SUCCESS" }]), stderr: "" }
+        }
+        throw new Error(`unexpected gh call: ${args}`)
+      })
+
+      const result = await readFlowStatus(12)
+      expect(result.ok).toBe(true)
+      const status = result.status
+      expect(status.flow.title).toBe("planning-baseline")
+      expect(status.stages.current).toBe("requirements")
+      expect(status.subtasks).toEqual([{ number: 13, title: "flow-record-layer", state: "OPEN" }])
+      expect(status.pullRequests).toHaveLength(1)
+      expect(status.pullRequests[0]).toMatchObject({ number: 99, state: "OPEN" })
+      expect(status.pullRequests[0].checks).toEqual([{ name: "CI", state: "SUCCESS" }])
+    })
+
+    it("returns NOT_FOUND when the flow record cannot be read", async () => {
+      setFlowGhExecutor(async () => {
+        throw new Error("not found")
+      })
+      const result = await readFlowStatus(12)
+      expect(result.ok).toBe(false)
+      expect(result.code).toBe("NOT_FOUND")
+    })
+  })
+})
+
+describe("planning worktree (spec §2.3 planning-start / planning-pr)", () => {
+  afterEach(() => {
+    setFlowGhExecutor(null)
+    setFlowGitExecutor(null)
+    setWorktreeGitExecutor(null)
+  })
+
+  it("planning-start creates the planning worktree with the flow slug", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "cabbage-flow-plan-"))
+    try {
+      setFlowGhExecutor(async args => {
+        if (args.includes("issue view")) return { stdout: "planning-baseline", stderr: "" }
+        if (args.includes("repo view")) return { stdout: "main", stderr: "" }
+        throw new Error(`unexpected gh: ${args}`)
+      })
+      setWorktreeGitExecutor(async args => {
+        if (args.includes("worktree add")) return { stdout: "", stderr: "" }
+        throw new Error(`unexpected git: ${args}`)
+      })
+
+      const result = await planningStart(dir, 12)
+      expect(result.ok).toBe(true)
+      expect(result.branch).toBe("docs/planning-planning-baseline")
+      expect(result.worktreePath).toBe(join(dir, ".worktree", "planning-planning-baseline"))
+    } finally {
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it("planning-start fails when the flow record cannot be read", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "cabbage-flow-plan-"))
+    try {
+      setFlowGhExecutor(async () => {
+        throw new Error("not found")
+      })
+      const result = await planningStart(dir, 12)
+      expect(result.ok).toBe(false)
+      expect(result.code).toBe("NOT_FOUND")
+    } finally {
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it("planning-pr commits, pushes, and creates the Planning PR", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "cabbage-flow-plan-"))
+    try {
+      await mkdir(join(dir, ".worktree", "planning-planning-baseline"), { recursive: true })
+      setFlowGhExecutor(async args => {
+        if (args.includes("issue view")) return { stdout: "planning-baseline", stderr: "" }
+        if (args.includes("repo view")) return { stdout: "main", stderr: "" }
+        if (args.includes("pr create")) return { stdout: "42", stderr: "" }
+        throw new Error(`unexpected gh: ${args}`)
+      })
+      setFlowGitExecutor(async args => {
+        if (args.includes("status --porcelain")) return { stdout: " M docs/planning.md\n", stderr: "" }
+        if (args.includes("add")) return { stdout: "", stderr: "" }
+        if (args.includes("commit")) return { stdout: "", stderr: "" }
+        if (args.includes("push")) return { stdout: "", stderr: "" }
+        throw new Error(`unexpected git: ${args}`)
+      })
+
+      const result = await planningPr(dir, 12)
+      expect(result.ok).toBe(true)
+      expect(result.prNumber).toBe(42)
+    } finally {
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it("planning-pr refuses when the planning worktree does not exist", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "cabbage-flow-plan-"))
+    try {
+      setFlowGhExecutor(async args => {
+        if (args.includes("issue view")) return { stdout: "planning-baseline", stderr: "" }
+        throw new Error(`unexpected gh: ${args}`)
+      })
+      const result = await planningPr(dir, 12)
+      expect(result.ok).toBe(false)
+      expect(result.code).toBe("PLANNING_WORKTREE_NOT_FOUND")
+    } finally {
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it("planning-pr refuses when there is nothing to commit", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "cabbage-flow-plan-"))
+    try {
+      await mkdir(join(dir, ".worktree", "planning-planning-baseline"), { recursive: true })
+      setFlowGhExecutor(async args => {
+        if (args.includes("issue view")) return { stdout: "planning-baseline", stderr: "" }
+        throw new Error(`unexpected gh: ${args}`)
+      })
+      setFlowGitExecutor(async args => {
+        if (args.includes("status --porcelain")) return { stdout: "", stderr: "" }
+        throw new Error(`unexpected git: ${args}`)
+      })
+      const result = await planningPr(dir, 12)
+      expect(result.ok).toBe(false)
+      expect(result.code).toBe("NOTHING_TO_COMMIT")
+    } finally {
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+})
+
+describe("flow lifecycle (spec §2.3 create-flow / complete-flow / cancel-flow / takeover)", () => {
+  afterEach(() => {
+    setFlowGhExecutor(null)
+    setFlowGitExecutor(null)
+    setRecordsGhExecutor(null)
+    setWorktreeGitExecutor(null)
+  })
+
+  async function withProjectDir(fn: (dir: string) => Promise<void>) {
+    const dir = await mkdtemp(join(tmpdir(), "cabbage-flow-lifecycle-"))
+    try {
+      await fn(dir)
+    } finally {
+      await rm(dir, { recursive: true, force: true })
+    }
+  }
+
+  async function writeProfile(dir: string) {
+    await writeFile(join(dir, "AGENTS.md"), "## Project Profile\n\n- test command: `cabbage-test-tool run`\n")
+  }
+
+  describe("createFlow", () => {
+    it("creates a draft parent issue, binds the session, and returns the ref", async () => {
+      await withProjectDir(async dir => {
+        await writeProfile(dir)
+        setRecordsGhExecutor(async args => {
+          if (args.includes("issue create")) return { stdout: "12", stderr: "" }
+          throw new Error(`unexpected gh: ${args}`)
+        })
+        setFlowGhExecutor(async () => {
+          throw new Error("unexpected gh")
+        })
+
+        const result = await createFlow({ projectDir: dir, title: "flow-control-tool", sessionID: "sess-1" })
+        expect(result.ok).toBe(true)
+        expect(result.ref.parentIssueNumber).toBe(12)
+
+        const index = await readIndex(dir)
+        expect(index.flows["12"]).toMatchObject({ sessionID: "sess-1", status: "active" })
+      })
+    })
+
+    it("rejects a bad title", async () => {
+      await withProjectDir(async dir => {
+        const result = await createFlow({ projectDir: dir, title: "Task 001", sessionID: "s" })
+        expect(result.ok).toBe(false)
+        expect(result.code).toBe("INVALID_FORMAT")
+      })
+    })
+
+    it("rejects when no project profile is confirmed", async () => {
+      await withProjectDir(async dir => {
+        const result = await createFlow({ projectDir: dir, title: "flow-control-tool", sessionID: "s" })
+        expect(result.ok).toBe(false)
+        expect(result.code).toBe("SETUP_REQUIRED")
+      })
+    })
+
+    it("detects a legacy FlowRun when binding an existing issue", async () => {
+      await withProjectDir(async dir => {
+        setFlowGhExecutor(async args => {
+          if (args.includes("issue view")) {
+            return { stdout: "<!-- cabbage-flow-run:start -->\n<!-- cabbage-flow-run:end -->", stderr: "" }
+          }
+          throw new Error(`unexpected gh: ${args}`)
+        })
+        const result = await createFlow({ projectDir: dir, title: "x", sessionID: "s", parentIssueNumber: 12 })
+        expect(result.ok).toBe(false)
+        expect(result.code).toBe("LEGACY_FLOW_DETECTED")
+      })
+    })
+
+    it("rejects a double-session bind with FLOW_SESSION_CONFLICT", async () => {
+      await withProjectDir(async dir => {
+        await writeIndex(dir, {
+          flows: { "12": { sessionID: "sess-A", status: "active", continuationCount: 0, updatedAt: 0 } },
+        })
+        setFlowGhExecutor(async args => {
+          if (args.includes("issue view")) return { stdout: "flow-x", stderr: "" }
+          throw new Error(`unexpected gh: ${args}`)
+        })
+        const result = await createFlow({ projectDir: dir, title: "x", sessionID: "sess-B", parentIssueNumber: 12 })
+        expect(result.ok).toBe(false)
+        expect(result.code).toBe("FLOW_SESSION_CONFLICT")
+      })
+    })
+  })
+
+  describe("completeFlow", () => {
+    it("closes the parent issue and marks the session completed", async () => {
+      await withProjectDir(async dir => {
+        await writeIndex(dir, {
+          flows: { "12": { sessionID: "sess-1", status: "active", continuationCount: 0, updatedAt: 0 } },
+        })
+        setFlowGhExecutor(async args => {
+          if (args.includes("issue close")) return { stdout: "", stderr: "" }
+          throw new Error(`unexpected gh: ${args}`)
+        })
+
+        const result = await completeFlow(dir, 12)
+        expect(result.ok).toBe(true)
+        const index = await readIndex(dir)
+        expect(index.flows["12"].status).toBe("completed")
+      })
+    })
+  })
+
+  describe("cancelFlow", () => {
+    it("requires user confirmation", async () => {
+      await withProjectDir(async dir => {
+        const result = await cancelFlow(dir, 12, false)
+        expect(result.ok).toBe(false)
+        expect(result.code).toBe("USER_CONFIRMATION_REQUIRED")
+      })
+    })
+
+    it("closes the issue and marks the session cancelled when confirmed", async () => {
+      await withProjectDir(async dir => {
+        await writeIndex(dir, {
+          flows: { "12": { sessionID: "sess-1", status: "active", continuationCount: 0, updatedAt: 0 } },
+        })
+        setFlowGhExecutor(async args => {
+          if (args.includes("issue close")) return { stdout: "", stderr: "" }
+          throw new Error(`unexpected gh: ${args}`)
+        })
+
+        const result = await cancelFlow(dir, 12, true)
+        expect(result.ok).toBe(true)
+        const index = await readIndex(dir)
+        expect(index.flows["12"].status).toBe("cancelled")
+      })
+    })
+  })
+
+  describe("takeoverFlow", () => {
+    it("requires user confirmation when another session is active", async () => {
+      await withProjectDir(async dir => {
+        await writeIndex(dir, {
+          flows: { "12": { sessionID: "sess-A", status: "active", continuationCount: 0, updatedAt: 0 } },
+        })
+        const result = await takeoverFlow(dir, 12, "sess-B", false)
+        expect(result.ok).toBe(false)
+        expect(result.code).toBe("TAKEOVER_NOT_CONFIRMED")
+      })
+    })
+
+    it("pauses the old session and binds the new one when confirmed", async () => {
+      await withProjectDir(async dir => {
+        await writeIndex(dir, {
+          flows: { "12": { sessionID: "sess-A", status: "active", continuationCount: 0, updatedAt: 0 } },
+        })
+        const result = await takeoverFlow(dir, 12, "sess-B", true)
+        expect(result.ok).toBe(true)
+        expect(result.oldSessionID).toBe("sess-A")
+
+        const index = await readIndex(dir)
+        expect(index.flows["12"]).toMatchObject({ sessionID: "sess-B", status: "active" })
+      })
+    })
+  })
+})
+
+describe("createFlowControlTool (spec §2.3 flow_control)", () => {
+  afterEach(() => {
+    setFlowGhExecutor(null)
+    setFlowGitExecutor(null)
+    setRecordsGhExecutor(null)
+    setWorktreeGitExecutor(null)
+  })
+
+  async function withProjectDir(fn: (dir: string, client: any) => Promise<void>) {
+    const dir = await mkdtemp(join(tmpdir(), "cabbage-flow-tool-"))
+    const client = makeSessionClient()
+    try {
+      await fn(dir, client)
+    } finally {
+      await rm(dir, { recursive: true, force: true })
+    }
+  }
+
+  function makeSessionClient(overrides: Record<string, unknown> = {}) {
+    return {
+      session: {
+        get: async () => ({ data: { parentID: null, metadata: {} } }),
+        update: async () => {},
+      },
+      ...overrides,
+    }
+  }
+
+  function makeCtx(agent: string, sessionID: string) {
+    return { agent, sessionID, messageID: "m1", directory: ".", worktree: "." }
+  }
+
+  async function executeOp(
+    dir: string,
+    client: any,
+    op: string,
+    args: Record<string, unknown> = {},
+    agent = "dev-lifecycle",
+    sessionID = "sess-1",
+  ): Promise<Record<string, any>> {
+    const t = createFlowControlTool({ projectDir: dir, sessionClient: client })
+    const out = await t.execute({ op, ...args }, makeCtx(agent, sessionID) as never)
+    return JSON.parse(String(out)) as Record<string, any>
+  }
+
+  it("defines the nine ops in the args schema", () => {
+    const t = createFlowControlTool({ projectDir: ".", sessionClient: makeSessionClient() })
+    const opSchema = t.args.op as { safeParse(value: unknown): { success: boolean } }
+    const ops = [
+      "create-flow", "status", "planning-start", "planning-pr",
+      "stage-start", "stage-complete", "complete-flow", "cancel-flow", "takeover",
+    ]
+    for (const op of ops) expect(opSchema.safeParse(op).success).toBe(true)
+    expect(opSchema.safeParse("bogus").success).toBe(false)
+  })
+
+  it("create-flow creates the draft issue, binds the session, and writes the minimal goal", async () => {
+    await withProjectDir(async (dir, client) => {
+      await writeFile(join(dir, "AGENTS.md"), "## Project Profile\n\n- test command: `cabbage-test-tool run`\n")
+      const updates: Array<{ sessionID: string; metadata: Record<string, unknown> }> = []
+      client.session.update = async (input: { sessionID: string; metadata: Record<string, unknown> }) => {
+        updates.push(input)
+      }
+      setRecordsGhExecutor(async args => {
+        if (args.includes("issue create")) return { stdout: "12", stderr: "" }
+        throw new Error(`unexpected gh: ${args}`)
+      })
+      setFlowGhExecutor(async () => {
+        throw new Error("unexpected gh")
+      })
+
+      const resp = await executeOp(dir, client, "create-flow", { title: "flow-control-tool" })
+      expect(resp.ok).toBe(true)
+      expect(resp.flow.parentIssueNumber).toBe(12)
+
+      const index = await readIndex(dir)
+      expect(index.flows["12"].sessionID).toBe("sess-1")
+      expect(updates).toHaveLength(1)
+      expect(updates[0].metadata.goal).toEqual({ parentIssueNumber: 12, status: "active", continuationCount: 0 })
+    })
+  })
+
+  it("create-flow rejects a non-primary caller", async () => {
+    await withProjectDir(async dir => {
+      const childClient = makeSessionClient({
+        session: { get: async () => ({ data: { parentID: "parent", metadata: {} } }) },
+      })
+      const resp = await executeOp(dir, childClient, "create-flow", { title: "flow-control-tool" }, "developer")
+      expect(resp.ok).toBe(false)
+      expect(resp.error.code).toBe("CALLER_NOT_AUTHORIZED")
+    })
+  })
+
+  it("status aggregates and returns the report to primary", async () => {
+    await withProjectDir(async dir => {
+      setFlowGhExecutor(async args => {
+        if (args.includes("issue view")) {
+          return {
+            stdout: JSON.stringify({
+              number: 12, title: "planning-baseline", state: "OPEN",
+              labels: ["cabbage:flow", "cabbage:stage:requirements"],
+              body: "## Stages\n\n- [x] requirements\n- [ ] design",
+            }),
+            stderr: "",
+          }
+        }
+        if (args.includes("issue list")) return { stdout: "[]", stderr: "" }
+        if (args.includes("pr list")) return { stdout: "[]", stderr: "" }
+        throw new Error(`unexpected gh: ${args}`)
+      })
+
+      const resp = await executeOp(dir, makeSessionClient(), "status", { parent_issue_number: 12 })
+      expect(resp.ok).toBe(true)
+      expect(resp.status.flow.title).toBe("planning-baseline")
+      expect(resp.status.stages.current).toBe("requirements")
+    })
+  })
+
+  it("status reports LEGACY_FLOW_DETECTED for an old-architecture issue", async () => {
+    await withProjectDir(async dir => {
+      setFlowGhExecutor(async args => {
+        if (args.includes("issue view")) {
+          return { stdout: "<!-- cabbage-flow-run:start -->\n<!-- cabbage-flow-run:end -->", stderr: "" }
+        }
+        throw new Error(`unexpected gh: ${args}`)
+      })
+      const resp = await executeOp(dir, makeSessionClient(), "status", { parent_issue_number: 12 })
+      expect(resp.ok).toBe(false)
+      expect(resp.error.code).toBe("LEGACY_FLOW_DETECTED")
+    })
+  })
+
+  it("stage-complete requirements requires user confirmation once", async () => {
+    await withProjectDir(async dir => {
+      const body = "## Stages\n\n- [ ] requirements\n- [ ] design"
+      setRecordsGhExecutor(async args => {
+        if (args.includes("--jq .body")) return { stdout: body, stderr: "" }
+        throw new Error(`unexpected gh: ${args}`)
+      })
+      setFlowGhExecutor(async args => {
+        if (args.includes("issue view")) return { stdout: '["cabbage:flow"]', stderr: "" }
+        throw new Error(`unexpected gh: ${args}`)
+      })
+
+      const resp = await executeOp(dir, makeSessionClient(), "stage-complete", {
+        parent_issue_number: 12,
+        stage: "requirements",
+      })
+      expect(resp.ok).toBe(false)
+      expect(resp.error.code).toBe("REQUIREMENTS_CONFIRMATION_REQUIRED")
+    })
+  })
+
+  it("stage-complete requirements with confirmation updates the checklist and label", async () => {
+    await withProjectDir(async dir => {
+      const body = "## Stages\n\n- [ ] requirements\n- [ ] design"
+      setRecordsGhExecutor(async args => {
+        if (args.includes("issue edit")) return { stdout: "", stderr: "" }
+        if (args.includes('join(" ")')) return { stdout: "", stderr: "" }
+        if (args.includes("--jq .body")) return { stdout: body, stderr: "" }
+        throw new Error(`unexpected gh: ${args}`)
+      })
+      setFlowGhExecutor(async args => {
+        if (args.includes("issue view")) return { stdout: '["cabbage:flow"]', stderr: "" }
+        throw new Error(`unexpected gh: ${args}`)
+      })
+
+      const resp = await executeOp(dir, makeSessionClient(), "stage-complete", {
+        parent_issue_number: 12,
+        stage: "requirements",
+        user_confirmed: true,
+      })
+      expect(resp.ok).toBe(true)
+      expect(resp.stage).toBe("requirements")
+      expect(resp.completedStages).toEqual(["requirements"])
+    })
+  })
+
+  it("stage-start tasks is blocked until design is completed", async () => {
+    await withProjectDir(async dir => {
+      const body = "## Stages\n\n- [x] requirements\n- [ ] design"
+      setRecordsGhExecutor(async args => {
+        if (args.includes("--jq .body")) return { stdout: body, stderr: "" }
+        throw new Error(`unexpected gh: ${args}`)
+      })
+      setFlowGhExecutor(async args => {
+        if (args.includes("issue view")) return { stdout: '["cabbage:flow", "cabbage:stage:requirements"]', stderr: "" }
+        throw new Error(`unexpected gh: ${args}`)
+      })
+
+      const resp = await executeOp(dir, makeSessionClient(), "stage-start", {
+        parent_issue_number: 12,
+        stage: "tasks",
+      })
+      expect(resp.ok).toBe(false)
+      expect(resp.error.code).toBe("DESIGN_NOT_COMPLETE")
+    })
+  })
+
+  it("complete-flow rejects callers other than goal-verify", async () => {
+    await withProjectDir(async dir => {
+      const resp = await executeOp(dir, makeSessionClient(), "complete-flow", { parent_issue_number: 12 }, "dev-lifecycle")
+      expect(resp.ok).toBe(false)
+      expect(resp.error.code).toBe("CALLER_NOT_AUTHORIZED")
+    })
+  })
+
+  it("complete-flow requires the goal to be verified complete", async () => {
+    await withProjectDir(async (dir, client) => {
+      await writeIndex(dir, { flows: { "12": { sessionID: "sess-parent", status: "active", continuationCount: 0, updatedAt: 0 } } })
+      const goalVerifyClient = makeSessionClient({
+        session: {
+          get: async () => ({
+            data: { parentID: "parent", metadata: { goal: { parentIssueNumber: 12, status: "active", continuationCount: 0 } } },
+          }),
+        },
+      })
+      const resp = await executeOp(dir, goalVerifyClient, "complete-flow", { parent_issue_number: 12 }, "goal-verify", "sess-gv")
+      expect(resp.ok).toBe(false)
+      expect(resp.error.code).toBe("GOAL_NOT_COMPLETE")
+      void client
+    })
+  })
+
+  it("complete-flow by goal-verify closes the parent issue", async () => {
+    await withProjectDir(async dir => {
+      await writeIndex(dir, { flows: { "12": { sessionID: "sess-parent", status: "active", continuationCount: 0, updatedAt: 0 } } })
+      const goalVerifyClient = makeSessionClient({
+        session: {
+          get: async () => ({
+            data: { parentID: "parent", metadata: { goal: { parentIssueNumber: 12, status: "complete", continuationCount: 0 } } },
+          }),
+        },
+      })
+      setFlowGhExecutor(async args => {
+        if (args.includes("issue close")) return { stdout: "", stderr: "" }
+        throw new Error(`unexpected gh: ${args}`)
+      })
+
+      const resp = await executeOp(dir, goalVerifyClient, "complete-flow", { parent_issue_number: 12 }, "goal-verify", "sess-gv")
+      expect(resp.ok).toBe(true)
+      const index = await readIndex(dir)
+      expect(index.flows["12"].status).toBe("completed")
+    })
+  })
+
+  it("cancel-flow requires user confirmation", async () => {
+    await withProjectDir(async dir => {
+      const resp = await executeOp(dir, makeSessionClient(), "cancel-flow", { parent_issue_number: 12 })
+      expect(resp.ok).toBe(false)
+      expect(resp.error.code).toBe("USER_CONFIRMATION_REQUIRED")
+    })
+  })
+
+  it("takeover with confirmation pauses the old session goal and rebinds", async () => {
+    await withProjectDir(async dir => {
+      await writeIndex(dir, { flows: { "12": { sessionID: "sess-A", status: "active", continuationCount: 0, updatedAt: 0 } } })
+      const updates: string[] = []
+      const client = makeSessionClient({
+        session: {
+          get: async ({ sessionID }: { sessionID: string }) => {
+            if (sessionID === "sess-A") {
+              return { data: { parentID: null, metadata: { goal: { parentIssueNumber: 12, status: "active", continuationCount: 0 } } } }
+            }
+            return { data: { parentID: null, metadata: {} } }
+          },
+          update: async (input: { sessionID: string }) => {
+            updates.push(input.sessionID)
+          },
+        },
+      })
+
+      const resp = await executeOp(dir, client, "takeover", { parent_issue_number: 12, user_confirmed: true }, "dev-lifecycle", "sess-B")
+      expect(resp.ok).toBe(true)
+      expect(resp.oldSessionID).toBe("sess-A")
+      expect(updates).toContain("sess-A")
+
+      const index = await readIndex(dir)
+      expect(index.flows["12"]).toMatchObject({ sessionID: "sess-B", status: "active" })
+    })
+  })
+
+  it("returns UNKNOWN_OP for an unknown op", async () => {
+    await withProjectDir(async dir => {
+      const resp = await executeOp(dir, makeSessionClient(), "bogus" as string)
+      expect(resp.ok).toBe(false)
+      expect(resp.error.code).toBe("UNKNOWN_OP")
+    })
+  })
+})
+
+describe("registerFlowControl", () => {
+  it("mounts flow_control on the tool registry", () => {
+    const registry: Record<string, unknown> = {}
+    const client = { session: { get: async () => ({ data: { parentID: null } }), update: async () => {} } }
+    registerFlowControl(registry, { projectDir: ".", sessionClient: client })
+    expect(registry.flow_control).toBeDefined()
+  })
+})