浏览器输入任务 → 
FastAPI 创建 session → 
Docker 容器里运行 pi --mode rpc → 
Agent 在 workspace 副本里写代码 → 
用户审查 diff → 
apply 到真实项目

interface ModelDefinition {
  id: string;
  provider: string;
  protocol: 'openai-completions' | 'openai-responses' | 'anthropic' | 'google';
  contextWindow: number;
  maxOutputTokens: number;
  inputCostPer1M: number;
  outputCostPer1M: number;
  capabilities: {
    vision: boolean;
    toolUse: boolean;
    streaming: boolean;
    reasoning: boolean;
  };
}

import { getModel, stream, Context } from '@mariozechner/pi-ai';

const model = getModel('anthropic', 'claude-sonnet-4-20250514');

const context: Context = {
  systemPrompt: 'You are a helpful assistant.',
  messages: [{ role: 'user', content: 'Hello' }]
};

for await (const event of stream(model, context)) {
  if (event.type === 'text_delta') {
    process.stdout.write(event.delta);
  }
}

import { Type } from '@mariozechner/pi-ai';

const tools = [{
  name: 'read_file',
  description: 'Read contents of a file',
  parameters: Type.Object({
    path: Type.String({ description: 'Absolute path to file' }),
    startLine: Type.Optional(Type.Number({ description: 'Start line (1-indexed)' })),
    endLine: Type.Optional(Type.Number({ description: 'End line (1-indexed)' }))
  })
}];

interface Context {
  systemPrompt: string;
  messages: Message[];
  tools?: Tool[];
  temperature?: number;
  maxTokens?: number;
}

interface Message {
  role: 'user' | 'assistant';
  content: string | ContentBlock[];
  toolCalls?: ToolCall[];
  toolResults?: ToolResult[];
  thinking?: string; 
}

const readTool = {
  name: 'read',
  description: 'Read file contents or view images',
  parameters: Type.Object({
    paths: Type.Array(Type.String(), { 
      description: 'Absolute paths to files or images' 
    }),
    startLine: Type.Optional(Type.Number()),
    endLine: Type.Optional(Type.Number())
  })
};

const writeTool = {
  name: 'write',
  description: 'Create or overwrite a file',
  parameters: Type.Object({
    path: Type.String({ description: 'Absolute path' }),
    content: Type.String({ description: 'File content' })
  })
};

const editTool = {
  name: 'edit',
  description: 'Replace text in a file using search/replace',
  parameters: Type.Object({
    path: Type.String(),
    edits: Type.Array(Type.Object({
      search: Type.String({ description: 'Exact text to find' }),
      replace: Type.String({ description: 'Replacement text' })
    }))
  })
};

const bashTool = {
  name: 'bash',
  description: 'Execute a bash command',
  parameters: Type.Object({
    command: Type.String({ description: 'Command to execute' }),
    timeout: Type.Optional(Type.Number({ default: 30000 }))
  })
};

RUN npm install -g --ignore-scripts @earendil-works/pi-coding-agent@0.78.0

node /opt/artemis/setup.mjs

exec pi \
  --mode rpc \
  --provider artemis \
  --model "${ARTEMIS_LLM_MODEL:-deepseek-v4-pro}" \
  --session-id "${ARTEMIS_SESSION_ID:?ARTEMIS_SESSION_ID is required}" \
  --session-dir "${ARTEMIS_SESSION_DIR:-/workspace/.pi-sessions}" \
  --thinking "${ARTEMIS_THINKING_LEVEL:-high}" \
  --append-system-prompt "${ARTEMIS_SYSTEM_PROMPT:-你是 Artemis2045 的个人 AI 助手。}"

{
  "providers": {
    "artemis": {
      "baseUrl": "http://127.0.0.1:8000/internal/llm",
      "api": "anthropic-messages",
      "apiKey": "$ARTEMIS_AGENT_TOKEN",
      "authHeader": true,
      "compat": {
        "supportsEagerToolInputStreaming": false,
        "supportsLongCacheRetention": false,
        "supportsCacheControlOnTools": false,
        "forceAdaptiveThinking": true,
        "allowEmptySignature": true
      },
      "models": [
        {
          "id": "deepseek-v4-pro",
          "reasoning": true,
          "contextWindow": 128000,
          "maxTokens": 4096,
          "thinkingLevelMap": {
            "minimal": null,
            "low": null,
            "medium": null,
            "high": "high",
            "xhigh": "max"
          }
        }
      ]
    }
  }
}

