集中 Run 文件路径公式

新增无状态的 run/layout.py，统一 Plan、Task、Candidate、Delivery、Validation、Stage、Checkpoint、工具输出和运行锁等路径计算。

各调用方只复用纯路径函数，不引入 Repository 或有状态 RunLayout；回归测试逐项锁定旧目录与文件名，确保历史 Run 的持久化布局完全不变。

production_build_agents/agents/executor/agent.py

@@ -26,6 +26,11 @@ from ...run.langgraph_checkpointer import (
     checkpoint_messages,
     create_sqlite_checkpointer,
 )
+from ...run.layout import (
+    delivery_artifact_path,
+    executor_candidate_path,
+    tool_output_dir,
+)
 from ...contracts.models import (
     Artifact,
     ArtifactExpectationBinding,
@@ -664,19 +669,15 @@ def run_executor_agent(
     skill: ExecutorSkill = load_executor_skill(planned_task.skill_id)
     validate_task_source_paths(planned_task, brief)
     executor_run_id = create_executor_run_id(task)
-    expected_manifest_path = (
-        run_dir
-        / "artifacts"
-        / task.task_id
-        / f"v{task.plan_version}"
-        / "executor_candidate.json"
+    expected_manifest_path = executor_candidate_path(
+        run_dir,
+        task.task_id,
+        task.plan_version,
     )
-    expected_artifact_path = (
-        run_dir
-        / "artifacts"
-        / task.task_id
-        / f"v{task.plan_version}"
-        / "delivery_artifact.json"
+    expected_artifact_path = delivery_artifact_path(
+        run_dir,
+        task.task_id,
+        task.plan_version,
     )
     payload = build_executor_user_payload(
         executor_run_id=executor_run_id,
@@ -697,7 +698,7 @@ def run_executor_agent(
             side_effect_tools=SIDE_EFFECT_TOOL_IDS,
         )
         registry = tool_registry or create_default_tool_registry(
-            output_dir=run_dir / "tool_outputs",
+            output_dir=tool_output_dir(run_dir),
             allowed_remote_tool_ids=(
                 set(skill.allowed_remote_tool_ids)
                 if skill.allowed_remote_tool_ids is not None

production_build_agents/agents/validator/stage_agent.py

@@ -25,6 +25,7 @@ from ..invocation import (
 )
 from ...tools.brief_reader import create_brief_reader_tool
 from ...run.langgraph_checkpointer import create_sqlite_checkpointer
+from ...run.layout import stage_validator_tool_output_dir
 from ...preprocess.brief_access import (
     build_allowed_source_paths,
     build_global_data_audit_paths,
@@ -141,7 +142,7 @@ def run_global_data_stage_validator(
         filename="agent_checkpoints.sqlite",
     ) as checkpointer:
         media_registry = create_default_tool_registry(
-            output_dir=run_dir / "stage_validator_tool_outputs",
+            output_dir=stage_validator_tool_output_dir(run_dir),
             allowed_remote_tool_ids=set(),
             run_dir=run_dir,
             executor_run_id=f"{validator_run_id}-evidence",

production_build_agents/agents/validator/task_agent.py

@@ -24,6 +24,7 @@ from ...run.langgraph_checkpointer import (
     checkpoint_messages,
     create_sqlite_checkpointer,
 )
+from ...run.layout import validator_tool_output_dir
 from ...contracts.models import (
     ExecutorDelivery,
     GlobalDataPlan,
@@ -255,13 +256,13 @@ def run_validator_agent(
         )
         if tool_registry is None:
             evidence_registry = create_default_tool_registry(
-                output_dir=run_dir / "validator_tool_outputs",
+                output_dir=validator_tool_output_dir(run_dir),
                 allowed_remote_tool_ids=set(),
                 run_dir=run_dir,
                 executor_run_id=f"{validator_run_id}-evidence",
             )
             agent_registry = create_default_tool_registry(
-                output_dir=run_dir / "validator_tool_outputs",
+                output_dir=validator_tool_output_dir(run_dir),
                 allowed_remote_tool_ids=set(),
                 run_dir=run_dir,
                 executor_run_id=f"{validator_run_id}-agent",

production_build_agents/global_data_stage_assembly.py

@@ -36,6 +36,10 @@ from .run.artifacts import (
     ArtifactIntegrityError,
     verify_delivery_artifacts,
 )
+from .run.layout import (
+    stage_validation_report_path,
+    task_package_path,
+)
 
 
 class GlobalDataStageError(ValueError):
@@ -82,13 +86,10 @@ def _read_task_evidence(
                 Path(record.validation_report_path).read_text(encoding="utf-8")
             )
             task = TaskPackage.model_validate_json(
-                (
-                    run_dir
-                    / "tasks"
-                    / (
-                        f"{planned.task_id}."
-                        f"v{record.accepted_plan_version}.json"
-                    )
+                task_package_path(
+                    run_dir,
+                    planned.task_id,
+                    record.accepted_plan_version,
                 ).read_text(encoding="utf-8")
             )
         except (OSError, ValueError) as exc:
@@ -338,10 +339,9 @@ def build_global_data_stage_delivery(
         plan_history=_plan_history(plan, plan_history),
         requirement_evaluations=evaluation.requirement_evaluations,
         stage_validation_report_uri=str(
-            (
-                run_dir
-                / "stage_validation_results"
-                / f"global_data.v{report.plan_version}.json"
+            stage_validation_report_path(
+                run_dir,
+                report.plan_version,
             ).resolve()
         ),
         tasks=rebuilt.tasks,

production_build_agents/graph_nodes/execute_task.py

@@ -11,6 +11,7 @@ from langgraph.graph import END
 from ..agents.executor.agent import ExecutorOutputError, run_executor_agent
 from ..agents.executor.context import create_executor_run_id
 from ..contracts.models import ExecutorDelivery, TaskPackage, TaskRecord
+from ..run.layout import executor_delivery_path
 from ..run.operation_journal import OperationOutcomeUnknownError
 from ..run.records import VersionConflictError, save_executor_delivery
 from ..state import ProductionState
@@ -18,11 +19,7 @@ from .common import terminal_failure
 
 
 def delivery_path(run_dir: Path, task: TaskPackage) -> Path:
-    return (
-        run_dir
-        / "executor_results"
-        / f"{task.task_id}.v{task.plan_version}.json"
-    )
+    return executor_delivery_path(run_dir, task.task_id, task.plan_version)
 
 
 def load_existing_delivery(

production_build_agents/graph_nodes/prepare_brief.py

@@ -8,6 +8,7 @@ from typing import Literal
 from langgraph.graph import END
 
 from ..preprocess.production_brief import preprocess_file
+from ..run.layout import preprocess_report_path, production_brief_path
 from ..state import ProductionState
 from .common import terminal_failure
 
@@ -25,8 +26,8 @@ def preprocess_node(state: ProductionState) -> dict[str, object]:
             event="拒绝恢复或启动非 0.3 Run",
         )
     output_dir = Path(state["output_dir"])
-    brief_path = output_dir / "production_brief.json"
-    report_path = output_dir / "preprocess_report.md"
+    brief_path = production_brief_path(output_dir)
+    report_path = preprocess_report_path(output_dir)
     try:
         outcome = preprocess_file(
             Path(state["input_path"]),

production_build_agents/graph_nodes/validate_stage.py

@@ -22,6 +22,7 @@ from ..global_data_stage_assembly import (
     GlobalDataStageError,
     build_global_data_stage_candidate,
 )
+from ..run.layout import stage_validation_report_path
 from ..run.operation_journal import OperationOutcomeUnknownError
 from ..run.records import (
     VersionConflictError,
@@ -33,11 +34,7 @@ from .common import terminal_failure
 
 
 def stage_validation_path(run_dir: Path, plan_version: int) -> Path:
-    return (
-        run_dir
-        / "stage_validation_results"
-        / f"global_data.v{plan_version}.json"
-    )
+    return stage_validation_report_path(run_dir, plan_version)
 
 
 def validate_global_data_stage_node(

production_build_agents/graph_nodes/validate_task.py

@@ -21,6 +21,7 @@ from ..contracts.models import (
     TaskRecord,
     ValidationReport,
 )
+from ..run.layout import validation_report_path
 from ..run.operation_journal import OperationOutcomeUnknownError
 from ..run.records import VersionConflictError, save_validation_report
 from ..state import ProductionState
@@ -28,11 +29,7 @@ from .common import terminal_failure
 
 
 def validation_path(run_dir: Path, task: TaskPackage) -> Path:
-    return (
-        run_dir
-        / "validation_results"
-        / f"{task.task_id}.v{task.plan_version}.json"
-    )
+    return validation_report_path(run_dir, task.task_id, task.plan_version)
 
 
 def validate_task_node(

production_build_agents/run/langgraph_checkpointer.py

@@ -9,6 +9,8 @@ from typing import Any, Iterator
 from langgraph.checkpoint.memory import InMemorySaver
 from langgraph.checkpoint.sqlite import SqliteSaver
 
+from .layout import checkpoint_path
+
 
 def create_in_memory_checkpointer() -> InMemorySaver:
     """为本地运行或单个测试创建隔离的内存 checkpoint。
@@ -30,7 +32,7 @@ def create_sqlite_checkpointer(
     root = run_dir.resolve()
     root.mkdir(parents=True, exist_ok=True)
     with SqliteSaver.from_conn_string(
-        str(root / filename)
+        str(checkpoint_path(root, filename))
     ) as saver:
         yield saver
 

production_build_agents/run/layout.py

@@ -0,0 +1,130 @@
+"""Production Run 的纯路径公式；不创建目录、不读写文件。"""
+
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+
+
+def run_directory(output_root: Path, thread_id: str) -> Path:
+    return output_root / thread_id
+
+
+def run_lock_path(run_dir: Path) -> Path:
+    return run_dir / ".run.lock"
+
+
+def checkpoint_path(run_dir: Path, filename: str = "checkpoints.sqlite") -> Path:
+    return run_dir / filename
+
+
+def production_brief_path(run_dir: Path) -> Path:
+    return run_dir / "production_brief.json"
+
+
+def preprocess_report_path(run_dir: Path) -> Path:
+    return run_dir / "preprocess_report.md"
+
+
+def plan_json_path(run_dir: Path, plan_version: int) -> Path:
+    return run_dir / "plans" / f"global_data_dag.v{plan_version}.json"
+
+
+def plan_mermaid_path(run_dir: Path, plan_version: int) -> Path:
+    return run_dir / "plans" / f"global_data_dag.v{plan_version}.mmd"
+
+
+def task_package_path(
+    run_dir: Path,
+    task_id: str,
+    plan_version: int,
+) -> Path:
+    return run_dir / "tasks" / f"{task_id}.v{plan_version}.json"
+
+
+def executor_candidate_path(
+    run_dir: Path,
+    task_id: str,
+    plan_version: int,
+) -> Path:
+    return (
+        run_dir
+        / "artifacts"
+        / task_id
+        / f"v{plan_version}"
+        / "executor_candidate.json"
+    )
+
+
+def delivery_artifact_path(
+    run_dir: Path,
+    task_id: str,
+    plan_version: int,
+) -> Path:
+    return (
+        run_dir
+        / "artifacts"
+        / task_id
+        / f"v{plan_version}"
+        / "delivery_artifact.json"
+    )
+
+
+def executor_delivery_path(
+    run_dir: Path,
+    task_id: str,
+    plan_version: int,
+) -> Path:
+    return run_dir / "executor_results" / f"{task_id}.v{plan_version}.json"
+
+
+def validation_report_path(
+    run_dir: Path,
+    task_id: str,
+    plan_version: int,
+) -> Path:
+    return run_dir / "validation_results" / f"{task_id}.v{plan_version}.json"
+
+
+def stage_candidate_path(run_dir: Path, plan_version: int) -> Path:
+    return run_dir / "stage_candidates" / f"global_data.v{plan_version}.json"
+
+
+def stage_validation_report_path(
+    run_dir: Path,
+    plan_version: int,
+) -> Path:
+    return (
+        run_dir
+        / "stage_validation_results"
+        / f"global_data.v{plan_version}.json"
+    )
+
+
+def global_data_delivery_path(run_dir: Path) -> Path:
+    return run_dir / "global_data_stage_delivery.json"
+
+
+def run_summary_path(run_dir: Path) -> Path:
+    return run_dir / "run_summary.json"
+
+
+def run_metrics_path(run_dir: Path) -> Path:
+    return run_dir / "run_metrics.json"
+
+
+def tool_output_dir(run_dir: Path) -> Path:
+    return run_dir / "tool_outputs"
+
+
+def validator_tool_output_dir(run_dir: Path) -> Path:
+    return run_dir / "validator_tool_outputs"
+
+
+def stage_validator_tool_output_dir(run_dir: Path) -> Path:
+    return run_dir / "stage_validator_tool_outputs"
+
+
+def tool_operation_path(run_dir: Path, operation_id: str) -> Path:
+    filename = hashlib.sha256(operation_id.encode("utf-8")).hexdigest()
+    return run_dir / "tool_operations" / f"{filename}.json"

production_build_agents/run/lock.py

@@ -7,6 +7,8 @@ from contextlib import contextmanager
 from pathlib import Path
 from typing import Iterator, TextIO
 
+from .layout import run_lock_path
+
 
 class RunLockError(RuntimeError):
     """同一 Run 已由另一个进程持有。"""
@@ -18,7 +20,7 @@ def acquire_run_lock(run_dir: Path) -> Iterator[Path]:
 
     root = run_dir.resolve()
     root.mkdir(parents=True, exist_ok=True)
-    path = root / ".run.lock"
+    path = run_lock_path(root)
     handle: TextIO = path.open("a+", encoding="utf-8")
     try:
         try:

production_build_agents/run/operation_journal.py

@@ -9,6 +9,7 @@ from pathlib import Path
 from threading import Lock
 from typing import Any, Callable, Mapping
 
+from .layout import tool_operation_path
 from .records import VersionConflictError, replace_json
 
 
@@ -51,10 +52,9 @@ class OperationJournal:
     def _next_operation(self) -> tuple[str, Path]:
         self._ordinal += 1
         operation_id = f"{self.operation_scope}:{self._ordinal}"
-        filename = hashlib.sha256(operation_id.encode("utf-8")).hexdigest()
         return (
             operation_id,
-            self.run_dir / "tool_operations" / f"{filename}.json",
+            tool_operation_path(self.run_dir, operation_id),
         )
 
     def execute(

production_build_agents/run/records.py

@@ -18,6 +18,18 @@ from ..contracts.models import (
     TaskPackage,
     ValidationReport,
 )
+from .layout import (
+    executor_candidate_path,
+    executor_delivery_path,
+    global_data_delivery_path,
+    plan_json_path,
+    plan_mermaid_path,
+    run_summary_path,
+    stage_candidate_path,
+    stage_validation_report_path,
+    task_package_path,
+    validation_report_path,
+)
 
 
 class VersionConflictError(RuntimeError):
@@ -91,12 +103,11 @@ def write_once_json(path: Path, payload: Any) -> Path:
 
 
 def save_global_data_plan(run_dir: Path, plan: GlobalDataPlan) -> tuple[Path, Path]:
-    plan_dir = run_dir / "plans"
     json_path = write_once_json(
-        plan_dir / f"global_data_dag.v{plan.plan_version}.json",
+        plan_json_path(run_dir, plan.plan_version),
         plan.model_dump(mode="json"),
     )
-    mermaid_path = plan_dir / f"global_data_dag.v{plan.plan_version}.mmd"
+    mermaid_path = plan_mermaid_path(run_dir, plan.plan_version)
     lines = ["flowchart LR"]
     for task in plan.tasks:
         label = f"{task.task_id}：{task.objective}".replace('"', "'")
@@ -110,7 +121,7 @@ def save_global_data_plan(run_dir: Path, plan: GlobalDataPlan) -> tuple[Path, Pa
 
 def save_task_package(run_dir: Path, task: TaskPackage) -> Path:
     return write_once_json(
-        run_dir / "tasks" / f"{task.task_id}.v{task.plan_version}.json",
+        task_package_path(run_dir, task.task_id, task.plan_version),
         task.model_dump(mode="json"),
     )
 
@@ -120,11 +131,11 @@ def save_executor_candidate(
     candidate: ExecutorCandidate,
 ) -> Path:
     return write_once_json(
-        run_dir
-        / "artifacts"
-        / candidate.task_id
-        / f"v{candidate.plan_version}"
-        / "executor_candidate.json",
+        executor_candidate_path(
+            run_dir,
+            candidate.task_id,
+            candidate.plan_version,
+        ),
         candidate.model_dump(mode="json"),
     )
 
@@ -161,9 +172,11 @@ def executor_delivery_artifact_text(candidate: ExecutorCandidate) -> str:
 
 def save_executor_delivery(run_dir: Path, delivery: ExecutorDelivery) -> Path:
     return write_once_json(
-        run_dir
-        / "executor_results"
-        / f"{delivery.task_id}.v{delivery.plan_version}.json",
+        executor_delivery_path(
+            run_dir,
+            delivery.task_id,
+            delivery.plan_version,
+        ),
         delivery.model_dump(mode="json"),
     )
 
@@ -173,9 +186,11 @@ def save_validation_report(
     report: ValidationReport,
 ) -> Path:
     return write_once_json(
-        run_dir
-        / "validation_results"
-        / f"{report.task_id}.v{report.plan_version}.json",
+        validation_report_path(
+            run_dir,
+            report.task_id,
+            report.plan_version,
+        ),
         report.model_dump(mode="json"),
     )
 
@@ -185,9 +200,7 @@ def save_global_data_stage_candidate(
     candidate: GlobalDataStageCandidate,
 ) -> Path:
     return write_once_json(
-        run_dir
-        / "stage_candidates"
-        / f"global_data.v{candidate.plan_version}.json",
+        stage_candidate_path(run_dir, candidate.plan_version),
         candidate.model_dump(mode="json"),
     )
 
@@ -197,9 +210,7 @@ def save_global_data_stage_validation_report(
     report: GlobalDataStageValidationReport,
 ) -> Path:
     return write_once_json(
-        run_dir
-        / "stage_validation_results"
-        / f"global_data.v{report.plan_version}.json",
+        stage_validation_report_path(run_dir, report.plan_version),
         report.model_dump(mode="json"),
     )
 
@@ -209,7 +220,7 @@ def save_global_data_stage_delivery(
     delivery: GlobalDataStageDelivery,
 ) -> Path:
     return write_once_json(
-        run_dir / "global_data_stage_delivery.json",
+        global_data_delivery_path(run_dir),
         delivery.model_dump(mode="json"),
     )
 
@@ -228,4 +239,4 @@ def save_run_summary(
             *list(state.get("event_log", [])),
             *list(update.get("event_log", [])),
         ]
-    return write_once_json(run_dir / "run_summary.json", merged)
+    return write_once_json(run_summary_path(run_dir), merged)

run_demo.py

@@ -18,6 +18,7 @@ from production_build_agents.run.langgraph_checkpointer import (
     create_sqlite_checkpointer,
 )
 from production_build_agents.run.lock import acquire_run_lock
+from production_build_agents.run.layout import run_directory
 from production_build_agents.run.registry import (
     file_sha256,
     register_thread,
@@ -99,7 +100,7 @@ def run_production(
     if not source.is_file():
         raise FileNotFoundError(f"输入文件不存在：{source}")
     input_sha256 = file_sha256(source)
-    run_dir = (output_root.resolve() / thread_id).resolve()
+    run_dir = run_directory(output_root.resolve(), thread_id).resolve()
     register_thread(
         registry_dir=(
             registry_dir

tests/run/test_layout.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+import hashlib
+import unittest
+from pathlib import Path
+
+from production_build_agents.run.layout import (
+    checkpoint_path,
+    delivery_artifact_path,
+    executor_candidate_path,
+    executor_delivery_path,
+    global_data_delivery_path,
+    plan_json_path,
+    plan_mermaid_path,
+    preprocess_report_path,
+    production_brief_path,
+    run_lock_path,
+    run_directory,
+    run_metrics_path,
+    run_summary_path,
+    stage_candidate_path,
+    stage_validation_report_path,
+    stage_validator_tool_output_dir,
+    task_package_path,
+    tool_operation_path,
+    tool_output_dir,
+    validation_report_path,
+    validator_tool_output_dir,
+)
+
+
+class RunLayoutTest(unittest.TestCase):
+    def test_all_path_formulas_preserve_existing_layout(self) -> None:
+        root = Path("/tmp/production-run")
+        self.assertEqual(
+            run_directory(Path("/tmp"), "production-run"),
+            root,
+        )
+        self.assertEqual(run_lock_path(root), root / ".run.lock")
+        self.assertEqual(checkpoint_path(root), root / "checkpoints.sqlite")
+        self.assertEqual(
+            checkpoint_path(root, "agent_checkpoints.sqlite"),
+            root / "agent_checkpoints.sqlite",
+        )
+        self.assertEqual(
+            production_brief_path(root),
+            root / "production_brief.json",
+        )
+        self.assertEqual(
+            preprocess_report_path(root),
+            root / "preprocess_report.md",
+        )
+        self.assertEqual(
+            plan_json_path(root, 2),
+            root / "plans" / "global_data_dag.v2.json",
+        )
+        self.assertEqual(
+            plan_mermaid_path(root, 2),
+            root / "plans" / "global_data_dag.v2.mmd",
+        )
+        self.assertEqual(
+            task_package_path(root, "Task3", 2),
+            root / "tasks" / "Task3.v2.json",
+        )
+        self.assertEqual(
+            executor_candidate_path(root, "Task3", 2),
+            root / "artifacts" / "Task3" / "v2" / "executor_candidate.json",
+        )
+        self.assertEqual(
+            delivery_artifact_path(root, "Task3", 2),
+            root / "artifacts" / "Task3" / "v2" / "delivery_artifact.json",
+        )
+        self.assertEqual(
+            executor_delivery_path(root, "Task3", 2),
+            root / "executor_results" / "Task3.v2.json",
+        )
+        self.assertEqual(
+            validation_report_path(root, "Task3", 2),
+            root / "validation_results" / "Task3.v2.json",
+        )
+        self.assertEqual(
+            stage_candidate_path(root, 2),
+            root / "stage_candidates" / "global_data.v2.json",
+        )
+        self.assertEqual(
+            stage_validation_report_path(root, 2),
+            root / "stage_validation_results" / "global_data.v2.json",
+        )
+        self.assertEqual(
+            global_data_delivery_path(root),
+            root / "global_data_stage_delivery.json",
+        )
+        self.assertEqual(run_summary_path(root), root / "run_summary.json")
+        self.assertEqual(run_metrics_path(root), root / "run_metrics.json")
+        self.assertEqual(tool_output_dir(root), root / "tool_outputs")
+        self.assertEqual(
+            validator_tool_output_dir(root),
+            root / "validator_tool_outputs",
+        )
+        self.assertEqual(
+            stage_validator_tool_output_dir(root),
+            root / "stage_validator_tool_outputs",
+        )
+
+    def test_operation_path_keeps_hashed_filename(self) -> None:
+        root = Path("/tmp/production-run")
+        operation_id = "run-executor-Task1-v1:3"
+        filename = hashlib.sha256(operation_id.encode("utf-8")).hexdigest()
+        self.assertEqual(
+            tool_operation_path(root, operation_id),
+            root / "tool_operations" / f"{filename}.json",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()