feat(agent): 新增规划/执行流水线、Skills API 与 Worker 支持

.gitignore

@@ -1,4 +1,12 @@
 *.env
+.env
+.env.*
+**/.env
+**/.env.local
+**/.env.secrets
+app/src/.env
+app/src/.env.local
+app/src/.env.secrets
 
 */node_modules/
 .vscode/
@@ -30,6 +38,7 @@
 docs/design
 .sandbox_*
 .test_*
+*.md
 
 app/artifacts/
 

app/.env.example

@@ -7,6 +7,7 @@ JWT_SECRET_KEY=please-change-me
 ACCESS_TOKEN_EXPIRE_MINUTES=4320
 REFRESH_TOKEN_EXPIRE_DAYS=7
 
+# 示例：redis://root:password@127.0.0.1:6379/0
 REDIS_URL=redis://localhost:6379/0
 # RabbitMQ is optional in this stage, kept for future publisher swap.
 # RABBITMQ_URL=amqp://guest:guest@localhost:5672/
@@ -38,7 +39,11 @@ WORKER_QUEUE_STREAM=fileflash:tasks
 WORKER_QUEUE_GROUP=fileflash-workers
 WORKER_QUEUE_BLOCK_MS=5000
 
-AGENT_ENABLED=false
+# 开发建议 true；生产可 false。503 Agent is disabled 时检查此项与 APP_ENV
+AGENT_ENABLED=true
+APP_ENV=development
+AGENT_INLINE_PROCESSING=true
+AGENT_ALLOW_WRITE_TOOLS=false
 AGENT_QUEUE_STREAM=fileflash:agents
 AGENT_QUEUE_GROUP=fileflash-agents
 AGENT_QUEUE_BLOCK_MS=5000
@@ -52,8 +57,10 @@ AGENT_USER_DAILY_LIMIT=50
 AGENT_USER_CONCURRENT_LIMIT=2
 AGENT_STAGING_TTL_SEC=86400
 AGENT_SSE_ENABLED=false
-AGENT_LLM_PROVIDER=anthropic
-AGENT_LLM_MODEL=claude-sonnet-4-6
+AGENT_LLM_PROVIDER=deepseek
+AGENT_LLM_MODEL=deepseek-chat
+AGENT_LLM_BASE_URL=https://api.deepseek.com
+# 密钥放 .env.local，见 .env.local.example
 # AGENT_LLM_API_KEY=
 AGENT_MCP_ENDPOINTS=[]
 

app/pyproject.toml

@@ -19,15 +19,21 @@ dependencies = [
     "sqlalchemy>=2.0.48",
     "uvicorn>=0.42.0",
     "python-multipart>=0.0.24",
+    "langchain-core>=0.3.0",
+    "langchain>=0.3.0",
+    "langchain-anthropic>=0.3.0",
+    "langchain-openai>=0.3.0",
 ]
 [project.scripts]
 fileflash = "fileflash.main:main"
 fileflash-dev = "fileflash.scripts.run_with_workers:main"