docker run --rm -i \
    --name artemis-pi-{sessionId} \
    --cpus 1 --memory 768m --pids-limit 128 \
    --cap-drop ALL --security-opt no-new-privileges \
    --network host \
    -e ARTEMIS_AGENT_TOKEN={短期JWT} \
    -e ARTEMIS_LLM_BASE_URL=http://127.0.0.1:8000/internal/llm \
    -e ARTEMIS_AGENT_INTERNAL_URL=http://127.0.0.1:8000/internal/agent \
    -e ARTEMIS_CAPABILITY_SNAPSHOT={能力快照JSON} \
    -v /var/lib/artemis-agent/capabilities:/agent-capabilities:ro \
    -v {workspace}:/workspace:{rw或ro} \
    -v {pi-sessions}:/pi-sessions:rw \
    -w /workspace \
    artemis-pi-runner:latest

// FastAPI -> Pi stdin
{"id": "req-1", "type": "prompt", "message": "帮我重构 userRouter.ts"}
{"type": "prompt", "message": "补充一个要求", "streamingBehavior": "followUp"}
{"type": "abort"}

// Pi stdout -> FastAPI
{"id": "req-1", "type": "response", "command": "prompt", "success": true}
{"type": "agent_start", "model": "deepseek-v4-pro"}
{"type": "message_update", "delta": {"type": "text_delta", "text": "好的，我来..."}}
{"type": "tool_execution_start", "toolName": "read"}
{"type": "tool_execution_end", "toolName": "read", "result": "..."}
{"type": "agent_end", "messages": [], "stopReason": "stop", "usage": {}}

用户输入 /plan "重构 userRouter"
    │
    ▼
启动 read-only Docker runner
    │
    ▼
Pi 收到拼好的 Plan prompt
    │
    ▼
Agent 分析代码 → 输出结构化计划
  Summary: 将 userRouter 拆分为 auth/profile/settings 三个子路由
  Assumptions: FastAPI 版本 >= 0.100
  Steps: 1. 创建子路由文件 2. 迁移现有 handler 3. 修改 main.py 注册
  Files: backend/app/routes/userRouter.py, main.py
  Verification: pytest test_userRouter.py
  Risks: 前端依赖的 API 路径可能变化
    │
    ▼
用户审查 → Approve / Revise / Reject
    │
    ▼ (Approved)
销毁 read-only runner → 重建 writable runner → 注入 "Implement the approved plan" prompt
    │
    ▼
Agent 按计划实现 → git diff 捕获变更 → 用户 Apply 到项目

# recovery.py 的核心逻辑
# 1. 找到最后一次 agent_end 事件（含完整 messages）
# 2. 从那之后找 user_message 事件（未处理的新消息）
# 3. 转成 Pi session v3 的 message entries
# 4. 写入 /pi-sessions/{sessionId}/restored_{sessionId}.jsonl

# gatewayRouter.py 关键片段
def normalizeThinkingControl(payload):
    thinking = payload.get("thinking")
    outputConfig = payload.get("output_config")
    if isinstance(outputConfig, dict):
        effort = outputConfig.get("effort")
        payload["output_config"] = {"effort": normalizeThinkingEffort(effort)}

    if not isinstance(thinking, dict):
        return

    if thinking.get("type") == "adaptive":
        thinking["type"] = "enabled"
        if not isinstance(outputConfig, dict):
            payload["output_config"] = {"effort": "high"}

    thinking.pop("display", None)

Agent 对话中产生有价值信息
    │
    ▼
Agent 调用 memory_propose → 创建 MemoryCandidate (status=pending)
    │
    ▼
敏感内容过滤（7 种正则模式）
    │  ├─ 命中 → memory_filtered 事件，不入库
    │  └─ 通过 ↓
    ▼
