# FR-056.3 Agent 執行能力擴充 Implementation Plan

> **For agentic workers / runner:** REQUIRED SUB-SKILL: superpowers:subagent-driven-development / executing-plans. `- [ ]` checkbox syntax.
>
> **依賴：FR-056.1（需 detection_tools 目錄）。可與 FR-056.2 並行。跨兩 repo：BE + evidence-agent。** 母案設計見 [`design.md`](./design.md)。

**Goal:** 讓客戶端 Agent 能領到掃描派工、執行 OpenVAS、回傳報告。雲端建 `agent_tasks` 派工表 + 狀態機 + 心跳夾帶待辦 + 結果回收 endpoint；Agent 端（`evidence-agent` repo）在心跳迴圈接 executor + OpenVAS connector。此階段做到「手動塞一筆派工 → Agent 跑完回報」全鏈通，不接 UI（UI 編排在 FR-056.4）。

**Architecture:** Pull 模式——Agent 主動心跳、雲端回應夾帶 `pending_tasks`、Agent 領走執行、再用獨立 result endpoint 回傳。認證全沿用既有 mTLS（agent→cloud 方向純 mTLS，不掛 user JWT，與既有 heartbeat/register 一致）。雲端狀態機 mirror `ssp_excel_parse_job` 的 `update_status()` + 語意 wrapper。Agent executor 與 OpenVAS connector 是**全新模組**（evidence-agent 內零 subprocess/外部 API 先例）。

**Tech Stack:** BE 同前；evidence-agent = Flask + httpx（mTLS client 已有）；OpenVAS 整合走其 API（GVM/gvm-tools）或 CLI，第一版擇一。

---

## 前置：關鍵發現與決策（runner 必讀）

| 項目 | 定案 | 來源 |
|------|------|------|
| 派工模式 | Pull（心跳夾帶待辦 + result endpoint），非雲端主動推 | D3 |
| agent→cloud 認證 | 純 mTLS，route **不掛 `@jwt_required()`**（同既有 heartbeat/register） | Explore |
| 心跳改法 | BE `heartbeat()` return 前插 `pending_tasks` 查詢；Agent `_heartbeat_once` 改 `return resp.json()` 交 executor | Explore |
| executor 執行緒 | **另開 thread**（`ThreadPoolExecutor`/individual thread），不阻塞心跳迴圈 | Explore |
| 狀態機範本 | `ssp_excel_parse_job_domain_service` 的 `update_status()` + `mark_running`/`write_result`/`write_error` wrapper | Explore |
| agent_tasks 狀態 | `pending → dispatched → running → succeeded / failed` | design §5 |
| executor 走不走 DI | evidence-agent 心跳邏輯目前是 plain module（非 DI）；executor **維持 plain module 一致**，除非要暴露 route | 見 Task 4 |
| **憑證下發（D9 定案）** | **(a) 雲端解密後隨派工下發**（走 mTLS 通道）。雲端在 `list_pending_for_agent` 組派工時，把該租戶 `tenant_detection_tool_configs` 憑證解密放進 task payload。**不採 (b) Agent 自持**——要客戶在 Agent 端自設不實際，雲端 UI 統一設定才符合產品體驗 | user 拍板 2026-07-26 |
| **OpenVAS 整合（D10 定案）** | **走 API（python-gvm，GVM protocol）**，非 CLI。程式化控制掃描比解析 CLI 輸出穩定；FR-056.1 seed 的 connection_type=`API` 一致 | user 拍板 2026-07-26 |

**⚠️ 全新模組警示：** `evidence-agent` repo 內 grep `subprocess`/`requests` 皆無，只有 `core/enroll.py` 用 httpx 打雲端。executor + OpenVAS connector **沒有可抄的先例**，是綠地。唯一可參考的 HTTP client 慣例是 `core/enroll.py` 的 `httpx.Client(cert=..., verify=...)`。

---

## Task 分佈（對應 design T-3.x）

