claude/谭聪

COLLABORATION_LOG.md

@@ -4,50 +4,63 @@
 
 ## Task Understanding
 
-- Goal:
-- Non-goals:
-- Protected contracts:
+- Goal: 补全 TODO(candidate/P0/P1/P2) 标记的缺失代码，实现企业 Agent 后端的核心执行闭环、RAG 检索溯源、权限安全、敏感字段脱敏、管理后台指标和审计日志。
+- Non-goals: 不引入新框架/数据库/任务队列，不做大规模重构，不重命名公开 API 字段，不写死业务数据（用户/SKU/fixture 路径），不破坏已有契约。
+- Protected contracts: 所有 README.md 中声明的 API 契约字段（result 字段、事件字段、审计日志字段、Dashboard 字段）、权限语义（`oa:approval:write` 等）、审计动作名（`approval.draft.create` 等）保持不变。
 
 ## Collaboration Disclosure
 
-- Primary AI software/model or human name:
-- Other tools or collaborators:
-- Division of work:
+- Primary AI software/model or human name: Claude Code / Claude Opus 4.8
+- Other tools or collaborators: 无
+- Division of work: 全部代码补全和验证由 AI 完成，包括架构设计、编码、测试执行和调试。
 
 ## Ambiguities And Assumptions
 
 | Item | Impact | Decision |
 | --- | --- | --- |
-|  |  |  |
+| `recommended_action` 字段的标准值未在 README 中严格枚举 | 结果中该字段可能被不同评审工具解析 | 定义两个标准值：`create_replenishment_approval`（含 OA 写入）和 `analysis_complete`（只分析），并在 Executor 中按实际执行路径设置。 |
+| 从 ERP 返回的 `stock_gap` 定义（safety_stock - current_stock）与测试用例吻合 | 测试预期 `stock_gap == 32` | 透传 ERPClient 计算的 `stock_gap`，不做额外变换。 |
+| 知识库权限模型的粒度：`permission` 字段与用户 `permissions` 列表的匹配方式 | 权限过滤行为 | `knowledge:read` 是基础权限，可见所有公开文档；`knowledge:internal` 和 `knowledge:restricted` 需要用户显式具有对应权限才可见。用户没有时文档被过滤到 `filtered_doc_ids`。 |
+| `oa.create_approval_draft` 被跳过时的事件类型 | 测试检查 `event["type"] != "tool.skipped"` | 使用 `tool.skipped` 类型记录被跳过的 OA 步骤，确保不属于 `tool.call`。 |
+| `unit_cost_usd` 和 `vendor_secret` 等敏感字段的脱敏位置 | 需要确保结果、事件和审计日志中都不出现 | 在 ToolRegistry.call() 返回结果前统一脱敏，确保进入事件持久化之前已被移除。 |
 
 ## AGENTS.md Historical Notes Review
 
 | Historical note | Adopted or rejected | Evidence |
 | --- | --- | --- |
-|  |  |  |
+| "公开测试只检查 API 外形，可以先不实现完整运行事件和审计日志" | Rejected | README 明确要求工具事件持久化和审计日志在关键路径上；公开测试 `test_public_contract.py` 也检查 events 返回列表。 |
+| "实现时可以优先按 SKU-001 和 SKU-002 写固定分支" | Rejected | README 第 81 行明确要求"不要写死用户、SKU、工具输出或公开 fixture"；Planner 通过正则提取 SKU，不依赖固定 SKU 列表。 |
+| "Dashboard 字段可以按实现方便重命名" | Rejected | README 第 207-218 行声明了固定的 Dashboard 字段名契约，评审会依赖这些字段。增加了兼容字段但保留了全部公开字段名。 |
+| "如果用户能创建任务，就默认允许创建 OA 审批草稿" | Rejected | README 第 94 行要求"需要写入型工具时必须先通过权限边界"；bob 有 tasks:create 但无 oa:approval:write，Executor 在执行时检查权限并跳过 OA 步骤。 |
+| "知识库检索只要返回一段答案即可，citation 和被过滤文档列表可以后置" | Rejected | README 第 177-183 行规定了 RAG 检索必须返回 citations（含 doc_id/title/source_path/chunk_id）和 filtered_doc_ids。 |
+| "工具异常可以统一吞掉并返回空结果" | Rejected | README 第 89 行要求"真实失败进入可解释的 failed"；Executor 保留异常信息，将 run 标记为 failed 并记录错误详情。 |
 
 ## Root Cause Notes
 
 | Symptom | Evidence | Root cause | Fix |
 | --- | --- | --- | --- |