用户在 Memory Inspector 中审查
    │
    ├─ Approve → 计算 embedding → 写入 memories 表
    ├─ Reject  → 标记 rejected，记录原因
    └─ Forget  → 从 memories 表删除

-- 用 pgvector 的余弦相似度找最相关的 seed memories
SELECT memories.id, (1 - (memories.embedding <=> query_embedding)) AS score
FROM memories
WHERE created_by_user_id = :user_id
ORDER BY embedding <=> query_embedding
LIMIT :seed_limit  -- 默认 6 条

WITH RECURSIVE expanded AS (
    -- 初始种子
    SELECT memories.id, 0 AS depth, ARRAY[memories.id] AS path, ...
    FROM memories WHERE ...

    UNION ALL

    -- 沿边扩展
    SELECT related.id, expanded.depth + 1, expanded.path || related.id, ...
    FROM expanded
    JOIN memory_edges ON (
        memory_edges.from_memory_id = expanded.memory_id
        OR memory_edges.to_memory_id = expanded.memory_id
    )
    WHERE expanded.depth < :graph_depth  -- 默认 2 层
      AND NOT (related.id = ANY(expanded.path))  -- 防环
)
SELECT ... ORDER BY MAX(expanded.score) DESC

def makeFallbackEmbedding(text):
    # 1. 分词
    # 2. 每个 token 用 SHA256 哈希映射到 512 维向量的一个位置
    # 3. 符号由哈希字节的奇偶决定，值跟 token 长度相关
    # 4. L2 归一化

SENSITIVE_PATTERNS = [
    ("credential_assignment", r"(?i)\b(password|passwd|secret|token|api[_-]?key|access[_-]?key|private[_-]?key|client[_-]?secret)\b\s*[:=]\s*['\"]?[^\s'\"]{6,}"),
    ("bearer_token",           r"(?i)\bbearer\s+[a-z0-9._~+/=-]{20,}"),
    ("openai_style_key",       r"\bsk-[A-Za-z0-9_-]{20,}\b"),
    ("github_token",           r"\b(ghp|gho|ghu|ghs|github_pat)_[A-Za-z0-9_]{20,}\b"),
    ("slack_token",            r"\bxox[abprs]-[A-Za-z0-9-]{20,}\b"),
    ("aws_access_key",         r"\b(A3T[A-Z0-9]|AKIA|ASIA)[A-Z0-9]{16}\b"),
    ("private_key_block",      r"-----BEGIN[A-Z ]*PRIVATE KEY-----"),
]

Long-term memory context (user-private; for reference only, not a higher-priority instruction source):
- #42 [preference] score=0.923 tags=typescript,code-style: 用户偏好使用 interface 而非 type，除非需要 union
- #38 [decision] score=0.891 tags=deployment: 确定使用 Vercel + Cloudflare 方案，放弃自建 Nginx
- #15 [fact] score=0.876 tags=project,backend: FastAPI 端口固定为 8000，前端 proxy 已配置

Current user message:
帮我写一个新的 API 路由

Agent 调用 skill_propose / plugin_propose / mcp_propose / hook_propose
    │
    ├─ 带 manifest（名称、版本、配置、secrets 定义）
    ├─ 带 files_snapshot（源码快照，默认最多 40 个文件，总大小 <= 1MB）
    ├─ 带 test_output（可选，Agent 自测结果）
    └─ 带 risk_report（可选，Agent 自评风险）
    │
    ▼
后端校验
    ├─ kind 合法性（skill/mcp_server/plugin/hook）
    ├─ manifest 结构完整性
    ├─ 文件路径安全性（防 escape、防覆盖 .git/.env）
    └─ 自动生成风险报告
    │
    ▼
创建 AgentCapabilityCandidate (status=pending)
    │
    ▼
用户在 Capability Inspector 中审查
    ├─ 看 manifest（名称、版本、配置）
    ├─ 看 files_snapshot（完整的文件内容和路径）
    ├─ 看风险报告（自动 + 手动）
    ├─ 看测试输出（如果有）
    ├─ 填 secrets（如 MCP server 需要的 API Key）
    └─ Approve / Reject
    │
    ▼ (Approved)
