"""
api.v1.agents — Agent CRUD + Execution endpoints.

POST /api/v1/agents                    Create agent
GET  /api/v1/agents                    List agents
GET  /api/v1/agents/{agent_id}         Get agent
PUT  /api/v1/agents/{agent_id}         Update agent (auto version)
DELETE /api/v1/agents/{agent_id}       Soft delete
POST /api/v1/agents/{agent_id}/run     Synchronous execution
POST /api/v1/agents/{agent_id}/rollback  Rollback version
"""
from __future__ import annotations
import json
import hashlib
import time
from typing import Any, Dict, Optional

from fastapi import APIRouter, Depends, Header, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

from agentpaas.api.deps import get_tenant, get_database
from agentpaas.api.errors import api_error
from agentpaas.api.middleware.auth import TenantContext
from agentpaas.db.models import Database, gen_id, now_utc
from agentpaas.observability.logging import logger
from agentpaas.tenant.rbac import require_permission
from agentpaas.tenant.quota import check_concurrency, check_and_reserve

router = APIRouter(prefix="/agents", tags=["agents"])


# ── Request/Response Models ──

class CreateAgentRequest(BaseModel):
    name: str = Field(..., max_length=256)
    description: str = Field(default="", max_length=4096)
    config: Dict[str, Any]
    tags: list = Field(default_factory=list)
    environment: str = Field(default="production", max_length=64)
    agent_dir: str = Field(default="", description="Agent template/system directory path")
    agent_template: str = Field(default="", description="Agent template name (e.g. qaagent67wiki)")
    instance_dir: str = Field(default="", description="Instance data directory for this agent")
    kb_ids: list = Field(default_factory=list, description="Linked knowledge base IDs")
    kb_search_mode: str = Field(default="bm25", description="bm25 | pageindex | fusion")
    work_dir: str = Field(default="", description="工作产物目录: Bash tool CWD; 最终输出文件落这里")
    source_dir: str = Field(default="", description="数据来源目录: 只读输入数据; 注入 system prompt 前缀")
    run_dir: str = Field(default="", description="运行日志目录: run workspace/run_{ts}/ 的父目录; 若空则用 work_dir")

class UpdateAgentRequest(BaseModel):
    config: Dict[str, Any]
    changelog: str = ""
    agent_dir: str = ""
    work_dir: str = ""
    source_dir: str = ""
    run_dir: str = ""
    kb_ids: Optional[list] = None
    kb_search_mode: Optional[str] = None

class UpdateKBsRequest(BaseModel):
    kb_ids: list = Field(default_factory=list)
    kb_search_mode: str = Field(default="bm25")

class RunRequest(BaseModel):
    input: str = Field(..., max_length=102400)  # 100KB max
    parameters: Dict[str, Any] = Field(default_factory=dict)
    context: Dict[str, Any] = Field(default_factory=dict)
    # Continue-run UI modes — see `_apply_mode_prompt()` below.
    # iterate: full pipeline (default, unchanged behavior)
    # edit:    bypass orchestrator's free decision, call target_subagent once
    # chat:    read workspace, answer question, never invoke sub-agents
    mode: str = Field(default="iterate", description="iterate | edit | chat")
    target_subagent: str = Field(
        default="",
        description="Required for mode=edit: name of the sub-agent to invoke (e.g. 'call_physwriter')",
    )
    # 会话记忆 P1：本消息归属的对话 thread；缺省则新建。与 continue-run
    # 正交（thread=对话语境，continue=产物复用，仅 context.run_id 显式驱动）。
    thread_id: str = Field(default="", description="对话线程 id；缺省自动新建")

class RollbackRequest(BaseModel):
    target_version: int


# ── S21: Security Audit Logging ──

def _audit_log(db: Database, event_type: str, tenant_id: str,
               agent_id: str = "", details: str = ""):
    """S21: Record security-relevant events to audit_log table."""
    try:
        db.execute(
            "INSERT INTO audit_log (id, event_type, tenant_id, agent_id, details, created_at) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            (gen_id("aud_"), event_type, tenant_id, agent_id, details[:4096], now_utc())
        )
        db.commit()
    except Exception:
        pass  # Audit logging should never break the main flow


# ── CRUD Endpoints ──