| Task | 對應 | 產出 | 依賴 | Repo |
|------|------|------|------|------|
| Task 1 | T-3.1 | agent_tasks 表 + DDD 全層 + 狀態機 + Agent capability | FR-056.1 | BE |
| Task 2 | T-3.2 | 心跳夾帶待辦 + ack + 結果回收 endpoint | Task 1 | BE |
| Task 3 | T-3.3 | Agent executor 模組骨架（領任務/回報/送 result） | Task 2 | evidence-agent |
| Task 4 | T-3.4 | OpenVAS connector（API/CLI） | Task 3 | evidence-agent |

---

## Task 1（T-3.1）: agent_tasks 表 + DDD 全層 + 狀態機

**Files:**
- Create: `scripts/sql/2026-07-26-fr056-3-agent-tasks.sql`
- Create: infra/domain/app 全層 `agent_task` 模組（複製 remote_agent 骨架）
- Modify: `infra/remote_agent/model/remote_agent.py`（加 `capabilities` 欄位標記能執行掃描）
- Create: `scripts/sql/2026-07-26-fr056-3-remote-agents-capabilities.sql`（remote_agents 加欄位）
- Test: `test/test_agent_task_service.py`

- [ ] **Step 1: migration 建 agent_tasks + remote_agents 加 capabilities**

`agent_tasks` 放 `compliance` schema（與 remote_agents 同 schema，派工是 agent 領域）:
```sql
-- Date: 2026-07-26
-- FR-056.3 Agent 派工表 + 狀態機
CREATE TABLE compliance.agent_tasks (
    id                 BIGSERIAL PRIMARY KEY,
    uid                VARCHAR(36)  NOT NULL UNIQUE,
    tenant_id          BIGINT       NOT NULL,
    agent_id           BIGINT       NOT NULL,             -- soft-ref → compliance.remote_agents.id
    job_execution_uid  VARCHAR(50),                        -- soft-ref → 對應稽核任務（FR-056.4 才填，本階段可 null）
    detection_tool_id  BIGINT       NOT NULL,             -- soft-ref → config.detection_tools.id
    params             JSONB        NOT NULL DEFAULT '{}'::jsonb,  -- 掃描參數快照
    status             VARCHAR(20)  NOT NULL DEFAULT 'pending',   -- pending/dispatched/running/succeeded/failed
    result_ref         JSONB,                              -- 結果檔參照（report upload uid 等）
    error_message      TEXT,
    dispatched_at      TIMESTAMPTZ,
    started_at         TIMESTAMPTZ,
    finished_at        TIMESTAMPTZ,
    created_user       VARCHAR(255), updated_user VARCHAR(255),
    created_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at         TIMESTAMPTZ NOT NULL DEFAULT now()
);
COMMENT ON TABLE compliance.agent_tasks IS 'FR-056.3 Agent 掃描派工單 + 狀態機';
GRANT SELECT, INSERT, UPDATE, DELETE ON compliance.agent_tasks TO cm_app;
GRANT USAGE, SELECT ON SEQUENCE compliance.agent_tasks_id_seq TO cm_app;
ALTER TABLE compliance.agent_tasks ENABLE ROW LEVEL SECURITY;
CREATE POLICY agent_tasks_tenant_isolation ON compliance.agent_tasks
    USING (current_setting('app.is_super_admin', TRUE) = 'true'
           OR tenant_id = ANY (string_to_array(current_setting('app.allowed_tenant_paths', TRUE), ',')::BIGINT[]));
INSERT INTO public.schema_migrations(filename, note) VALUES
  ('2026-07-26-fr056-3-agent-tasks.sql', 'FR-056.3 Agent 派工表') ON CONFLICT (filename) DO NOTHING;
```
另一支 migration 給 remote_agents 加 capabilities（標記哪些 agent 能執行掃描）:
```sql
-- Date: 2026-07-26
-- FR-056.3 remote_agents 加 capabilities（agent 能力標記，如 ["file_storage","detection_scan"]）
ALTER TABLE compliance.remote_agents ADD COLUMN IF NOT EXISTS capabilities JSONB NOT NULL DEFAULT '["file_storage"]'::jsonb;
COMMENT ON COLUMN compliance.remote_agents.capabilities IS 'FR-056.3 agent 能力清單，派掃描任務只給含 detection_scan 者';
INSERT INTO public.schema_migrations(filename, note) VALUES
  ('2026-07-26-fr056-3-remote-agents-capabilities.sql', 'FR-056.3 remote_agents 加 capabilities') ON CONFLICT (filename) DO NOTHING;
```
套進 DEV + psql 驗證。