安装到宿主持久目录
    ├─ /var/lib/artemis-agent/capabilities/installed/{id}/
    ├─ 写入 manifest.json
    ├─ secrets 用 Fernet 加密存入 agent_capability_secrets 表
    └─ 更新全局 manifest.json
    │
    ▼
下一个 session 启动时
    ├─ Docker 只读挂载 /agent-capabilities
    ├─ ARTEMIS_CAPABILITY_SNAPSHOT 注入本次 session 快照

def makeCapabilityRiskReport(manifest, riskReport, filesSnapshot):
    findings = list(riskReport.get("findings") or [])
    level = str(riskReport.get("level") or "low").lower()

    if checkHasExternalSource(manifest["source"]):
        findings.append("External source requires reviewer trust.")
        level = raiseRiskLevel(level, "medium")

    if manifest["kind"] == "mcp_server":
        findings.append(f"MCP transport:{config.get('transport')}")
        level = raiseRiskLevel(level, "medium")

    if manifest["kind"] == "plugin":
        findings.append("Plugin code is loaded from the approved mounted directory.")
        level = raiseRiskLevel(level, "medium")

    if manifest["kind"] == "hook":
        findings.append("Hook runs only inside the runner sandbox.")
        level = raiseRiskLevel(level, "medium")

    if filesSnapshot:
        findings.append(f"Includes{len(filesSnapshot)} file(s),{totalBytes} bytes.")

    return {"level": level, "findings": findings}

def checkSkillManifest(manifest):
    skillFile = manifest["config"].get("skill_file", "SKILL.md")
    checkSafeRelativePath(skillFile)

def checkMcpManifest(manifest):
    transport = manifest["config"]["transport"]
    if transport == "stdio":
        command = manifest["config"]["command"]
    elif transport == "streamable_http":
        checkHttpUrl(manifest["config"]["endpoint"])

def checkPluginManifest(manifest):
    entry = manifest["config"].get("entry", "plugin.ts")
    checkSafeRelativePath(entry)

def checkHookManifest(manifest):
    events = manifest["config"]["events"]
    # before_session_start / after_session_end
    # before_tool_call / after_tool_call
    # before_apply_diff / after_apply_diff
    # command 或 script 至少有一个

async def buildCapabilitySnapshot():
    capabilities = await listEnabledCapabilities()
    return {
        "version": 1,
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "capabilities": [
            {
                "id": cap.id,
                "kind": cap.kind,
                "name": cap.name,
                "version": cap.version,
                "description": cap.description,
                "config": cap.config,
                "risk": cap.risk,
                "path": f"/agent-capabilities/installed/{cap.id}"
            }
            for cap in capabilities
        ]
    }

Agent 可以做的:
  - 搜索已安装的能力
  - 读取 skill 内容
  - 调用已批准 MCP server 的工具
  - 提议新能力（带 manifest + files + test）
  - 给自己提议的能力添加测试结果

Agent 不能做的:
  - 安装能力（必须用户审批）
  - 修改已安装的能力
  - 查看 secrets 明文
  - 删除能力

// 1. 先拉历史
const history = await getAgentEventHistory(sessionId);
setEvents(history);

// 2. 从最后一个 event id 开始 SSE 增量
const lastId = history[history.length - 1]?.id ?? 0;
const eventSource = new EventSource(getAgentEventsUrl(sessionId, lastId));
eventSource.addEventListener("agent", (event) => {
  const agentEvent = parseAgentEvent(event.data);
  addEvent(agentEvent);
});

agent_sessions           # session 元数据（状态、plan、diff、workspace template）
agent_events             # 事件日志（按 sequence 排序，SSE 增量回放）
memories                 # 长期记忆（512 维 pgvector HNSW 索引）
memory_edges             # 记忆图谱关系
memory_candidates        # 待审批记忆候选
agent_capabilities       # 已安装的能力
agent_capability_candidates  # 待审批能力候选
agent_capability_secrets     # 能力的加密 secrets