Browse Source

feat(web-api): read-only config endpoints, effect status fix, walk edge execution facts

- 新增 GET /config/rule-packs|walk-strategy|query-prompts(固定白名单路径,
  只读返回 product_documents 配置 JSON 原文 + source_file)
- 修复 rule_application_summary 的 content_effect_status 恒 None
  (decision 记录无该字段;改为 search_query_effect_status 优先,与 M6C 同口径)
- walk_graph 游走动作边增补 rule_pack_executed / executed_rule_pack_id
  (取自 raw_payload.rule_pack_execution,支撑归属/执行分离展示)
- tests/test_api.py 追加配置端点与修复断言(301 passed)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Sam Lee 3 weeks ago
parent
commit
86517cc22e
4 changed files with 89 additions and 1 deletions
  1. 37 0
      content_agent/api.py
  2. 7 1
      content_agent/dashboard_service.py
  3. 5 0
      content_agent/schemas.py
  4. 40 0
      tests/test_api.py

+ 37 - 0
content_agent/api.py

@@ -12,6 +12,7 @@ from content_agent.dashboard_service import DashboardService
 from content_agent.errors import ErrorCode, error_response, sanitize_error_detail
 from content_agent.run_service import RunService
 from content_agent.schemas import (
+    ConfigFileResponse,
     ContentItemsResponse,
     DashboardResponse,
     JsonFileResponse,
@@ -117,6 +118,42 @@ def get_run(run_id: str) -> RunSummaryResponse:
     return RunSummaryResponse(**service.get_summary(run_id))
 
 
+_CONFIG_FILES = {
+    "rule-packs": "product_documents/规则包/douyin_rule_packs.v1.json",
+    "walk-strategy": "product_documents/抖音游走策略/douyin_walk_strategy.v1.json",
+    "query-prompts": "product_documents/配置/query_prompts.v1.json",
+}
+
+
+def _config_file_response(key: str) -> ConfigFileResponse:
+    from pathlib import Path
+
+    from content_agent.integrations import config_store
+
+    path = Path(_CONFIG_FILES[key])
+    if not path.exists():
+        raise HTTPException(status_code=404, detail=error_response(
+            ErrorCode.RUN_NOT_FOUND, f"config file missing: {key}", {"source_file": str(path)}
+        ))
+    data, _ = config_store.load_json(path)
+    return ConfigFileResponse(source_file=str(path), data=data)
+
+
+@app.get("/config/rule-packs", response_model=ConfigFileResponse)
+def get_config_rule_packs() -> ConfigFileResponse:
+    return _config_file_response("rule-packs")
+
+
+@app.get("/config/walk-strategy", response_model=ConfigFileResponse)
+def get_config_walk_strategy() -> ConfigFileResponse:
+    return _config_file_response("walk-strategy")
+
+
+@app.get("/config/query-prompts", response_model=ConfigFileResponse)
+def get_config_query_prompts() -> ConfigFileResponse:
+    return _config_file_response("query-prompts")
+
+
 @app.get("/runs/{run_id}/dashboard", response_model=DashboardResponse)
 def get_run_dashboard(run_id: str) -> DashboardResponse:
     _ensure_web_run_exists(run_id)

+ 7 - 1
content_agent/dashboard_service.py

@@ -711,7 +711,10 @@ def _rule_application_summary(
             "score": decision.get("score") or scorecard.get("total_score"),
             "decision_action": decision.get("decision_action"),
             "decision_reason_code": decision.get("decision_reason_code"),
-            "content_effect_status": decision.get("content_effect_status"),
+            # 与 _effect_status_counts 同口径: decision 正式字段是 search_query_effect_status,
+            # content_effect_status 仅旧数据回退(decision 记录从无该字段时原代码恒读 None)。
+            "content_effect_status": decision.get("search_query_effect_status")
+            or decision.get("content_effect_status"),
             "primary_reason": _reason_label(decision.get("decision_reason_code")),
             "technical_ref": {
                 "decision_id": decision.get("decision_id"),
@@ -786,6 +789,7 @@ def _walk_graph(
             "label": action.get("to_node_id") or action.get("edge_type") or "下一跳",
             "status": action.get("walk_status") or "pending",
         })
+        execution = (action.get("raw_payload") or {}).get("rule_pack_execution") or {}
         edges.append({
             "id": action.get("walk_action_id") or f"{source_id}->{target_id}",
             "source": source_id,
@@ -793,6 +797,8 @@ def _walk_graph(
             "label": action.get("edge_id") or action.get("edge_type") or action.get("walk_action"),
             "status": action.get("walk_status") or "pending",
             "rule_pack": action.get("rule_pack_id"),
+            "rule_pack_executed": execution.get("executed"),
+            "executed_rule_pack_id": execution.get("executed_rule_pack_id"),
             "budget_tier": action.get("budget_tier"),
             "reason_code": action.get("reason_code"),
         })

+ 5 - 0
content_agent/schemas.py

@@ -144,6 +144,11 @@ class QueryListResponse(BaseModel):
     data_origin: str
 
 
+class ConfigFileResponse(BaseModel):
+    source_file: str
+    data: dict[str, Any]
+
+
 class TimelineResponse(BaseModel):
     run_id: str
     items: list[dict[str, Any]]

+ 40 - 0
tests/test_api.py

@@ -225,3 +225,43 @@ def test_api_timeline_includes_summary(tmp_path, monkeypatch):
     }
     assert summary["stage_duration_ms"]
     assert "stalled" not in timeline and "is_blocked" not in timeline
+
+
+def test_api_config_readonly_endpoints():
+    client = TestClient(api.app)
+
+    rule_packs = client.get("/config/rule-packs").json()
+    assert rule_packs["source_file"].endswith("douyin_rule_packs.v1.json")
+    assert len(rule_packs["data"]["rule_pack_dispatch"]) == 5
+    assert len(rule_packs["data"]["effect_status_mapping"]) == 5
+
+    walk = client.get("/config/walk-strategy").json()
+    assert len(walk["data"]["walk_rule_pack_binding"]) == 8
+    assert len(walk["data"]["walk_edge_catalog"]) == 10
+
+    prompts = client.get("/config/query-prompts").json()
+    assert "douyin/V1" in prompts["data"]["profiles"]
+
+
+def test_api_dashboard_rule_summary_effect_status_not_none(tmp_path, monkeypatch):
+    monkeypatch.setattr(
+        api,
+        "service",
+        RunService(
+            runtime_root=tmp_path / "runtime" / "v1",
+            demand_source=FakeDemandSource(),
+            query_variant_client=FakeQueryVariantClient(),
+        ),
+    )
+    client = TestClient(api.app)
+    run_id = client.post("/runs", json={"platform": "douyin", "platform_mode": "mock"}).json()["run_id"]
+
+    dashboard = client.get(f"/runs/{run_id}/dashboard").json()
+
+    summaries = dashboard["rule_application_summary"]
+    assert summaries
+    # 修复前 content_effect_status 恒为 None(decision 记录无该字段)。
+    assert all(row["content_effect_status"] is not None for row in summaries)
+    # walk_graph edge 带归属/执行分离字段。
+    walk_edges = [e for e in dashboard["walk_graph"]["edges"] if e.get("rule_pack")]
+    assert any("rule_pack_executed" in e for e in walk_edges)