- [ ] **Step 2: DDD 全層 agent_task 模組**（複製 remote_agent 骨架，tenant-scoped）。

- [ ] **Step 3: 狀態機 domain service（mirror ssp_excel_parse_job）**

`domain/agent_task/service/agent_task_domain_service.py`，統一入口 + 語意 wrapper：
```python
def update_status(self, uid, status, user, **fields):
    entity = AgentTaskEntity(uid=uid, status=status, updated_user=user, **fields)
    return self._repo.update(entity)

def mark_dispatched(self, uid, user):
    return self.update_status(uid, "dispatched", user, dispatched_at=datetime.now())
def mark_running(self, uid, user):
    return self.update_status(uid, "running", user, started_at=datetime.now())
def write_success(self, uid, result_ref, user):
    return self.update_status(uid, "succeeded", user, result_ref=result_ref, finished_at=datetime.now())
def write_failure(self, uid, error_message, user):
    return self.update_status(uid, "failed", user, error_message=error_message, finished_at=datetime.now())

def list_pending_for_agent(self, agent_id, tenant_id):
    return self._repo.get_all_by_fields(AgentTaskQueryEntity(agent_id=agent_id, tenant_id=tenant_id, status="pending"))
```

- [ ] **Step 4: 寫 failing test（狀態轉換 + list_pending）** → FAIL → 實作 app service（`@transaction` 包）→ PASS。

- [ ] **Step 5: Commit**
```bash
git commit -m "feat(fr056): agent_tasks table + state machine + agent capabilities (T-3.1)"
```

---

## Task 2（T-3.2）: 心跳夾帶待辦 + ack + 結果回收 endpoint

**Files:**
- Modify: `app/remote_agent/service/agent_enrollment_service.py`（heartbeat 回應加 pending_tasks）
- Modify: `api/remote_agent/routes/remote_agent_route.py`（加 ack + result route）
- Modify: `api/remote_agent/__init__.py`（掛新 route）
- Modify: `api/remote_agent/serializers/`（result payload schema）
- Test: `test/test_agent_heartbeat_dispatch.py`

- [ ] **Step 1: 寫 failing test（心跳回應含 pending_tasks + result 回收改狀態）** → FAIL。

- [ ] **Step 2: heartbeat 回應夾帶待辦**

`app/remote_agent/service/agent_enrollment_service.py` 的 `heartbeat()`，在 `return {"status": "ok"}` 前改（注意此 method 已在 `@transaction` + 顯式帶 tenant_id）:
```python
pending = self._agent_task_domain.list_pending_for_agent(agent.id, agent.tenant_id)
tasks = []
for t in pending:
    # D9：解密該租戶工具憑證，隨派工下發（走 mTLS 通道，Agent 用完即丟不落地）
    creds = self._resolve_tool_credentials(t.detection_tool_id, agent.tenant_id)  # 查 tenant_detection_tool_configs → crypto.decrypt
    tasks.append({"uid": t.uid, "detection_tool_id": t.detection_tool_id,
                  "params": t.params, "credentials": creds})
return {"status": "ok", "pending_tasks": tasks}
```
constructor 注入 `agent_task_domain_service` + `tenant_config_domain_service` + `crypto`（DI 加）。`_resolve_tool_credentials` 查該租戶對應工具的 `tenant_detection_tool_configs`、`crypto.decrypt(credentials_encrypted)` 回明文 dict。
> **D9 安全註記：** 憑證只在 mTLS 加密通道下發，Agent 端不持久化（executor 用完即丟）。雲端仍是憑證唯一保管處（加密存放），符合「租戶在雲端 UI 統一設定」的產品體驗。

