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

-- 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;

-- 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;

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"))

git commit -m "feat(fr056): agent_tasks table + state machine + agent capabilities (T-3.1)"

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}

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.add_resource(AgentTaskAckRoute, '/agents/tasks/<string:uid>/ack')
api.add_resource(AgentTaskResultRoute, '/agents/tasks/<string:uid>/result')

@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")

# 1. 手動 INSERT 一筆 agent_tasks（pending）到 DEV
# 2. 模擬 agent 心跳（mTLS）→ 應回 pending_tasks 含該筆
# 3. 模擬 ack → status 變 dispatched
# 4. 模擬 result（succeeded）→ status 變 succeeded
# 每步 psql 查 status 確認

git commit -m "feat(fr056): heartbeat task dispatch + ack + result endpoints (T-3.2)"

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

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)

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）
    ...

git commit -m "feat(fr056): agent task executor skeleton (T-3.3)

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

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。"""

git commit -m "feat(fr056): OpenVAS connector (T-3.4)"