class Intent(BaseModel):
    id: str                           # Intent 的唯一标识
    from_: list[str] = Field(alias="from")  # 源 Fact ID 列表（支持多个）
    to: str | None = None            # 目标 Fact ID（结论）
    description: str                  # 探索方向的描述
    creator: str                      # 创建者
    worker: str | None = None        # 当前认领者
    last_heartbeat_at: str | None = None  # 最后心跳时间
    created_at: str                   # 创建时间
    concluded_at: str | None = None  # 结论时间

class ProjectDetail(BaseModel):
    project: ProjectMeta    # 项目元数据
    facts: list[Fact]       # 图的节点集合
    intents: list[Intent]   # 图的边集合
    hints: list[Hint]       # 图外输入（不属于图结构）

status: Literal["active", "stopped", "completed"]

# services.py:173-178
if body.status == "stopped":
    # 清空所有未结论的 Intent 的 worker
    conn.execute(
        "UPDATE intents SET worker = NULL WHERE project_id = ? AND concluded_at IS NULL",
        (project_id,),
    )
    # 清空项目的 reason lease
    clear_project_reason(conn, project_id)

class UpdateProjectStatusRequest(BaseModel):
    status: Literal["active", "stopped"]
    
    @router.post("/projects/{project_id}/intents/{intent_id}/heartbeat")
def heartbeat(project_id: str, intent_id: str, body: HeartbeatRequest):
    with get_conn() as conn:
        # 检查项目必须是 active 状态
        check_project_active(conn, project_id)
        
        # 检查 intent 是否可以被认领
        get_claimable_open_intent_or_404(conn, project_id, intent_id, body.worker)

        now = utcnow()
        # 重新设置 worker（认领）
        conn.execute(
            "UPDATE intents SET worker = ?, last_heartbeat_at = ? WHERE id = ? AND project_id = ?",
            (body.worker, now, intent_id, project_id),
        )

# claim_project_reason  互斥锁
current_worker = row["reason_worker"]
if current_worker is not None and current_worker != body.worker:
    raise HTTPException(409, f"Project reason is currently claimed by {current_worker}")

class ProjectReason(BaseModel):
    worker: str              # 当前执行 reason 的消费者
    trigger: str             # 本次 reason 的触发原因
    started_at: str          # 本次 reason 开始时间
    last_heartbeat_at: str   # 最近一次 reason 心跳时间

class ProjectMeta(BaseModel):
    id: str
    title: str
    status: Literal["active", "stopped", "completed"]
    created_at: str
    reason: ProjectReason | None = None  # reason 字段

changes: list[str] = []

# 条件 1：Fact 数量增加（发现了新事实）
if len(project.facts) > checkpoint.fact_count:
    changes.append(f"facts:{checkpoint.fact_count}->{len(project.facts)}")

# 条件 2：Hint 数量增加（收到新提示）
if len(project.hints) > checkpoint.hint_count:
    changes.append(f"hints:{checkpoint.hint_count}->{len(project.hints)}")

# 条件 3：从"有 open intents"变为"无 open intents"（所有任务完成）
if checkpoint.open_intent_count > 0 and open_intent_count == 0:
    changes.append(f"open_intents:{checkpoint.open_intent_count}->0")

if not changes:
    return None  # 没有实质性变化，不触发 reason
return ",".join(changes)

@dataclass(slots=True)
class ReasonCheckpoint:
    fact_count: int
    hint_count: int
    open_intent_count: int

class Hint(BaseModel):
    id: str           # 提示的唯一标识
    content: str      # 提示内容
    creator: str      # 创建者
    created_at: str   # 创建时间

def check_project_hint_writable(conn: sqlite3.Connection, project_id: str) -> sqlite3.Row:
    row = get_project_or_404(conn, project_id)
    # Hint 在 active、stopped、completed 状态下都允许写入
    if row["status"] not in ("active", "stopped", "completed"):
        raise HTTPException(403, f"Project is {row['status']}")
    return row
    
    # reason 上下文准备prompt = render_prompt(
    load_prompt(config.runtime.prompt_group, "reason.md"),
    {
        "graph_yaml": export_yaml.strip(),      # 包含 hints 的完整图数据
        "fact_ids": format_fact_ids(allowed_fact_ids),  # 可用 Fact ID 列表
        "open_intents": format_open_intents(open_intents),  # 未完成的 Intents
        "max_intents": str(config.tasks.reason.max_intents),  # 最多创建几个新 intent
    },
)

def _is_initial_project(self, project: ProjectDetail) -> bool:
    # 只有 origin 和 goal 两个事实,且所有 intent 都是 bootstrap 类型
    fact_ids = {fact.id for fact in project.facts}
    if fact_ids != {"origin", "goal"} or len(project.facts) != 2:
        return False
    if not project.intents:
        return True
    return all(self._is_bootstrap_intent(intent) for intent in project.intents)

def build_execute(self, worker: WorkerConfig, prompt: str, session: str | None) -> DriverResult:
    assert session is not None
    return DriverResult(
        argv=[
            "claude",                        # Claude Code CLI
            "--session-id", session,         # 会话 ID,用于保持上下文
            "--dangerously-skip-permissions", # 跳过权限确认
            "-p",                            # 表示后面是 prompt
            "--",                            # 参数分隔符
            prompt,                          # ← 这里就是渲染后的完整提示词
        ],
        session=session,
    )

# containers.py#L181-L201