+fileflash-agent-worker = "fileflash.workers.agent_consumer:main"
 
 [dependency-groups]
 dev = [
     "httpx>=0.28.1",
     "hypothesis>=6.151.9",
+    "psycopg2-binary>=2.9.10",
     "pytest-asyncio>=1.2.0",
     "pytest>=9.0.2",
     "ruff>=0.15.6",
@@ -50,4 +56,4 @@ requires = ["hatchling"]
 build-backend = "hatchling.build"
 
 [tool.hatch.build.targets.wheel]
-packages = ["src/fileflash"]
+packages = ["src/fileflash"]

app/scripts/flyway_migrate.py

@@ -0,0 +1,164 @@
+#!/usr/bin/env python3
+"""Apply docker/flyway/migrations/*.sql in order (Flyway-compatible history table).
+
+Usage (from app/):
+  uv run python scripts/flyway_migrate.py
+  uv run python scripts/flyway_migrate.py --dry-run
+"""
+
+from __future__ import annotations
+
+import argparse
+import re
+import sys
+from pathlib import Path
+
+_APP_ROOT = Path(__file__).resolve().parents[1]
+if str(_APP_ROOT) not in sys.path:
+    sys.path.insert(0, str(_APP_ROOT))
+
+from fileflash.core.settings import get_settings
+
+HISTORY_DDL = """
+CREATE TABLE IF NOT EXISTS flyway_schema_history (
+    installed_rank SERIAL PRIMARY KEY,
+    version VARCHAR(50) NOT NULL,
+    description VARCHAR(200) NOT NULL,
+    type VARCHAR(20) NOT NULL DEFAULT 'SQL',
+    script VARCHAR(1000) NOT NULL,
+    checksum INTEGER NULL,
+    installed_by VARCHAR(100) NOT NULL DEFAULT 'flyway_migrate.py',
+    installed_on TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    success BOOLEAN NOT NULL DEFAULT TRUE
+);
+"""
+
+
+def _repo_migrations_dir() -> Path:
+    # app/scripts -> repo root/docker/flyway/migrations
+    return _APP_ROOT.parent / "docker" / "flyway" / "migrations"
+
+
+def _parse_version(path: Path) -> tuple[int, str, str]:
+    # V4__worker.sql -> (4, "4", "worker")
+    match = re.match(r"V(\d+)__(.+)\.sql$", path.name, re.IGNORECASE)
+    if not match:
+        raise ValueError(f"Unexpected migration file name: {path.name}")
+    number = int(match.group(1))
+    description = match.group(2)
+    return number, str(number), description
+
+
+def _sync_database_url(async_url: str) -> str:
+    if async_url.startswith("postgresql+asyncpg://"):
+        return "postgresql://" + async_url[len("postgresql+asyncpg://") :]
+    return async_url
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Run FileFlash SQL migrations")
+    parser.add_argument("--dry-run", action="store_true", help="List pending migrations only")
+    parser.add_argument(
+        "--mark-through",
+        type=int,
+        default=0,
+        help="Mark migrations <= this version as applied without executing (baseline existing DB)",
+    )
+    parser.add_argument(
+        "--from-version",
+        type=int,
+        default=1,
+        help="Only consider migrations with version >= this number",
+    )
+    args = parser.parse_args()
+
+    try:
+        import psycopg2
+    except ImportError:
+        print(
+            "psycopg2 is required: uv add --dev psycopg2-binary\n"
+            "Or run on Linux VM: bash docker/flyway/run-migrate.sh",
+            file=sys.stderr,
+        )
+        return 1
+
+    settings = get_settings()
+    db_url = _sync_database_url(settings.async_database_url)
+    migrations_dir = _repo_migrations_dir()
+    if not migrations_dir.is_dir():
+        print(f"Migrations not found: {migrations_dir}", file=sys.stderr)
+        return 1
+
+    files = sorted(migrations_dir.glob("V*.sql"), key=lambda p: _parse_version(p)[0])
+    if not files:
+        print("No migration files found.", file=sys.stderr)
+        return 1
+
+    conn = psycopg2.connect(db_url)
+    conn.autocommit = True
+    cur = conn.cursor()
+    cur.execute(HISTORY_DDL)
+
+    cur.execute("SELECT version FROM flyway_schema_history WHERE success = TRUE")
+    applied = {row[0] for row in cur.fetchall()}
+
+    pending = []
+    for path in files:
+        rank, version, description = _parse_version(path)
+        if rank < args.from_version:
+            continue
+        if version in applied:
+            print(f"skip V{version} ({description})")
+            continue
+        if args.mark_through and rank <= args.mark_through:
+            print(f"baseline V{version} ({description})")
+            cur.execute(
+                """
+                INSERT INTO flyway_schema_history (version, description, script, success)
+                VALUES (%s, %s, %s, TRUE)
+                ON CONFLICT DO NOTHING
+                """,
+                (version, description, path.name),
+            )
+            applied.add(version)
+            continue
+        pending.append((path, version, description))
+
+    if not pending:
+        print("All migrations already applied.")
+        return 0
+
+    print(f"Database: {db_url.split('@')[-1]}")
+    print(f"Pending: {len(pending)} migration(s)")
+    for path, version, description in pending:
+        print(f"  - V{version} {description} ({path.name})")
+
+    if args.dry_run:
+        return 0
+
+    for path, version, description in pending:
+        sql = path.read_text(encoding="utf-8")
+        print(f"Applying V{version} {description}...")
+        try:
+            cur.execute(sql)
+        except Exception as exc:
+            print(f"FAILED V{version}: {exc}", file=sys.stderr)
+            conn.close()
+            return 2
+        cur.execute(
+            """
+            INSERT INTO flyway_schema_history (version, description, script, success)
+            VALUES (%s, %s, %s, TRUE)
+            """,
+            (version, description, path.name),
+        )
+        print(f"OK V{version}")
+
+    cur.close()
+    conn.close()
+    print("Migrations complete. Restart uvicorn and check: agent.db ok table=background_job")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

app/src/fileflash/agents/harness/policy.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
+from typing import Any
 
 
 @dataclass(slots=True)
@@ -10,5 +11,26 @@ class PolicyDecision:
 
 
 class PolicyGuard:
-    async def evaluate_tool_call(self, *args, **kwargs) -> PolicyDecision:
-        raise NotImplementedError("PolicyGuard is scaffolded only in this stage")
+    def __init__(
+        self,
+        *,
+        allow_writes: bool = False,
+        data_policy: dict[str, Any] | None = None,
+        execution_policy: str = "confirm",
+    ) -> None:
+        self.allow_writes = allow_writes
+        self.data_policy = data_policy or {}
+        self.execution_policy = execution_policy
+
+    def evaluate_tool_call(self, tool: str, side_effect: str) -> PolicyDecision:  # noqa: ARG002
+        if side_effect == "write":
+            if self.execution_policy == "planOnly":
+                return PolicyDecision(allowed=False, reasons=["executionPolicy is planOnly"])
+            if not self.allow_writes:
+                return PolicyDecision(
+                    allowed=False,
+                    reasons=["Write tools disabled (set AGENT_ALLOW_WRITE_TOOLS=true to enable)"],
+                )
+            if not self.data_policy.get("allowFileContent", self.data_policy.get("allow_file_content", True)):
+                return PolicyDecision(allowed=False, reasons=["dataPolicy disallows write operations"])
+        return PolicyDecision(allowed=True)

app/src/fileflash/agents/harness/router.py

@@ -3,13 +3,18 @@
 from dataclasses import dataclass
 from typing import Any
 
+from ..tools.drive import DriveToolContext
+
 
 @dataclass(slots=True)
 class ToolCall:
-    tool_name: str
-    arguments: dict[str, Any]
+    tool: str
+    inputs: dict[str, Any]
 
 
 class ToolRouter:
-    async def dispatch(self, call: ToolCall) -> dict[str, Any]:
-        raise NotImplementedError("ToolRouter is scaffolded only in this stage")
+    def __init__(self, *, drive: DriveToolContext) -> None:
+        self._drive = drive
+
+    async def execute(self, call: ToolCall) -> dict[str, Any]:
+        return await self._drive.invoke(call.tool, call.inputs)

app/src/fileflash/agents/llm.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from typing import Any
+
+from ..core.settings import Settings
+
+
+def build_chat_model(settings: Settings) -> Any | None:
+    """Return a LangChain chat model when configured; otherwise None (use mock planner)."""
+    api_key = (settings.agent_llm_api_key or "").strip()
+    if not api_key:
+        return None
+
+    provider = (settings.agent_llm_provider or "deepseek").strip().lower()
+    model = settings.agent_llm_model
+
+    try:
+        if provider == "anthropic":
+            from langchain_anthropic import ChatAnthropic
+
+            return ChatAnthropic(model=model, api_key=api_key, temperature=0.2)
+        if provider in {"openai", "azure_openai"}:
+            from langchain_openai import ChatOpenAI
+
+            return ChatOpenAI(model=model, api_key=api_key, temperature=0.2)
+        if provider == "deepseek":
+            from langchain_openai import ChatOpenAI
+
+            base_url = (settings.agent_llm_base_url or "https://api.deepseek.com").rstrip("/")
+            if not base_url.endswith("/v1"):
+                base_url = f"{base_url}/v1"
+            return ChatOpenAI(
+                model=model or "deepseek-chat",
+                api_key=api_key,
+                base_url=base_url,
+                temperature=0.2,
+            )
+    except Exception:
+        return None
+    return None