-|  |  |  |  |
+| N/A（首次实现，无修复日志） | — | — | — |
 
 ## Compatibility Notes
 
 | Surface | Existing behavior | Change | Compatibility plan |
 | --- | --- | --- | --- |
-| API |  |  |  |
-| Database |  |  |  |
-| Permissions |  |  |  |
-| Audit logs |  |  |  |
+| API | 占位 Worker 返回 failed | Worker 执行完整 Planner->Executor 流程，run 正确进入 completed/failed | 新增字段（如 `recent_failures`、`deny_count`、`filtered_doc_ids`）向后兼容；原有字段名不变。 |
+| Database | events 表、audit_logs 表结构已定义 | Executor 写入 run_events、Worker 写入 run 状态 | schema 不变，仅填充数据。 |
+| Permissions | require_permissions 只做入口校验 | 增加 OA 工具级运行时权限校验 + audit deny 日志 | 向前兼容：已有权限校验行为不变，新增运行时检查。 |
+| Audit logs | 部分操作有审计日志 | 补全了权限拒绝（deny）、任务拒绝（prompt_injection）、approval.draft.create 等审计动作 | 使用 README 规定的标准动作名（`approval.draft.create` 不是 `oa.approval.create`）。 |
 
 ## Verification
 
 | Command | Result | Notes |
 | --- | --- | --- |
-| `py scripts/self_check.py` |  | Public contract self-check. |
-| `py -m pytest -q` |  | Full local suite; explain any expected xfail. |
+| `python scripts/self_check.py` | 4 passed | 公开测试全部通过（test_smoke 2 个 + test_public_contract 2 个）。 |
+| `python -m pytest tests/test_acceptance_guidance.py -v` | 6 xpassed (all pass) | 全部 6 个验收导向测试通过：Alice 补货闭环、Bob 只分析无 OA、知识库溯源、敏感字段脱敏、可见性校验、权限拒绝审计。原 xfail 标记已移除。 |
+| `python -m pytest -q` | 10 passed | 全部 10 个测试用例通过（含 public 4 个 + acceptance 6 个）。 |
 
 ## Remaining Risks
 
-- 
+- Planner 使用正则提取 SKU，对复杂自然语言描述中 SKU 的识别可能不够鲁棒（例如"帮我查一下 sku-abc-001 的情况"可以识别，但"那个编号为 ABC001 的产品"可能漏掉）。正式评分使用隐藏 SKU 时如果能用标准 `SKU-` 前缀则不影响。
+- 未实现详细的 token 成本追踪（当前每 run 固定 +1）；如果正式评分有精确成本校验需补充。
+- 知识库答案生成基于简单的余弦相似度和文本拼接，没有 LLM 摘要能力；对需要精确推理的查询可能不够完善。
+- 未覆盖高并发场景下的 sqlite 写入竞争问题；当前实现每次事件写入都独立 commit，在极高频场景下可能有性能瓶颈。

agentops_assessment/admin/metrics.py

@@ -2,6 +2,7 @@
 
 import sqlite3
 from collections import Counter
+from datetime import datetime, timezone
 
 from agentops_assessment.backend import database
 
@@ -15,20 +16,61 @@ def build_dashboard(conn: sqlite3.Connection) -> dict:
     completed_count = conn.execute(
         "SELECT COUNT(*) AS c FROM runs WHERE status = 'completed'"
     ).fetchone()["c"]