@router.post("", status_code=201)
async def create_agent(
    req: CreateAgentRequest,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Create a new agent. Auto-compiles config via lambdagent."""
    _audit_log(db, "agent.create", tenant.tenant_id, details=f"name={req.name}")
    agent_id = gen_id("ag_")
    config_json = json.dumps(req.config, ensure_ascii=False)
    config_hash = hashlib.sha256(config_json.encode()).hexdigest()

    # Validate config by compiling (fail fast)
    try:
        _compile_agent(req.config)
    except Exception as e:
        api_error(400, "INVALID_CONFIG", str(e))

    # Resolve instance: if instance_dir is set, load and merge config
    agent_dir = req.agent_dir
    instance_dir = req.instance_dir
    if instance_dir:
        from agentpaas.engine.instance import load_instance_from_dirs
        if agent_dir:
            merged = load_instance_from_dirs(agent_dir, instance_dir)
            # Use instance-merged config for validation
            config_json = json.dumps(merged, ensure_ascii=False, default=str)
            config_hash = hashlib.sha256(config_json.encode()).hexdigest()
            # Re-validate
            try:
                _compile_agent(merged)
            except Exception as e:
                api_error(400, "INVALID_CONFIG", f"Instance config error: {e}")

    # 自动开辟工作目录（参照 Workspace/multi-reviewer67 布局）：
    # 三个目录全空时建 <workspace_base>/<名>/{data,workspace}；
    # 用户显式给了任何一个则完全尊重输入。
    work_dir, source_dir, run_dir = req.work_dir, req.source_dir, req.run_dir
    if not (work_dir or source_dir or run_dir):
        from agentpaas.engine.workspace_provision import provision_agent_workspace
        dirs = provision_agent_workspace(req.name)
        work_dir, source_dir, run_dir = (
            dirs["work_dir"], dirs["source_dir"], dirs["run_dir"])

    now = now_utc()
    db.execute(
        "INSERT INTO agents (id, tenant_id, name, description, current_version, tags, environment, "
        "agent_dir, agent_template, instance_dir, kb_ids, kb_search_mode, "
        "work_dir, source_dir, run_dir, status, created_at, updated_at) "
        "VALUES (?, ?, ?, ?, 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', ?, ?)",
        (agent_id, tenant.tenant_id, req.name, req.description,
         json.dumps(req.tags), req.environment,
         agent_dir, req.agent_template, instance_dir,
         json.dumps(req.kb_ids), req.kb_search_mode,
         work_dir, source_dir, run_dir, now, now)
    )
    db.execute(
        "INSERT INTO agent_versions (agent_id, version, config, config_hash, changelog, created_by, created_at) "
        "VALUES (?, 1, ?, ?, 'Initial version', ?, ?)",
        (agent_id, config_json, config_hash, tenant.user_id, now)
    )
    db.commit()

    return {
        "agent_id": agent_id,
        "version": 1,
        "created_at": now,
        "endpoint": f"/api/v1/agents/{agent_id}",
        "work_dir": work_dir,
        "source_dir": source_dir,
        "run_dir": run_dir,
    }


@router.get("")
async def list_agents(
    tag: Optional[str] = None,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """List all agents for this tenant."""
    sql = "SELECT * FROM agents WHERE tenant_id = ? AND status = 'active'"
    params = [tenant.tenant_id]
    if tag:
        # SEC-08/SEC-12: Escape SQL LIKE wildcards
        safe_tag = tag.replace("%", "\\%").replace("_", "\\_")
        sql += " AND tags LIKE ? ESCAPE '\\'"
        params.append(f'%"{safe_tag}"%')
    sql += " ORDER BY updated_at DESC"
    agents = db.fetchall(sql, tuple(params))
    for a in agents:
        a["tags"] = json.loads(a.get("tags", "[]"))
        a["kb_ids"] = json.loads(a.get("kb_ids") or "[]")
    return {"agents": agents, "count": len(agents)}


@router.get("/{agent_id}")
async def get_agent(
    agent_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Get agent details including current version config."""
    agent = db.fetchone(
        "SELECT * FROM agents WHERE id = ? AND tenant_id = ? AND status = 'active'",
        (agent_id, tenant.tenant_id)
    )
    if not agent:
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")

    version = db.fetchone(
        "SELECT * FROM agent_versions WHERE agent_id = ? AND version = ?",
        (agent_id, agent["current_version"])
    )
    agent["tags"] = json.loads(agent.get("tags", "[]"))
    agent["kb_ids"] = json.loads(agent.get("kb_ids") or "[]")
    agent["config"] = json.loads(version["config"]) if version else {}
    return agent


@router.get("/{agent_id}/subagents/{sub_name}/config")
async def get_subagent_config(
    agent_id: str,
    sub_name: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Return the raw config file contents of a sub-agent (multi-agent orchestrator).

    The orchestrator's config has a `subAgents` map of
    `{name: {tool, config: ./relative/path.yml}}`. We resolve the relative path
    against the orchestrator's `_config_dir` and return the file text so the Web
    UI can display it. Path traversal is blocked — the resolved file must live
    under `_config_dir`.
    """
    import os

    agent = db.fetchone(
        "SELECT * FROM agents WHERE id = ? AND tenant_id = ? AND status = 'active'",
        (agent_id, tenant.tenant_id),
    )
    if not agent:
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")

    version = db.fetchone(
        "SELECT * FROM agent_versions WHERE agent_id = ? AND version = ?",
        (agent_id, agent["current_version"]),
    )
    cfg = json.loads(version["config"]) if version else {}
    sub_agents = cfg.get("subAgents") or cfg.get("sub_agents") or {}

    entry = sub_agents.get(sub_name)
    if entry is None:
        api_error(404, "SUBAGENT_NOT_FOUND", f"Sub-agent '{sub_name}' not found")

    # An entry may be an inline definition (no file) or a file reference.
    if not isinstance(entry, dict) or not entry.get("config"):
        # Inline sub-agent — return the definition itself as YAML-ish JSON.
        import yaml  # local import; PyYAML is a project dependency
        return {
            "agent_id": agent_id,
            "sub_name": sub_name,
            "path": None,
            "inline": True,
            "content": yaml.safe_dump(entry, allow_unicode=True, sort_keys=False),
        }

    rel_path = entry["config"]
    config_dir = cfg.get("_config_dir") or agent.get("source_dir") or agent.get("agent_dir") or ""
    if not config_dir or not os.path.isdir(config_dir):
        api_error(
            404, "CONFIG_DIR_MISSING",
            "Orchestrator config directory is unknown on this machine; "
            "cannot resolve the sub-agent file.",
        )

    base = os.path.realpath(config_dir)
    target = os.path.realpath(os.path.join(base, rel_path))

    # Block path traversal: the resolved file must stay under config_dir.
    if os.path.commonpath([base, target]) != base:
        api_error(400, "INVALID_PATH", "Sub-agent path escapes the config directory.")
    if not os.path.isfile(target):
        api_error(404, "FILE_NOT_FOUND", f"Sub-agent file not found: {rel_path}")

    try:
        with open(target, "r", encoding="utf-8") as f:
            content = f.read()
    except OSError as e:
        api_error(500, "READ_FAILED", f"Could not read sub-agent file: {e}")

    return {
        "agent_id": agent_id,
        "sub_name": sub_name,
        "tool": entry.get("tool", f"call_{sub_name}"),
        "path": target,
        "rel_path": rel_path,
        "inline": False,
        "content": content,
    }


@router.put("/{agent_id}")
async def update_agent(
    agent_id: str,
    req: UpdateAgentRequest,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Update agent config. Auto-increments version."""
    require_permission(tenant, "agents:write")
    _audit_log(db, "agent.update", tenant.tenant_id, agent_id)
    agent = db.fetchone(
        "SELECT * FROM agents WHERE id = ? AND tenant_id = ? AND status = 'active'",
        (agent_id, tenant.tenant_id)
    )
    if not agent:
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")

    config_json = json.dumps(req.config, ensure_ascii=False)
    config_hash = hashlib.sha256(config_json.encode()).hexdigest()

    # Check if config actually changed
    last_version = db.fetchone(
        "SELECT config_hash FROM agent_versions WHERE agent_id = ? ORDER BY version DESC LIMIT 1",
        (agent_id,)
    )
    config_changed = not (last_version and last_version["config_hash"] == config_hash)

    now = now_utc()
    new_version = agent["current_version"]

    if config_changed:
        # Validate before inserting a new version
        try:
            _compile_agent(req.config)
        except Exception as e:
            api_error(400, "INVALID_CONFIG", str(e))

        new_version = agent["current_version"] + 1
        db.execute(
            "INSERT INTO agent_versions (agent_id, version, config, config_hash, changelog, created_by, created_at) "
            "VALUES (?, ?, ?, ?, ?, ?, ?)",
            (agent_id, new_version, config_json, config_hash, req.changelog, tenant.user_id, now)
        )

    # Always persist directory and metadata fields regardless of whether
    # config changed — prior bug: early return on "config unchanged" meant
    # work_dir / source_dir / run_dir edits were silently dropped.
    update_fields = "agent_dir = ?, work_dir = ?, source_dir = ?, run_dir = ?, updated_at = ?"
    update_vals   = [req.agent_dir, req.work_dir, req.source_dir, req.run_dir, now]
    if config_changed:
        update_fields = "current_version = ?, " + update_fields
        update_vals   = [new_version] + update_vals
    if req.kb_ids is not None:
        update_fields += ", kb_ids = ?"
        update_vals.append(json.dumps(req.kb_ids))
    if req.kb_search_mode is not None:
        update_fields += ", kb_search_mode = ?"
        update_vals.append(req.kb_search_mode)
    update_vals.append(agent_id)
    db.execute(f"UPDATE agents SET {update_fields} WHERE id = ?", tuple(update_vals))
    db.commit()

    if config_changed:
        return {"agent_id": agent_id, "version": new_version, "updated_at": now}
    return {"agent_id": agent_id, "version": new_version, "updated_at": now,
            "message": "Config unchanged; directory/metadata updated"}


@router.delete("/{agent_id}")
async def delete_agent(
    agent_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Soft delete agent."""
    require_permission(tenant, "agents:write")
    _audit_log(db, "agent.delete", tenant.tenant_id, agent_id)
    db.execute(
        "UPDATE agents SET status = 'deleted', updated_at = ? WHERE id = ? AND tenant_id = ?",
        (now_utc(), agent_id, tenant.tenant_id)
    )
    db.commit()
    return {"message": "Agent deleted", "agent_id": agent_id}


class SwitchModelRequest(BaseModel):
    provider: str = Field(..., description="Provider id, e.g. dashscope / ollama / claude-code")
    name: str = Field(..., description="Model name, e.g. qwen-max / qwen2.5:7b / sonnet")


@router.put("/{agent_id}/model")
async def switch_agent_model(
    agent_id: str,
    req: SwitchModelRequest,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """轻量模型切换（模型页「显示与切换」功能）。

    只改当前配置的 model.provider / model.name，temperature、maxTokens
    等其余字段原样保留；生成新 agent_version（可在版本历史回滚）。
    """
    require_permission(tenant, "agents:write")
    agent = db.fetchone(
        "SELECT * FROM agents WHERE id = ? AND tenant_id = ? AND status = 'active'",
        (agent_id, tenant.tenant_id))
    if not agent:
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")

    ver = db.fetchone(
        "SELECT config FROM agent_versions WHERE agent_id = ? AND version = ?",
        (agent_id, agent["current_version"]))
    if not ver:
        api_error(404, "VERSION_NOT_FOUND", "current version config missing")

    import hashlib
    config = json.loads(ver["config"] or "{}")
    old_model = config.get("model") or {}
    if isinstance(old_model, str):
        old_model = {"name": old_model}
    old_desc = f"{old_model.get('provider', '?')}/{old_model.get('name', '?')}"

    new_model = dict(old_model)
    new_model["provider"] = req.provider
    new_model["name"] = req.name
    config["model"] = new_model

    config_json = json.dumps(config, ensure_ascii=False)
    config_hash = hashlib.sha256(config_json.encode()).hexdigest()
    new_version = int(agent["current_version"]) + 1
    now = now_utc()
    db.execute(
        "INSERT INTO agent_versions (agent_id, version, config, config_hash, "
        "changelog, created_by, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
        (agent_id, new_version, config_json, config_hash,
         f"模型切换: {old_desc} → {req.provider}/{req.name}",
         tenant.user_id, now))
    db.execute(
        "UPDATE agents SET current_version = ?, updated_at = ? WHERE id = ?",
        (new_version, now, agent_id))
    db.commit()
    return {"agent_id": agent_id, "version": new_version,
            "provider": req.provider, "model": req.name,
            "previous": old_desc}


@router.post("/{agent_id}/rollback")
async def rollback_agent(
    agent_id: str,
    req: RollbackRequest,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Rollback to a previous version.

    audit/AUDIT_2026-06-05.md #14 fix: previously this endpoint had three
    holes that the sibling `update_agent` / `delete_agent` did not:
      (1) no `require_permission` check — read-only keys could trigger;
      (2) version SELECT lacked `tenant_id` filter — tenant A could
          enumerate tenant B's version numbers via 404-vs-200 oracle;
      (3) UPDATE's `tenant_id = ?` clause silently no-op'd on mismatch
          but the endpoint still returned `"Rolled back"`;
      (4) no `_audit_log` call — silent breach.
    All four now closed below.
    """
    require_permission(tenant, "agents:write")

    # Tenant-scoped version lookup: subquery anchors to this tenant's agents
    # so a successful 200 vs 404 is the SAME for "wrong tenant" and
    # "version does not exist" — no oracle.
    version = db.fetchone(
        "SELECT v.* FROM agent_versions v "
        "JOIN agents a ON v.agent_id = a.id "
        "WHERE v.agent_id = ? AND v.version = ? AND a.tenant_id = ?",
        (agent_id, req.target_version, tenant.tenant_id)
    )
    if not version:
        api_error(404, "VERSION_NOT_FOUND", f"Version {req.target_version} not found for agent {agent_id}")

    cur = db.execute(
        "UPDATE agents SET current_version = ?, updated_at = ? WHERE id = ? AND tenant_id = ?",
        (req.target_version, now_utc(), agent_id, tenant.tenant_id)
    )
    # rowcount == 0 means tenant_id mismatch (the agent exists but belongs
    # to another tenant) — same 404 as "not found" to avoid id oracle.
    if cur.rowcount == 0:
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")
    db.commit()

    _audit_log(db, "agent.rollback", tenant.tenant_id, agent_id,
               details=f"version={req.target_version}")

    return {"agent_id": agent_id, "version": req.target_version, "message": "Rolled back"}


# ── Execution Endpoint ──

# 工作区对话模式必须绑定真实工作文件夹。曾出过最坑的失败模式:用户从非
# 「工作区对话」入口(或历史会话)进来,URL 无 ?dir → 前端不发 context.work_dir
# → 后端静默回退到 agent 空家目录 → agent 读不到用户文件、把产物写进日志区、
# 凭会话记忆编造,却以 completed 收尾(假装成功)。这里把它变成显式拒绝:
# 模板为 workspace.assistant 且无有效 context.work_dir 时直接 400,提示先选文件夹。
_WORKSPACE_CONV_TEMPLATES = {"workspace.assistant"}


def _resolve_session_cwd(inplace_dir: str, workspace_path: str, work_dir: str) -> str:
    """会话 CWD(Bash/文件工具相对路径基准)必须 == 给模型看的 [工作目录] 提示
    (= overrides["workspace_path"]),否则相对写入落错目录。见 _execute_agent 注释。
    工作区模式 → F;专项 agent → 本 run 的 workspace_path;兜底 → work_dir。"""
    import os as _os
    if inplace_dir and _os.path.isdir(inplace_dir):
        return _os.path.abspath(inplace_dir)
    if workspace_path and _os.path.isdir(workspace_path):
        return workspace_path
    return work_dir if (work_dir and _os.path.isdir(work_dir)) else workspace_path


def _is_native_workspace(agent, config) -> bool:
    """是否走原生 claude-code 车道:工作区对话型模板 **且** provider=claude-code。
    native 车道是 `claude -p`(只认 claude 自己的模型);切到 qwen/ollama/openai 等
    必须退回 react 车道(text-provider+平台工具),否则会把 --model qwen-max 传给
    claude-code、跑不起来。"""
    if (agent.get("agent_template") or "") not in _WORKSPACE_CONV_TEMPLATES:
        return False
    provider = ((config or {}).get("model") or {}).get("provider") or ""
    return provider == "claude-code"


def _require_workspace_dir(agent, req):
    """workspace 对话型 agent 必须带有效 context.work_dir,否则 400 拒绝(不回退空家目录)。"""
    import os as _os
    if (agent.get("agent_template") or "") not in _WORKSPACE_CONV_TEMPLATES:
        return
    ctx_wd = (req.context or {}).get("work_dir") if isinstance(req.context, dict) else ""
    if not (ctx_wd and isinstance(ctx_wd, str) and _os.path.isdir(ctx_wd)):
        api_error(
            400, "WORKSPACE_DIR_REQUIRED",
            "工作区助手需要先选择工作文件夹再开始对话。请从「💻工作区对话」入口选择文件夹"
            "(URL 会带上 ?dir=…),或点击聊天页顶部的 📁 选择/切换工作文件夹。"
            "(未绑定文件夹时不会运行,以免读不到你的文件还'假装成功'。)",
            details={"agent_template": agent.get("agent_template") or "",
                     "got_work_dir": ctx_wd or ""},
        )


@router.post("/{agent_id}/run")
async def run_agent(
    agent_id: str,
    req: RunRequest,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
    idempotency_key: Optional[str] = Header(None, alias="X-Idempotency-Key"),
):
    """
    Synchronous agent execution.

    AgentPaaS.serve(request) =
        let yaml  = Registry.load(agent_id)        in
        let term  = lambdagent.from_config(yaml)    in
        let result = Runtime.reduce(term, input)    in
        (result, trace)
    """
    # SEC-04: RBAC check
    require_permission(tenant, "agents:execute")

    # Idempotency check — return cached response if already completed
    if idempotency_key:
        existing = db.fetchone(
            "SELECT id, output, status FROM runs WHERE idempotency_key = ? AND tenant_id = ?",
            (idempotency_key, tenant.tenant_id)
        )
        if existing and existing["status"] == "completed":
            return existing

    # SEC-03: Rate limiting — concurrency and token quota
    await check_concurrency(tenant.tenant_id)
    await check_and_reserve(tenant.tenant_id)
    # S21: Audit log
    _audit_log(db, "agent.run", tenant.tenant_id, agent_id)
    # Load agent
    agent = db.fetchone(
        "SELECT * FROM agents WHERE id = ? AND tenant_id = ? AND status = 'active'",
        (agent_id, tenant.tenant_id)
    )
    if not agent:
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")

    # 工作区对话型 agent 必须绑定工作文件夹,否则拒绝(不回退空家目录假装成功)。
    _require_workspace_dir(agent, req)

    version_rec = db.fetchone(
        "SELECT * FROM agent_versions WHERE agent_id = ? AND version = ?",
        (agent_id, agent["current_version"])
    )
    config = json.loads(version_rec["config"])

    # Apply runtime parameter overrides
    for key, val in req.parameters.items():
        parts = key.split(".")
        target = config
        for p in parts[:-1]:
            target = target.setdefault(p, {})
        target[parts[-1]] = val

    # 会话记忆 P1：解析/新建 thread，run 归属之
    from agentpaas.engine import thread_memory as _tm
    thread_id = _tm.get_or_create_thread(
        db, tenant.tenant_id, agent_id, req.thread_id, req.input)

    # Create run record
    run_id = gen_id("run_")
    now = now_utc()
    db.execute(
        "INSERT INTO runs (id, agent_id, agent_version, tenant_id, input, status, "
        "idempotency_key, thread_id, created_at) "
        "VALUES (?, ?, ?, ?, ?, 'running', ?, ?, ?)",
        (run_id, agent_id, agent["current_version"], tenant.tenant_id, req.input,
         idempotency_key, thread_id, now)
    )
    db.commit()

    # Resolve agent_dir: prefer instance_dir (per-instance data), fallback to agent_dir
    instance_dir = agent.get("instance_dir", "") or ""
    agent_dir = instance_dir or agent.get("agent_dir", "") or ""
    work_dir = agent.get("work_dir", "") or ""
    source_dir = agent.get("source_dir", "") or ""
    run_dir = agent.get("run_dir", "") or ""

    # 工作区对话模式（Dashboard「工作区对话」）：本次 run 经 context.work_dir
    # 覆盖工作文件夹 —— Bash CWD + 读写就地在该文件夹（同时作 source_dir 注入
    # [数据来源目录]）。run_dir 保持 agent 自己的，trace/日志不污染用户项目文件夹。
    import os as _os
    _ctx_wd = (req.context or {}).get("work_dir") if isinstance(req.context, dict) else ""
    _inplace_dir = _ctx_wd if (_ctx_wd and _os.path.isdir(_ctx_wd)) else ""
    if _inplace_dir:
        work_dir = source_dir = _inplace_dir
    # 车道 B:工作区对话型模板 **且 provider=claude-code** 才走原生车道。native 车道是
    # `claude -p`(只认 sonnet/opus/haiku);切成 qwen/ollama 等必须退回 react 车道
    # (text-provider+平台工具,对非 agentic-CLI 模型本来就是对的模式)。
    _native_ws = _is_native_workspace(agent, config)

    # 注入顺序（MEMORY_DESIGN §2.2）：本会话前情 → 相关历史(L3) → KB → 用户问题
    enriched_input = req.input
    preamble = _tm.build_preamble(db, thread_id)
    recall = _tm.search_recall(agent_dir, req.input)

    # KB context injection: enrich input with top-K retrieved passages
    kb_ids = json.loads(agent.get("kb_ids") or "[]")
    kb_search_mode = agent.get("kb_search_mode") or "bm25"
    kb_ctx = ""
    if kb_ids:
        # audit #30: pass tenant_id so KB lookup is scoped, not global.
        kb_ctx = _build_kb_context(db, kb_ids, req.input,
                                   tenant_id=tenant.tenant_id,
                                   search_mode=kb_search_mode)
    _segments = [s for s in (preamble, recall, kb_ctx) if s]
    if _segments:
        enriched_input = "\n\n".join(_segments) + f"\n\n[用户问题]\n{req.input}"

    # Continue-run not currently wired on the sync /run endpoint — only
    # /run/stream supports it. Initialize empty so _execute_agent's
    # kwarg-pass below doesn't NameError.
    continue_workspace = ""

    # Execute via lambdagent.
    # audit #10: _execute_agent is sync and runs multi-step LLM/tool work
    # for seconds-to-minutes; calling it directly from this async handler
    # freezes the entire FastAPI event loop (all health checks, status
    # polls, /cancel endpoints stall). Offload to the default thread pool.
    # The streaming variant (~line 671) already uses a daemon thread; this
    # brings the sync /run path in line.
    import asyncio as _asyncio
    t0 = time.time()
    try:
        _loop = _asyncio.get_event_loop()
        result, trace_info = await _loop.run_in_executor(
            None,
            lambda: _execute_agent(
                config, enriched_input, agent_dir=agent_dir, run_id=run_id,
                continue_workspace=continue_workspace,
                work_dir=work_dir, source_dir=source_dir, run_dir=run_dir,
                kb_tools=_make_platform_kb_tools(
                    db, kb_ids, tenant.tenant_id, kb_search_mode),
                inplace_dir=_inplace_dir,
                native_workspace=_native_ws,
            ),
        )
        duration_ms = int((time.time() - t0) * 1000)
        workspace_path = trace_info.get("workspace_path", "")

        db.execute(
            "UPDATE runs SET status='completed', output=?, duration_ms=?, "
            "input_tokens=?, output_tokens=?, steps=?, workspace_path=?, trace_json=?, "
            "cost_usd=?, cache_read_tokens=?, cache_creation_tokens=?, completed_at=? WHERE id=?",
            (str(result), duration_ms,
             trace_info.get("input_tokens", 0), trace_info.get("output_tokens", 0),
             trace_info.get("steps", 0), workspace_path,
             trace_info.get("trace_json", "[]"),
             trace_info.get("cost_usd", 0.0),
             trace_info.get("cache_read_tokens", 0),
             trace_info.get("cache_creation_tokens", 0),
             now_utc(), run_id)
        )
        db.commit()

        # 会话记忆：刷新 thread 时间 + 排入滚动压缩（串行后台队列）
        _tm.touch_thread(db, thread_id)
        _tm.enqueue_compress(thread_id, config, tenant.tenant_id)
        # 记忆 P3：核心记忆候选提炼（同串行队列，不自动转正）
        _tm.enqueue_distill(agent_dir or work_dir, config,
                            req.input, str(result), run_id)

        resp = {
            "run_id": run_id,
            "status": "completed",
            "output": str(result),
            "thread_id": thread_id,
            "usage": {
                "input_tokens": trace_info.get("input_tokens", 0),
                "output_tokens": trace_info.get("output_tokens", 0),
                "total_tokens": trace_info.get("total_tokens", 0),
                "cache_read_tokens": trace_info.get("cache_read_tokens", 0),
                "cache_creation_tokens": trace_info.get("cache_creation_tokens", 0),
                "cost_usd": trace_info.get("cost_usd", 0.0),
                "steps": trace_info.get("steps", 0),
                "duration_ms": duration_ms,
            },
        }
        if workspace_path:
            resp["workspace_path"] = workspace_path
        return resp

    except Exception as e:
        duration_ms = int((time.time() - t0) * 1000)
        # SEC-04: Log full detail server-side, return sanitized error to client
        from agentpaas.observability.logging import logger
        logger.error("Agent execution failed: run=%s agent=%s error=%s", run_id, agent_id, str(e), exc_info=True)
        db.execute(
            "UPDATE runs SET status='failed', error=?, duration_ms=?, completed_at=? WHERE id=?",
            (json.dumps({"code": type(e).__name__}), duration_ms, now_utc(), run_id)
        )
        db.commit()
        api_error(500, "EXECUTION_ERROR", "Agent execution failed", details={"run_id": run_id})


# ── Streaming Execution (SSE) ──

@router.post("/{agent_id}/run/stream")
async def run_agent_stream(
    agent_id: str,
    req: RunRequest,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """
    SSE stream agent execution — emits step events in real time.

    Events:
      event: think    — agent's reasoning for this step
      event: tool_call — tool invocation (name + input)
      event: tool_result — tool output
      event: error    — tool error
      event: answer   — final answer
      event: done     — execution complete with usage stats
    """
    agent = db.fetchone(
        "SELECT * FROM agents WHERE id = ? AND tenant_id = ? AND status = 'active'",
        (agent_id, tenant.tenant_id)
    )
    if not agent:
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")

    # 工作区对话型 agent 必须绑定工作文件夹,否则拒绝(不回退空家目录假装成功)。
    _require_workspace_dir(agent, req)

    version_rec = db.fetchone(
        "SELECT * FROM agent_versions WHERE agent_id = ? AND version = ?",
        (agent_id, agent["current_version"])
    )
    config = json.loads(version_rec["config"])

    for key, val in req.parameters.items():
        parts = key.split(".")
        target = config
        for p in parts[:-1]:
            target = target.setdefault(p, {})
        target[parts[-1]] = val

    require_permission(tenant, "agents:execute")

    # Resolve agent_dir: prefer instance_dir, fallback to agent_dir
    instance_dir = agent.get("instance_dir", "") or ""
    agent_dir = instance_dir or agent.get("agent_dir", "") or ""
    work_dir = agent.get("work_dir", "") or ""
    source_dir = agent.get("source_dir", "") or ""
    run_dir = agent.get("run_dir", "") or ""

    # 工作区对话模式：本次 run 经 context.work_dir 覆盖工作文件夹（见 run_agent 同段注释）。
    import os as _os
    _ctx_wd = (req.context or {}).get("work_dir") if isinstance(req.context, dict) else ""
    _inplace_dir = _ctx_wd if (_ctx_wd and _os.path.isdir(_ctx_wd)) else ""
    if _inplace_dir:
        work_dir = source_dir = _inplace_dir
    _native_ws = _is_native_workspace(agent, config)

    # 会话记忆 P1：thread 解析（run 记录在下方流式分支创建，这里先拿 id）
    from agentpaas.engine import thread_memory as _tm
    thread_id = _tm.get_or_create_thread(
        db, tenant.tenant_id, agent_id, req.thread_id, req.input)

    # 注入顺序：本会话前情 → KB 检索 → 用户问题
    kb_ids = json.loads(agent.get("kb_ids") or "[]")
    kb_search_mode = agent.get("kb_search_mode") or "bm25"
    preamble = _tm.build_preamble(db, thread_id)
    recall = _tm.search_recall(agent_dir, req.input)
    kb_ctx = ""
    if kb_ids:
        # audit #30: pass tenant_id so KB lookup is scoped, not global.
        kb_ctx = _build_kb_context(db, kb_ids, req.input,
                                   tenant_id=tenant.tenant_id,
                                   search_mode=kb_search_mode)
    _segments = [s for s in (preamble, recall, kb_ctx) if s]
    if _segments:
        stream_input = "\n\n".join(_segments) + f"\n\n[用户问题]\n{req.input}"
    else:
        stream_input = req.input

    # Default empty — the continue-run injection block (when present) sets
    # this to the previous run's workspace. Initialized here so the inner
    # _run() closure doesn't NameError when no continue context is sent.
    continue_workspace = ""

    # ── Continue-run context injection ──
    # When the UI sends context.run_id, the user clicked "继续此 run". We
    # look up the previous run's workspace, point _execute_agent at it so
    # cycle dirs are shared, compute the next cycle number, and inject a
    # prompt prefix telling the orchestrator the workspace path + cycle to
    # use. Skipped silently if the prior run / workspace can't be found.
    ctx_run_id = (req.context or {}).get("run_id", "")
    if ctx_run_id:
        logger.info("continue-run requested: ctx_run_id=%s agent=%s", ctx_run_id, agent_id)
        prev = db.fetchone(
            "SELECT workspace_path, input FROM runs WHERE id = ? AND agent_id = ?",
            (ctx_run_id, agent_id),
        )
        if prev and prev.get("workspace_path"):
            wp = prev["workspace_path"]
            prev_input = (prev.get("input") or "")[:200]
            continue_workspace = wp
            logger.info("continue-run injecting workspace: %s", wp)

            # Discover existing cycle_N dirs so we can name the next one.
            import os as _os, re as _re
            next_cycle = 1
            try:
                existing = [
                    int(m.group(1))
                    for n in _os.listdir(wp)
                    if (m := _re.match(r"^cycle_(\d+)$", n))
                    and _os.path.isdir(_os.path.join(wp, n))
                ]
                if existing:
                    next_cycle = max(existing) + 1
            except OSError:
                pass
            next_workspace = _os.path.join(wp, f"cycle_{next_cycle}")
            prev_cycle_path = (
                _os.path.join(wp, f"cycle_{next_cycle - 1}")
                if next_cycle > 1 else ""
            )

            stream_input = (
                f"[本次执行是【上一次的延续】 — 工作目录已复用，禁止新建顶层目录]\n"
                f"工作根目录（不要修改不要新建）：{wp}\n"
                f"本次必须使用的工作区（第 {next_cycle} 轮的隔离目录）：\n"
                f"  workspace = {next_workspace}\n"
                f"上一轮的目录（读取上下文用）：\n"
                f"  prev_workspace = {prev_cycle_path}\n"
                f"  cycle = {next_cycle}\n"
                f"\n"
                f"上一次输入摘要（仅供参考）：{prev_input}\n"
                f"\n"
                f"⚠️ 调用 sub-agent 时输入 JSON 里的 workspace/prev_workspace/cycle 字段"
                f"必须严格使用上面给的 3 个值，**不要凭直觉改名**"
                f"（例如不要加 _revision、_v2、_new 等后缀）。\n"
                f"⚠️ 不要因为'看到已有 cycle_*' 就直接 terminate；本次有【新的指令】，"
                f"必须按指令调用相关 sub-agent。\n"
                f"\n"
                f"[本次新指令]\n{stream_input}"
            )

            # NOTE: enforceLoop is now mode-aware (see mode block below).
            # Previously we unconditionally dropped it on every continue,
            # which made `mode=iterate` silently behave like edit/chat —
            # the orchestrator was terminating after one sub-agent attempt
            # because no enforceLoop floor was holding it to the full
            # pipeline. iterate now keeps the floor; chat / edit drop it.
        else:
            logger.warning("continue-run prev run not found or has no workspace: %s", ctx_run_id)

    # ── Mode-specific prompt + config tweaks (Chat/Edit/Iterate) ──
    # These are LOCAL overrides applied to the in-memory config dict for
    # this single run, never persisted. iterate is a no-op.
    mode = (req.mode or "iterate").lower()
    target_sub = (req.target_subagent or "").strip()
    if mode == "edit":
        if not target_sub:
            api_error(400, "EDIT_REQUIRES_TARGET",
                      "mode=edit requires target_subagent (e.g. 'call_physwriter')")
        stream_input = (
            f"[MODE=EDIT — 只调用一次 sub-agent，然后立刻 terminate]\n"
            f"目标 sub-agent：{target_sub}\n"
            f"⚠️ 严格规则：\n"
            f"  - 只输出一个 JSON 调用 {target_sub}，input 字段照搬用户指令。\n"
            f"  - 调用完成后立刻 terminate，不要补凑其他 sub-agent。\n"
            f"  - 禁止全 pipeline、禁止循环、禁止 enforceLoop 的 18 步硬卡。\n"
            f"\n[用户指令]\n{stream_input}"
        )
        # Disable any enforceLoop floor for this run — Edit mode should
        # terminate after a single sub-agent call.
        if isinstance(config.get("react"), dict):
            config["react"] = {**config["react"]}
            config["react"].pop("enforceLoop", None)
    elif mode == "chat":
        stream_input = (
            f"[MODE=CHAT — 不调用任何 sub-agent，直接回答用户问题]\n"
            f"工作目录（可读）：{continue_workspace or '(无 continue 上下文)'}\n"
            f"⚠️ 严格规则：\n"
            f"  - 你可以用 ListFiles / ReadFile / Bash 查看 workspace 文件。\n"
            f"  - **禁止**调用任何 call_phys* sub-agent。\n"
            f"  - **禁止**修改任何文件（不要用 WriteFile/EditFile）。\n"
            f"  - 看够文件理解上下文后，立刻输出 terminate，"
            f"summary 字段写完整答案给用户。\n"
            f"\n[用户问题]\n{stream_input}"
        )
        # Temporarily expand localTools so the model can read workspace
        # without going through a sub-agent. We don't add Write/Edit —
        # chat mode is read-only.
        if isinstance(config.get("mcp"), dict):
            config["mcp"] = {**config["mcp"]}
            old_tools = list(config["mcp"].get("localTools", []))
            chat_tools = []
            for t in old_tools:
                # Drop call_phys* sub-agents — chat shouldn't route through them.
                if not t.startswith("call_"):
                    chat_tools.append(t)
            # Add read-only tools if not already there.
            for t in ("ReadFile", "ListFiles", "Bash"):
                if t not in chat_tools:
                    chat_tools.append(t)
            if "terminate" not in chat_tools:
                chat_tools.append("terminate")
            config["mcp"]["localTools"] = chat_tools
        # Always drop enforceLoop in chat — terminate after first useful answer.
        if isinstance(config.get("react"), dict):
            config["react"] = {**config["react"]}
            config["react"].pop("enforceLoop", None)
    # mode == "iterate" or unknown → no changes, current behavior

    # Create run record before streaming starts
    run_id = gen_id("run_")
    now = now_utc()
    db.execute(
        "INSERT INTO runs (id, agent_id, agent_version, tenant_id, input, status, "
        "thread_id, created_at) VALUES (?, ?, ?, ?, ?, 'running', ?, ?)",
        (run_id, agent_id, agent["current_version"], tenant.tenant_id, req.input,
         thread_id, now)
    )
    db.commit()

    import queue
    # FIX-09: Bounded queue prevents OOM from slow clients
    event_queue = queue.Queue(maxsize=1000)
    t0 = time.time()

    def generate():
        import threading
        from lambdagent.agentruntime import cancel as _cancel

        # Emit run_id to the client immediately so the Stop button can call
        # /runs/<id>/cancel without having to wait for the final 'done' event.
        # Without this the user might click Stop before any other event lands
        # and the frontend wouldn't know which run to cancel.
        yield f"event: started\ndata: {json.dumps({'run_id': run_id, 'thread_id': thread_id}, ensure_ascii=False)}\n\n"

        def _persist_subprocess_pid(pid):
            """Hook fired by cancel.bind_proc/unbind_proc — write the live
            claude pid to runs.subprocess_pid so a *future* backend (post
            restart) can still kill the orphan subprocess via os.kill().
            Cleared back to NULL on unbind / run end."""
            try:
                db.execute(
                    "UPDATE runs SET subprocess_pid=? WHERE id=?",
                    (pid, run_id),
                )
                db.commit()
            except Exception:
                # Persisting pid is best-effort. A failure here doesn't
                # break the in-process cancel path, only the restart-
                # resilient fallback.
                pass

        def _run():
            # Register this run with the cancel registry so a POST to the
            # cancel endpoint can reach in, kill the active claude subprocess,
            # and unwind the engine via CancelledRun. Without this the only
            # way to stop a wedged run is pkill + manual DB update.
            #
            # The persist hook closes the second hole: cross-backend-restart
            # cancellation. Without it, killing an orphan subprocess after
            # the backend that started it has died requires looking up its
            # pid via lsof / pgrep and `kill -9` by hand.
            _cancel.begin_run(run_id, on_proc_change=_persist_subprocess_pid)
            try:
                result, trace_info = _execute_agent(
                    config, stream_input, on_step=event_queue.put,
                    agent_dir=agent_dir, run_id=run_id,
                    continue_workspace=continue_workspace,
                    work_dir=work_dir, source_dir=source_dir, run_dir=run_dir,
                    kb_tools=_make_platform_kb_tools(
                        db, kb_ids, tenant.tenant_id, kb_search_mode),
                    inplace_dir=_inplace_dir,
                    native_workspace=_native_ws,
                )
                duration_ms = int((time.time() - t0) * 1000)
                workspace_path = trace_info.get("workspace_path", "")

                # ── status semantic correction ──
                # _execute_agent returns "successfully" even when every
                # claude call stalled — the engine catches ProviderError,
                # records the error message as the "thought", and lets
                # terminate fire because no tool was selected. From the
                # user's POV the run produced nothing usable. Mark such
                # cases as 'failed' so the UI doesn't show a green
                # "completed" badge for runs where the model literally
                # said "[CLAUDE-CODE_ERROR] stalled".
                final = str(result)
                _ERROR_PREFIXES = (
                    "[CLAUDE-CODE_ERROR]",
                    "[STALL]",
                    "[ProviderError]",
                    "[ERROR]",
                )
                # Substring fail markers — for cases where the orchestrator
                # composes a natural-language failure report (no machine
                # prefix). Catches "Sub-agent 调用全部超时", "系统工具链存在
                # 问题", "无法完成", "未能完成", "all sub-agents failed",
                # etc. Conservative: only fires when paired with the
                # `not made_progress` check below.
                _FAILURE_KEYWORDS = (
                    "全部超时", "全部 timeout", "全部失败",
                    "无法完成", "未能完成", "无法继续",
                    "系统工具链存在问题",
                    "all sub-agents failed", "all calls failed",
                    "unable to complete",
                )
                # AUDIT_2026-06-11 ②: 结构化优先。lambdagent Context 现在在
                # provider 抛错时记录结构化事件（trace_info["provider_errors"]），
                # 泛化的 [XXX_ERROR]/[XXX_TIMEOUT] 前缀正则只在有结构化佐证时
                # 才判失败 —— 否则 LLM 正常输出恰好以这类字样开头会被误判。
                # [MCP_*] 是工具网关的纯文本通道（无结构化事件），单独保留。
                import re as _re_err
                _lead = final.lstrip()
                _structured_provider_failure = bool(trace_info.get("provider_errors"))
                _generic_err_prefix = bool(
                    _re_err.match(r"\[[A-Z0-9_\-]*(?:ERROR|TIMEOUT)\]", _lead))
                output_is_error = (
                    any(_lead.startswith(p) for p in _ERROR_PREFIXES)
                    or (_generic_err_prefix and _structured_provider_failure)
                    or bool(_re_err.match(r"\[MCP_[A-Z0-9_\-]*(?:ERROR|TIMEOUT)", _lead))
                    or any(kw in final for kw in _FAILURE_KEYWORDS)
                )
                # Scan trace entries for evidence of any successful tool
                # call ("[Step K] X done" — see _compress_state in the
                # engine). The engine writes "done" markers even when the
                # tool returned an error (e.g. STALL), so we must also
                # inspect each entry's observation to filter those out.
                # Otherwise an all-error run gets status=completed because
                # the trace technically has "[Step 1] call_physreview done"
                # markers despite physreview producing only [CLAUDE-CODE_ERROR].
                trace_json = trace_info.get("trace_json", "[]")
                made_progress = False
                # Markers that prove a tool's observation was an error rather
                # than real output. Kept in sync with compiler._observation_failed.
                _FAIL_MARKERS = (
                    "[tool_error]", "[validation_error]", "[claude code error]",
                    "[claude-code_error]", "[error]", "[mcp_error", "[mcp_timeout",
                    " timed out", " timeout", " not found", " compile error",
                    '"status":"fail"', '"status": "fail"',
                    '"ok":false', '"ok": false',
                )
                try:
                    import re as _re
                    for entry in json.loads(trace_json) or []:
                        out = str(entry.get("output", ""))
                        if not _re.search(r"\[Step \d+\] [^\n]*\bdone\b", out):
                            continue
                        # A done marker exists; check the observation in the
                        # same entry to see if the tool actually succeeded.
                        # The engine concatenates observation into the state
                        # output, so we look at the whole entry text.
                        out_low = out.lower()
                        if any(m in out_low for m in _FAIL_MARKERS):
                            continue  # done marker but content is an error
                        made_progress = True
                        break
                except Exception:
                    pass

                if output_is_error and not made_progress:
                    final_status = "failed"
                    err_payload = json.dumps(
                        {"code": "AGENT_FAILED_NO_PROGRESS",
                         "message": final[:500]},
                        ensure_ascii=False,
                    )
                    db.execute(
                        "UPDATE runs SET status='failed', output=?, error=?, duration_ms=?, "
                        "input_tokens=?, output_tokens=?, steps=?, workspace_path=?, "
                        "cost_usd=?, cache_read_tokens=?, cache_creation_tokens=?, "
                        "trace_json=?, completed_at=? WHERE id=?",
                        (final, err_payload, duration_ms,
                         trace_info.get("input_tokens", 0),
                         trace_info.get("output_tokens", 0),
                         trace_info.get("steps", 0), workspace_path,
                         trace_info.get("cost_usd", 0.0),
                         trace_info.get("cache_read_tokens", 0),
                         trace_info.get("cache_creation_tokens", 0),
                         trace_json, now_utc(), run_id),
                    )
                else:
                    final_status = "completed"
                    db.execute(
                        "UPDATE runs SET status='completed', output=?, duration_ms=?, "
                        "input_tokens=?, output_tokens=?, steps=?, workspace_path=?, "
                        "cost_usd=?, cache_read_tokens=?, cache_creation_tokens=?, "
                        "trace_json=?, completed_at=? WHERE id=?",
                        (final, duration_ms,
                         trace_info.get("input_tokens", 0),
                         trace_info.get("output_tokens", 0),
                         trace_info.get("steps", 0), workspace_path,
                         trace_info.get("cost_usd", 0.0),
                         trace_info.get("cache_read_tokens", 0),
                         trace_info.get("cache_creation_tokens", 0),
                         trace_json, now_utc(), run_id),
                    )
                db.commit()
                # 会话记忆：刷新 thread + 排入滚动压缩 + 核心记忆提炼
                try:
                    _tm.touch_thread(db, thread_id)
                    _tm.enqueue_compress(thread_id, config, tenant.tenant_id)
                    _tm.enqueue_distill(agent_dir or work_dir, config,
                                        req.input, str(final), run_id)
                except Exception:
                    pass
                event_queue.put({"event": "done", "data": {
                    "run_id": run_id,
                    "thread_id": thread_id,
                    "status": final_status, "output": final,
                    "steps": trace_info.get("steps", 0),
                    "total_tokens": trace_info.get("total_tokens", 0),
                    "cost_usd": trace_info.get("cost_usd", 0.0),
                    "workspace_path": workspace_path,
                }})
            except _cancel.CancelledRun:
                # User-requested stop. Run was already marked cancel_requested=1
                # by the /cancel endpoint; transition status to 'cancelled'.
                duration_ms = int((time.time() - t0) * 1000)
                db.execute(
                    "UPDATE runs SET status='cancelled', duration_ms=?, completed_at=? WHERE id=?",
                    (duration_ms, now_utc(), run_id)
                )
                db.commit()
                event_queue.put({"event": "cancelled", "data": {"run_id": run_id}})
                event_queue.put({"event": "done", "data": {"run_id": run_id, "status": "cancelled"}})
            except Exception as e:
                duration_ms = int((time.time() - t0) * 1000)
                db.execute(
                    "UPDATE runs SET status='failed', error=?, duration_ms=?, completed_at=? WHERE id=?",
                    (json.dumps({"code": type(e).__name__, "message": str(e)[:500]}),
                     duration_ms, now_utc(), run_id)
                )
                db.commit()
                event_queue.put({"event": "error", "data": {"message": str(e)}})
                event_queue.put({"event": "done", "data": {"run_id": run_id, "status": "failed"}})
            finally:
                _cancel.end_run(run_id)
                # Safety net: a stream run must never be left as 'running' once
                # the worker thread exits (client disconnect, exception inside an
                # except handler, etc.). If still 'running', force a terminal
                # status — cancelled if the user requested stop, else failed.
                try:
                    cur_row = db.fetchone(
                        "SELECT status, cancel_requested FROM runs WHERE id=?", (run_id,)
                    )
                    if cur_row and cur_row.get("status") == "running":
                        term = "cancelled" if cur_row.get("cancel_requested") else "failed"
                        db.execute(
                            "UPDATE runs SET status=?, error=COALESCE(error, ?), "
                            "duration_ms=?, completed_at=? WHERE id=? AND status='running'",
                            (term,
                             json.dumps({"code": "ORPHANED",
                                         "message": "stream worker exited without finalizing run status"}),
                             int((time.time() - t0) * 1000), now_utc(), run_id),
                        )
                        db.commit()
                        event_queue.put({"event": "done",
                                         "data": {"run_id": run_id, "status": term}})
                except Exception:
                    pass
            event_queue.put(None)  # sentinel

        threading.Thread(target=_run, daemon=True).start()

        while True:
            try:
                # Use a timeout so we can send SSE heartbeat comments while
                # waiting.  Without this, the generator blocks silently for
                # 13+ seconds between LLM calls; proxies and browsers see the
                # connection as dead and close it → "Failed to fetch" on the
                # client even though the backend is still running fine.
                # SSE comment lines (": …") are ignored by EventSource /
                # fetch readers but keep the TCP connection alive.
                item = event_queue.get(timeout=15)
            except queue.Empty:
                yield ": heartbeat\n\n"
                continue
            if item is None:
                break
            evt_type = item.get("event", "message")
            data = json.dumps(item.get("data", item), ensure_ascii=False)
            yield f"event: {evt_type}\ndata: {data}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache",
                                      "X-Accel-Buffering": "no",
                                      "Connection": "keep-alive"})


# ── Agent Versions ──

@router.get("/{agent_id}/versions")
async def list_versions(
    agent_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """List all versions of an agent."""
    require_permission(tenant, "agents:read")
    versions = db.fetchall(
        "SELECT v.*, a.current_version FROM agent_versions v "
        "JOIN agents a ON v.agent_id = a.id "
        "WHERE v.agent_id = ? AND a.tenant_id = ? ORDER BY v.version DESC",
        (agent_id, tenant.tenant_id)
    )
    for v in versions:
        v["is_current"] = v["version"] == v.pop("current_version", None)
    return {"versions": versions}


# ── Record-Only Endpoint (for external execution backends like Claude Code CLI) ──

class RecordRunRequest(BaseModel):
    """记录一次外部执行结果（不触发 lambdagent 执行）"""
    input: str
    output: str
    status: str = "completed"
    duration_ms: int = 0
    steps: int = 0
    source: str = "external"


@router.post("/{agent_id}/runs/record", status_code=201)
async def record_run(
    agent_id: str,
    req: RecordRunRequest,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """
    Record an externally-executed run (e.g., from Claude Code CLI).
    Does NOT trigger lambdagent execution — only writes to the runs table.
    """
    agent = db.fetchone(
        "SELECT * FROM agents WHERE id = ? AND tenant_id = ? AND status = 'active'",
        (agent_id, tenant.tenant_id)
    )
    if not agent:
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")

    run_id = gen_id("run_")
    now = now_utc()
    db.execute(
        "INSERT INTO runs (id, agent_id, agent_version, tenant_id, input, output, "
        "status, duration_ms, steps, created_at, completed_at) "
        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
        (run_id, agent_id, agent["current_version"], tenant.tenant_id,
         req.input, req.output, req.status, req.duration_ms, req.steps, now, now)
    )
    db.commit()

    return {
        "run_id": run_id,
        "status": req.status,
        "source": req.source,
        "message": "Run recorded (external execution)",
    }


# ── Run Cancellation ──

@router.post("/{agent_id}/runs/{run_id}/cancel")
async def cancel_run(
    agent_id: str,
    run_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Request cancellation of a running run.

    Sets `runs.cancel_requested=1` and signals the in-process cancel
    registry. The agent's background thread polls cancel state both
    before each ReAct step and inside the claude subprocess polling
    loop, so latency from click → kill is typically <1s.

    Returns 200 even if the run already finished (idempotent). The
    response includes `was_running` so the UI can distinguish
    "successfully signaled" from "already gone".
    """
    run = db.fetchone(
        "SELECT id, status FROM runs WHERE id = ? AND agent_id = ? AND tenant_id = ?",
        (run_id, agent_id, tenant.tenant_id),
    )
    if not run:
        api_error(404, "RUN_NOT_FOUND", f"Run {run_id} not found")

    # Mark intent in DB unconditionally (cheap, idempotent).
    db.execute(
        "UPDATE runs SET cancel_requested = 1 WHERE id = ?",
        (run_id,),
    )
    db.commit()

    # Signal the in-process registry — kills the active claude subprocess
    # immediately. No-op if the run isn't in this process's registry (e.g.
    # already finished, or running on another worker, or started by a
    # previous backend instance that has since restarted).
    from lambdagent.agentruntime import cancel as _cancel
    was_running = _cancel.request_cancel(run_id)
    killed_via_pid = False

    # Restart-resilient fallback: if the in-memory registry didn't know
    # about this run but a subprocess_pid is persisted, the subprocess
    # is almost certainly an orphan (its launching backend died). Kill
    # it directly by pid. This closes the "restart loses cancel reach"
    # gap exposed by run_5cbbb1d1dc54.
    if not was_running:
        pid_row = db.fetchone(
            "SELECT subprocess_pid FROM runs WHERE id=?", (run_id,)
        )
        pid = (pid_row or {}).get("subprocess_pid")
        if pid:
            import os as _os
            import signal as _signal
            try:
                _os.kill(int(pid), _signal.SIGKILL)
                killed_via_pid = True
                # Clear the stale pid + mark cancelled. The original _run
                # thread is gone (its backend died), so no one else will
                # do this for us.
                db.execute(
                    "UPDATE runs SET subprocess_pid=NULL, status='cancelled', "
                    "completed_at=strftime('%Y-%m-%dT%H:%M:%S+00:00','now'), "
                    "error=COALESCE(NULLIF(error,''),'') || "
                    "  '[orphan killed by pid fallback after backend restart]' "
                    "WHERE id=? AND status='running'",
                    (run_id,),
                )
                db.commit()
            except ProcessLookupError:
                # Already dead, just clear the stale pid.
                db.execute(
                    "UPDATE runs SET subprocess_pid=NULL WHERE id=?",
                    (run_id,),
                )
                db.commit()
            except (PermissionError, ValueError):
                # Wrong user / corrupt pid — give up gracefully.
                pass

    return {
        "run_id": run_id,
        "status": run["status"],
        "cancel_requested": True,
        "was_running": was_running,
        # NEW: tells the UI we found + killed an orphan even though the
        # in-process registry was empty. Useful for telemetry.
        "killed_via_pid": killed_via_pid,
    }


# ── Runs History ──

@router.get("/{agent_id}/runs")
async def list_runs(
    agent_id: str,
    limit: int = 20,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """List recent runs for an agent."""
    runs = db.fetchall(
        "SELECT * FROM runs WHERE agent_id = ? AND tenant_id = ? ORDER BY created_at DESC LIMIT ?",
        (agent_id, tenant.tenant_id, limit)
    )
    return {"runs": runs}


# ── 会话记忆 thread API ──

@router.get("/{agent_id}/threads")
async def list_threads(
    agent_id: str,
    limit: int = 20,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """该智能体最近的对话线程（工作台「历史会话」用）。"""
    rows = db.fetchall(
        "SELECT t.id, t.title, t.status, t.updated_at, "
        "(SELECT COUNT(*) FROM runs r WHERE r.thread_id = t.id) AS run_count "
        "FROM threads t WHERE t.agent_id = ? AND t.tenant_id = ? AND t.status = 'active' "
        "ORDER BY t.updated_at DESC LIMIT ?",
        (agent_id, tenant.tenant_id, limit))
    return {"threads": rows}


@router.get("/{agent_id}/threads/{thread_id}/runs")
async def thread_runs(
    agent_id: str,
    thread_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """thread 内的消息流（恢复历史会话气泡）。"""
    th = db.fetchone(
        "SELECT id FROM threads WHERE id = ? AND tenant_id = ? AND agent_id = ?",
        (thread_id, tenant.tenant_id, agent_id))
    if not th:
        raise HTTPException(404, {"error": {"code": "THREAD_NOT_FOUND"}})
    runs = db.fetchall(
        "SELECT id, input, output, status, workspace_path, created_at "
        "FROM runs WHERE thread_id = ? ORDER BY created_at ASC",
        (thread_id,))
    return {"thread_id": thread_id, "runs": runs}


@router.post("/{agent_id}/threads/{thread_id}/archive")
async def archive_thread(
    agent_id: str,
    thread_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """归档对话线程（不删，列表隐藏）。"""
    n = db.execute(
        "UPDATE threads SET status = 'archived', updated_at = ? "
        "WHERE id = ? AND tenant_id = ? AND agent_id = ?",
        (now_utc(), thread_id, tenant.tenant_id, agent_id)).rowcount
    db.commit()
    if not n:
        raise HTTPException(404, {"error": {"code": "THREAD_NOT_FOUND"}})
    return {"ok": True}


# ── Workspace File API ──

@router.get("/{agent_id}/runs/{run_id}/workspace")
async def list_workspace_files(
    agent_id: str,
    run_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """List files in a run's workspace directory."""
    import os as _os
    require_permission(tenant, "agents:read")
    run = db.fetchone(
        "SELECT workspace_path FROM runs WHERE id = ? AND agent_id = ? AND tenant_id = ?",
        (run_id, agent_id, tenant.tenant_id)
    )
    if not run:
        api_error(404, "RUN_NOT_FOUND", f"Run {run_id} not found")
    workspace = run["workspace_path"] or ""
    if not workspace or not _os.path.isdir(workspace):
        return {"files": [], "workspace_path": workspace}
    files = []
    for dirpath, dirnames, filenames in _os.walk(workspace):
        dirnames.sort()
        for fname in sorted(filenames):
            full = _os.path.join(dirpath, fname)
            rel = _os.path.relpath(full, workspace)
            try:
                size = _os.path.getsize(full)
                modified = _os.path.getmtime(full)
            except OSError:
                size, modified = 0, 0
            files.append({"name": fname, "path": rel, "size": size, "modified": modified})
    return {"files": files, "workspace_path": workspace}


@router.get("/{agent_id}/runs/{run_id}/workspace/{file_path:path}")
async def download_workspace_file(
    agent_id: str,
    run_id: str,
    file_path: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Download a file from a run's workspace (path-traversal protected)."""
    import os as _os
    from fastapi.responses import FileResponse as _FileResponse
    require_permission(tenant, "agents:read")
    run = db.fetchone(
        "SELECT workspace_path FROM runs WHERE id = ? AND agent_id = ? AND tenant_id = ?",
        (run_id, agent_id, tenant.tenant_id)
    )
    if not run:
        api_error(404, "RUN_NOT_FOUND", f"Run {run_id} not found")
    workspace = run["workspace_path"] or ""
    if not workspace:
        api_error(404, "NO_WORKSPACE", "This run has no workspace")
    workspace_real = _os.path.realpath(workspace)
    full_path = _os.path.realpath(_os.path.join(workspace, file_path))
    if not full_path.startswith(workspace_real + _os.sep):
        api_error(403, "FORBIDDEN", "Path traversal not allowed")
    if not _os.path.isfile(full_path):
        api_error(404, "FILE_NOT_FOUND", f"File not found: {file_path}")
    return _FileResponse(full_path, filename=_os.path.basename(full_path))


# ── Agent Memory API ──

@router.get("/{agent_id}/memory")
async def list_agent_memory(
    agent_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """List non-expired memory entries for an agent."""
    require_permission(tenant, "agents:read")
    if not db.fetchone(
        "SELECT id FROM agents WHERE id = ? AND tenant_id = ?",
        (agent_id, tenant.tenant_id)
    ):
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")
    from agentpaas.services.memory_service import MemoryStore
    store = MemoryStore()
    entries = store.list_all(source=agent_id)
    valid = [e.to_dict() for e in entries if not e.is_expired()]
    return {"entries": valid, "count": len(valid)}


@router.delete("/{agent_id}/memory/{key}")
async def delete_agent_memory(
    agent_id: str,
    key: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Delete a specific memory entry for an agent."""
    require_permission(tenant, "agents:write")
    if not db.fetchone(
        "SELECT id FROM agents WHERE id = ? AND tenant_id = ?",
        (agent_id, tenant.tenant_id)
    ):
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")
    from agentpaas.services.memory_service import MemoryStore
    deleted = MemoryStore().forget(key)
    if not deleted:
        api_error(404, "MEMORY_NOT_FOUND", f"Memory key {key} not found")
    return {"deleted": key}


# ── Agent-KB Linking API ──

@router.get("/{agent_id}/knowledge")
async def get_agent_knowledge(
    agent_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """List knowledge bases linked to this agent, with their index status."""
    require_permission(tenant, "agents:read")
    agent = db.fetchone(
        "SELECT kb_ids, kb_search_mode FROM agents WHERE id = ? AND tenant_id = ?",
        (agent_id, tenant.tenant_id)
    )
    if not agent:
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")
    kb_ids = json.loads(agent.get("kb_ids") or "[]")
    kbs = []
    for kb_id in kb_ids:
        kb = db.fetchone(
            "SELECT id, name, root_dir, description FROM knowledge_bases WHERE id = ?",
            (kb_id,)
        )
        if kb:
            kbs.append(dict(kb))
    return {
        "kb_ids": kb_ids,
        "kb_search_mode": agent.get("kb_search_mode", "bm25"),
        "knowledge_bases": kbs,
    }


@router.put("/{agent_id}/knowledge")
async def update_agent_knowledge(
    agent_id: str,
    req: UpdateKBsRequest,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Replace the list of knowledge bases linked to this agent."""
    require_permission(tenant, "agents:write")
    if not db.fetchone(
        "SELECT id FROM agents WHERE id = ? AND tenant_id = ?",
        (agent_id, tenant.tenant_id)
    ):
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")
    # Validate that all kb_ids exist (KB is a shared resource; no tenant restriction)
    for kb_id in req.kb_ids:
        if not db.fetchone("SELECT id FROM knowledge_bases WHERE id = ?", (kb_id,)):
            api_error(404, "KB_NOT_FOUND", f"Knowledge base {kb_id} not found")
    db.execute(
        "UPDATE agents SET kb_ids = ?, kb_search_mode = ?, updated_at = ? WHERE id = ?",
        (json.dumps(req.kb_ids), req.kb_search_mode, now_utc(), agent_id)
    )
    db.commit()
    return {"kb_ids": req.kb_ids, "kb_search_mode": req.kb_search_mode}


# ── KB Context Injection (RAG helper for agent execution) ──

def _build_kb_context(
    db: "Database",
    kb_ids: list,
    query: str,
    tenant_id: str,
    search_mode: str = "bm25",
    top_k: int = 5,
) -> str:
    """Search linked KBs and return a formatted context block.

    Tries pageindex first (fast, in-process), then falls back to BM25
    subprocess.  Returns empty string when nothing useful is found.

    audit/AUDIT_2026-06-05.md #30 fix: previously the per-kb SELECT was
      `SELECT * FROM knowledge_bases WHERE id = ?`
    with NO tenant scope. The agent CRUD endpoints DO enforce tenant
    isolation when you write `kb_ids`, but at RUN time any tenant could
    have linked another tenant's kb_id (or had it injected via a
    cross-tenant update bug) and the runtime would happily fan out a
    search over it. Now: each lookup is `AND tenant_id = ?`; KBs that
    don't belong to the caller are silently skipped.
    """
    if not kb_ids or not query.strip():
        return ""

    all_results: list = []
    for kb_id in kb_ids:
        kb = db.fetchone(
            "SELECT * FROM knowledge_bases WHERE id = ? AND tenant_id = ?",
            (kb_id, tenant_id),
        )
        if not kb:
            continue
        kb_root = kb.get("root_dir", "")
        if not kb_root:
            continue
        kb_name = kb.get("name", kb_id)

        results: list = []
        try:
            from pathlib import Path as _Path
            from agentpaas.api.v1.knowledge import (
                _page_index_search,
                _find_scripts_dir,
                _search_subprocess,
            )
            page_idx_exists = (_Path(kb_root) / "rag_page_index.json").exists()

            if search_mode == "pageindex" and page_idx_exists:
                results = _page_index_search(kb_root, query, top_k)
            else:
                # BM25 subprocess
                try:
                    sd = _find_scripts_dir(_Path(kb_root))
                    if sd:
                        results = _search_subprocess(kb_root, sd, "bm25", query, top_k)
                except Exception:
                    pass
                # Fallback to pageindex
                if not results and page_idx_exists:
                    results = _page_index_search(kb_root, query, top_k)
        except Exception:
            continue

        for r in results:
            r["_kb_name"] = kb_name
            all_results.append(r)

    if not all_results:
        return ""

    # Sort by score descending (results may come from multiple KBs)
    all_results.sort(key=lambda x: -(x.get("score") or 0))

    lines = ["[知识库检索结果]"]
    for i, r in enumerate(all_results[:top_k], 1):
        src = r.get("_kb_name", "")
        doc = r.get("source", "") or r.get("doc_id", "")
        if doc:
            src += f" / {doc}"
        pg = r.get("page")
        if pg:
            src += f" 第{pg}页"
        text = (r.get("text") or r.get("chunk_text") or r.get("content") or "")[:600].strip()
        lines.append(f"\n[来源{i}: {src}]\n{text}")

    return "\n".join(lines)


def _make_platform_kb_tools(db: "Database", kb_ids: list, tenant_id: str,
                            search_mode: str = "bm25") -> dict:
    """构造平台版 KBSearch / KBList 工具实现（KB 调用链核查修复）。

    背景：lambdagent 内置的 KBSearch 工具查的是它自己的
    ``~/.lambdagent/knowledge_bases/<name>/store.json`` 向量库——与平台
    「知识库」页面管理的 knowledge_bases 表 + root_dir 索引是两套互不
    相通的系统。此前 agent 运行时调 KBSearch/KBList 永远看不到用户
    挂接的资料库，只能靠运行入口的一次性被动注入（_build_kb_context）。

    这里把两个工具替换成查询**该 agent 挂接的平台知识库**的闭包，经
    overrides["tools"] 注入编译器（custom tools 优先于 builtin），让
    agent 在推理中途可以按需多次主动检索。

    Returns {} when the agent has no linked KBs（不注入，保留内置行为）。
    """
    if not kb_ids:
        return {}

    def _parse_query(input_val) -> str:
        """LLM 工具入参可能是裸字符串或 JSON {"query"/"q"/"keyword": ...}"""
        s = str(input_val or "").strip()
        if s.startswith("{"):
            try:
                d = json.loads(s)
                for k in ("query", "q", "keyword", "search", "input"):
                    if d.get(k):
                        return str(d[k])
            except Exception:
                pass
        return s

    def kb_search(input_val) -> str:
        query = _parse_query(input_val)
        if not query:
            return "[ERROR] KBSearch 需要查询词。用法: {\"query\": \"...\"} 或直接给关键词。"
        try:
            ctx = _build_kb_context(db, kb_ids, query,
                                    tenant_id=tenant_id,
                                    search_mode=search_mode, top_k=5)
        except Exception as e:
            return f"[KB_ERROR] 检索失败: {e}"
        if not ctx:
            return (f"[NO_MATCH] 在挂接的知识库中没有检索到与「{query}」"
                    f"相关的内容。可换关键词重试，或确认知识库已建索引。")
        return ctx

    def kb_list(_input_val=None) -> str:
        lines = ["[挂接的知识库]"]
        for kb_id in kb_ids:
            kb = db.fetchone(
                "SELECT * FROM knowledge_bases WHERE id = ? AND tenant_id = ?",
                (kb_id, tenant_id))
            if not kb:
                continue
            n = db.fetchone(
                "SELECT COUNT(*) AS c FROM kb_files WHERE kb_id = ?", (kb_id,))
            lines.append(f"- {kb.get('name', kb_id)} (id={kb_id}, "
                         f"文件 {(n or {}).get('c', 0)} 个) — "
                         f"{kb.get('description', '') or '无描述'}")
        if len(lines) == 1:
            return "[挂接的知识库] (无 — 该智能体未关联任何知识库)"
        lines.append("用 KBSearch({\"query\": \"...\"}) 在以上知识库中检索。")
        return "\n".join(lines)

    return {"KBSearch": kb_search, "KBList": kb_list}


# ── Persistent Memory API (Layer 1 Core + Layer 2 Recall) ──

def _get_agent_dir(agent_id: str, tenant_id: str, db: "Database") -> str:
    """Resolve the stable memory directory (instance_dir > agent_dir > work_dir).

    与 _execute_agent 的记忆落盘目录一致：手建 agent 的 instance/agent_dir
    为空时记忆落在自动开辟的稳定 work_dir（Workspace/<名>/），记忆/候选
    面板必须读同一处。"""
    row = db.fetchone(
        "SELECT instance_dir, agent_dir, work_dir FROM agents WHERE id = ? AND tenant_id = ?",
        (agent_id, tenant_id)
    )
    if not row:
        return ""
    return (row.get("instance_dir") or row.get("agent_dir")
            or row.get("work_dir") or "")


@router.get("/{agent_id}/memory/core")
async def get_core_memory(
    agent_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Return Core Memory (Layer 1) — the persistent key-value dict."""
    require_permission(tenant, "agents:read")
    if not db.fetchone(
        "SELECT id FROM agents WHERE id = ? AND tenant_id = ?",
        (agent_id, tenant.tenant_id)
    ):
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")
    from agentpaas.engine.memory_store import load_memory
    agent_dir = _get_agent_dir(agent_id, tenant.tenant_id, db)
    core = load_memory(agent_dir)
    return {"core": core, "agent_dir": agent_dir}


# ── 核心记忆候选（L2 自动提炼，记忆 P3）──

class CandidateAction(BaseModel):
    fact: str


@router.get("/{agent_id}/memory/candidates")
async def get_memory_candidates(
    agent_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """待确认的核心记忆候选（自动提炼，需用户确认才转正，评审#10）。"""
    require_permission(tenant, "agents:read")
    if not db.fetchone("SELECT id FROM agents WHERE id = ? AND tenant_id = ?",
                       (agent_id, tenant.tenant_id)):
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")
    d = _get_agent_dir(agent_id, tenant.tenant_id, db)
    from agentpaas.engine.memory_store import load_candidates, load_memory
    cands = load_candidates(d) if d else []
    cands.sort(key=lambda c: -(c.get("confidence") or 0))
    facts = (load_memory(d).get("_facts") or []) if d else []
    return {"candidates": cands, "confirmed_facts": facts}


@router.post("/{agent_id}/memory/candidates/confirm")
async def confirm_memory_candidate(
    agent_id: str,
    body: CandidateAction,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """确认一条候选 → 转正并入 core._facts。"""
    require_permission(tenant, "agents:write")
    if not db.fetchone("SELECT id FROM agents WHERE id = ? AND tenant_id = ?",
                       (agent_id, tenant.tenant_id)):
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")
    d = _get_agent_dir(agent_id, tenant.tenant_id, db)
    from agentpaas.engine.memory_store import confirm_candidate
    if not d or not confirm_candidate(d, body.fact):
        api_error(404, "CANDIDATE_NOT_FOUND", "候选不存在")
    return {"ok": True}


@router.post("/{agent_id}/memory/candidates/reject")
async def reject_memory_candidate(
    agent_id: str,
    body: CandidateAction,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """拒绝一条候选（删除，不转正）。"""
    require_permission(tenant, "agents:write")
    if not db.fetchone("SELECT id FROM agents WHERE id = ? AND tenant_id = ?",
                       (agent_id, tenant.tenant_id)):
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")
    d = _get_agent_dir(agent_id, tenant.tenant_id, db)
    from agentpaas.engine.memory_store import reject_candidate
    if not d or not reject_candidate(d, body.fact):
        api_error(404, "CANDIDATE_NOT_FOUND", "候选不存在")
    return {"ok": True}


@router.put("/{agent_id}/memory/core")
async def update_core_memory(
    agent_id: str,
    body: Dict[str, Any],
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Replace Core Memory with the given key-value dict (full overwrite)."""
    require_permission(tenant, "agents:write")
    if not db.fetchone(
        "SELECT id FROM agents WHERE id = ? AND tenant_id = ?",
        (agent_id, tenant.tenant_id)
    ):
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")
    from agentpaas.engine.memory_store import save_memory
    agent_dir = _get_agent_dir(agent_id, tenant.tenant_id, db)
    if not agent_dir:
        api_error(400, "NO_AGENT_DIR", "Agent has no persistent directory configured")
    save_memory(agent_dir, body)
    return {"core": body}


@router.delete("/{agent_id}/memory/core")
async def clear_core_memory(
    agent_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Clear (wipe) Core Memory for this agent."""
    require_permission(tenant, "agents:write")
    if not db.fetchone(
        "SELECT id FROM agents WHERE id = ? AND tenant_id = ?",
        (agent_id, tenant.tenant_id)
    ):
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")
    from agentpaas.engine.memory_store import save_memory
    agent_dir = _get_agent_dir(agent_id, tenant.tenant_id, db)
    if not agent_dir:
        api_error(400, "NO_AGENT_DIR", "Agent has no persistent directory configured")
    save_memory(agent_dir, {})
    return {"cleared": True}


@router.get("/{agent_id}/memory/recall")
async def get_recall_log(
    agent_id: str,
    n: int = 20,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Return the last N Recall Memory entries (Layer 2)."""
    require_permission(tenant, "agents:read")
    if not db.fetchone(
        "SELECT id FROM agents WHERE id = ? AND tenant_id = ?",
        (agent_id, tenant.tenant_id)
    ):
        api_error(404, "AGENT_NOT_FOUND", f"Agent {agent_id} not found")
    import json as _json
    from pathlib import Path as _Path
    agent_dir = _get_agent_dir(agent_id, tenant.tenant_id, db)
    entries: list = []
    if agent_dir:
        recall_file = _Path(agent_dir) / "memory" / "recall_log.jsonl"
        if recall_file.exists():
            lines = recall_file.read_text(encoding="utf-8").splitlines()
            recent = lines[-n:] if len(lines) >= n else lines
            for line in reversed(recent):  # newest first
                try:
                    entries.append(_json.loads(line))
                except Exception:
                    pass
    return {"entries": entries, "total": len(entries)}


# ── Internal helpers ──

def _compile_agent(config: dict):
    """Compile agent config via lambdagent. Raises on invalid config."""
    import tempfile, yaml, os
    with tempfile.NamedTemporaryFile(mode='w', suffix='.yml', delete=False, encoding='utf-8') as f:
        yaml.dump(config, f, allow_unicode=True)
        tmp_path = f.name
    try:
        from lambdagent.fromconfig import from_config
        return from_config(tmp_path)
    finally:
        os.unlink(tmp_path)


# 原生 claude-code 模式的 system 追加(纪律+安全,不含 react JSON 协议——native
# 用自己的结构化工具,讲"输出 JSON"反而会让它吐文本 JSON 而不调真工具)。
_NATIVE_SYSTEM_APPEND = (
    "你是在用户指定**工作文件夹**(即你的当前目录)里干活的结对助手。"
    "所有读、写、改、跑命令都在这个文件夹内就地完成,直接用你的原生工具"
    "(Read/Write/Edit/Bash/Glob/Grep)。\n"
    "- 用户提到的文件/目录(如 04_paper)先用 ls/glob 在工作文件夹里实际找到再动手;"
    "找不到就如实报告并询问正确位置,**绝不凭记忆编造内容或伪造数据**。\n"
    "- 危险/不可逆命令(rm -rf、git reset --hard、强制 push、覆盖重要文件)先说明影响再做。\n"
    "- 完成任务要真正把产物写到工作文件夹里,不要只在回答里描述。"
)


def _run_native_workspace(*, config: dict, input_text: str, inplace_dir: str,
                          workspace_path: str, on_step=None, run_id: str = ""):
    """车道 B:原生 claude-code 执行(workspace.assistant)。返回 (result, trace_info)。"""
    import os as _os
    from lambdagent.providers.claude_code_native import run_native

    model_cfg = (config.get("model") or {})
    model = model_cfg.get("name") or "sonnet"
    extra = model_cfg.get("extra") or {}

    # native 事件 → 平台 SSE 形状(前端复用现有 think_chunk/tool_call/tool_result)。
    def _native_on_event(ev: dict):
        if not on_step:
            return
        et = ev.get("event")
        if et == "think_chunk":
            on_step({"event": "think_chunk", "data": {"content": ev.get("text", "")}})
        elif et == "tool_call":
            on_step({"event": "tool_call",
                     "data": {"tool": ev.get("tool", ""),
                              "content": str(ev.get("input", ""))[:1000]}})
        elif et == "tool_result":
            on_step({"event": "tool_result", "data": {"content": ev.get("output", "")}})

    cancel = None
    if run_id:
        try:
            from agentpaas.engine.cancellation import get_cancel_token
            cancel = get_cancel_token(run_id)
        except Exception:
            cancel = None

    res = run_native(
        input_text, cwd=_os.path.abspath(inplace_dir), model=model,
        system_append=_NATIVE_SYSTEM_APPEND, on_event=_native_on_event,
        config_extra=extra, cancel=cancel,
    )

    trace = [{"term_name": "claude-code.native", "term_id": "native",
              "input": str(t.get("input", ""))[:500],
              "output": "", "duration_ms": 0, "model": model,
              "tokens_used": 0, "tool": t.get("tool", "")} for t in res.trace]
    import json as _json
    trace_info = {
        "steps": res.num_turns or len(res.trace),
        "input_tokens": res.input_tokens,
        "output_tokens": res.output_tokens,
        "cache_read_tokens": res.cache_read_tokens,
        "cache_creation_tokens": res.cache_creation_tokens,
        "cost_usd": round(res.cost_usd, 6),
        "trace_json": _json.dumps(trace, ensure_ascii=False),
        "workspace_path": workspace_path,
        "provider_error": None,
        "native_session_id": res.session_id,
    }
    if workspace_path:
        try:
            save_run_artifacts(
                workspace_path, str(res.text), trace, 0,
                res.input_tokens, res.output_tokens, trace_info["steps"],
            )
        except Exception as _se:
            logger.warning("native save_run_artifacts failed: %s", _se)
    return res.text, trace_info


def _execute_agent(config: dict, input_text: str, on_step=None,
                   agent_dir: str = "", run_id: str = "",
                   continue_workspace: str = "", work_dir: str = "",
                   source_dir: str = "", run_dir: str = "",
                   kb_tools: dict = None, inplace_dir: str = "",
                   native_workspace: bool = False):
    """Execute agent via lambdagent. Returns (result, trace_info).

    3-directory system
    ─────────────────
    source_dir : 数据来源目录. Read-only input data location.
        When set, a preamble is injected into the system prompt informing
        the agent where to find source documents/data.
    work_dir   : 工作产物目录. Final-output directory; shell Bash CWD.
        The agent writes its deliverables here.
    run_dir    : 运行日志目录. Base for run workspace creation
        (workspace/run_{ts}/ sub-dirs with traces, configs, intermediate
        artifacts). Falls back to work_dir, then agent_dir.

    Other args
    ──────────
    config: Agent YAML config dict.
    input_text: User prompt.
    on_step: Optional callback for streaming.
    agent_dir: Agent template/system directory (fallback for run_dir).
    run_id: Run ID for workspace metadata.
    continue_workspace: If set, reuse this existing workspace dir directly.
    """
    import yaml, os
    from agentpaas.engine.sandbox import create_run_workspace, save_run_artifacts

    # MCP 注册中心：把启用的 server 连接注入 config，供 _compile_mcp_caller
    # 解析。每条消息独立编译，禁用的 server 不注入 → 立即生效（评审#14）。
    try:
        from agentpaas.engine import mcp_registry
        config = mcp_registry.inject_into_config(config)
    except Exception as _mcp_e:
        logger.warning("MCP inject skipped: %s", _mcp_e)

    # 能力插件（prompt skill）：把 agent 声明的 skills 在编译前挂载进 config
    # （prompt 带边界包裹追加 systemPrompt、工具并进 localTools，高风险不
    # 自动授予）。Skill P2，docs/MCP_SKILL_DESIGN §2.2。
    try:
        from agentpaas.engine import skill_registry
        config, _skill_report = skill_registry.mount_into_config(config)
        if _skill_report.get("skipped_injection") or _skill_report.get("high_risk_skipped"):
            logger.info("skill mount report: %s", _skill_report)
    except Exception as _sk_e:
        logger.warning("skill mount skipped: %s", _sk_e)

    # ── source_dir injection ──────────────────────────────────────────────────
    # Prepend a brief preamble to the system prompt so the agent knows where
    # its input data lives. This is purely informational — the agent still
    # uses normal Bash/Read tools to access those files.
    if source_dir and os.path.isdir(source_dir):
        preamble = (
            f"[数据来源目录]\n"
            f"你的输入数据位于：{source_dir}\n"
            f"请优先从该目录读取所需文件。\n\n"
        )
        existing_prompt = config.get("systemPrompt") or ""
        config = dict(config)  # shallow copy to avoid mutating caller's dict
        config["systemPrompt"] = preamble + existing_prompt

    # 创建 run workspace
    # Priority: continue_workspace > run_dir > work_dir > agent_dir
    workspace_path = ""
    if continue_workspace and os.path.isdir(continue_workspace):
        # Reuse the previous run's workspace dir. Don't call
        # create_run_workspace — it'd build a new timestamped sibling and
        # the sub-agents would write into an empty tree, exactly the
        # confusing behavior the user reported. We still drop a small
        # marker file recording which run_id is iterating in here.
        workspace_path = os.path.abspath(continue_workspace)
        # CRITICAL: overwrite the stale config.yml in the reused workspace.
        # Otherwise from_config(workspace/config.yml) below picks up the
        # config snapshot from the ORIGINAL run, missing any DB-level edits
        # made since (toolTimeout bumps, enforceLoop tweaks, etc.).
        # Observed once with toolTimeout still 600 after we'd bumped to 1200.
        try:
            with open(os.path.join(workspace_path, "config.yml"), "w", encoding="utf-8") as f:
                yaml.dump(config, f, allow_unicode=True, default_flow_style=False)
        except Exception as _e:
            logger.warning("could not refresh config.yml in reused workspace: %s", _e)
        try:
            with open(os.path.join(workspace_path, f"continue_{run_id}.json"), "w", encoding="utf-8") as f:
                import json as _json, time as _time
                _json.dump({
                    "run_id": run_id,
                    "input": input_text,
                    "timestamp": _time.strftime("%Y%m%d_%H%M%S"),
                    "mode": "continue",
                }, f, ensure_ascii=False, indent=2)
        except Exception as _e:
            logger.warning("could not write continue marker: %s", _e)
        logger.info("reusing workspace for continue-run: %s", workspace_path)
    elif run_dir and os.path.isdir(run_dir):
        # run_dir is the dedicated log/trace base; workspace lives DIRECTLY here
        # (no extra "workspace/" subdir — the user already named this dir whatever
        # they want, e.g. "…/review67/workspace"). bare=True skips the subdir.
        workspace_path = create_run_workspace(run_dir, run_id, input_text, config, bare=True)
    elif work_dir and os.path.isdir(work_dir):
        # Fallback: create run workspace inside work_dir when run_dir is not set.
        workspace_path = create_run_workspace(work_dir, run_id, input_text, config)
    elif agent_dir:
        workspace_path = create_run_workspace(agent_dir, run_id, input_text, config)

    # CRITICAL: Bash/文件工具的会话 CWD 必须 == 给模型看的 [工作目录] 提示
    # (compiler step-0 注入的 = overrides["workspace_path"])。否则模型把产物写成
    # 相对路径时落到 CWD,而提示叫它写到另一个目录 → 产物落错地方、run 查看器空、
    # 跨 run 互相覆盖(run_bf2405e28547 实锤:literature_map.md 落进 agent 家目录而非
    # 本 run 工作区)。对齐规则:
    #   · 工作区对话模式(inplace_dir=F):CWD = F(就地读写用户文件夹)。
    #   · 专项 agent:CWD = 本 run 的 workspace_path(per-run 隔离,查看器可见,不覆盖);
    #     读输入靠 [数据来源目录] 的绝对路径(step-0 提示已明确禁止相对/无参读)。
    _cwd_to_set = _resolve_session_cwd(inplace_dir, workspace_path, work_dir)
    # P1 工作文件夹硬沙箱：工作区模式（inplace_dir 有值）下，把 WriteFile/EditFile
    # 的写限定在所选文件夹 F 内，越界写直接拒绝。仅工作区模式启用——专项智能体
    # 写各自 run 工作区，root=None 不受影响。finally(下方)清空，防泄漏到下个 run。
    try:
        from lambdagent.builtin_tools._sandbox import set_sandbox_root
        set_sandbox_root(inplace_dir if (inplace_dir and os.path.isdir(inplace_dir)) else None)
    except Exception as _sbe:
        logger.warning("set sandbox root failed: %s", _sbe)

    if _cwd_to_set:
        try:
            from lambdagent.builtin_tools.shell_tools import _set_cwd as _shell_set_cwd
            _shell_set_cwd(_cwd_to_set)
            logger.info(
                "shell CWD → %s (%s)",
                _cwd_to_set,
                "work_dir" if _cwd_to_set == work_dir else "workspace",
            )
        except Exception as _e:
            logger.warning("failed to set shell session CWD: %s", _e)

    # 配置写入 workspace 或 /tmp
    if workspace_path:
        config_path = os.path.join(workspace_path, "config.yml")
    else:
        import tempfile
        with tempfile.NamedTemporaryFile(mode='w', suffix='.yml', delete=False, encoding='utf-8') as f:
            yaml.dump(config, f, allow_unicode=True)
            config_path = f.name

    try:
        from lambdagent.fromconfig import from_config
        from lambdagent.core import Context

        # Wire on_step callback into the compiled term's ReActEngine
        overrides = {}
        if workspace_path:
            overrides["workspace_path"] = workspace_path
        # 工作区对话模式：给模型看的 [工作目录] = 用户选的文件夹 F（就地读写改），
        # 而 config.yml / trace 仍写到上面的 workspace_path（agent run_dir，不污染 F）。
        # 即把"模型操作目录"与"日志/产物目录"解耦——仅工作区模式（inplace_dir 有值）生效。
        if inplace_dir and os.path.isdir(inplace_dir):
            overrides["workspace_path"] = os.path.abspath(inplace_dir)
        if on_step:
            def _step_adapter(step_event):
                """Convert StepEvent dataclass to SSE dict for the queue."""
                on_step({
                    "event": step_event.type,
                    "data": {
                        "step": step_event.step,
                        "content": step_event.content,
                        "tool": step_event.tool,
                        "duration_ms": step_event.duration_ms,
                    },
                })
            overrides["on_step"] = _step_adapter

            # Bridge on_chunk for ClaudeLam streaming (provider: claude-code)
            def _chunk_adapter(text):
                """Push ClaudeLam streaming chunks as SSE think_chunk events."""
                on_step({
                    "event": "think_chunk",
                    "data": {"content": text},
                })
            overrides["on_chunk"] = _chunk_adapter

        # AUDIT_2026-06-11 ③: SEC-12/30 闭环 — HIGH-risk 工具调用确认。
        # ToolGateway 产生 Action.CONFIRM 时回调到这里：先推一个
        # confirm_required 事件给前端（流式 run 可见），然后阻塞等待
        # POST /traces/{run_id}/confirm，超时按拒绝处理（fail-closed）。
        if run_id:
            from agentpaas.engine.confirmations import request_confirmation

            def _confirm_cb(_tool, _input, _reason):
                if on_step:
                    on_step({
                        "event": "confirm_required",
                        "data": {"run_id": run_id, "tool": _tool,
                                 "reason": _reason, "input": str(_input)[:500]},
                    })
                return request_confirmation(run_id, _tool, str(_input), _reason)

            overrides["_confirm_callback"] = _confirm_cb

        # KB 调用链修复: 注入平台版 KBSearch/KBList（查该 agent 挂接的
        # 知识库），替换 lambdagent 内置的同名工具（那套查的是另一个
        # 互不相通的 ~/.lambdagent 本地库）。compiler 侧 custom tools
        # 优先于 builtin，且与 subAgents 的 call_* 工具合并共存。
        if kb_tools:
            overrides["tools"] = {**(overrides.get("tools") or {}), **kb_tools}

        # ── 车道 B：原生 claude-code（workspace.assistant 专用）──
        # 见 docs/NATIVE_CLAUDE_CODE_DESIGN.md。让 claude-code 当 agent 本身,在
        # 工作文件夹里用 native 工具干活,而不是套进 react-over-text 封装(那套实测
        # 工具调用 0 执行、丢上下文、空转)。仅工作区模板 + 有效文件夹时启用。
        if native_workspace and inplace_dir and os.path.isdir(inplace_dir):
            return _run_native_workspace(
                config=config, input_text=input_text, inplace_dir=inplace_dir,
                workspace_path=workspace_path, on_step=on_step, run_id=run_id,
            )

        term = from_config(config_path, **overrides)
        ctx = Context(workspace_path=workspace_path, run_id=run_id)

        # P1-2: Load persistent Core Memory into ctx.memory before execution
        from agentpaas.engine.memory_store import load_memory, save_memory
        if agent_dir:
            ctx.memory = load_memory(agent_dir)

        t0_exec = time.time()
        result = term.apply(input_text, ctx)
        duration_ms = int((time.time() - t0_exec) * 1000)

        # ── Accurate token counts ──
        # react: _compile_react 把 _think_ref(ConversationLam) 挂在 Loop 上。
        # simple (P3 token 透传修复): term 本身就是 ConversationLam，或被
        # Guard/Memory 包装一层 —— 此前只认 _think_ref，单轮智能体的
        # usage 恒为 None，运行记录 token/成本一律记 0。现在沿包装链
        # 找到第一个带 get_usage 的节点。
        def _find_usage_source(t, depth=0):
            if t is None or depth > 5:
                return None
            ref = getattr(t, "_think_ref", None)
            if ref is not None and callable(getattr(ref, "get_usage", None)):
                return ref
            if callable(getattr(t, "get_usage", None)):
                return t
            return _find_usage_source(getattr(t, "agent", None), depth + 1)

        usage = None
        usage_src = _find_usage_source(term)
        if usage_src is not None:
            try:
                usage = usage_src.get_usage()
            except Exception as _gu_err:
                logger.warning("get_usage failed: %s", _gu_err)

        # Extract trace info
        # BUG-06: serialize ctx.trace to JSON for DB persistence
        import json as _json
        def _trace_entry_to_dict(e):
            return {
                "term_name": getattr(e, 'term_name', ''),
                "term_id": getattr(e, 'term_id', ''),
                "input": str(getattr(e, 'input', ''))[:500],
                "output": str(getattr(e, 'output', ''))[:500],
                "duration_ms": getattr(e, 'duration_ms', 0),
                "model": getattr(e, 'model', ''),
                "tokens_used": getattr(e, 'tokens_used', 0),
            }
        try:
            trace_json = _json.dumps([_trace_entry_to_dict(e) for e in ctx.trace], ensure_ascii=False)
        except Exception:
            trace_json = "[]"

        real_input        = (usage or {}).get("input_tokens", 0) or sum(getattr(e, 'tokens_used', 0) for e in ctx.trace)
        real_output       = (usage or {}).get("output_tokens", 0)
        cache_read        = (usage or {}).get("cache_read_input_tokens", 0)
        cache_create      = (usage or {}).get("cache_creation_input_tokens", 0)
        from agentpaas.engine.sandbox import _compute_cost_usd
        cost_usd          = _compute_cost_usd(usage) if usage else 0.0

        trace_info = {
            "steps": len(ctx.trace),
            "input_tokens": real_input,
            "output_tokens": real_output,
            "cache_read_tokens": cache_read,
            "cache_creation_tokens": cache_create,
            "total_tokens": real_input + real_output + cache_read + cache_create,
            "cost_usd": round(cost_usd, 6),
            "duration_ms": duration_ms,
            "workspace_path": workspace_path,
            "trace_json": trace_json,
            # AUDIT_2026-06-11 ②: 结构化 provider 错误（lambdagent Context
            # 记录的 API 级失败），供 run 边界成败判定优先使用，
            # 替代对输出文本的正则猜测。
            "provider_errors": list(getattr(ctx, "provider_errors", []) or []),
        }

        # P1-2: Persist updated Core Memory + append Recall log entry
        if agent_dir:
            save_memory(agent_dir, ctx.memory, run_id=run_id,
                        input_text=input_text, output_text=str(result))

        # 保存 run 产物（含真实 token 数 + USD 成本估算）
        if workspace_path:
            save_run_artifacts(
                workspace_path, str(result), ctx.trace,
                duration_ms, real_input, real_output, trace_info["steps"],
                usage=usage,
            )

        return result, trace_info
    finally:
        # P1 沙箱：清空写沙箱 root，防泄漏到下一个 run（模块级全局状态）。
        try:
            from lambdagent.builtin_tools._sandbox import set_sandbox_root
            set_sandbox_root(None)
        except Exception:
            pass
        # 只删除 /tmp 临时文件，workspace 里的保留
        if not workspace_path:
            try:
                os.unlink(config_path)
            except OSError:
                pass