- [ ] **Step 3: ack + result endpoint（不掛 jwt，純 mTLS）**

`api/remote_agent/routes/remote_agent_route.py` 加兩支（mirror `AgentHeartbeatRoute`，**無 `@jwt_required()`**）:
```python
class AgentTaskAckRoute(MethodResource):
    @inject
    def post(self, uid, agent_task_service=Provide[...]):
        # agent 領到任務後 ack → dispatched
        return return_response(True, agent_task_service.ack_task(uid))

class AgentTaskResultRoute(MethodResource):
    @inject
    def post(self, uid, agent_task_service=Provide[...]):
        payload = request.get_json(silent=True) or {}
        # payload: {status: "succeeded"|"failed", result_ref?: {...}, error_message?: str}
        # 報告檔上傳走既有 blob 資料面（Agent 端已有），此處只收結果 metadata
        return return_response(True, agent_task_service.receive_result(uid, payload))
```
掛進 `api/remote_agent/__init__.py`:
```python
api.add_resource(AgentTaskAckRoute, '/agents/tasks/<string:uid>/ack')
api.add_resource(AgentTaskResultRoute, '/agents/tasks/<string:uid>/result')
```
> **報告檔怎麼傳：** Agent 端已有 blob 上傳能力（`/blob*` 資料面）。掃描報告先走既有 blob 通道上傳成 upload_file，Agent 再把回傳的 upload uid 放進 result payload 的 `result_ref`。**FR-056.3 不新建檔案傳輸通道**，複用既有 blob。FR-056.4 的轉證據 handler 再把這個 upload 轉成 job_evidences。

- [ ] **Step 4: app service `ack_task` / `receive_result`**（`@transaction`，呼叫 Task 1 的狀態 wrapper）:
```python
@transaction
def ack_task(self, uid):
    return self._domain.mark_dispatched(uid, "agent")

@transaction
def receive_result(self, uid, payload):
    if payload.get("status") == "succeeded":
        return self._domain.write_success(uid, payload.get("result_ref"), "agent")
    return self._domain.write_failure(uid, payload.get("error_message") or "unknown", "agent")
```
> **FR-056.4 掛鉤點：** `receive_result` 成功時，FR-056.4 會在此觸發「轉證據 + 通知 + 完成模式分岔」。本階段先只更新狀態，留 TODO(FR-056.4) 註記。

- [ ] **Step 5: 跑 test 通過 + 手測整鏈**

手測（脫離 UI，直接塞派工驗證）:
```bash
# 1. 手動 INSERT 一筆 agent_tasks（pending）到 DEV
# 2. 模擬 agent 心跳（mTLS）→ 應回 pending_tasks 含該筆
# 3. 模擬 ack → status 變 dispatched
# 4. 模擬 result（succeeded）→ status 變 succeeded
# 每步 psql 查 status 確認
```

- [ ] **Step 6: Commit**
```bash
git commit -m "feat(fr056): heartbeat task dispatch + ack + result endpoints (T-3.2)"
```

---

## Task 3（T-3.3）: Agent executor 模組骨架（evidence-agent repo）

> **此 Task 在 `evidence-agent` repo** `~/Projects/Billows/Audit-Manager/evidence-agent/`。全新模組，無先例可抄，唯一參考 `core/enroll.py` 的 httpx mTLS client 慣例。

**決策點（runner 動工前定，或問 user）：** executor 走不走 DI 容器？evidence-agent 現況心跳是 plain module（`core/enroll.py` function-based，未走 DI）。**建議：executor 維持 plain module 與心跳一致**（心跳迴圈直接 import 呼叫），除非之後要暴露「手動觸發」route 才需 DI wire。本計畫按 plain module 寫。