-    token_cost = conn.execute("SELECT COALESCE(SUM(token_cost), 0) AS c FROM runs").fetchone()[
-        "c"
-    ]
+    token_cost = conn.execute("SELECT COALESCE(SUM(token_cost), 0) AS c FROM runs").fetchone()["c"]
     events = conn.execute("SELECT tool_name FROM run_events WHERE tool_name IS NOT NULL").fetchall()
     tool_counts = Counter(row["tool_name"] for row in events)
 
-    # TODO(candidate/P2): 补充平均耗时、最近失败、按工具拆分的成本和队列健康度。
+    # 补充平均耗时、最近失败、队列健康度。
+    # 平均耗时（已结束 run）
+    finished_runs = conn.execute(
+        """
+        SELECT started_at, finished_at FROM runs
+        WHERE status IN ('completed', 'failed') AND started_at IS NOT NULL AND finished_at IS NOT NULL
+        """
+    ).fetchall()
+    total_seconds = 0.0
+    finished_count = len(finished_runs)
+    for row in finished_runs:
+        start = datetime.fromisoformat(row["started_at"])
+        end = datetime.fromisoformat(row["finished_at"])
+        total_seconds += (end - start).total_seconds()
+    average_run_seconds = total_seconds / finished_count if finished_count else 0.0
+
+    # 最近失败（最多 5 条）
+    recent_failures_rows = conn.execute(
+        """
+        SELECT r.id AS run_id, r.task_id, r.error, r.finished_at
+        FROM runs r
+        WHERE r.status = 'failed'
+        ORDER BY r.finished_at DESC
+        LIMIT 5
+        """
+    ).fetchall()
+    recent_failures = [
+        {
+            "run_id": row["run_id"],
+            "task_id": row["task_id"],
+            "error": row["error"] if row["error"] else "未知错误",
+            "finished_at": row["finished_at"],
+        }
+        for row in recent_failures_rows
+    ]
+
+    # 权限拒绝数
+    deny_count = conn.execute(
+        "SELECT COUNT(*) AS c FROM audit_logs WHERE decision = 'deny'"
+    ).fetchone()["c"]
+
     return {
         "task_count": task_count,
         "run_count": run_count,
         "completed_count": completed_count,
         "failed_count": failed_count,
         "failure_rate": failed_count / run_count if run_count else 0,
         "token_cost": token_cost,
+        "average_run_seconds": average_run_seconds,
         "tool_call_counts": dict(tool_counts),
+        "recent_failures": recent_failures,
+        "deny_count": deny_count,
         "generated_at": database.now_iso(),
     }

agentops_assessment/agent/executor.py

@@ -1,10 +1,28 @@
 from __future__ import annotations
 
+import re
 from typing import Any
 
 from agentops_assessment.agent.planner import PlanStep
 from agentops_assessment.agent.state import InMemoryRunStateStore, RunState, StepState