def build_exec_process(
    self,
    container_name: str,
    env: dict[str, str],      # 环境变量 (API Key 等)
    command: list[str],       # 如 ["claude", "-p", "--", "<prompt>"]
    timeout_seconds: int | None = None,
    kill_after_seconds: int = 5,
) -> ManagedProcess:
    container = self._require_container(container_name)
    
    argv: list[str] = []
    if timeout_seconds is not None:
        # 添加 Linux timeout 命令包装
        # 300秒超时，超时后等5秒发 SIGKILL
        argv.extend([
            "timeout",
            "-k", f"{kill_after_seconds}s",  # -k: 超时后强制kill
            f"{timeout_seconds}s",           # 超时时间
        ])
    
    argv.extend(command)
    # 最终: ["timeout", "-k", "5s", "300s", "claude", "-p", "--", "..."]
    
    return ManagedProcess(container, argv, env)

if timeout_seconds is not None:
       argv.extend([
           "timeout",          # Linux 的 timeout 命令
           "-k", f"{kill_after_seconds}s",  # 强制杀死的等待时间
           f"{timeout_seconds}s",           # 超时时间

def run_bootstrap_task(..., intent, ...):
   创建 session（内存中的 UUID）
    session = driver.prepare_session()
        # execute 阶段使用 session
    execute = driver.build_execute(worker, prompt, session)
    first = run_worker_process(..., execute.argv, ...)
        # 提取 session
    session = driver.extract_session(session, first.stderr)
        # 如果 execute 失败，使用同一个 session 恢复对话
    # fallback 到 conclude
    return _try_conclude_fallback(
        ...,
        intent,
        session,  # ← 传递同一个 session
        ...
    )
        # 如果 execute 成功，直接返回结果
    # session 被丢弃，不再需要
    return "success"

def run_startup_healthchecks(config, container_manager, *, show_commands=False):
    # 创建临时容器
    container_name = container_manager.create_startup_container()
        # 并行检查所有 worker
    with ThreadPoolExecutor(max_workers=parallelism) as executor:
        future_map = {
            executor.submit(
                _run_worker_healthcheck,
                container_manager, container_name, worker,
                config.runtime.healthcheck_timeout,
            ): worker.name
            for worker in workers
        }
        # 最后清理临时容器
    container_manager.remove_container(container_name, force=True)

class ManagedProcess:  # ← 自定义类
    def __init__(self, container: Container, command: list[str], env: dict[str, str]):
        self.command = command                              # 自定义
        self.env = env                                      # 自定义
        self._container = container                         # Docker SDK 原生对象
        self._api = container.client.api                   # Docker SDK 原生 API 客户端
        
        self._exec_id: str | None = None                    # Docker 返回的 exec ID
        self._reader: threading.Thread | None = None        # 自定义线程
        self._stdout: list[str] = []                        # 自定义缓冲
        self._stderr: list[str] = []                        # 自定义缓冲
        self._returncode: int | None = None                 # Docker 返回的退出码
        self._timed_out = False                             # 自定义标志
        self._cancel_reason: str | None = None              # 自定义
        self._read_error: str | None = None                 # 自定义
        self._done = threading.Event()                      # 自定义同步原语

try:
    payload = parse_json_output(first.stdout)
    kind, data = validate_bootstrap_execute_payload(payload)
except Exception as exc:
    # 解析失败时
    LOG.warning(
        "bootstrap parse failed project=%s intent=%s worker=%s error=%s ... stdout_preview=%s stderr_preview=%s",
        project.project.id,
        intent.id,
        worker.name,
        exc,
        preview(first.stdout),      # 只记录日志（截取 1200 字符）
        preview(first.stderr),
    )
    # 触发 conclude fallback，尝试让 Agent 重新输出
    return _try_conclude_fallback(...)  # 不保留非 JSON 信息

{
  "phase": "reason",
  "fact_ids": {fact_ids},
  "open_intents": {open_intents},
  "max_intents": {max_intents}
}

# loop.py 第 71-100 行
def run(self, once: bool = False) -> None:
    try:
        self.run_startup_healthchecks()
        while True:  # 主调度循环
            try:
                if not self._settings_checked:
                    self._validate_server_settings()
                    self._settings_checked = True
                
                self._reap_futures()              # 1. 收集已完成的任务
                self._reap_cleanup_futures()      # 2. 收集已完成的清理任务
                summaries = self.client.list_projects()
                self._initialize_reason_checkpoints(summaries)
                self._refresh_runtime_projects(summaries)
                self._cancel_inactive_tasks(summaries)
                self._queue_container_cleanups(summaries)  # 3. 队列化容器清理
                self._dispatch_available(summaries)        # 4. 调度新任务
            except requests.RequestException as exc:
                if once:
                    raise
                LOG.warning("dispatcher server request failed ...")
                time.sleep(self.config.runtime.interval)
                continue
            
            if once:
                break
            time.sleep(self.config.runtime.interval)  # 每隔 interval 秒循环一次
    finally:
        ...

# 主循环中的清理操作
def _queue_container_cleanups(self, summaries: list[ProjectSummary]) -> None:
    # 只是提交任务到线程池，立即返回
    future = self.cleanup_executor.submit(
        self.container_manager.cleanup_completed, 
        summary.id
    )
    # 不等待 future 完成，继续执行后续代码

def create_executor(config: DispatchConfig) -> ExecutorProtocol:
    if config.execution_mode == "container":
        return ContainerManager(config.container)
    if config.execution_mode == "local":
        return LocalExecutor(config.workspace_root)