**Files:**
- Create: `core/task_executor.py`（領任務 → 呼叫 connector → 回報）
- Modify: `core/enroll.py`（`_heartbeat_once` 讀 response、`_heartbeat_loop` 交給 executor）
- Create: `common/code/agent_task_error_code.py`（若 repo 有 error code 慣例，比照既有 `common/code/agent_error_code.py`）

- [ ] **Step 1: `_heartbeat_once` 改回傳 response body**

`core/enroll.py`:
```python
def _heartbeat_once(cfg, device_uuid, state) -> dict:
    ...
    resp = client.post(url, json={...})
    resp.raise_for_status()
    return resp.json()   # 原本丟棄，改回傳
```

- [ ] **Step 2: `_heartbeat_loop` 接 executor（fire-and-forget thread）**

```python
from core.task_executor import dispatch_tasks
...
def _heartbeat_loop(cfg, device_uuid, state):
    while True:
        try:
            body = _heartbeat_once(cfg, device_uuid, state)
            tasks = body.get("pending_tasks") or []
            if tasks:
                dispatch_tasks(tasks, cfg, state)   # 內部另開 thread，不阻塞迴圈
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 404:
                # 既有重註冊邏輯
                ...
        time.sleep(cfg.heartbeat_interval)
```

- [ ] **Step 3: `core/task_executor.py` 骨架**

```python
import json, threading
import httpx
from core.task_executor_connectors import get_connector  # Task 4 提供

def dispatch_tasks(tasks, cfg, state):
    for task in tasks:
        threading.Thread(target=_run_one, args=(task, cfg, state), daemon=True).start()

def _run_one(task, cfg, state):
    task_uid = task["uid"]
    _ack(task_uid, cfg, state)                    # → dispatched
    try:
        connector = get_connector(task["detection_tool_id"], task["params"])  # Task 4
        report_bytes = connector.run()            # 執行掃描（阻塞在這個 thread，不影響心跳）
        upload_uid = _upload_report(report_bytes, cfg, state)  # 走既有 blob 通道
        _result(task_uid, {"status": "succeeded", "result_ref": {"upload_uid": upload_uid}}, cfg, state)
    except Exception as e:
        _result(task_uid, {"status": "failed", "error_message": str(e)}, cfg, state)

def _ack(task_uid, cfg, state):
    _post(f"/api/1.0/agents/tasks/{task_uid}/ack", {}, cfg, state)

def _result(task_uid, payload, cfg, state):
    _post(f"/api/1.0/agents/tasks/{task_uid}/result", payload, cfg, state)

def _post(path, payload, cfg, state):
    url = f"{cfg.cloud_endpoint}{path}"
    with httpx.Client(cert=(state.cert_file, state.key_file), verify=state.ca_file, timeout=60) as client:
        r = client.post(url, json=payload); r.raise_for_status(); return r.json()

def _upload_report(report_bytes, cfg, state):
    # 走既有 blob 資料面上傳（參考 api/blob 的既有上傳；回傳 upload uid）
    ...
```
> `_upload_report` 細節 runner 依 evidence-agent 既有 blob 上傳實作補（既有能力，非新建）。

- [ ] **Step 4: 手測（mock connector）**

先用假 connector（回固定 bytes）驗證整鏈：塞派工 → agent 心跳領到 → ack → 假掃描 → 上傳 → result → 雲端 status succeeded。connector 真身在 Task 4。

- [ ] **Step 5: Commit（evidence-agent repo，顯式 git add）**
```bash
git commit -m "feat(fr056): agent task executor skeleton (T-3.3)

心跳迴圈接 executor：領 pending_tasks → thread 執行 → ack/result 回報。connector 為 Task 4。"
```

---

## Task 4（T-3.4）: OpenVAS connector（evidence-agent repo）

> **全新模組，evidence-agent 無 subprocess/外部 API 先例。** OpenVAS **走 API（python-gvm，GVM protocol）**（D10 定案），非 CLI。

**Files:**
- Create: `core/task_executor_connectors/__init__.py`（`get_connector` factory）
- Create: `core/task_executor_connectors/base.py`（connector 介面）
- Create: `core/task_executor_connectors/openvas.py`（OpenVAS 實作）