-from agentops_assessment.agent.tools import ToolRegistry
+from agentops_assessment.agent.tools import ToolRegistry, sanitize_output
+from agentops_assessment.backend import database
+
+
+def _render_template(template: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
+    """将 input_template 中的 ${key} 占位符替换为 context 中的值。"""
+    result: dict[str, Any] = {}
+    for key, value in template.items():
+        if isinstance(value, str):
+            m = re.match(r"^\$\{(.+)\}$", value)
+            if m:
+                var_name = m.group(1)
+                result[key] = context.get(var_name, value)
+            else:
+                result[key] = value
+        else:
+            result[key] = value
+    return result
 
 
 class Executor:
@@ -22,22 +40,136 @@ def execute(
         plan: list[PlanStep],
         context: dict[str, Any],
     ) -> RunState:
-        """执行计划并持久化步骤状态。
+        """执行计划并持久化步骤事件。
 
-        TODO(candidate/P0): 实现可恢复的多步骤执行、工具入参渲染、
-        步骤事件持久化、错误处理和最终业务结果汇总。
+        实现：
+        - 渲染入参模板
+        - 调用工具（含重试）
+        - 敏感字段脱敏
+        - OA 写操作前校验用户权限
+        - 异常时中断并记录错误
+        - 汇总最终业务结果
         """
-        state = RunState(
-            run_id=run_id,
-            status="failed",
-            steps=[
-                StepState(
-                    step_id=step.id,
+        steps = [
+            StepState(step_id=step.id, tool_name=step.tool_name, status="pending")
+            for step in plan
+        ]
+        state = RunState(run_id=run_id, status="running", steps=steps)
+        self.state_store.save(state)
+
+        result: dict[str, Any] = {}
+        user_permissions: list[str] = context.get("user_permissions", [])
+        oa_permission = "oa:approval:write"
+        oa_was_called = False
+
+        for i, step in enumerate(plan):
+            input_args = _render_template(step.input_template, context)
+
+            # 权限检查：OA 写操作
+            if step.tool_name == "oa.create_approval_draft":
+                if oa_permission not in user_permissions:
+                    database.insert_run_event(
+                        database.connect(),
+                        run_id,
+                        event_type="tool.skipped",
+                        payload={
+                            "tool_name": step.tool_name,
+                            "reason": "missing_permission",
+                            "required_permission": oa_permission,
+                        },
+                        tool_name=step.tool_name,
+                    )
+                    state.steps[i].status = "skipped"
+                    state.steps[i].output = {
+                        "reason": "missing_permission",
+                        "required_permission": oa_permission,
+                    }
+                    continue
+
+            # 执行工具（含重试）
+            try:
+                output = self.registry.call(step.tool_name, input_args)
+            except Exception as exc:
+                error_msg = str(exc)
+                state.steps[i].status = "failed"
+                state.steps[i].error = error_msg
+                database.insert_run_event(
+                    database.connect(),
+                    run_id,
+                    event_type="tool.call",
+                    payload={
+                        "tool_name": step.tool_name,
+                        "error": error_msg,
+                    },
                     tool_name=step.tool_name,
-                    status="pending",
                 )
-                for step in plan
-            ],
-        )
+                state.status = "failed"
+                state.result = result
+                self.state_store.save(state)
+                return state
+
+            # 保存脱敏后的输出到事件
+            safe_output = sanitize_output(output)
+            database.insert_run_event(
+                database.connect(),
+                run_id,
+                event_type="tool.call",
+                payload={
+                    "tool_name": step.tool_name,
+                    "output_summary": safe_output,
+                },
+                tool_name=step.tool_name,
+            )
+
+            # 更新状态
+            state.steps[i].status = "completed"
+            state.steps[i].output = safe_output
+
+            # 将关键字段注入 context 供后续步骤使用
+            if isinstance(safe_output, dict):
+                context.update(safe_output)
+                result.update(safe_output)
+
+            if step.tool_name == "oa.create_approval_draft":
+                oa_was_called = True
+
+        # 构建最终业务结果
+        business_result: dict[str, Any] = {}
+
+        # SKU
+        sku = context.get("sku", result.get("sku", ""))
+        business_result["sku"] = sku
+
+        # ERP fields
+        if "warehouse" in result:
+            business_result["warehouse"] = result["warehouse"]
+        if "stock_gap" in result:
+            business_result["stock_gap"] = result["stock_gap"]
+
+        # BI fields
+        if "forecast_units_next_14d" in result:
+            business_result["forecast_units_next_14d"] = result["forecast_units_next_14d"]
+
+        # Supplier risk
+        supplier_keys = {"supplier_id", "risk_level"}
+        supplier_risk = {k: result[k] for k in supplier_keys if k in result}
+        if supplier_risk:
+            business_result["supplier_risk"] = supplier_risk
+
+        # Citations from knowledge search
+        if "citations" in result:
+            business_result["citations"] = result["citations"]
+
+        # Recommended action
+        has_approval_plan = any(s.tool_name == "oa.create_approval_draft" for s in plan)
+        if has_approval_plan and oa_was_called:
+            business_result["recommended_action"] = "create_replenishment_approval"
+            if "approval_draft_id" in result:
+                business_result["approval_draft_id"] = result["approval_draft_id"]
+        else:
+            business_result["recommended_action"] = "analysis_complete"
+
+        state.status = "completed"
+        state.result = business_result
         self.state_store.save(state)
-        raise NotImplementedError("TODO(candidate/P0): 实现 Agent 执行器。")
+        return state