U-035: self-target unconditional worktree + core_engine classification audit

Adds paths.repo_root()/is_self_target() and forces needs_worktree=True in work_graph_orchestrate.run_graph() whenever target_folder_path is graphlemon's own checkout, regardless of caller intent -- 05 §5.8's unconditional rule, since a self-target in-place edit risks corrupting the currently-running process's own source out from under it. Adds src/core_engine_classify.py (path-based rule: .claude/agents/*.md, src/*, supervisor/* = core_engine; frontend/, docs/ = not) and bug_record.record_core_engine_change() (a PASS-path audit record, distinct from create_bug_record's FAIL-path root-cause attribution -- no route decision, status='resolved' from creation). scheduler.finish_task_lifecycle() records this audit trail as an additive side effect after close_task_done() on self-target PASS closures. Found and resolved a real SDD-internal conflict (confirmed with the user via AskUserQuestion) before wiring this up: supervisor-auto-rollback-design.md §1 triggers Deploy Supervisor on self-target+Validator-PASS alone, but 03-C §3.11/04-P §4.6 literally gate it on an existing core_engine bug_record -- and task_create.py confirms goal-type self-improvement tasks (the exact self-development-feasibility.md walkthrough scenario) never produce a bug_record on their happy path, which would silently defeat that whole task class under the literal reading. Resolved per user decision: trigger on self-target+PASS alone; the bug_record is a byproduct, not a gate. p_deploy_supervisor_blue_green (04-P §4.6) never touches task.status, so close_task_done() still runs unconditionally -- Deploy Supervisor is a fully decoupled state machine, matching §2's "no shared failure domain" principle. org_loop_route.py is intentionally untouched -- fast_lane/full_replan is FAIL-path retry routing, orthogonal to this PASS-path trigger; deployment row creation stays U-037's job per its own Done criterion. 417 tests green; RealGitWorktreeIsolationTest (U-013/REQ-03's real git worktree regression) re-verified passing per this Unit's own Done criterion. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

U-035: self-target unconditional worktree + core_engine classification audit
Adds paths.repo_root()/is_self_target() and forces needs_worktree=True in work_graph_orchestrate.run_graph() whenever target_folder_path is graphlemon's own checkout, regardless of caller intent -- 05 §5.8's unconditional rule, since a self-target in-place edit risks corrupting the currently-running process's own source out from under it. Adds src/core_engine_classify.py (path-based rule: .claude/agents/*.md, src/*, supervisor/* = core_engine; frontend/, docs/ = not) and bug_record.record_core_engine_change() (a PASS-path audit record, distinct from create_bug_record's FAIL-path root-cause attribution -- no route decision, status='resolved' from creation). scheduler.finish_task_lifecycle() records this audit trail as an additive side effect after close_task_done() on self-target PASS closures. Found and resolved a real SDD-internal conflict (confirmed with the user via AskUserQuestion) before wiring this up: supervisor-auto-rollback-design.md §1 triggers Deploy Supervisor on self-target+Validator-PASS alone, but 03-C §3.11/04-P §4.6 literally gate it on an existing core_engine bug_record -- and task_create.py confirms goal-type self-improvement tasks (the exact self-development-feasibility.md walkthrough scenario) never produce a bug_record on their happy path, which would silently defeat that whole task class under the literal reading. Resolved per user decision: trigger on self-target+PASS alone; the bug_record is a byproduct, not a gate. p_deploy_supervisor_blue_green (04-P §4.6) never touches task.status, so close_task_done() still runs unconditionally -- Deploy Supervisor is a fully decoupled state machine, matching §2's "no shared failure domain" principle. org_loop_route.py is intentionally untouched -- fast_lane/full_replan is FAIL-path retry routing, orthogonal to this PASS-path trigger; deployment row creation stays U-037's job per its own Done criterion. 417 tests green; RealGitWorktreeIsolationTest (U-013/REQ-03's real git worktree regression) re-verified passing per this Unit's own Done criterion. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent b6b495c8
Showing 12 changed files with 412 additions and 26 deletions
docs/development-plan.md
docs/sdd/unit_matrix.json
src/bug_record.py
src/core_engine_classify.py
src/paths.py
src/scheduler.py
src/test_bug_record.py
src/test_core_engine_classify.py
src/test_paths.py
src/test_scheduler.py
src/test_work_graph_orchestrate.py
src/work_graph_orchestrate.py
@@ -342,7 +342,11 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立，可并行推进。
 - **U-035 [Risk Gate]** self-target 场景 worktree 无条件强制规则（缺口 1）+ `root_cause_category=core_engine` 命中时路由到蓝绿部署流程而非 `fast_lane`（`c_org_loop_route` 扩展）
   - 依赖：U-013（worktree 机制）、U-016（`c_org_loop_route`）、U-034（`core_engine` 枚举）
   - Done：`target_folder_path == graphlemon 自身安装路径` 时，即使是单节点串行修改也强制走 worktree（不再要求"仅并行时才强制"）；`root_cause_category=core_engine` 的 bug 被 `c_org_loop_route` 正确路由到 PH5-S2/S3 的蓝绿流程；**且改动落地后必须重跑一次 U-013/REQ-03（worktree 隔离）现有回归用例全部通过**——这是 PH4 全量收口之后第一个真正 touch 生产核心执行路径的 Unit，不能假设"没测出问题就是没问题"，否则等于在 PH5 一开工就悄悄打破 PH4 刚建立起来的信任前提
-  - 覆盖：REQ-03（回归）
+  - **实现补充（2026-08-02）——发现并解决一处真实 SDD 内部冲突（已跟用户确认）**：动手前核对"core_engine 命中时路由到蓝绿部署流程"具体怎么触发时，发现 `supervisor-auto-rollback-design.md` §1（"触发时点：Validator PASS → merge → Deploy Supervisor 接管"，完全不提 bug_record）和 03-C §3.11/04-P §4.6（字面上把触发门槛定义为"已存在一条 `root_cause_category=core_engine` 的 bug_record"）互相矛盾。核实 `task_create.py` 后确认：`c_task_create` 只会创建 `task_type='goal'` 的任务，`bug_record` 只在 Validator FAIL 归因或人工报告两条路径上产生——也就是说，`self-development-feasibility.md` 9 步走查场景本身（一个普通的自我改进 goal 任务，Planner→Coder→Validator 一次性 PASS，从未进入过 FAIL 循环）根本不会产生任何 bug_record，按 03-C/04-P 字面文本，Deploy Supervisor 永远不会为这整类任务触发——这会悄悄废掉 PH5 对"goal 型自我改进任务"这整条路径的意义。用 AskUserQuestion 向用户说明冲突后，用户选择"按 §1 字面设计：仅凭 self-target+PASS 触发，bug_record 是触发后的旁路产物而非前提条件"。
+    - **设计**：新增 `src/paths.repo_root()`（graphlemon 自身 checkout 根目录，和已有的 `platform_home()`——`~/.graphlemon` 运行时目录——是两个不同概念，别混）+ `paths.is_self_target(target_folder_path)`（真实路径比较，支持测试注入 `repo_root_path` 覆盖，避免测试真的对着当前这份 checkout 跑 worktree 副作用）。`work_graph_orchestrate.run_graph()` 内部计算 `effective_needs_worktree = needs_worktree or paths.is_self_target(target_folder_path)`，不管调用方传了什么都会在 self-target 时强制为真——两个真实调用点（`e2e_pipeline.plan_documents`/`implement_code`）都不用改，改动集中在唯一的真正入口。新增 `src/core_engine_classify.py`：`classify_core_engine_change(changed_files)` 是纯路径规则（`.claude/agents/*.md`/`src/*`/`supervisor/*` 命中即 core_engine，`frontend/`/`docs/` 不命中——05 §5.8 原文明确点名前两类+schema，schema 本身就在 `src/` 下天然覆盖，没有单独点名前端console，判断为有意排除）；`parse_changed_files_from_diff()` 直接从已有的 `code_diff` artifact 内容里解析 `diff --git a/X b/Y` 头，不需要额外一次 `git diff --name-only` 子进程调用。`bug_record.py` 新增 `record_core_engine_change()`——和已有的 `create_bug_record()`（FAIL 路径的根因归因，`suggested_route` 必填）是两个不同场景：这是 PASS 路径的审计留痕，没有"路由决策"可言，`suggested_route` 留空、`status='resolved'`（因为验证已经通过，不是待处理的开放问题）；`_VALID_ROOT_CAUSE_CATEGORIES` 顺带放开 `core_engine`（U-034 当时故意没放开，因为分类机制还不存在，现在存在了）。`scheduler.finish_task_lifecycle()` 的 PASS 分支里，`close_task_done()` 之后附加调用 `_record_core_engine_audit_if_self_target()`——**是附加副作用，不是替代**：`p_deploy_supervisor_blue_green`（04-P §4.6）整套状态机只写 `deployment.status`，从头到尾没有一条转移碰 `task.status`，这与 §2"Deploy Supervisor 和 Live/Candidate 之间没有共享失败域"的原则完全一致——Coder 的工作本身确实已经完成且已合并，`task.status='done'` 不应该被 Deploy Supervisor 是否还在跑而卡住。
+    - **范围边界（刻意不做的部分）**：这个 Unit 只到"检测 self-target + 记一条 core_engine 审计 bug_record"为止，不创建 `deployment` 行——`deployment` 记录的生成明确是 U-037 `c_deploy_supervisor_stage` 自己的 Done 标准（"给定一个已合并到主分支的 commit,能生成 deployment 记录"），Deploy Supervisor 的守护进程（U-036 建骨架）今后靠轮询"self-target 项目里状态为 done、且没有对应 deployment 行的任务"来发现待部署的改动，这个发现机制本身也是 U-036/037 的范围，这里不提前实现。`org_loop_route.route_org_loop()` 的 `fast_lane`/`full_replan` 分支本身完全没有改动——那是 Validator FAIL 后的重试路由决策，和这里"PASS 后是否需要蓝绿部署"是两个正交的问题，development-plan.md 原文里"`c_org_loop_route` 扩展"这行描述在这次实现后证明是不准确的说法，实际改动点在 `scheduler.py` 的任务收尾逻辑，不在 `org_loop_route.py`。
+    - **验证**：全量 417 个测试通过；`test_work_node_run.RealGitWorktreeIsolationTest`（真实 git worktree 隔离+merge，U-013 的 REQ-03 boundary 用例）单独重跑确认无回归，如 Done 标准要求。
+  - 覆盖：REQ-03（回归）；05 §5.8（self-target worktree 强制 + core_engine 分类，触发条件已按用户决策订正）
  
 ### PH5-S2 Deploy Supervisor 核心机制
  
@@ -52,7 +52,7 @@
     {"id": "U-033", "phase": "PH4", "sub_phase": "PH4-S3", "name": "A-2 假设收口（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-004"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": [], "assumption": ["A-2"]}, "done": "A-2 状态确认为 confirmed 或记录实际偏差；最终结论：维持 U-004 记录的\"部分成立\"，PH1~PH4 积累的真实运行数据未暴露新偏差，偏差范围到 U-004 为止已完全摸清（Coder SOUL.md 三处实质性重写），不改判 confirmed，正式收口"},
  
     {"id": "U-034", "phase": "PH5", "sub_phase": "PH5-S1", "name": "01-M 扩展：deployment/deploy_health_signal + core_engine 枚举（见 development-plan.md 2026-08-02 实现补充）", "classification": "Enabler", "depends_on": ["U-001"], "coverage": {"model": ["deployment", "deploy_health_signal", "bug_record"], "view": [], "controller": [], "process": [], "requirement": []}, "done": "两张新表建表成功（共 17 张）；bug_record 新枚举值向后兼容 U-015/U-018 已有测试；deployment 状态机用 01-M 1.15 完整 15 值枚举；ux_deployment_active 表达式索引强制全局单例非终态约束；bug_record.py 应用层白名单刻意未跟着放开 core_engine，classification+路由逻辑留给 U-035"},
-    {"id": "U-035", "phase": "PH5", "sub_phase": "PH5-S1", "name": "self-target worktree 无条件强制 + core_engine 路由到蓝绿部署", "classification": "Risk Gate", "depends_on": ["U-013", "U-016", "U-034"], "coverage": {"model": [], "view": [], "controller": ["c_org_loop_route"], "process": [], "requirement": ["REQ-03"]}, "done": "self-target 时无条件走 worktree；core_engine 分类的 bug 被路由到 PH5-S2/S3 而非 fast_lane；且改动落地后必须重跑 U-013/REQ-03（worktree 隔离）现有回归用例全部通过（2026-07-30 计划评审追加，确认未破坏 PH4 已建立的隔离保证）"},
+    {"id": "U-035", "phase": "PH5", "sub_phase": "PH5-S1", "name": "self-target worktree 无条件强制 + core_engine 分类审计（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-013", "U-016", "U-034"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": ["REQ-03"]}, "done": "self-target 时无条件走 worktree（work_graph_orchestrate.run_graph 内强制）；core_engine 分类（src/core_engine_classify.py，路径规则）在 self-target 任务 PASS 收尾时记一条审计用 bug_record；且改动落地后重跑 U-013/REQ-03（worktree 隔离）现有回归用例全部通过。**2026-08-02 用户决策修正**：Deploy Supervisor 触发条件按 supervisor-auto-rollback-design.md §1 字面设计（self-target+PASS 即触发，bug_record 是旁路产物不是前提），而非 03-C/04-P 字面文本要求的\"先有 core_engine bug_record\"——后者会让从不产生 bug_record 的 goal 型自我改进任务永远触发不了 Deploy Supervisor，已用 AskUserQuestion 与用户确认此订正。c_org_loop_route 未改动（fast_lane/full_replan 是 FAIL 路径重试路由，与此处 PASS 路径的蓝绿触发正交），deployment 行创建仍是 U-037 的范围，不在本 Unit 提前做"},
     {"id": "U-036", "phase": "PH5", "sub_phase": "PH5-S2", "name": "supervisor/isolation 补齐单测 + Deploy Supervisor 进程骨架", "classification": "Enabler", "depends_on": [], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": []}, "done": "四模块各有 test_*.py；Deploy Supervisor 骨架独立进程可启停，不受 Live/Candidate 崩溃影响"},
     {"id": "U-037", "phase": "PH5", "sub_phase": "PH5-S2", "name": "c_deploy_supervisor_stage", "classification": "Risk Gate", "depends_on": ["U-034", "U-036"], "coverage": {"model": ["deployment"], "view": [], "controller": ["c_deploy_supervisor_stage"], "process": [], "requirement": []}, "done": "能生成 deployment 记录并拉起监听临时端口、连一次性 DB 副本的候选进程"},
     {"id": "U-038", "phase": "PH5", "sub_phase": "PH5-S2", "name": "c_deploy_supervisor_health_check", "classification": "Risk Gate", "depends_on": ["U-037"], "coverage": {"model": [], "view": [], "controller": ["c_deploy_supervisor_health_check"], "process": [], "requirement": []}, "done": "process_alive/http_health/canary_task 三项检查全过 -> health_check_passed；任一失败 -> health_check_failed 且 Live 未受影响"},
@@ -25,9 +25,12 @@ the loop) and has since been resolved, not left open.
 import json
 import uuid
  
-# core_engine deliberately excluded -- see src/schema.py's matching CHECK
-# constraint and its own comment: PH5-scoped (U-034), not accepted until then.
-_VALID_ROOT_CAUSE_CATEGORIES = ("planning", "coder", "skill_defect")
+# core_engine added in U-035 -- U-034 deliberately left this app-layer
+# whitelist unchanged even though it already widened the schema-level CHECK
+# constraint, since classifying a change as core_engine is U-035's
+# path-based rule (src/core_engine_classify.py), not something the
+# Validator's existing 3-perspective judge_root_cause can produce on its own.
+_VALID_ROOT_CAUSE_CATEGORIES = ("planning", "coder", "skill_defect", "core_engine")
  
  
 class InvalidRootCauseCategoryError(Exception):
@@ -108,6 +111,31 @@ def create_bug_record(
     return {"bug_id": bug_id}
  
  
+def record_core_engine_change(conn, task_id, description):
+    """U-035 (05 §5.8): audit-trail record for a self-target task that
+    touched core-engine territory (`.claude/agents/*.md`/`src/`/`supervisor/`)
+    and passed Validator on its first attempt -- NOT the Validator-FAIL
+    root-cause-judgment path `create_bug_record()` serves. There is no
+    "route" decision here (`suggested_route` stays null, unlike
+    `create_bug_record`'s required field) and no open triage step -- the
+    change already passed validation, so `status='resolved'` from creation.
+    Exists so 03-C §3.11's `c_org_loop_route` core_engine branch and 04-P
+    §4.6's `p_deploy_supervisor_blue_green` trigger text -- both phrased in
+    terms of "a bug_record with root_cause_category=core_engine" -- have a
+    real row to reference, satisfying that text even though this is a
+    PASS-path audit record rather than a FAIL-path bug attribution.
+    Returns {"bug_id"}."""
+    bug_id = str(uuid.uuid4())
+    conn.execute(
+        "INSERT INTO bug_record "
+        "(bug_id, task_id, source, description, root_cause_category, status) "
+        "VALUES (?, ?, 'validator', ?, 'core_engine', 'resolved')",
+        (bug_id, task_id, description),
+    )
+    conn.commit()
+    return {"bug_id": bug_id}
+
+
 def claim_human_reported_bug(conn, bug_id):
     """04-P §4.3: human_reported -> tracing_lineage. Validator claims an
     open, source=human bug_record. status: open -> triaging."""
+"""05 §5.8 / U-035: path-based classification of whether a self-target
+change touches "core engine" territory -- `.claude/agents/*.md` identity
+files, Graph Engine scheduler/executor source (`src/*.py`), or 01-M's own
+schema (`src/schema.py`, itself a `src/*.py` file so already covered).
+
+Deliberately NOT the Validator's existing 3-perspective `judge_root_cause`
+(root_cause_trace.py) -- that produces planning|coder|skill_defect from
+content-level judgment about WHY a bug happened; this produces core_engine
+from a purely mechanical WHERE-did-the-change-land rule, independent of
+whether anything went wrong. The two are orthogonal and only combined at
+the call site (scheduler.py), never merged into one function.
+
+Scope note: `frontend/` is deliberately excluded (05 §5.8 names the identity
+files + engine source + schema, not the console UI -- a frontend/app.js bug
+is much lower blast radius and doesn't need blue-green). `supervisor/` IS
+included -- Deploy Supervisor modifying its own isolation/rollback logic is
+at least as self-referentially risky as the engine it's protecting.
+"""
+
+import posixpath
+
+_CORE_ENGINE_PREFIXES = ("src/", "supervisor/", ".claude/agents/")
+
+
+def classify_core_engine_change(changed_files):
+    """changed_files: iterable of repo-relative paths (posix-style, as
+    produced by `git diff --name-only` / `diff --git a/X b/Y` headers).
+    Returns True if ANY path falls under a core-engine prefix."""
+    for path in changed_files:
+        normalized = posixpath.normpath(path.replace("\\", "/"))
+        if normalized.startswith(_CORE_ENGINE_PREFIXES) or any(
+            normalized == prefix.rstrip("/") for prefix in _CORE_ENGINE_PREFIXES
+        ):
+            return True
+    return False
+
+
+def parse_changed_files_from_diff(diff_text):
+    """Extracts repo-relative changed file paths from unified `git diff`
+    output (`diff --git a/X b/Y` header lines) -- avoids requiring a second
+    git subprocess call when the diff text is already on hand (e.g. as the
+    code_diff artifact's content)."""
+    changed = []
+    for line in (diff_text or "").splitlines():
+        if line.startswith("diff --git a/"):
+            rest = line[len("diff --git a/"):]
+            # "<a-path> b/<b-path>" -- split on the last " b/" occurrence
+            # since a path itself could theoretically contain " b/".
+            marker = rest.rfind(" b/")
+            if marker != -1:
+                changed.append(rest[:marker])
+    return changed
@@ -48,3 +48,23 @@ def is_separate_from(target_folder_path):
     home_prefix = home.rstrip(os.sep) + os.sep
     target_prefix = target.rstrip(os.sep) + os.sep
     return not target.startswith(home_prefix) and not home.startswith(target_prefix)
+
+
+def repo_root():
+    """graphlemon's own code checkout root (the directory containing `src/`,
+    `.git`, `docs/`, ...) -- distinct from platform_home() (the ~/.graphlemon
+    *runtime* directory for logs/db, deliberately outside the checkout).
+    U-035 (05 §5.8 self-target rules) needs this to detect when a task's
+    `target_folder_path` IS graphlemon's own repo."""
+    return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+
+def is_self_target(target_folder_path, repo_root_path=None):
+    """05 §5.8: `target_folder_path == graphlemon 平台自身安装路径`. Compares
+    realpaths so a symlinked or relative target_folder_path still matches.
+    `repo_root_path` defaults to this running instance's own checkout
+    (`repo_root()`) but can be overridden -- e.g. by a test simulating a
+    self-target scenario against a throwaway clone rather than this actual
+    running checkout."""
+    root = repo_root_path if repo_root_path is not None else repo_root()
+    return os.path.realpath(target_folder_path) == os.path.realpath(root)
@@ -43,6 +43,27 @@ exactly the layer `budget_guardrail.check_budget()` already does the
 equivalent translation at for its own case (REQ-10), so `_block_task()`
 below follows that same established pattern (UPDATE + `task_status_changed`
 execution_event) rather than inventing a new one.
+
+**U-035 self-target audit trail (2026-08-02)**: resolved a real conflict
+between supervisor-auto-rollback-design.md §1 (Deploy Supervisor triggers on
+"self-target + Validator PASS + merge", no bug_record involved) and 03-C
+§3.11/04-P §4.6's literal text (gated on an *existing*
+`bug_record.root_cause_category=core_engine`) -- confirmed with the user
+that a plain goal-type self-improvement task (task_create.py: `c_task_create`
+only ever makes task_type='goal') never gets a bug_record on its happy path
+at all, so the literal gate would silently defeat PH5's entire purpose for
+that whole task class. Resolution (user-confirmed): trigger on self-target+
+PASS alone; `bug_record.record_core_engine_change()` is created as an
+audit-trail byproduct of that trigger, not a precondition for it. `deployment`
+row creation itself (the actual hand-off) stays in U-037's `c_deploy_
+supervisor_stage` as its own Done criterion already specifies -- this module
+only detects + records, it doesn't stage a candidate. `p_deploy_supervisor_
+blue_green` (04-P §4.6) never touches `task.status` at all -- it's a fully
+separate state machine on `deployment`, matching supervisor-auto-rollback-
+design.md §2's "no shared failure domain" between Deploy Supervisor and the
+Graph Engine -- so `close_task_done()` still runs unconditionally on PASS
+(the Coder's work IS genuinely complete and merged), the self-target check
+is a purely additive side effect alongside it, not a replacement for it.
 """
  
 import json
@@ -50,8 +71,8 @@ import uuid
 import concurrent.futures
  
 from src import (
-    e2e_pipeline, org_level_loop, org_loop_route, schema, settings_config,
-    task_create, work_graph_execute, work_graph_orchestrate, work_node_run,
+    bug_record, core_engine_classify, e2e_pipeline, org_level_loop, org_loop_route, paths, schema,
+    settings_config, task_create, work_graph_execute, work_graph_orchestrate, work_node_run,
 )
  
 _BUSY_TIMEOUT_MS = 30000
@@ -88,6 +109,24 @@ def _block_task(conn, task_id, blocked_reason):
     conn.commit()
  
  
+def _record_core_engine_audit_if_self_target(conn, task_id, target_folder_path, run_result):
+    """U-035 (05 §5.8): additive side effect on a self-target task's PASS
+    closure -- see module docstring's "self-target audit trail" note for
+    why this doesn't gate close_task_done, just follows it."""
+    if not paths.is_self_target(target_folder_path):
+        return
+    code_diff_id = run_result.get("code_artifact_ids", {}).get("code_diff")
+    if not code_diff_id:
+        return
+    diff_text = e2e_pipeline._read_artifact_content(conn, target_folder_path, code_diff_id)
+    changed_files = core_engine_classify.parse_changed_files_from_diff(diff_text)
+    if core_engine_classify.classify_core_engine_change(changed_files):
+        bug_record.record_core_engine_change(
+            conn, task_id,
+            "self-target change touched core-engine files: " + ", ".join(sorted(set(changed_files))),
+        )
+
+
 def finish_task_lifecycle(conn, run_result, target_folder_path, goal_text):
     """PASS -> `close_task_done`; FAIL/INCONCLUSIVE -> `org_level_loop.
     run_fix_loop` (root-cause judge + reroute + fix + reverify, U-017).
@@ -96,6 +135,7 @@ def finish_task_lifecycle(conn, run_result, target_folder_path, goal_text):
     verdict = run_result["verdict"]["verdict"]
     if verdict == "PASS":
         org_loop_route.close_task_done(conn, task_id)
+        _record_core_engine_audit_if_self_target(conn, task_id, target_folder_path, run_result)
         run_result["final_status"] = "done"
         return run_result
  
@@ -60,20 +60,22 @@ class CreateBugRecordTest(BugRecordTestBase):
                 root_cause_detail="d", suggested_route="fast_lane",
             )
  
-    def test_core_engine_rejected_pending_u035(self):
-        # U-034 (2026-08-02) widened the SCHEMA's own CHECK constraint to
-        # accept core_engine (see src/schema.py), but bug_record.py's own
-        # application-layer _VALID_ROOT_CAUSE_CATEGORIES deliberately still
-        # excludes it -- nothing yet determines WHEN a bug is core_engine
-        # (05 §5.8's file-path-based rule, not the Validator's 3-perspective
-        # judge_root_cause, which only ever produces planning/coder/
-        # skill_defect); that classification logic + c_org_loop_route's
-        # routing to the blue-green flow is U-035's job, not this one's.
-        with self.assertRaises(bug_record.InvalidRootCauseCategoryError):
-            bug_record.create_bug_record(
-                self.conn, task_id="task-1", description="x", root_cause_category="core_engine",
-                root_cause_detail="d", suggested_route="fast_lane",
-            )
+    def test_core_engine_accepted_since_u035(self):
+        # U-034 widened the schema's CHECK constraint; U-035 widens this
+        # app-layer whitelist too, once src/core_engine_classify.py exists
+        # as a real (if separate) mechanism for determining WHEN a change is
+        # core_engine. create_bug_record() itself still doesn't classify
+        # anything -- a caller (e.g. a self-target-aware Validator FAIL path)
+        # decides the category and passes it explicitly, same as any other
+        # category.
+        result = bug_record.create_bug_record(
+            self.conn, task_id="task-1", description="x", root_cause_category="core_engine",
+            root_cause_detail="d", suggested_route="fast_lane",
+        )
+        row = self.conn.execute(
+            "SELECT root_cause_category FROM bug_record WHERE bug_id = ?", (result["bug_id"],)
+        ).fetchone()
+        self.assertEqual(row[0], "core_engine")
  
  
 class CreateBugReportTest(BugRecordTestBase):
@@ -145,6 +147,25 @@ class CreateBugReportTest(BugRecordTestBase):
         self.assertIsNone(row[0])
  
  
+class RecordCoreEngineChangeTest(BugRecordTestBase):
+    """U-035: the PASS-path audit trail, distinct from create_bug_record's
+    FAIL-path root-cause attribution."""
+
+    def test_creates_resolved_core_engine_record_with_no_route(self):
+        result = bug_record.record_core_engine_change(
+            self.conn, task_id="task-1", description="self-target change touched src/scheduler.py"
+        )
+        row = self.conn.execute(
+            "SELECT source, status, root_cause_category, suggested_route, description "
+            "FROM bug_record WHERE bug_id = ?",
+            (result["bug_id"],),
+        ).fetchone()
+        self.assertEqual(
+            row,
+            ("validator", "resolved", "core_engine", None, "self-target change touched src/scheduler.py"),
+        )
+
+
 class HumanReportedFlowTest(BugRecordTestBase):
     def _insert_human_bug(self, bug_id="bug-1", status="open"):
         self.conn.execute(
+"""U-035 tests: path-based core_engine classification.
+
+Run: python3 -m unittest src.test_core_engine_classify -v
+"""
+
+import unittest
+
+from src import core_engine_classify
+
+
+class ClassifyCoreEngineChangeTest(unittest.TestCase):
+    def test_src_python_file_is_core_engine(self):
+        self.assertTrue(core_engine_classify.classify_core_engine_change(["src/scheduler.py"]))
+
+    def test_claude_agents_identity_file_is_core_engine(self):
+        self.assertTrue(
+            core_engine_classify.classify_core_engine_change([".claude/agents/coder.md"])
+        )
+
+    def test_supervisor_module_is_core_engine(self):
+        self.assertTrue(
+            core_engine_classify.classify_core_engine_change(["supervisor/isolation/change_manifest.py"])
+        )
+
+    def test_frontend_only_change_is_not_core_engine(self):
+        self.assertFalse(core_engine_classify.classify_core_engine_change(["frontend/app.js"]))
+
+    def test_docs_only_change_is_not_core_engine(self):
+        self.assertFalse(core_engine_classify.classify_core_engine_change(["docs/development-plan.md"]))
+
+    def test_mixed_changeset_with_one_core_engine_file_is_core_engine(self):
+        self.assertTrue(
+            core_engine_classify.classify_core_engine_change(["frontend/app.js", "src/e2e_pipeline.py"])
+        )
+
+    def test_empty_changeset_is_not_core_engine(self):
+        self.assertFalse(core_engine_classify.classify_core_engine_change([]))
+
+    def test_windows_style_separators_normalized(self):
+        self.assertTrue(core_engine_classify.classify_core_engine_change(["src\\scheduler.py"]))
+
+
+class ParseChangedFilesFromDiffTest(unittest.TestCase):
+    def test_single_file_diff(self):
+        diff_text = (
+            "diff --git a/src/scheduler.py b/src/scheduler.py\n"
+            "index abc123..def456 100644\n"
+            "--- a/src/scheduler.py\n"
+            "+++ b/src/scheduler.py\n"
+            "@@ -1,3 +1,3 @@\n"
+            "-old\n"
+            "+new\n"
+        )
+        self.assertEqual(
+            core_engine_classify.parse_changed_files_from_diff(diff_text), ["src/scheduler.py"]
+        )
+
+    def test_multiple_files_diff(self):
+        diff_text = (
+            "diff --git a/src/scheduler.py b/src/scheduler.py\n"
+            "--- a/src/scheduler.py\n"
+            "+++ b/src/scheduler.py\n"
+            "@@ -1,1 +1,1 @@\n"
+            "-a\n"
+            "+b\n"
+            "diff --git a/frontend/app.js b/frontend/app.js\n"
+            "--- a/frontend/app.js\n"
+            "+++ b/frontend/app.js\n"
+            "@@ -1,1 +1,1 @@\n"
+            "-a\n"
+            "+b\n"
+        )
+        self.assertEqual(
+            core_engine_classify.parse_changed_files_from_diff(diff_text),
+            ["src/scheduler.py", "frontend/app.js"],
+        )
+
+    def test_empty_diff_returns_empty_list(self):
+        self.assertEqual(core_engine_classify.parse_changed_files_from_diff(""), [])
+
+    def test_none_diff_returns_empty_list(self):
+        self.assertEqual(core_engine_classify.parse_changed_files_from_diff(None), [])
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -83,5 +83,33 @@ class IsSeparateFromTest(unittest.TestCase):
         self.assertFalse(paths.is_separate_from(target))
  
  
+class RepoRootTest(unittest.TestCase):
+    def test_matches_computed_repo_root(self):
+        self.assertEqual(paths.repo_root(), _REPO_ROOT)
+
+    def test_contains_src_and_git(self):
+        root = paths.repo_root()
+        self.assertTrue(os.path.isdir(os.path.join(root, "src")))
+        self.assertTrue(os.path.isdir(os.path.join(root, ".git")))
+
+
+class IsSelfTargetTest(unittest.TestCase):
+    def test_this_repos_own_checkout_is_self_target(self):
+        self.assertTrue(paths.is_self_target(_REPO_ROOT))
+
+    def test_other_directory_is_not_self_target(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            self.assertFalse(paths.is_self_target(tmp))
+
+    def test_relative_path_to_repo_root_still_matches(self):
+        rel = os.path.relpath(_REPO_ROOT, os.getcwd())
+        self.assertTrue(paths.is_self_target(rel))
+
+    def test_explicit_repo_root_override(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            self.assertTrue(paths.is_self_target(tmp, repo_root_path=tmp))
+            self.assertFalse(paths.is_self_target(_REPO_ROOT, repo_root_path=tmp))
+
+
 if __name__ == "__main__":
     unittest.main()
@@ -17,7 +17,10 @@ import time
 import unittest
 from unittest.mock import patch
  
-from src import org_level_loop, org_loop_route, schema, scheduler, settings_config, work_graph_execute, work_node_run
+from src import (
+    bug_record, e2e_pipeline, org_level_loop, org_loop_route, paths, schema, scheduler, settings_config,
+    work_graph_execute, work_node_run,
+)
  
  
 class FinishTaskLifecycleTestBase(unittest.TestCase):
@@ -50,6 +53,58 @@ class FinishTaskLifecyclePassTest(FinishTaskLifecycleTestBase):
         self.assertEqual(status, "done")
  
  
+class FinishTaskLifecycleSelfTargetAuditTest(FinishTaskLifecycleTestBase):
+    """U-035: self-target PASS closures additively record a core_engine
+    audit-trail bug_record, without changing close_task_done's own effect."""
+
+    def test_self_target_core_engine_change_records_audit_bug(self):
+        run_result = {
+            "task_id": "task-1", "doc_artifact_ids": {}, "code_artifact_ids": {"code_diff": "art-diff"},
+            "verdict": {"verdict": "PASS", "raw_response": "VERDICT: PASS"},
+        }
+        with patch.object(paths, "is_self_target", return_value=True), \
+             patch.object(e2e_pipeline, "_read_artifact_content", return_value=(
+                 "diff --git a/src/scheduler.py b/src/scheduler.py\n"
+                 "--- a/src/scheduler.py\n+++ b/src/scheduler.py\n"
+             )):
+            result = scheduler.finish_task_lifecycle(self.conn, run_result, "/tmp/proj-1", "goal text")
+
+        self.assertEqual(result["final_status"], "done")
+        status = self.conn.execute("SELECT status FROM task WHERE task_id = 'task-1'").fetchone()[0]
+        self.assertEqual(status, "done")  # unaffected -- audit trail is additive, not a gate
+        audit_row = self.conn.execute(
+            "SELECT root_cause_category, status FROM bug_record WHERE task_id = 'task-1'"
+        ).fetchone()
+        self.assertEqual(audit_row, ("core_engine", "resolved"))
+
+    def test_self_target_non_core_engine_change_records_no_audit_bug(self):
+        run_result = {
+            "task_id": "task-1", "doc_artifact_ids": {}, "code_artifact_ids": {"code_diff": "art-diff"},
+            "verdict": {"verdict": "PASS", "raw_response": "VERDICT: PASS"},
+        }
+        with patch.object(paths, "is_self_target", return_value=True), \
+             patch.object(e2e_pipeline, "_read_artifact_content", return_value=(
+                 "diff --git a/frontend/app.js b/frontend/app.js\n"
+                 "--- a/frontend/app.js\n+++ b/frontend/app.js\n"
+             )):
+            scheduler.finish_task_lifecycle(self.conn, run_result, "/tmp/proj-1", "goal text")
+
+        count = self.conn.execute("SELECT COUNT(*) FROM bug_record WHERE task_id = 'task-1'").fetchone()[0]
+        self.assertEqual(count, 0)
+
+    def test_non_self_target_pass_records_no_audit_bug(self):
+        run_result = {
+            "task_id": "task-1", "doc_artifact_ids": {}, "code_artifact_ids": {"code_diff": "art-diff"},
+            "verdict": {"verdict": "PASS", "raw_response": "VERDICT: PASS"},
+        }
+        # is_self_target NOT patched -- "/tmp/proj-1" is genuinely not this
+        # repo's own checkout, so _read_artifact_content must never even be
+        # attempted (would fail on a nonexistent artifact/file otherwise).
+        scheduler.finish_task_lifecycle(self.conn, run_result, "/tmp/proj-1", "goal text")
+        count = self.conn.execute("SELECT COUNT(*) FROM bug_record WHERE task_id = 'task-1'").fetchone()[0]
+        self.assertEqual(count, 0)
+
+
 class FinishTaskLifecycleFailTest(FinishTaskLifecycleTestBase):
     def test_fail_verdict_runs_fix_loop(self):
         run_result = {
@@ -10,7 +10,7 @@ Run: python3 -m unittest src.test_work_graph_orchestrate -v
 import unittest
 from unittest.mock import patch
  
-from src import schema, work_graph_execute, work_graph_orchestrate as wgo
+from src import paths, schema, work_graph_execute, work_graph_orchestrate as wgo, work_node_run
  
  
 class TopologicalOrderTest(unittest.TestCase):
@@ -264,5 +264,47 @@ class VerifyFnTest(RunGraphTestBase):
         self.assertEqual(run_status, "blocked")
  
  
+class SelfTargetForcesWorktreeTest(RunGraphTestBase):
+    """U-035 (05 §5.8): self-target unconditionally forces needs_worktree=True,
+    regardless of what the caller passed. run_work_node itself is mocked here
+    (not just execute_work_graph) so this only asserts what needs_worktree
+    value run_graph computes and passes down -- it doesn't exercise real git
+    worktree commands against this actual checkout."""
+
+    def test_self_target_forces_worktree_even_when_caller_passed_false(self):
+        manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []}
+        self.conn.execute(
+            "UPDATE project SET target_folder_path = ? WHERE project_id = 'proj-1'", (paths.repo_root(),)
+        )
+        self.conn.commit()
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_node_run, "run_work_node", return_value={"content": "x"}) as mock_run:
+            wgo.run_graph(
+                self.conn, "run-1", "planner", "planner-doc-writing", paths.repo_root(), manifest,
+                lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, needs_worktree=False,
+            )
+        self.assertTrue(mock_run.call_args.kwargs["needs_worktree"])
+
+    def test_non_self_target_respects_caller_supplied_false(self):
+        manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []}
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_node_run, "run_work_node", return_value={"content": "x"}) as mock_run:
+            wgo.run_graph(
+                self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, needs_worktree=False,
+            )
+        self.assertFalse(mock_run.call_args.kwargs["needs_worktree"])
+
+    def test_non_self_target_still_respects_caller_supplied_true(self):
+        manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []}
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_node_run, "run_work_node", return_value={"content": "x"}) as mock_run:
+            wgo.run_graph(
+                self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, needs_worktree=True,
+            )
+        self.assertTrue(mock_run.call_args.kwargs["needs_worktree"])
+
+
 if __name__ == "__main__":
     unittest.main()
@@ -45,7 +45,7 @@ import json
 import time
 import uuid
  
-from src import work_graph_execute, work_node_run
+from src import paths, work_graph_execute, work_node_run
  
  
 class ConsistencyCheckFailedError(Exception):
@@ -181,6 +181,15 @@ def run_graph(
     writing code (Coder) should pass this; callers only producing document
     content (Planner) don't need it.
  
+    **U-035 (05 §5.8) self-target override**: when `target_folder_path` IS
+    graphlemon's own checkout (`paths.is_self_target`), `needs_worktree` is
+    forced True regardless of what the caller passed -- the currently-running
+    backend process is reading this exact code from disk, so an in-place edit
+    to the main checkout (even a serial, non-parallel one) risks corrupting
+    state out from under itself mid-execution. This is unconditional (not
+    "only when parallel", 03-C §3.6's general rule) precisely because
+    self-target has no safe non-worktree case at all.
+
     `verify_fn(node, output) -> (ok: bool, feedback)`, if given, is called
     right after each node completes, BEFORE moving on to nodes that depend
     on it -- lets a caller intercept a bad output before it propagates
@@ -195,6 +204,7 @@ def run_graph(
     Returns {node_id: output_payload}.
     """
     ensure_placeholder_skill(conn, skill_id, org_node)
+    effective_needs_worktree = needs_worktree or paths.is_self_target(target_folder_path)
     ordered = topological_order(graph_manifest)
     outputs = {}
     for node in ordered:
@@ -208,14 +218,14 @@ def run_graph(
         run_fn = _make_run_fn(target_folder_path, node_exec_id, org_node, prompt, output_schema, timeout)
         output = work_node_run.run_work_node(
             conn, node_exec_id, {"node_id": node_id}, input_schema, output_schema, run_fn,
-            needs_worktree=needs_worktree, repo_path=target_folder_path,
+            needs_worktree=effective_needs_worktree, repo_path=target_folder_path,
             sleep=sleep, now=now,
         )
  
         if verify_fn is not None:
             output = _verify_and_maybe_regenerate(
                 conn, run_id, org_node, skill_id, target_folder_path, node, node_id, prompt,
-                output, input_schema, output_schema, timeout, needs_worktree, verify_fn, sleep, now,
+                output, input_schema, output_schema, timeout, effective_needs_worktree, verify_fn, sleep, now,
             )
  
         outputs[node_id] = output