- [ ] **Step 1: connector 介面（三層 adapter 的介面層）**

`core/task_executor_connectors/base.py`:
```python
from abc import ABC, abstractmethod

class DetectionConnector(ABC):
    def __init__(self, params: dict, credentials: dict):
        self.params = params
        self.credentials = credentials
    @abstractmethod
    def run(self) -> bytes:
        """執行檢測，回傳報告原始 bytes（XML/PDF）。失敗 raise。"""
```
> **憑證怎麼到 agent（D9 定案＝(a)）：** 雲端在 Task 2 的 `list_pending_for_agent` 組派工時，把該租戶 `tenant_detection_tool_configs.credentials_encrypted` **解密後放進 task payload**（走既有 mTLS 加密通道下發）。Agent 執行時從 payload 取憑證用完即用，**不落地持久化**。`get_connector` 收到的 `credentials` 就來自 payload。此決定已回寫 Task 2 Step 2 的 pending_tasks 組裝（需在心跳查詢後補解密步驟）。

- [ ] **Step 2: factory**

`get_connector(detection_tool_id, params, credentials)` 依工具 code 回對應 connector。第一版只有 openvas。

- [ ] **Step 3: OpenVAS connector 實作**

`core/task_executor_connectors/openvas.py`——用 `python-gvm`（GVM protocol）連 OpenVAS：連線 → 建 target（掃描 IP/範圍來自 params）→ 建 task → 啟動 → 輪詢完成 → 取報告 XML → 回 bytes。逾時/連線失敗 raise。

> **依賴：** evidence-agent `pyproject.toml` 要加 `python-gvm`（或走 CLI 就不用）。這是新依賴，runner 加。

- [ ] **Step 4: 手測整鏈（真 OpenVAS）**

塞派工（含真實掃描目標）→ agent 領到 → 連真 OpenVAS 跑掃描 → 報告上傳 → 雲端 status succeeded + result_ref 有 upload uid。這是 FR-056.3 的最終驗收：**脫離 UI 的完整 dispatch→scan→result 鏈通**。

- [ ] **Step 5: Commit（evidence-agent repo）**
```bash
git commit -m "feat(fr056): OpenVAS connector (T-3.4)"
```

---

## 完成後收尾（等 user 下令）

- [ ] STG/POC 套 agent_tasks + capabilities migration。
- [ ] evidence-agent 部署到客戶端測試機（含 python-gvm 依賴）。
- [ ] FR-056.4 掛鉤：`receive_result` 成功時觸發轉證據（本階段留的 TODO(FR-056.4)）。
- [ ] Notion FR-056.3 子卡標「修正待驗證」+ 回寫。

---

## 決策狀態（D9/D10 已定案）

1. **憑證下發到 Agent** → **D9 定案：(a) 雲端解密後隨派工下發（走 mTLS，Agent 用完即丟不落地）。** 已回寫 Task 2 Step 2 的 pending_tasks 組裝。（原因：要客戶在 Agent 端自設不實際，雲端 UI 統一設定才符合產品體驗。）
2. **OpenVAS 整合方式** → **D10 定案：API（python-gvm，GVM protocol）。**
3. **executor DI vs plain module**（Task 3）：維持 plain module 與既有心跳一致——實作風格，runner 逕行，非阻塞決策。

---

## 附錄：範本來源

- 狀態機：`domain/oscal/service/ssp_excel_parse_job_domain_service.py`（`update_status` + 語意 wrapper）
- 心跳既有：`app/remote_agent/service/agent_enrollment_service.py::heartbeat`
- agent route 慣例（不掛 jwt）：`api/remote_agent/routes/remote_agent_route.py::AgentHeartbeatRoute`
- 雲端→agent mTLS client（反向參考）：`infra/upload_file/remote_agent_adapter.py::_request`
- Agent 心跳迴圈：`evidence-agent/core/enroll.py::_heartbeat_once` / `_heartbeat_loop`
- DDD 九層：同 FR-056.1 附錄 A（remote_agent 模組）