From 420f67a261f10364b52d8a42f65cb5e4f5cdf0a9 Mon Sep 17 00:00:00 2001 From: createpjf <113523690+createpjf@users.noreply.github.com> Date: Thu, 26 Feb 2026 22:32:12 +0800 Subject: [PATCH] V0.04: dashboard overhaul, SSE streaming, episode scoring chain MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dashboard: - KB horizontal grid with scrollable cards and tag colors - Session sidebar (create/rename/delete conversations) - Tab badges with live counts on Episodes/Cases/KB - Episode score + timestamp columns - Daily log content filtering (hide tool artifacts) SSE Streaming: - Per-task stream files (lockless append, cursor-based read) - GET /v1/stream/:task_id endpoint - Auto-cleanup on task complete/fail/cancel Episode Scoring: - Baseline scores at creation (success=8, partial=5, other=2) - Alic critique backfill via EpisodicMemory.update_episode_score() - Orchestrator wires critique to evaluated agent's episode New modules: - adapters/memo/ — Memo Protocol integration - adapters/memory/consolidator.py — 3-phase memory consolidation - reputation/textgrad.py — TextGrad critique → skill patches - cli/memo_cmd.py — Memo CLI - docs/Cleo_V0.01_*.md — Product and technical documentation Other: generate_doc 8 formats, empty LLM guards, session history 200, dashboard session API, KB insight dedup, cron timeout improvements Co-Authored-By: Claude Opus 4.6 --- .gitignore | 7 + PLAN.md | 416 ---------- README.md | 55 +- adapters/channels/manager.py | 17 +- adapters/channels/session.py | 116 ++- adapters/llm/minimax.py | 17 +- adapters/memo/__init__.py | 16 + adapters/memo/client.py | 142 ++++ adapters/memo/config.py | 150 ++++ adapters/memo/deidentifier.py | 201 +++++ adapters/memo/exporter.py | 346 ++++++++ adapters/memo/hooks.py | 159 ++++ adapters/memo/importer.py | 195 +++++ adapters/memo/quality_scorer.py | 180 +++++ adapters/memo/tracking.py | 94 +++ adapters/memo/transformer.py | 398 +++++++++ adapters/memory/consolidator.py | 337 ++++++++ adapters/memory/episodic.py | 40 +- adapters/memory/extractor.py | 11 + adapters/memory/knowledge_base.py | 71 +- cli/memo_cmd.py | 334 ++++++++ core/agent.py | 36 +- core/cron.py | 14 +- core/dashboard.html | 786 ++++++++++++++++-- core/gateway.py | 265 +++++- core/orchestrator.py | 138 +++- core/task_board.py | 55 +- core/tools.py | 699 ++++++++++++++-- docs/Cleo_V0.01_Product_Logic.md | 570 +++++++++++++ docs/Cleo_V0.01_Product_Narrative.md | 258 ++++++ docs/Cleo_V0.01_Technical_Architecture.md | 943 ++++++++++++++++++++++ pyproject.toml | 22 +- reputation/textgrad.py | 271 +++++++ skills/pdf.md | 94 ++- 34 files changed, 6783 insertions(+), 670 deletions(-) delete mode 100644 PLAN.md create mode 100644 adapters/memo/__init__.py create mode 100644 adapters/memo/client.py create mode 100644 adapters/memo/config.py create mode 100644 adapters/memo/deidentifier.py create mode 100644 adapters/memo/exporter.py create mode 100644 adapters/memo/hooks.py create mode 100644 adapters/memo/importer.py create mode 100644 adapters/memo/quality_scorer.py create mode 100644 adapters/memo/tracking.py create mode 100644 adapters/memo/transformer.py create mode 100644 adapters/memory/consolidator.py create mode 100644 cli/memo_cmd.py create mode 100644 docs/Cleo_V0.01_Product_Logic.md create mode 100644 docs/Cleo_V0.01_Product_Narrative.md create mode 100644 docs/Cleo_V0.01_Technical_Architecture.md create mode 100644 reputation/textgrad.py diff --git a/.gitignore b/.gitignore index 2a624d3..d0dfecd 100644 --- a/.gitignore +++ b/.gitignore @@ -99,3 +99,10 @@ config/agents.yaml.lock cleo-completion.sh adapters/voice/sherpa-onnx/ =3.1.0 + +# ── Runtime data (consolidation/streaming) ── +memory/consolidation_log.jsonl +memory/critique_log.jsonl +memo_export/ +.task_streams/ +.task_signals/ diff --git a/PLAN.md b/PLAN.md deleted file mode 100644 index 8dbbfcf..0000000 --- a/PLAN.md +++ /dev/null @@ -1,416 +0,0 @@ -# Cleo 架構重構計劃：從串行 Review 到智能協作 - -## 目標 - -將 Planner → Executor → Reviewer 的固定串行 pipeline，改為： -- Planner 前後都管（拆解 + 收口合成） -- Reviewer 從必經的打分機器 → 按需的專家顧問（智庫角色） -- 打回不是全部重做，而是定向修正（critique → fix 循環） -- 加入複雜度判斷 — 簡單任務跳過 review - -## 架構變化總覽 - -``` -舊: Planner(拆解) → Executor(執行) → Reviewer(打分 pass/fail) - -新: Planner(拆解+派工) - → Executor(執行) - → [簡單任務] Planner 收口合成 → 完成 - → [複雜任務] Reviewer 給 critique + 修復建議 - → Executor 定向修正（不是全部重做） - → Planner 收口合成 → 完成 -``` - ---- - -## Phase 1: TaskBoard 新增狀態 + 定向修正流程 - -**文件**: `core/task_board.py` - -### 1.1 新增 TaskStatus.CRITIQUE (line ~90) - -```python -CRITIQUE = "critique" # 智庫給了修復建議，等 Executor 定向修正 -``` - -### 1.2 Task dataclass 新增欄位 (line ~96) - -```python -complexity: str = "normal" # "simple" | "normal" | "complex" -critique: dict | None = None # {reviewer, passed, suggestions, comment, ts} -critique_round: int = 0 # 當前修正輪次 (max=1) -``` - -### 1.3 新增 `add_critique()` 方法 - -替代原來純分數的 `add_review()`，智庫反饋變為結構化 critique： - -```python -def add_critique(self, task_id, reviewer_id, passed, suggestions, comment): - """智庫提交 critique：通過 or 帶修復建議的打回""" - t["critique"] = { - "reviewer": reviewer_id, - "passed": passed, - "suggestions": suggestions or [], - "comment": comment, - "ts": time.time(), - } - if passed: - t["status"] = TaskStatus.COMPLETED.value - t["completed_at"] = time.time() - else: - t["status"] = TaskStatus.CRITIQUE.value - t["critique_round"] = t.get("critique_round", 0) + 1 -``` - -### 1.4 新增 `claim_critique()` — Executor 認領修正任務 - -```python -def claim_critique(self, agent_id, agent_role=None): - """Executor 認領 status=CRITIQUE 的任務做定向修正""" - # 找 status=CRITIQUE 且 agent_id 匹配原執行者的任務 - # 設 status=CLAIMED，保留 critique 和原 result -``` - -### 1.5 修改 `complete()` (line 254) - -```python -def complete(self, task_id): - """簡化：直接標記完成，不再檢查 review score""" - # 移除 avg_review_score < 60 打回邏輯 - # 直接設 status=COMPLETED, completed_at=now -``` - -### 1.6 保留 `submit_for_review()` (line 226) — 不改，仍用於送 critique - -### 1.7 修改 `recover_stale_tasks()` (line 374) - -```python -# 新增 CRITIQUE 超時回收: -# stale CRITIQUE (> 5 min): 強制完成（使用原 result） -``` - ---- - -## Phase 2: Orchestrator 流程重構 - -**文件**: `core/orchestrator.py` - -### 2.1 Planner 拆解時標記複雜度 (修改 _extract_and_create_subtasks, line 179) - -```python -def _extract_and_create_subtasks(board, planner_output, parent_id): - # 現有: 解析 TASK: 行 - # 新增: 解析 COMPLEXITY: simple|normal|complex (從 planner output) - # 預設規則: - # 含 "review"/"audit"/"verify"/"analyze" → complex - # 含 "fix"/"update"/"change" → normal - # 含 "list"/"show"/"get" → simple -``` - -### 2.2 Planner 不再 auto-complete (修改 line 380-387) - -```python -# 舊: planner auto-completes 自己 -# 新: planner 完成拆解後進入 "waiting" 狀態 -# 記錄 parent_task_id → subtask_ids 的映射 -# 等所有 subtasks completed → 觸發收口 -``` - -### 2.3 新增 Planner 收口函數 - -```python -async def _planner_close_out(agent, board, parent_task_id, config): - """Planner 收口：合成所有子任務結果為最終輸出""" - results = board.collect_results(parent_task_id) - prompt = f"你之前拆解了任務。以下是各子任務的執行結果：\n\n{results}\n\n" - f"請合成為一個完整、連貫的最終答案，直接面向用戶。" - messages = [{"role": "system", "content": agent.cfg.role}, - {"role": "user", "content": prompt}] - final = await agent.llm.chat(messages, agent.cfg.model) - board.complete(parent_task_id) - # 更新 parent task 的 result 為合成結果 -``` - -### 2.4 Executor 完成後的路由邏輯 (重寫 line 389-414) - -```python -# 舊: 一律 submit_for_review → 發 mailbox 給 reviewer -# 新: -is_simple = task.complexity == "simple" -if is_simple: - board.complete(task.task_id) # 跳過 review - logger.info("simple task %s auto-completed", task.task_id) -else: - board.submit_for_review(task.task_id, result) - # 發 critique_request (不是 review_request) - for r_id in reviewers: - if r_id != agent.cfg.agent_id: - agent.send_mail(r_id, - _json_critique_request(task, result), - msg_type="critique_request") -``` - -### 2.5 重寫 review handler → critique handler (替換 line 118-175) - -```python -async def _handle_critique_request(agent, board, mail, sched): - """智庫模式：不打分，給結構化 critique""" - payload = json.loads(mail["content"]) - task_id, description, result = payload["task_id"], payload["description"], payload["result"] - - prompt = ( - f"Review the following task output.\n\n" - f"## Task\n{description}\n\n" - f"## Output\n{result}\n\n" - f"Decide: is this ready to deliver?\n" - f'If YES: {{"passed": true, "comment": "brief praise"}}\n' - f'If NO: {{"passed": false, "suggestions": ["fix1", "fix2"], "comment": "why"}}\n' - f"Max 3 suggestions, each must be specific and actionable." - ) - raw = await agent.llm.chat([ - {"role": "system", "content": agent.cfg.role}, - {"role": "user", "content": prompt} - ], agent.cfg.model) - - critique = json.loads(raw) - passed = critique.get("passed", True) - suggestions = critique.get("suggestions", []) - comment = critique.get("comment", "") - - board.add_critique(task_id, agent.cfg.agent_id, passed, suggestions, comment) - await sched.on_critique(agent.cfg.agent_id, passed) - - if not passed: - logger.info("critique REJECTED task %s with %d suggestions", task_id, len(suggestions)) - else: - logger.info("critique APPROVED task %s", task_id) -``` - -### 2.6 Executor 處理 CRITIQUE 修正 (在 _agent_loop claim 邏輯中新增) - -```python -# 在主循環 claim_next 之前，先檢查 CRITIQUE 任務 -critique_task = board.claim_critique(agent_id) -if critique_task: - suggestions = critique_task.critique.get("suggestions", []) - fix_prompt = ( - f"你之前提交了以下結果:\n{critique_task.result}\n\n" - f"智庫給了修正建議:\n" - + "\n".join(f"- {s}" for s in suggestions) + - f"\n\n請針對以上建議修正輸出，只修改需要改的部分。" - ) - result = await agent.run_with_prompt(fix_prompt, bus) - - # 修正後: 如果已經是第 1 輪 critique → 直接完成（不再送 review） - if critique_task.critique_round >= 1: - board.complete(critique_task.task_id) # 強制完成 - else: - board.submit_for_review(critique_task.task_id, result) # 可再送一次 -``` - -### 2.7 Planner 監控子任務 + 觸發收口 (在 Planner 的 _agent_loop 中) - -```python -# Planner 每次循環額外檢查: -# 1. 找到自己創建的 parent tasks -# 2. 如果所有 subtasks 都 completed → 呼叫 _planner_close_out() -# 這讓 Planner 持續「值班」直到所有工作完成 -``` - -### 2.8 mailbox 消息類型更新 - -```python -# 舊: msg_type="review_request" → _handle_review_request() -# 新: msg_type="critique_request" → _handle_critique_request() -# 保留舊類型作為 fallback 以防相容性問題 -``` - ---- - -## Phase 3: Agent Config + Skills 更新 - -**文件**: `config/agents.yaml`, `skills/` - -### 3.1 Reviewer 角色 prompt 更新 (agents.yaml line 101-105) - -```yaml -- id: reviewer - role: > - Quality advisor. Review task outputs and provide structured feedback. - If output is ready to ship: {"passed": true, "comment": "..."} - If needs revision: {"passed": false, - "suggestions": ["specific fix 1", "specific fix 2"], - "comment": "..."} - Be specific with actionable fix recommendations. Max 3 suggestions. -``` - -### 3.2 Planner 角色 prompt 新增收口職責 (agents.yaml line 54-58) - -```yaml -- id: planner - role: > - Strategic planner. Decompose user requests into subtasks. - Write TASK: per line for each subtask. Do not implement yourself. - For each task, add COMPLEXITY: simple|normal|complex. - After all subtasks complete, synthesize a final unified answer. -``` - -### 3.3 更新 skills/review.md - -```markdown -## Quality Advisor Guidelines -- Decision: PASS or NEEDS REVISION (不用數字分數) -- If PASS: briefly explain what was done well -- If NEEDS REVISION: - - List specific, actionable suggestions (max 3) - - Each suggestion = a concrete fix, not vague criticism - - Prioritize by importance -- Always respond JSON: - - {"passed": true, "comment": "..."} - - {"passed": false, "suggestions": ["...", "..."], "comment": "..."} -``` - -### 3.4 skills/planning.md 追加收口指令 - -```markdown -## Closing Out Tasks -When all subtasks are completed, synthesize a final answer: -- Combine outputs, resolve contradictions -- Present as one unified user-facing response -- Remove internal task references -``` - -### 3.5 skills/coding.md 追加修正指令 - -```markdown -## Handling Review Feedback -When you receive critique suggestions: -- Address EACH suggestion specifically -- Only modify parts that need fixing (don't rewrite everything) -- Explain what you changed -``` - ---- - -## Phase 4: 聲譽系統適配 - -**文件**: `reputation/scheduler.py`, `reputation/peer_review.py` - -### 4.1 scheduler.py — 新增 on_critique() (替代 on_review) - -```python -async def on_critique(self, reviewer_id, passed): - """智庫提交了 critique""" - # 更新 reviewer 的 review_accuracy: - # 合理的 critique (有具體 suggestions) → 85 - # 總是 pass → 60 (可能太寬鬆) - # 總是 reject → 65 (可能太嚴格) - -async def on_critique_result(self, agent_id, passed_first_time, had_revision): - """Executor 的任務被 critique 後的結果""" - # passed first time → output_quality = 90 - # passed after revision → output_quality = 70 - # forced complete after max rounds → output_quality = 50 -``` - -### 4.2 peer_review.py — 簡化 anti-cheating - -```python -# 移除: mutual_inflation (單 reviewer 無意義) -# 移除: consensus_deviation (單 reviewer 無意義) -# 保留: extreme_bias → 改為 always_pass_bias (>80% pass rate → 警告) -# 新增: suggestion_quality → 如果 suggestions 總是空/重複 → 降權 -``` - ---- - -## Phase 5: 前端 Dashboard 更新 - -**文件**: `core/dashboard.html` - -### 5.1 Header Bar Agent Chips - -``` -Reviewer chip: icon 🔍 → 🧠, 名稱 "Reviewer" → "Advisor" -狀態: "reviewing" → "advising" -``` - -### 5.2 updateWorkflow() 適配新狀態 (~line 1778) - -```javascript -// 新增: task.status === 'critique' → executor chip 高亮 "fixing" -// 修改: task.status === 'review' → advisor chip active (不是 reviewer) -``` - -### 5.3 diffAndRoute() 新增 critique 狀態 dispatch 消息 - -```javascript -// 新增處理: -// review → "🧠 Advisor reviewing..." -// critique → "📝 Revision needed: 2 suggestions" (帶 suggestions 預覽) -// critique 修正完成 → "✓ Revised and resubmitted" -// planner 收口 → "📋 Planner synthesizing final answer..." -// simple 任務跳過 → "⚡ Simple task auto-completed" -``` - -### 5.4 renderChatMsgHtml() — 新增 critique 展示 (~line 1201) - -```javascript -// assistant bubble 中: -// 舊: score/100 badge -// 新: ✓ Approved / ⚠ Needs revision badge -// suggestions 列表顯示 (如果有) -// "1st attempt" / "Revised" 標記 -``` - -### 5.5 Welcome 文案 (~line 567-571) - -``` -舊: "planned, executed, and reviewed" -新: "planned, executed, and quality-checked" -``` - -### 5.6 Chat live status 適配 - -```javascript -// reviewer working → "🧠 Advisor analyzing..." -// executor 在 critique 後 → "⚙️ Executor fixing..." -``` - ---- - -## Phase 6: 測試更新 - -**文件**: `tests/test_task_board.py`, `tests/test_p2_p3.py` - -### 6.1 test_task_board.py 新增 - -- `test_critique_flow`: submit → critique(not passed) → claim_critique → fix → complete -- `test_simple_task_skip_review`: simple complexity → 直接完成 -- `test_critique_max_rounds`: 超過 1 次 critique → 強制完成 -- `test_critique_passed`: critique passed → 直接 completed -- `test_recover_stale_critique`: CRITIQUE 超時 → 強制完成 - -### 6.2 修改現有測試 - -- `test_submit_review_complete`: 適配 critique 結構 -- peer review anti-cheating tests: 適配新邏輯 - ---- - -## 實施順序 - -| 階段 | 內容 | 依賴 | 預估改動 | -|------|------|------|---------| -| Phase 1 | TaskBoard 新狀態+方法 | 無 | ~80 行 | -| Phase 3 | Config/Skills prompts | 無 | ~40 行 | -| Phase 2 | Orchestrator 核心重構 | Phase 1 | ~150 行 | -| Phase 4 | Reputation 適配 | Phase 2 | ~50 行 | -| Phase 5 | Dashboard 前端 | Phase 2 | ~80 行 | -| Phase 6 | Tests | Phase 1-4 | ~120 行 | - -**總計**: ~520 行改動（新增+修改） - -Phase 1+3 可以先做，不破壞現有流程（新狀態和 prompts 是增量的）。 -Phase 2 是核心斷裂點，需要和 Phase 4-6 一起完成。 diff --git a/README.md b/README.md index 09820c7..2a50ca8 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # ⬡ Cleo -![version](https://img.shields.io/badge/version-0.03-blue) +![version](https://img.shields.io/badge/version-0.04-blue) ![python](https://img.shields.io/badge/python-3.11%2B-green) ![license](https://img.shields.io/badge/license-MIT-grey) @@ -97,6 +97,33 @@ Role-based routing via `_ROLE_TO_AGENTS` mapping. Timeout recovery: claimed > 18 | **Episodic Memory** | `adapters/memory/episodic.py` | 3-layer progressive: L0 atomic (~100 tok) → L1 overview (~500 tok) → L2 full detail | | **Knowledge Base** | `adapters/memory/knowledge_base.py` | Shared Zettelkasten-style notes + insights | | **Context Bus** | `core/context_bus.py` | 4-layer KV store (TASK/SESSION/SHORT/LONG) with TTL | +| **Memory Consolidation** | `adapters/memory/consolidator.py` | 3-phase pipeline: cluster old episodes (>3d) → compress → promote to KB | + +### Episode Scoring + +Two-stage quality scoring for every task: + +1. **Baseline** — Agent self-assigns at episode creation (`success=8, partial=5, other=2`) +2. **Critique backfill** — Alic's score retroactively written to evaluated agent's episode via `update_episode_score()` + +### SSE Streaming (`core/task_board.py` + `core/gateway.py`) + +Real-time token streaming for chat responses: + +- Per-task `.stream` files with lockless append + cursor-based reads +- `GET /v1/stream/:task_id` — Server-Sent Events endpoint +- Auto-cleanup on task complete/fail/cancel + +### Dashboard (`core/dashboard.html`) + +Web UI at `http://127.0.0.1:19789`: + +- **Session sidebar** — multi-session conversation management +- **KB grid** — horizontal 2-column layout with scrollable cards +- **Tab badges** — live count indicators on Episodes/Cases/KB tabs +- **Episode table** — score + timestamp columns +- **Daily log** — auto-filters tool call artifacts +- **SSE streaming** — real-time token-by-token chat display ### Tools (37 tools × 10 groups) @@ -104,6 +131,8 @@ Role-based routing via `_ROLE_TO_AGENTS` mapping. Timeout recovery: claimed > 18 Access control: profiles (`minimal` / `coding` / `full`) + per-agent allow/deny lists. Audit log at `.logs/tool_audit.log`. +`generate_doc` supports 8 output formats: PDF, DOCX, XLSX, PPTX, CSV, TXT, MD, HTML. + ### Channels | Channel | Auth | Config | @@ -140,6 +169,9 @@ Gateway on port **19789** (+ WebSocket on **19790**). Auth: `Authorization: Bear | GET | `/v1/memory/*` | Memory status / episodes / cases | | GET | `/v1/chain/*` | Blockchain status / balance | | POST | `/v1/cron` | Create scheduled job | +| GET | `/v1/stream/:id` | SSE token stream | +| GET/POST | `/v1/sessions` | Dashboard sessions | +| GET/PUT/DELETE | `/v1/sessions/:id` | Session CRUD | | GET | `/health` | Gateway health | 30+ endpoints total — see [ARCHITECTURE.md](docs/ARCHITECTURE.md) for the full list. @@ -200,12 +232,20 @@ cleo-dev/ │ └── doctor.py # Health check + auto-repair ├── adapters/ │ ├── llm/minimax.py # MiniMax SSE streaming + truncation recovery -│ ├── memory/ # hybrid (BM25+ChromaDB), episodic, embedding -│ └── channels/ # manager, telegram, discord, feishu, slack -├── reputation/scorer.py # 5-dim EMA scoring +│ ├── memory/ # hybrid, episodic, embedding, consolidator +│ ├── channels/ # manager, telegram, discord, feishu, slack +│ └── memo/ # Memo Protocol integration +├── reputation/ +│ ├── scorer.py # 5-dim EMA scoring +│ └── textgrad.py # TextGrad critique → skill patches ├── skills/ # 56+ hot-reload markdown skills ├── tests/ # 399 tests -└── docs/ARCHITECTURE.md # Full technical architecture +├── cli/memo_cmd.py # Memo CLI +└── docs/ # Architecture + product docs + ├── ARCHITECTURE.md + ├── Cleo_V0.01_Product_Logic.md + ├── Cleo_V0.01_Product_Narrative.md + └── Cleo_V0.01_Technical_Architecture.md ``` --- @@ -216,13 +256,16 @@ cleo-dev/ - One LLM API key (MiniMax, OpenAI, or local Ollama) **Core:** `pyyaml` `filelock` `requests` `chromadb` `websockets` -**Optional:** `python-telegram-bot` · `discord.py` · `web3` · `rich` +**Optional:** `python-telegram-bot` · `discord.py` · `web3` · `rich` · `python-pptx` · `slack-sdk` --- ## Docs - **[ARCHITECTURE.md](docs/ARCHITECTURE.md)** — Full technical architecture with code details +- **[Product Logic](docs/Cleo_V0.01_Product_Logic.md)** — Product design and decision logic +- **[Product Narrative](docs/Cleo_V0.01_Product_Narrative.md)** — Product vision and narrative +- **[Technical Architecture V0.01](docs/Cleo_V0.01_Technical_Architecture.md)** — Original technical architecture --- diff --git a/adapters/channels/manager.py b/adapters/channels/manager.py index b55f99f..2de7a38 100644 --- a/adapters/channels/manager.py +++ b/adapters/channels/manager.py @@ -908,14 +908,27 @@ async def _wait_for_result(self, task_id: str, board = TaskBoard() data = board._read() - # Check if all tasks are done (including subtasks) + # Check if all tasks in THIS tree are done if not data: continue + # Build task tree: root + all descendant subtasks (BFS via parent_id) + # Same logic as TaskBoard.collect_results() — scopes to this task only + tree_ids = {task_id} + changed = True + while changed: + changed = False + for tid, t in data.items(): + if tid not in tree_ids and t.get("parent_id") in tree_ids: + tree_ids.add(tid) + changed = True + active_states = {"pending", "claimed", "review", "critique", "blocked", "paused", "synthesizing"} has_active = any( - t.get("status") in active_states for t in data.values()) + data[tid].get("status") in active_states + for tid in tree_ids + if tid in data) if not has_active: # All done — prefer root task result (Leo's synthesis) diff --git a/adapters/channels/session.py b/adapters/channels/session.py index 891278b..94daeae 100644 --- a/adapters/channels/session.py +++ b/adapters/channels/session.py @@ -8,6 +8,9 @@ Tracks per-user/group sessions across channel interactions. Stores conversation history per session in separate JSONL files. Uses the same FileLock pattern as ContextBus and TaskBoard. + +V0.03+: Also serves as the persistence layer for Dashboard sessions +(multi-session support — ChatGPT-style conversation list). """ from __future__ import annotations @@ -16,6 +19,7 @@ import logging import os import time +import uuid from dataclasses import dataclass, field, asdict from typing import Optional @@ -23,10 +27,14 @@ logger = logging.getLogger(__name__) -SESSIONS_FILE = "memory/channel_sessions.json" -SESSIONS_LOCK = "memory/channel_sessions.lock" -HISTORY_DIR = "memory/sessions" -MAX_HISTORY_MESSAGES = 50 # FIFO limit per session +# ── Absolute paths (Gateway CWD may differ from project root) ── +_PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname( + os.path.abspath(__file__)))) # adapters/channels/ → adapters/ → project root +SESSIONS_FILE = os.path.join(_PROJECT_ROOT, "memory", "channel_sessions.json") +SESSIONS_LOCK = os.path.join(_PROJECT_ROOT, "memory", "channel_sessions.lock") +HISTORY_DIR = os.path.join(_PROJECT_ROOT, "memory", "sessions") + +MAX_HISTORY_MESSAGES = 200 # FIFO limit per session (increased for dashboard) CHARS_PER_TOKEN = 3 # conservative (English ~4, CJK ~1.5) SESSION_EXPIRE_HOURS = 24 # start fresh if idle longer than this GROUP_USER_ISOLATION = True # isolate per-user contexts in group chats @@ -36,14 +44,18 @@ class ChannelSession: """Represents a channel conversation session.""" session_id: str # "{channel}:{chat_id}" - channel: str # "telegram" | "discord" | "feishu" - chat_id: str # platform chat/group ID + channel: str # "telegram" | "discord" | "feishu" | "dashboard" + chat_id: str # platform chat/group ID (UUID for dashboard) user_ids: list[str] = field(default_factory=list) user_names: list[str] = field(default_factory=list) message_count: int = 0 last_task_id: str = "" last_active: float = 0.0 created_at: float = field(default_factory=time.time) + # V0.03+: Dashboard session fields + title: str = "" # user-visible session name + pinned: bool = False # pinned to top of list + no_expire: bool = False # skip auto-expiry (dashboard sessions) class SessionStore: @@ -147,6 +159,7 @@ def get_history(self, session_id: str, Returns list of {role, content, ts, user?} dicts, oldest first. Returns empty list if session is expired (idle > SESSION_EXPIRE_HOURS). + Dashboard sessions (no_expire=True) skip the expiry check. """ history_path = self._history_path(session_id) if not os.path.exists(history_path): @@ -169,12 +182,19 @@ def get_history(self, session_id: str, return [] # Check session expiry — if idle too long, start fresh - last_ts = messages[-1].get("ts", 0) - idle_hours = (time.time() - last_ts) / 3600 - if idle_hours > SESSION_EXPIRE_HOURS: - logger.info("Session %s expired (idle %.1fh), starting fresh", - session_id, idle_hours) - return [] + # (skip for dashboard sessions with no_expire=True) + skip_expiry = False + data = self._read() + if session_id in data: + skip_expiry = data[session_id].get("no_expire", False) + + if not skip_expiry: + last_ts = messages[-1].get("ts", 0) + idle_hours = (time.time() - last_ts) / 3600 + if idle_hours > SESSION_EXPIRE_HOURS: + logger.info("Session %s expired (idle %.1fh), starting fresh", + session_id, idle_hours) + return [] # Return last max_turns*2 messages (user + assistant pairs) limit = max_turns * 2 @@ -235,6 +255,7 @@ def cleanup_expired(self, max_idle_hours: float = SESSION_EXPIRE_HOURS) -> int: """Remove sessions idle longer than max_idle_hours. Also deletes the associated conversation history JSONL files. + Skips sessions with no_expire=True (dashboard sessions). Returns the number of sessions removed. """ cutoff = time.time() - (max_idle_hours * 3600) @@ -244,6 +265,7 @@ def cleanup_expired(self, max_idle_hours: float = SESSION_EXPIRE_HOURS) -> int: expired_keys = [ k for k, v in data.items() if v.get("last_active", 0) < cutoff + and not v.get("no_expire", False) ] for key in expired_keys: # Delete history file @@ -261,6 +283,73 @@ def cleanup_expired(self, max_idle_hours: float = SESSION_EXPIRE_HOURS) -> int: "(idle > %.1fh)", removed, max_idle_hours) return removed + # ── Dashboard Session Methods ───────────────────────────── + + def create_dashboard_session(self, title: str = "") -> ChannelSession: + """Create a new dashboard session with a UUID chat_id. + + Dashboard sessions have no_expire=True and channel='dashboard'. + """ + chat_id = uuid.uuid4().hex[:12] + session_id = f"dashboard:{chat_id}" + now = time.time() + session = ChannelSession( + session_id=session_id, + channel="dashboard", + chat_id=chat_id, + message_count=0, + last_active=now, + created_at=now, + title=title, + no_expire=True, + ) + with self.lock: + data = self._read() + data[session_id] = asdict(session) + self._write(data) + return session + + def list_dashboard_sessions(self) -> list[ChannelSession]: + """Return all dashboard sessions, sorted by last_active descending.""" + data = self._read() + sessions = [] + for s in data.values(): + if s.get("channel") == "dashboard": + sessions.append(self._from_dict(s)) + sessions.sort(key=lambda s: s.last_active, reverse=True) + return sessions + + def rename_session(self, session_id: str, title: str): + """Update a session's title.""" + with self.lock: + data = self._read() + if session_id in data: + data[session_id]["title"] = title + self._write(data) + + def pin_session(self, session_id: str, pinned: bool = True): + """Pin or unpin a session.""" + with self.lock: + data = self._read() + if session_id in data: + data[session_id]["pinned"] = pinned + self._write(data) + + def delete_session(self, session_id: str): + """Delete a session and its history file.""" + with self.lock: + data = self._read() + if session_id in data: + # Delete history file + history_path = self._history_path(session_id) + try: + if os.path.exists(history_path): + os.remove(history_path) + except OSError: + pass + del data[session_id] + self._write(data) + # ── Internal ── def _history_path(self, session_id: str) -> str: @@ -302,4 +391,7 @@ def _from_dict(d: dict) -> ChannelSession: last_task_id=d.get("last_task_id", ""), last_active=d.get("last_active", 0), created_at=d.get("created_at", 0), + title=d.get("title", ""), + pinned=d.get("pinned", False), + no_expire=d.get("no_expire", False), ) diff --git a/adapters/llm/minimax.py b/adapters/llm/minimax.py index dee412f..9501343 100644 --- a/adapters/llm/minimax.py +++ b/adapters/llm/minimax.py @@ -293,7 +293,14 @@ async def chat(self, messages: list[dict], model: str, **kwargs) -> str: content = message.get("content") or "" tc_text = _tool_calls_to_text(tool_calls) return f"{content}\n{tc_text}" if content else tc_text - return message["content"] + content = message.get("content") or "" + if not content.strip(): + logger.warning("[minimax] empty content in response " + "(possible content filter)") + raise RuntimeError( + "Minimax returned empty content " + "(content filter or API issue)") + return content except httpx.HTTPStatusError as e: code = e.response.status_code body = "" @@ -446,7 +453,13 @@ async def chat_with_usage(self, messages: list[dict], tc_text = _tool_calls_to_text(tool_calls) content = f"{content}\n{tc_text}" if content else tc_text else: - content = message["content"] + content = message.get("content") or "" + if not content.strip(): + logger.warning("[minimax] empty content in " + "chat_with_usage (possible content filter)") + raise RuntimeError( + "Minimax returned empty content " + "(content filter or API issue)") usage = data.get("usage", {}) return content, { "prompt_tokens": usage.get("prompt_tokens", 0), diff --git a/adapters/memo/__init__.py b/adapters/memo/__init__.py new file mode 100644 index 0000000..b1176a3 --- /dev/null +++ b/adapters/memo/__init__.py @@ -0,0 +1,16 @@ +""" +adapters/memo/ — Memo Protocol integration for Cleo. + +Packages Cleo agent memories (episodes, cases, patterns, KB notes) +into Memo MemoryObject format for export / upload to the Memo platform. + +Public API: + MemoConfig — configuration from agents.yaml + MemoExporter — batch / selective export pipeline + MemoImporter — Memo Skill pull → Cleo skill directory injection + MemoClient — Memo REST API HTTP client +""" + +from adapters.memo.config import MemoConfig # noqa: F401 + +__all__ = ["MemoConfig"] diff --git a/adapters/memo/client.py b/adapters/memo/client.py new file mode 100644 index 0000000..540337a --- /dev/null +++ b/adapters/memo/client.py @@ -0,0 +1,142 @@ +""" +adapters/memo/client.py — Memo Platform REST API client. + +Wraps the Memo v1 API endpoints: + POST /memories Upload new memory + GET /memories/search Semantic search + GET /memories/{id} Get full content (paid) + POST /skills/sync Bulk pull purchased skills + POST /memories/similarity Provenance similarity check + +Uses ``httpx.AsyncClient`` for async HTTP. +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, Optional + +logger = logging.getLogger(__name__) + +if TYPE_CHECKING: + from adapters.memo.config import MemoConfig + + +class MemoClient: + """HTTP client for the Memo Protocol REST API.""" + + def __init__(self, config: "MemoConfig"): + self.base_url = config.api_base_url.rstrip("/") + self.api_key = config.api_key + self.wallet = config.wallet_address + self.agent_id = config.erc8004_agent_id + + def _headers(self) -> dict: + h = { + "Content-Type": "application/json", + "X-Agent-ID": self.agent_id, + } + if self.api_key: + h["Authorization"] = f"Bearer {self.api_key}" + if self.wallet: + h["X-Wallet-Address"] = self.wallet + return h + + # ── memories ────────────────────────────────────────────────────────── + + async def upload_memory(self, payload: dict) -> dict: + """POST /memories — upload a new MemoryObject. + + Returns API response (id, status, quality_score, etc.). + """ + import httpx + async with httpx.AsyncClient(timeout=30.0) as client: + resp = await client.post( + f"{self.base_url}/v1/memories", + headers=self._headers(), + json=payload, + ) + resp.raise_for_status() + return resp.json() + + async def search_memories( + self, + query: str, + type: str = "", + min_quality: float = 0.6, + domain: str = "", + limit: int = 10, + offset: int = 0, + ) -> list[dict]: + """GET /memories/search — semantic search.""" + import httpx + params: dict = { + "q": query, + "min_quality": min_quality, + "limit": limit, + } + if type: + params["type"] = type + if domain: + params["domain"] = domain + if offset: + params["offset"] = offset + + async with httpx.AsyncClient(timeout=15.0) as client: + resp = await client.get( + f"{self.base_url}/v1/memories/search", + headers=self._headers(), + params=params, + ) + resp.raise_for_status() + data = resp.json() + return data if isinstance(data, list) else data.get("results", []) + + async def get_memory(self, memory_id: str, + subscription_token: str = "") -> dict: + """GET /memories/{id} — get full content (may require payment).""" + import httpx + headers = self._headers() + if subscription_token: + headers["X-Subscription-Token"] = subscription_token + + async with httpx.AsyncClient(timeout=15.0) as client: + resp = await client.get( + f"{self.base_url}/v1/memories/{memory_id}", + headers=headers, + ) + resp.raise_for_status() + return resp.json() + + # ── skills ──────────────────────────────────────────────────────────── + + async def sync_skills(self, memory_ids: list[str]) -> list[dict]: + """POST /skills/sync — bulk pull purchased skills.""" + import httpx + async with httpx.AsyncClient(timeout=30.0) as client: + resp = await client.post( + f"{self.base_url}/v1/skills/sync", + headers=self._headers(), + json={"memory_ids": memory_ids}, + ) + resp.raise_for_status() + data = resp.json() + return data if isinstance(data, list) else data.get("skills", []) + + # ── provenance ──────────────────────────────────────────────────────── + + async def check_similarity(self, content: str) -> dict: + """Check content similarity against existing memories. + + Returns ``{max_similarity, most_similar_id, root_id, generation}``. + If max_similarity > 0.85, the upload MUST declare parent_id. + """ + import httpx + async with httpx.AsyncClient(timeout=15.0) as client: + resp = await client.post( + f"{self.base_url}/v1/memories/similarity", + headers=self._headers(), + json={"content": content[:4000]}, + ) + resp.raise_for_status() + return resp.json() diff --git a/adapters/memo/config.py b/adapters/memo/config.py new file mode 100644 index 0000000..d422c41 --- /dev/null +++ b/adapters/memo/config.py @@ -0,0 +1,150 @@ +""" +adapters/memo/config.py — MemoConfig: Memo Protocol integration configuration. + +Reads from the ``memo:`` section of ``config/agents.yaml``. +Sensitive values (API key, wallet, private key) are resolved from +environment variables at runtime, following the same ``*_env`` pattern +used by ``adapters/memory/embedding.py``. + +When the ``memo:`` section is absent, ``MemoConfig()`` returns a default +instance with ``enabled=False`` — all Memo features are silently disabled. +""" + +from __future__ import annotations + +import os +import logging +from dataclasses import dataclass, field +from typing import Any + +logger = logging.getLogger(__name__) + + +@dataclass +class MemoConfig: + """Memo Protocol integration settings.""" + + # ── master switch ───────────────────────────────────────────────────── + enabled: bool = False + + # ── API connection ──────────────────────────────────────────────────── + api_base_url: str = "https://api.memo.ac" + api_key: str = "" # resolved from env + wallet_address: str = "" # resolved from env + private_key: str = "" # resolved from env + + # ── identity ────────────────────────────────────────────────────────── + erc8004_agent_id: str = "" + display_name: str = "Cleo Agent" + + # ── auto-upload (post-task hook) ────────────────────────────────────── + auto_upload_enabled: bool = False + auto_upload_min_quality: float = 0.6 + auto_upload_types: list[str] = field( + default_factory=lambda: ["procedural", "semantic"]) + + # ── export defaults ─────────────────────────────────────────────────── + default_domain: str = "python" + default_language: str = "zh" + default_access_tier: str = "developer" # free | developer | team + default_price_usdc: float = 0.0 + + # ── deidentification ────────────────────────────────────────────────── + deidentification_use_llm: bool = False + deidentification_llm_model: str = "minimax-m2.5" + + # ── skill sync ──────────────────────────────────────────────────────── + skill_sync_enabled: bool = False + skill_sync_interval_hours: int = 24 + + # ── company names to redact (user-configurable) ─────────────────────── + company_names: list[str] = field(default_factory=list) + + # ────────────────────────────────────────────────────────────────────── + # Factory + # ────────────────────────────────────────────────────────────────────── + + @classmethod + def from_yaml(cls, config: dict[str, Any] | None) -> "MemoConfig": + """Build MemoConfig from the top-level ``config/agents.yaml`` dict. + + If the ``memo:`` key is missing, returns a default (disabled) instance. + """ + if not config: + return cls() + + memo: dict = config.get("memo", {}) + if not memo: + return cls() + + # Resolve sensitive values from environment variables + def _env(key: str) -> str: + env_name = memo.get(key, "") + return os.environ.get(env_name, "") if env_name else "" + + auto = memo.get("auto_upload", {}) + export = memo.get("export", {}) + deident = memo.get("deidentification", {}) + skill = memo.get("skill_sync", {}) + + cfg = cls( + enabled=memo.get("enabled", False), + api_base_url=memo.get("api_base_url", cls.api_base_url), + api_key=_env("api_key_env"), + wallet_address=_env("wallet_address_env"), + private_key=_env("private_key_env"), + erc8004_agent_id=memo.get("erc8004_agent_id", ""), + display_name=memo.get("display_name", cls.display_name), + # auto-upload + auto_upload_enabled=auto.get("enabled", False), + auto_upload_min_quality=auto.get("min_quality", 0.6), + auto_upload_types=auto.get("types", ["procedural", "semantic"]), + # export + default_domain=export.get("default_domain", cls.default_domain), + default_language=export.get("default_language", cls.default_language), + default_access_tier=export.get("default_access_tier", + cls.default_access_tier), + default_price_usdc=export.get("default_price_usdc", 0.0), + # deidentification + deidentification_use_llm=deident.get("use_llm", False), + deidentification_llm_model=deident.get("llm_model", + cls.deidentification_llm_model), + # skill sync + skill_sync_enabled=skill.get("enabled", False), + skill_sync_interval_hours=skill.get("interval_hours", 24), + # company names + company_names=memo.get("company_names", []), + ) + + if cfg.enabled: + missing = [] + if not cfg.api_key: + missing.append("api_key_env") + if not cfg.wallet_address: + missing.append("wallet_address_env") + if missing: + logger.warning("[memo] enabled but missing env vars: %s", + ", ".join(missing)) + return cfg + + # ────────────────────────────────────────────────────────────────────── + # Helpers + # ────────────────────────────────────────────────────────────────────── + + @property + def author_info(self) -> dict: + """Author block for Memo MemoryObject.""" + return { + "erc8004_agent_id": self.erc8004_agent_id, + "wallet_address": self.wallet_address, + "display_name": self.display_name, + } + + @property + def default_access(self) -> dict: + """Default access control block for Memo MemoryObject.""" + return { + "tier": self.default_access_tier, + "price_usdc": self.default_price_usdc, + "subscription_bypass": True, + } diff --git a/adapters/memo/deidentifier.py b/adapters/memo/deidentifier.py new file mode 100644 index 0000000..ed0b5dd --- /dev/null +++ b/adapters/memo/deidentifier.py @@ -0,0 +1,201 @@ +""" +adapters/memo/deidentifier.py — PII removal for Memo export. + +Two layers: + 1. Regex layer (mandatory, zero cost) — emails, IPs, API keys, etc. + 2. LLM-assisted layer (optional) — business context, product names. + +Usage:: + + text_out, stats = deidentify_regex(text) + # or with LLM: + text_out, stats = await deidentify(text, config, llm_adapter) +""" + +from __future__ import annotations + +import re +import logging +from typing import TYPE_CHECKING, Optional + +if TYPE_CHECKING: + from adapters.memo.config import MemoConfig + +logger = logging.getLogger(__name__) + +# ══════════════════════════════════════════════════════════════════════════════ +# Regex patterns +# ══════════════════════════════════════════════════════════════════════════════ + +_PATTERNS: list[tuple[str, re.Pattern, str]] = [ + # ── credentials & secrets ───────────────────────────────────────────── + ("private_key_block", + re.compile( + r"-----BEGIN (?:RSA |EC |DSA |OPENSSH )?PRIVATE KEY-----" + r".*?" + r"-----END (?:RSA |EC |DSA |OPENSSH )?PRIVATE KEY-----", + re.DOTALL), + "[PRIVATE_KEY_REDACTED]"), + + ("env_var_secret", + re.compile( + r"(?:export\s+)?" + r"(?:API_KEY|SECRET|TOKEN|PASSWORD|PRIVATE_KEY|ACCESS_KEY|AUTH)" + r"\s*=\s*[\"']?[^\s\"']+[\"']?", + re.IGNORECASE), + "[ENV_REDACTED]"), + + ("bearer_token", + re.compile(r"Bearer\s+[A-Za-z0-9._~+/=-]{20,}", re.IGNORECASE), + "Bearer [REDACTED]"), + + ("api_key_inline", + re.compile( + r"(?:sk|pk|api|key|token|secret|password|auth)[_-]?" + r"[A-Za-z0-9]{20,}", + re.IGNORECASE), + "[REDACTED]"), + + # ── URLs with credentials ───────────────────────────────────────────── + ("url_with_token", + re.compile( + r"https?://[^\s]*[?&]" + r"(?:token|key|secret|api_key|access_token|auth)" + r"=[^\s&]+"), + "[URL_WITH_CREDENTIALS]"), + + # ── personal identifiers ────────────────────────────────────────────── + ("email", + re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b"), + "[EMAIL]"), + + ("ip_v4", + re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"), + "[IP]"), + + ("ip_v6", + re.compile(r"\b(?:[0-9a-fA-F]{1,4}:){7}[0-9a-fA-F]{1,4}\b"), + "[IP]"), + + ("wallet_address", + re.compile(r"\b0x[0-9a-fA-F]{40}\b"), + "[WALLET]"), + + # ── UUIDs (task IDs, user IDs etc) ──────────────────────────────────── + ("uuid", + re.compile( + r"\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-" + r"[0-9a-f]{4}-[0-9a-f]{12}\b", + re.IGNORECASE), + "[UUID]"), +] + +# ── company name patterns (populated at runtime) ───────────────────────── + +_company_patterns: list[tuple[re.Pattern, str]] = [] + + +def set_company_names(names: list[str]): + """Configure company / product names to redact.""" + _company_patterns.clear() + for name in names: + if name.strip(): + _company_patterns.append(( + re.compile(re.escape(name.strip()), re.IGNORECASE), + "[Company]", + )) + + +# ══════════════════════════════════════════════════════════════════════════════ +# Regex layer (mandatory) +# ══════════════════════════════════════════════════════════════════════════════ + +def deidentify_regex(text: str) -> tuple[str, dict[str, int]]: + """Apply regex-based PII removal. + + Returns ``(cleaned_text, replacement_stats)``. + """ + stats: dict[str, int] = {} + result = text + + for name, pattern, replacement in _PATTERNS: + matches = pattern.findall(result) + if matches: + stats[name] = len(matches) + result = pattern.sub(replacement, result) + + for pattern, replacement in _company_patterns: + matches = pattern.findall(result) + if matches: + stats["company_name"] = stats.get("company_name", 0) + len(matches) + result = pattern.sub(replacement, result) + + return result, stats + + +# ══════════════════════════════════════════════════════════════════════════════ +# LLM-assisted layer (optional) +# ══════════════════════════════════════════════════════════════════════════════ + +_LLM_PROMPT = """\ +请对以下文本进行脱敏处理，用于公开发布到 AI 记忆市场。 + +规则： +1. 将所有公司名替换为 [Company] +2. 将所有人名替换为 [User] +3. 将所有内部项目名替换为 [Project] +4. 将所有具体业务金额/数据替换为 [DATA] +5. 保留所有技术细节（代码、算法、错误消息、架构模式、框架名称） +6. 保留所有通用技术知识和最佳实践 + +输出要求：直接返回脱敏后的文本，不要添加任何说明性文字。 + +原文： +{text}""" + + +async def _deidentify_llm(text: str, llm_adapter, model: str) -> str: + """Run LLM-assisted deidentification pass (business context removal).""" + # Truncate to avoid blowing up context + truncated = text[:4000] + messages = [ + {"role": "system", "content": "你是一个数据脱敏专家。直接输出脱敏后的文本。"}, + {"role": "user", "content": _LLM_PROMPT.format(text=truncated)}, + ] + result = await llm_adapter.chat(messages, model) + if isinstance(result, dict): + result = result.get("content", "") + return result.strip() + + +# ══════════════════════════════════════════════════════════════════════════════ +# Combined pipeline +# ══════════════════════════════════════════════════════════════════════════════ + +async def deidentify( + text: str, + config: "MemoConfig", + llm_adapter=None, +) -> tuple[str, dict]: + """Full deidentification: regex first, then optional LLM pass. + + Returns ``(cleaned_text, stats_dict)``. + """ + # Populate company names from config (idempotent) + if config.company_names and not _company_patterns: + set_company_names(config.company_names) + + # Step 1 — regex (mandatory) + result, stats = deidentify_regex(text) + + # Step 2 — LLM (optional) + if config.deidentification_use_llm and llm_adapter: + try: + result = await _deidentify_llm( + result, llm_adapter, config.deidentification_llm_model) + stats["llm_pass"] = 1 + except Exception as e: + logger.debug("[memo-deident] LLM pass failed: %s", e) + stats["llm_pass_error"] = str(e) + + return result, stats diff --git a/adapters/memo/exporter.py b/adapters/memo/exporter.py new file mode 100644 index 0000000..f033869 --- /dev/null +++ b/adapters/memo/exporter.py @@ -0,0 +1,346 @@ +""" +adapters/memo/exporter.py — Batch / selective Memo export pipeline. + +Full pipeline: + [Cleo raw memories] + → filter (agent / date / type / score) + → content assembly (L2 expand) + → deidentify (regex + optional LLM) + → quality score (≥ 0.6 gate) + → MemoObject transform + → idempotent tracking + → output (JSON files / Memo API upload) + +Usage:: + + exporter = MemoExporter(config) + result = await exporter.export_batch( + ExportFilter(agents=["jerry"], min_score=7), + output_dir="memo_export/", + ) + print(result) +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +import time +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Optional + +from adapters.memo.tracking import ExportTracker +from adapters.memo.deidentifier import deidentify +from adapters.memo.quality_scorer import score_memory +from adapters.memo.transformer import ( + MemoObject, + CONTENT_BUILDERS, + CONVERTERS, +) + +if TYPE_CHECKING: + from adapters.memo.config import MemoConfig + +logger = logging.getLogger(__name__) + + +# ══════════════════════════════════════════════════════════════════════════════ +# Filter & Result +# ══════════════════════════════════════════════════════════════════════════════ + +@dataclass +class ExportFilter: + """Criteria for selecting memories to export.""" + agents: list[str] = field(default_factory=list) # empty = all + types: list[str] = field(default_factory=list) # episodic/semantic/procedural + date_from: str = "" # YYYY-MM-DD inclusive + date_to: str = "" # YYYY-MM-DD inclusive + min_score: int = 0 # Cleo episode score minimum + min_quality: float = 0.6 # Memo quality composite minimum + include_kb: bool = True + include_patterns: bool = True + exclude_archived: bool = True + + +@dataclass +class ExportResult: + """Statistics from an export run.""" + total_scanned: int = 0 + total_eligible: int = 0 + total_exported: int = 0 + skipped_quality: int = 0 + skipped_duplicate: int = 0 + skipped_error: int = 0 + by_type: dict = field(default_factory=dict) + output_path: str = "" + errors: list[str] = field(default_factory=list) + duration_seconds: float = 0.0 + + +# ══════════════════════════════════════════════════════════════════════════════ +# Exporter +# ══════════════════════════════════════════════════════════════════════════════ + +class MemoExporter: + """Batch export pipeline: Cleo memories → Memo MemoryObjects.""" + + def __init__(self, config: "MemoConfig", + tracker: Optional[ExportTracker] = None): + self.config = config + self.tracker = tracker or ExportTracker() + + # ── main entry ──────────────────────────────────────────────────────── + + async def export_batch( + self, + filt: ExportFilter, + output_dir: str = "memo_export", + upload: bool = False, + dry_run: bool = False, + ) -> ExportResult: + """Run the full export pipeline. + + Args: + filt: filter criteria + output_dir: where to write JSON files + upload: also upload to Memo API (requires client) + dry_run: preview only — no writes, no uploads + + Returns: + ExportResult with statistics + """ + t0 = time.monotonic() + result = ExportResult() + + if not dry_run: + os.makedirs(output_dir, exist_ok=True) + + # Phase 1 — collect candidates + candidates = self._collect_candidates(filt) + result.total_scanned = len(candidates) + + # Phase 2 — process each candidate + memo_objects: list[MemoObject] = [] + for candidate in candidates: + src_type = candidate["_source_type"] + src_id = candidate["_source_id"] + + # Idempotent check + if self.tracker.is_exported(src_type, src_id): + result.skipped_duplicate += 1 + continue + + try: + obj = await self._process_one(candidate, filt) + except Exception as e: + result.skipped_error += 1 + result.errors.append(f"{src_type}:{src_id}: {e}") + continue + + if obj is None: + result.skipped_quality += 1 + continue + + memo_objects.append(obj) + + result.total_eligible = len(memo_objects) + + # Phase 3 — output + if not dry_run: + for obj in memo_objects: + # Write JSON file + path = os.path.join(output_dir, f"{obj.id}.json") + with open(path, "w") as f: + json.dump(obj.to_api_payload(), f, + ensure_ascii=False, indent=2) + result.total_exported += 1 + result.by_type[obj.type] = result.by_type.get(obj.type, 0) + 1 + + # Track + self.tracker.record(obj._cleo_source_type, + obj._cleo_source_id, obj.id) + + # Optional upload + if upload: + await self._upload(obj, result) + + self.tracker.save() + else: + # Dry run — just count + for obj in memo_objects: + result.total_exported += 1 + result.by_type[obj.type] = result.by_type.get(obj.type, 0) + 1 + + result.output_path = output_dir + result.duration_seconds = round(time.monotonic() - t0, 2) + return result + + # ── single-item processing ──────────────────────────────────────────── + + async def _process_one(self, candidate: dict, + filt: ExportFilter) -> Optional[MemoObject]: + """Process one candidate: content → deident → score → transform. + + Returns MemoObject if quality passes, else None. + """ + src_type = candidate["_source_type"] + + # Build raw content + builder = CONTENT_BUILDERS.get(src_type) + if not builder: + return None + raw_content = builder(candidate) + + if not raw_content or len(raw_content.strip()) < 50: + return None # too short, skip + + # Deidentify + deidentified, _stats = await deidentify(raw_content, self.config) + if not deidentified or len(deidentified.strip()) < 30: + return None + + # Quality score + quality = score_memory(deidentified, src_type, candidate) + if not quality["passed"] or quality["composite"] < filt.min_quality: + return None + + # Transform + converter = CONVERTERS.get(src_type) + if not converter: + return None + memo_obj = converter(candidate, self.config, deidentified) + + # Set quality score + memo_obj.signals["quality_score"] = quality["composite"] + + # Type filter (if specified) + if filt.types and memo_obj.type not in filt.types: + return None + + return memo_obj + + # ── candidate collection ────────────────────────────────────────────── + + def _collect_candidates(self, filt: ExportFilter) -> list[dict]: + """Gather all candidate memories from Cleo storage.""" + candidates: list[dict] = [] + + agent_ids = filt.agents or self._list_agents() + + for agent_id in agent_ids: + try: + self._collect_agent_memories( + agent_id, filt, candidates) + except Exception as e: + logger.debug("[memo-export] agent %s scan error: %s", + agent_id, e) + + # KB Notes (shared, not per-agent) + if filt.include_kb: + self._collect_kb_notes(filt, candidates) + + return candidates + + def _collect_agent_memories(self, agent_id: str, + filt: ExportFilter, + out: list[dict]): + """Collect episodes, cases, patterns for one agent.""" + try: + from adapters.memory.episodic import EpisodicMemory + except ImportError: + logger.debug("[memo-export] episodic module not available") + return + + ep = EpisodicMemory(agent_id) + + # ── Episodes ───────────────────────────────────────────────── + for episode in ep.list_episodes(limit=500, level=2): + # Date filter + date = episode.get("date", "") + if filt.date_from and date < filt.date_from: + continue + if filt.date_to and date > filt.date_to: + continue + + if filt.exclude_archived and episode.get("archived"): + continue + + # Summary episodes → semantic + if episode.get("type") == "summary_episode": + if episode.get("source_count", 0) >= 2: + episode["_source_type"] = "summary" + episode["_source_id"] = episode.get( + "task_id", + f"summary_{int(episode.get('created_at', 0))}") + out.append(episode) + continue + + # Regular episodes → episodic (success + high score) + outcome = episode.get("outcome", "") + score = episode.get("score") + if outcome != "success": + continue + if score is not None and score < max(filt.min_score, 7): + continue + + episode["_source_type"] = "episode" + episode["_source_id"] = episode.get("task_id", "") + out.append(episode) + + # ── Cases ──────────────────────────────────────────────────── + for case in ep.list_cases(limit=200): + if len(case.get("solution", "")) <= 100: + continue + case["_source_type"] = "case" + case["_source_id"] = case.get("id", "") + out.append(case) + + # ── Patterns ───────────────────────────────────────────────── + if filt.include_patterns: + for pattern in ep.list_patterns(limit=100): + if pattern.get("occurrences", 0) < 3: + continue + pattern["_source_type"] = "pattern" + pattern["_source_id"] = pattern.get("id", "") + out.append(pattern) + + def _collect_kb_notes(self, filt: ExportFilter, out: list[dict]): + """Collect shared KB notes.""" + try: + from adapters.memory.knowledge_base import KnowledgeBase + except ImportError: + logger.debug("[memo-export] knowledge_base module not available") + return + + kb = KnowledgeBase() + for note in kb.list_notes(limit=200): + density = note.get("density", "NORMAL") + update_count = note.get("update_count", 1) + if density != "HIGH" and update_count < 2: + continue + note["_source_type"] = "kb_note" + note["_source_id"] = note.get("slug", "") + out.append(note) + + # ── helpers ─────────────────────────────────────────────────────────── + + def _list_agents(self) -> list[str]: + """Discover agent IDs from memory directory.""" + agents_dir = os.path.join("memory", "agents") + if not os.path.isdir(agents_dir): + return [] + return [d for d in os.listdir(agents_dir) + if os.path.isdir(os.path.join(agents_dir, d)) + and not d.startswith(".")] + + async def _upload(self, obj: MemoObject, result: ExportResult): + """Upload a MemoObject to the Memo API.""" + try: + from adapters.memo.client import MemoClient + client = MemoClient(self.config) + await client.upload_memory(obj.to_api_payload()) + logger.info("[memo] uploaded %s (%s)", obj.id, obj.type) + except Exception as e: + result.errors.append(f"upload {obj.id}: {e}") diff --git a/adapters/memo/hooks.py b/adapters/memo/hooks.py new file mode 100644 index 0000000..8a9e30a --- /dev/null +++ b/adapters/memo/hooks.py @@ -0,0 +1,159 @@ +""" +adapters/memo/hooks.py — Runtime hooks for Memo Protocol integration. + +Post-task hook: + Called after task completion in orchestrator._extract_and_store_memories(). + Auto-uploads successful, high-quality episodes to Memo. + +Pre-task hook (future): + Searches skills/memo/ for relevant skills to inject into task context. + +Both hooks are non-blocking and fault-tolerant — Memo failures never +affect core task execution. +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from adapters.memo.config import MemoConfig + +logger = logging.getLogger(__name__) + + +# ══════════════════════════════════════════════════════════════════════════════ +# Post-task auto-upload hook +# ══════════════════════════════════════════════════════════════════════════════ + +async def post_task_memo_hook( + agent_id: str, + task_id: str, + outcome: str, + score: int | None, + config: "MemoConfig", +) -> None: + """Auto-upload hook called after task completion. + + Injection point: ``core/orchestrator.py`` after + ``_extract_and_store_memories()``. + + Conditions for upload: + - config.memo.enabled AND config.memo.auto_upload.enabled + - outcome == "success" + - score is None or score >= 7 + - quality composite >= config.auto_upload_min_quality + + This function is meant to be wrapped in ``asyncio.create_task()`` + so it never blocks the main pipeline. + """ + if not config.enabled or not config.auto_upload_enabled: + return + + try: + from adapters.memo.tracking import ExportTracker + tracker = ExportTracker() + + # Already exported? + if tracker.is_exported("episode", task_id): + return + + # Load full episode (L2) + from adapters.memory.episodic import EpisodicMemory + ep = EpisodicMemory(agent_id) + episode = ep.load_episode(task_id, level=2) + if not episode: + return + + # Outcome / score gate + ep_outcome = episode.get("outcome", "") + ep_score = episode.get("score") + if ep_outcome != "success": + return + if ep_score is not None and ep_score < 7: + return + + # Build content + from adapters.memo.transformer import _build_episode_content, episode_to_memo + raw_content = _build_episode_content(episode) + if len(raw_content.strip()) < 50: + return + + # Deidentify + from adapters.memo.deidentifier import deidentify + deidentified, _stats = await deidentify(raw_content, config) + if len(deidentified.strip()) < 30: + return + + # Quality score + from adapters.memo.quality_scorer import score_memory + quality = score_memory(deidentified, "episode", episode) + if not quality["passed"]: + return + if quality["composite"] < config.auto_upload_min_quality: + return + + # Transform + memo_obj = episode_to_memo(episode, config, deidentified) + memo_obj.signals["quality_score"] = quality["composite"] + + # Upload + from adapters.memo.client import MemoClient + client = MemoClient(config) + resp = await client.upload_memory(memo_obj.to_api_payload()) + + # Track + tracker.record("episode", task_id, memo_obj.id) + tracker.save() + + logger.info("[memo] auto-uploaded %s → %s (quality=%.2f)", + task_id, memo_obj.id, quality["composite"]) + + except ImportError as e: + logger.debug("[memo] hook skipped (missing dep): %s", e) + except Exception as e: + # Never let Memo failures affect core pipeline + logger.debug("[memo] auto-upload failed (non-critical): %s", e) + + +# ══════════════════════════════════════════════════════════════════════════════ +# Pre-task skill injection hook (future enhancement) +# ══════════════════════════════════════════════════════════════════════════════ + +def find_relevant_memo_skills(task_description: str, + max_skills: int = 3) -> list[str]: + """Search local ``skills/memo/`` for relevant skills. + + Returns list of skill file paths (most relevant first). + This is a simple keyword-match implementation; a future version + could use embedding similarity. + """ + import os + + skills_dir = os.path.join("skills", "memo") + if not os.path.isdir(skills_dir): + return [] + + desc_lower = task_description.lower() + scored: list[tuple[float, str]] = [] + + for fname in os.listdir(skills_dir): + if not fname.endswith(".md") or fname.startswith("."): + continue + path = os.path.join(skills_dir, fname) + try: + with open(path) as f: + content = f.read(2000) # read head only + except OSError: + continue + + # Simple keyword overlap score + content_lower = content.lower() + words = set(desc_lower.split()) + matches = sum(1 for w in words if w in content_lower and len(w) > 3) + if matches > 0: + scored.append((matches, path)) + + scored.sort(key=lambda x: x[0], reverse=True) + return [path for _, path in scored[:max_skills]] diff --git a/adapters/memo/importer.py b/adapters/memo/importer.py new file mode 100644 index 0000000..b9f7e74 --- /dev/null +++ b/adapters/memo/importer.py @@ -0,0 +1,195 @@ +""" +adapters/memo/importer.py — Import Memo Skills into Cleo's skill system. + +Purchased Memo Skills are written to ``skills/memo/`` as Markdown files +with YAML frontmatter, compatible with Cleo's ``SkillLoader``. + +An optional symlink mechanism lets skills be shared to specific agents +under ``skills/agents/{agent_id}/``. +""" + +from __future__ import annotations + +import json +import logging +import os +from typing import TYPE_CHECKING, Optional + +if TYPE_CHECKING: + from adapters.memo.client import MemoClient + from adapters.memo.config import MemoConfig + +logger = logging.getLogger(__name__) + +MEMO_SKILLS_DIR = os.path.join("skills", "memo") + + +class MemoImporter: + """Pull Memo Skills and inject them into Cleo's skill directory.""" + + def __init__(self, config: "MemoConfig", client: "MemoClient"): + self.config = config + self.client = client + os.makedirs(MEMO_SKILLS_DIR, exist_ok=True) + + # ── public API ──────────────────────────────────────────────────────── + + async def sync_skills( + self, + memory_ids: Optional[list[str]] = None, + ) -> dict: + """Synchronize Memo Skills to local directory. + + If ``memory_ids`` is None, uses the export tracker to get + previously exported IDs. + + Returns stats dict: ``{fetched, written, updated, errors}``. + """ + stats = {"fetched": 0, "written": 0, "updated": 0, "errors": 0} + + if not memory_ids: + try: + from adapters.memo.tracking import ExportTracker + tracker = ExportTracker() + memory_ids = tracker.all_memo_ids() + except Exception: + memory_ids = [] + + if not memory_ids: + logger.debug("[memo-import] no memory IDs to sync") + return stats + + try: + skills = await self.client.sync_skills(memory_ids) + stats["fetched"] = len(skills) + except Exception as e: + stats["errors"] += 1 + logger.error("[memo-import] skill sync failed: %s", e) + return stats + + for skill in skills: + try: + written = self._write_skill_file(skill) + if written == "new": + stats["written"] += 1 + elif written == "updated": + stats["updated"] += 1 + except Exception as e: + stats["errors"] += 1 + logger.debug("[memo-import] skill write failed: %s", e) + + return stats + + def inject_skill_to_agent(self, skill_filename: str, agent_id: str): + """Create a symlink from ``skills/memo/`` to an agent's skill dir. + + This makes the Memo skill visible to a specific agent without + duplicating the file. + """ + agent_skills_dir = os.path.join("skills", "agents", agent_id) + os.makedirs(agent_skills_dir, exist_ok=True) + + source = os.path.abspath( + os.path.join(MEMO_SKILLS_DIR, skill_filename)) + target = os.path.join(agent_skills_dir, skill_filename) + + if not os.path.exists(source): + logger.warning("[memo-import] source not found: %s", source) + return + + if os.path.exists(target): + return # already linked + + try: + os.symlink(source, target) + logger.info("[memo-import] linked %s → %s", source, target) + except OSError as e: + # Fallback: copy instead of symlink (Windows compat) + import shutil + shutil.copy2(source, target) + logger.info("[memo-import] copied %s → %s", source, target) + + # ── internal ────────────────────────────────────────────────────────── + + def _write_skill_file(self, skill: dict) -> str: + """Write a Memo Skill as a Markdown file with YAML frontmatter. + + Returns "new", "updated", or "skipped". + """ + skill_id = skill.get("id", "unknown") + title = skill.get("title", "Memo Skill") + content = skill.get("content", "") + tags = skill.get("tags", []) + source_memory = skill.get("source_memory_id", "") + version = skill.get("source_version", 1) + quality = skill.get("quality_score", 0.0) + + # YAML frontmatter (compatible with SkillLoader) + frontmatter = ( + f"---\n" + f"name: \"{title}\"\n" + f"description: \"Imported from Memo Protocol\"\n" + f"tags: {json.dumps(tags)}\n" + f"source: memo\n" + f"memo_skill_id: \"{skill_id}\"\n" + f"memo_memory_id: \"{source_memory}\"\n" + f"memo_version: {version}\n" + f"quality_score: {quality}\n" + f"---\n\n" + ) + + md_content = frontmatter + content + + # Filename: memo_{skill_id_prefix}.md + safe_id = skill_id.replace("/", "_")[:30] + filename = f"memo_{safe_id}.md" + path = os.path.join(MEMO_SKILLS_DIR, filename) + + # Check if exists and needs update + if os.path.exists(path): + try: + with open(path) as f: + existing = f.read() + if f"memo_version: {version}" in existing: + return "skipped" # same version + except OSError: + pass + # Version changed → update + with open(path, "w") as f: + f.write(md_content) + logger.info("[memo-import] updated skill: %s", path) + return "updated" + + with open(path, "w") as f: + f.write(md_content) + logger.info("[memo-import] new skill: %s", path) + return "new" + + def list_local_skills(self) -> list[dict]: + """List all locally stored Memo skills with metadata.""" + skills = [] + if not os.path.isdir(MEMO_SKILLS_DIR): + return skills + + for fname in os.listdir(MEMO_SKILLS_DIR): + if not fname.endswith(".md") or fname.startswith("."): + continue + path = os.path.join(MEMO_SKILLS_DIR, fname) + try: + with open(path) as f: + head = f.read(500) + # Parse basic frontmatter + if head.startswith("---"): + end = head.find("---", 3) + if end > 0: + fm = head[3:end].strip() + info = {"filename": fname} + for line in fm.split("\n"): + if ": " in line: + k, v = line.split(": ", 1) + info[k.strip()] = v.strip().strip('"') + skills.append(info) + except OSError: + continue + + return skills diff --git a/adapters/memo/quality_scorer.py b/adapters/memo/quality_scorer.py new file mode 100644 index 0000000..3d127bb --- /dev/null +++ b/adapters/memo/quality_scorer.py @@ -0,0 +1,180 @@ +""" +adapters/memo/quality_scorer.py — Memo 3-dimension quality scoring. + +Dimensions (matching Memo Protocol spec): + completeness 35 % — executable without additional context? + utility 35 % — solves real problems? immediately reusable? + uniqueness 30 % — novel approach? unique insights? + +Minimum threshold: **0.6** (enforced at export time). + +Reuses ``core/protocols.classify_density()`` as a uniqueness signal. +""" + +from __future__ import annotations + +import logging + +logger = logging.getLogger(__name__) + +# ── signal word lists ───────────────────────────────────────────────────── + +_UTILITY_SIGNALS: list[str] = [ + "solution", "解决", "fix", "修复", "workaround", "步骤", + "implementation", "实现", "code", "command", "命令", + "install", "安装", "配置", "config", "deploy", "部署", + "answer", "回答", "result", "结果", "output", "输出", +] + +_EDGE_CASE_SIGNALS: list[str] = [ + "edge case", "边界", "exception", "异常", "workaround", + "pitfall", "陷阱", "caveat", "注意", "warning", "警告", + "gotcha", "trap", "limitation", "限制", "trade-off", "权衡", + "lesson", "教训", "root cause", "根因", +] + + +# ══════════════════════════════════════════════════════════════════════════════ +# Scoring functions +# ══════════════════════════════════════════════════════════════════════════════ + +def _score_completeness(content: str, source_type: str, + meta: dict) -> float: + """Content completeness — can it be understood/executed standalone?""" + score = 0.45 # baseline + + # Length signals + length = len(content) + if length > 500: + score += 0.08 + if length > 1500: + score += 0.08 + if length > 3000: + score += 0.04 + + # Structure signals + if "## " in content or "# " in content: + score += 0.05 # has heading structure + if "```" in content: + score += 0.08 # has code blocks + if "- " in content or "1. " in content: + score += 0.04 # has lists/steps + + # Procedural completeness: does it have steps? + if source_type in ("case", "procedural"): + cl = content.lower() + if "step" in cl or "步骤" in cl or "step 1" in cl: + score += 0.08 + if "result" in cl or "结果" in cl: + score += 0.04 + + # Cleo-specific: episode score + cleo_score = meta.get("score") + if cleo_score is not None: + if cleo_score >= 7: + score += 0.08 + if cleo_score >= 9: + score += 0.04 + + return min(score, 1.0) + + +def _score_utility(content: str, source_type: str, + meta: dict) -> float: + """Practical utility — does it solve real problems?""" + score = 0.40 # baseline + + # Case type is naturally high utility (problem→solution) + if source_type == "case": + score += 0.15 + use_count = meta.get("use_count", 0) + if use_count >= 2: + score += 0.08 + if use_count >= 5: + score += 0.08 + + # Utility signal words + cl = content.lower() + hit = sum(1 for s in _UTILITY_SIGNALS if s in cl) + score += min(hit * 0.04, 0.16) + + # Episode outcome=success + if meta.get("outcome") == "success": + score += 0.08 + + # KB note with multiple contributors / updates = validated knowledge + if source_type == "kb_note": + if meta.get("update_count", 1) >= 3: + score += 0.10 + if len(meta.get("contributors", [])) >= 2: + score += 0.06 + + # Pattern with high occurrences = repeatedly validated + if source_type == "pattern": + occ = meta.get("occurrences", 0) + if occ >= 5: + score += 0.10 + elif occ >= 3: + score += 0.06 + + return min(score, 1.0) + + +def _score_uniqueness(content: str, meta: dict) -> float: + """Uniqueness — novel approach or insight?""" + score = 0.50 # baseline + + # DensityTag signal (reuse core/protocols.classify_density) + try: + from core.protocols import classify_density, DensityLevel + density = classify_density(content, meta.get("tags", [])) + if density == DensityLevel.HIGH: + score += 0.18 + elif density == DensityLevel.LOW: + score -= 0.18 + except ImportError: + pass # fallback: no density signal + + # Edge case / pitfall signals = rarer, more unique knowledge + cl = content.lower() + hit = sum(1 for s in _EDGE_CASE_SIGNALS if s in cl) + score += min(hit * 0.04, 0.15) + + # Pattern occurrences = repeatedly validated = higher value + if meta.get("occurrences", 0) >= 5: + score += 0.08 + + return min(max(score, 0.0), 1.0) + + +# ══════════════════════════════════════════════════════════════════════════════ +# Public API +# ══════════════════════════════════════════════════════════════════════════════ + +def score_memory(content: str, source_type: str, + metadata: dict) -> dict: + """Score a memory candidate across 3 Memo dimensions. + + Args: + content: deidentified content text + source_type: one of episode / summary / case / pattern / kb_note + metadata: original Cleo metadata dict (score, tags, use_count, etc.) + + Returns: + dict with keys: completeness, utility, uniqueness, composite, passed + """ + completeness = _score_completeness(content, source_type, metadata) + utility = _score_utility(content, source_type, metadata) + uniqueness = _score_uniqueness(content, metadata) + + composite = (completeness * 0.35 + + utility * 0.35 + + uniqueness * 0.30) + + return { + "completeness": round(completeness, 3), + "utility": round(utility, 3), + "uniqueness": round(uniqueness, 3), + "composite": round(composite, 3), + "passed": composite >= 0.6, + } diff --git a/adapters/memo/tracking.py b/adapters/memo/tracking.py new file mode 100644 index 0000000..fb7a542 --- /dev/null +++ b/adapters/memo/tracking.py @@ -0,0 +1,94 @@ +""" +adapters/memo/tracking.py — Idempotent export tracker. + +Maintains a ``memory/memo_export_tracking.json`` file that maps +Cleo source IDs (episode task_id, case hash, pattern hash, KB slug) +to Memo ``mem_*`` IDs, preventing duplicate exports / uploads. +""" + +from __future__ import annotations + +import json +import os +import time +import logging +from typing import Optional + +logger = logging.getLogger(__name__) + +TRACKING_FILE = os.path.join("memory", "memo_export_tracking.json") + + +class ExportTracker: + """Cleo source_id ↔ Memo mem_id mapping with persistence.""" + + def __init__(self, path: str = TRACKING_FILE): + self.path = path + self._data: dict = self._load() + + # ── persistence ─────────────────────────────────────────────────────── + + def _load(self) -> dict: + if os.path.exists(self.path): + try: + with open(self.path) as f: + return json.load(f) + except (json.JSONDecodeError, OSError) as e: + logger.debug("[memo-tracking] load failed: %s", e) + return {"exports": {}, "meta": {"created_at": time.time()}} + + def save(self): + os.makedirs(os.path.dirname(self.path) or ".", exist_ok=True) + self._data["meta"]["updated_at"] = time.time() + self._data["meta"]["total_exports"] = len(self._data["exports"]) + tmp = self.path + ".tmp" + with open(tmp, "w") as f: + json.dump(self._data, f, ensure_ascii=False, indent=2) + os.replace(tmp, self.path) + + # ── key helpers ─────────────────────────────────────────────────────── + + @staticmethod + def _key(source_type: str, source_id: str) -> str: + return f"{source_type}:{source_id}" + + # ── query / mutate ──────────────────────────────────────────────────── + + def is_exported(self, source_type: str, source_id: str) -> bool: + return self._key(source_type, source_id) in self._data["exports"] + + def record(self, source_type: str, source_id: str, memo_id: str): + self._data["exports"][self._key(source_type, source_id)] = { + "memo_id": memo_id, + "exported_at": time.time(), + "source_type": source_type, + } + + def get_memo_id(self, source_type: str, source_id: str) -> Optional[str]: + entry = self._data["exports"].get(self._key(source_type, source_id)) + return entry["memo_id"] if entry else None + + def all_memo_ids(self) -> list[str]: + """Return all exported Memo IDs (useful for skill sync).""" + return [e["memo_id"] for e in self._data["exports"].values() + if "memo_id" in e] + + # ── stats ───────────────────────────────────────────────────────────── + + def stats(self) -> dict: + exports = self._data["exports"] + by_type: dict[str, int] = {} + for entry in exports.values(): + t = entry.get("source_type", "unknown") + by_type[t] = by_type.get(t, 0) + 1 + return { + "total": len(exports), + "by_type": by_type, + "created_at": self._data["meta"].get("created_at"), + "updated_at": self._data["meta"].get("updated_at"), + } + + def reset(self): + """Clear all tracking data (for testing or re-export).""" + self._data = {"exports": {}, "meta": {"created_at": time.time()}} + self.save() diff --git a/adapters/memo/transformer.py b/adapters/memo/transformer.py new file mode 100644 index 0000000..0a749c1 --- /dev/null +++ b/adapters/memo/transformer.py @@ -0,0 +1,398 @@ +""" +adapters/memo/transformer.py — Cleo memory → Memo MemoryObject conversion. + +Handles 5 source types: + episode → Memo EPISODIC (full task execution journey) + summary → Memo SEMANTIC (consolidated multi-episode summary) + case → Memo PROCEDURAL (problem → solution = executable skill) + pattern → Memo SEMANTIC (recurring generalizable observation) + kb_note → Memo SEMANTIC (cross-agent distilled knowledge) + +Each converter returns a ``MemoObject`` dataclass ready for +``to_api_payload()`` serialization. +""" + +from __future__ import annotations + +import hashlib +import json +import secrets +import time +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from adapters.memo.config import MemoConfig + + +# ══════════════════════════════════════════════════════════════════════════════ +# Helpers +# ══════════════════════════════════════════════════════════════════════════════ + +_NANOID_ALPHABET = ( + "0123456789" + "ABCDEFGHIJKLMNOPQRSTUVWXYZ" + "abcdefghijklmnopqrstuvwxyz" +) + + +def _nanoid(prefix: str = "mem_", length: int = 21) -> str: + """Generate a Memo-compatible nanoid.""" + return prefix + "".join(secrets.choice(_NANOID_ALPHABET) + for _ in range(length)) + + +def _content_hash(content: str) -> str: + """SHA-256 hash of content (Memo requirement).""" + return hashlib.sha256(content.encode("utf-8")).hexdigest() + + +def _iso_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _truncate(text: str, max_len: int = 280) -> str: + if len(text) <= max_len: + return text + return text[: max_len - 3] + "..." + + +# ══════════════════════════════════════════════════════════════════════════════ +# MemoObject +# ══════════════════════════════════════════════════════════════════════════════ + +@dataclass +class MemoObject: + """Python representation of a Memo MemoryObject (v1.0).""" + + memo_version: str = "1.0" + id: str = "" + type: str = "episodic" # episodic | semantic | procedural + status: str = "draft" + content: str = "" + content_hash: str = "" + title: str = "" + summary: str = "" # max 280 chars + tags: list[str] = field(default_factory=list) + domain: str = "python" + language: str = "zh" + author: dict = field(default_factory=dict) + provenance: dict = field(default_factory=dict) + signals: dict = field(default_factory=lambda: { + "quality_score": 0.0, + "community_score": 1.0, + "call_count": 0, + "helpful_count": 0, + "not_helpful_count": 0, + "freshness_score": 1.0, + }) + access: dict = field(default_factory=lambda: { + "tier": "developer", + "price_usdc": 0.0, + "subscription_bypass": True, + }) + created_at: str = "" + updated_at: str = "" + + # ── Cleo-internal tracking (not sent to API) ───────────────────────── + _cleo_source_type: str = "" + _cleo_source_id: str = "" + + def to_api_payload(self) -> dict: + """Serialize to Memo API POST /memories request body.""" + d: dict[str, Any] = {} + for k, v in self.__dict__.items(): + if k.startswith("_"): + continue + d[k] = v + return d + + +# ══════════════════════════════════════════════════════════════════════════════ +# Content builders +# ══════════════════════════════════════════════════════════════════════════════ + +def _build_episode_content(episode: dict) -> str: + """Assemble L2 episode into structured Markdown for Memo.""" + parts: list[str] = [] + + title = episode.get("title", "Untitled Task") + parts.append(f"# Task: {title}") + + desc = episode.get("description", "") + if desc: + parts.append(f"\n## Description\n{desc}") + + # Execution context + ctx = episode.get("context", {}) + if ctx: + try: + ctx_str = json.dumps(ctx, indent=2, ensure_ascii=False) + parts.append(f"\n## Context\n```json\n{ctx_str}\n```") + except (TypeError, ValueError): + pass + + # Result + result = episode.get("result_full", + episode.get("result_preview", "")) + if result: + parts.append(f"\n## Result\n{result}") + + # Metadata + parts.append("\n## Metadata") + parts.append(f"- Outcome: {episode.get('outcome', 'unknown')}") + score = episode.get("score") + if score is not None: + parts.append(f"- Score: {score}") + parts.append(f"- Model: {episode.get('model', 'unknown')}") + err = episode.get("error_type") + if err: + parts.append(f"- Error Type: {err}") + + return "\n".join(parts) + + +def _build_case_content(case: dict) -> str: + """Convert Case (problem→solution) into Skill Document format.""" + problem = case.get("problem", "") + solution = case.get("solution", "") + tags = case.get("tags", []) + + return f"""# SKILL: {problem[:80]} + +## Trigger Conditions +- {problem} + +## Solution Steps +{solution} + +## Metadata +- Agent: {case.get('agent_id', 'unknown')} +- Usage Count: {case.get('use_count', 0)} +- Tags: {', '.join(tags)} +""" + + +def _build_pattern_content(pattern: dict) -> str: + """Convert Pattern into semantic knowledge text.""" + desc = pattern.get("description", "") + evidence = pattern.get("evidence", []) + + parts = [f"# Pattern: {desc[:100]}"] + parts.append(f"\n## Description\n{desc}") + + if evidence: + parts.append("\n## Evidence") + for i, ev in enumerate(evidence[:10], 1): + if isinstance(ev, str): + parts.append(f"{i}. {ev}") + elif isinstance(ev, dict): + parts.append(f"{i}. {ev.get('text', str(ev))}") + + parts.append(f"\n## Occurrences: {pattern.get('occurrences', 0)}") + return "\n".join(parts) + + +def _build_kb_note_content(note: dict) -> str: + """Convert KB Note into semantic knowledge text.""" + topic = note.get("topic", "") + content = note.get("content", "") + links = note.get("links", []) + + parts = [f"# {topic}"] + parts.append(f"\n{content}") + + if links: + parts.append("\n## Related Notes") + for link in links[:10]: + parts.append(f"- [[{link}]]") + + contributors = note.get("contributors", []) + if contributors: + parts.append(f"\n## Contributors: {', '.join(contributors)}") + + return "\n".join(parts) + + +def _build_summary_content(summary: dict) -> str: + """Convert Summary Episode into semantic knowledge text.""" + parts = [f"# Consolidated Summary"] + + titles = summary.get("titles", []) + if titles: + parts.append("\n## Source Tasks") + for t in titles[:10]: + parts.append(f"- {t}") + + content = summary.get("content_summary", "") + if content: + parts.append(f"\n## Summary\n{content}") + + dist = summary.get("outcome_distribution", {}) + if dist: + parts.append("\n## Outcome Distribution") + for k, v in dist.items(): + parts.append(f"- {k}: {v}") + + parts.append(f"\n## Source Count: {summary.get('source_count', 0)}") + avg = summary.get("avg_score") + if avg is not None: + parts.append(f"## Average Score: {avg:.1f}") + + return "\n".join(parts) + + +# ══════════════════════════════════════════════════════════════════════════════ +# Converters (Cleo → MemoObject) +# ══════════════════════════════════════════════════════════════════════════════ + +def episode_to_memo(episode: dict, config: "MemoConfig", + deidentified_content: str) -> MemoObject: + """Convert Cleo Episode → Memo EPISODIC MemoryObject.""" + title = episode.get("title", "Untitled")[:120] + desc = episode.get("description", "") + now = _iso_now() + + return MemoObject( + id=_nanoid(), + type="episodic", + content=deidentified_content, + content_hash=_content_hash(deidentified_content), + title=title, + summary=_truncate(desc, 280), + tags=episode.get("tags", [])[:10], + domain=config.default_domain, + language=config.default_language, + author=config.author_info, + access=config.default_access, + created_at=now, + updated_at=now, + _cleo_source_type="episode", + _cleo_source_id=episode.get("task_id", ""), + ) + + +def summary_to_memo(summary: dict, config: "MemoConfig", + deidentified_content: str) -> MemoObject: + """Convert Cleo Summary Episode → Memo SEMANTIC MemoryObject.""" + titles = summary.get("titles", []) + title = f"Summary: {titles[0][:80]}" if titles else "Consolidated Summary" + now = _iso_now() + + return MemoObject( + id=_nanoid(), + type="semantic", + content=deidentified_content, + content_hash=_content_hash(deidentified_content), + title=title[:120], + summary=_truncate(summary.get("content_summary", ""), 280), + tags=list(set(summary.get("tags", [])))[:10], + domain=config.default_domain, + language=config.default_language, + author=config.author_info, + access=config.default_access, + created_at=now, + updated_at=now, + _cleo_source_type="summary", + _cleo_source_id=summary.get("task_id", + f"summary_{int(summary.get('created_at', 0))}"), + ) + + +def case_to_memo(case: dict, config: "MemoConfig", + deidentified_content: str) -> MemoObject: + """Convert Cleo Case → Memo PROCEDURAL MemoryObject.""" + problem = case.get("problem", "") + title = f"Case: {problem[:100]}" + now = _iso_now() + + return MemoObject( + id=_nanoid(), + type="procedural", + content=deidentified_content, + content_hash=_content_hash(deidentified_content), + title=title[:120], + summary=_truncate(problem, 280), + tags=case.get("tags", [])[:10], + domain=config.default_domain, + language=config.default_language, + author=config.author_info, + access=config.default_access, + created_at=now, + updated_at=now, + _cleo_source_type="case", + _cleo_source_id=case.get("id", ""), + ) + + +def pattern_to_memo(pattern: dict, config: "MemoConfig", + deidentified_content: str) -> MemoObject: + """Convert Cleo Pattern → Memo SEMANTIC MemoryObject.""" + desc = pattern.get("description", "") + title = f"Pattern: {desc[:100]}" + now = _iso_now() + + return MemoObject( + id=_nanoid(), + type="semantic", + content=deidentified_content, + content_hash=_content_hash(deidentified_content), + title=title[:120], + summary=_truncate(desc, 280), + tags=pattern.get("tags", [])[:10], + domain=config.default_domain, + language=config.default_language, + author=config.author_info, + access=config.default_access, + created_at=now, + updated_at=now, + _cleo_source_type="pattern", + _cleo_source_id=pattern.get("id", ""), + ) + + +def kb_note_to_memo(note: dict, config: "MemoConfig", + deidentified_content: str) -> MemoObject: + """Convert Cleo KB Note → Memo SEMANTIC MemoryObject.""" + topic = note.get("topic", "") + now = _iso_now() + + return MemoObject( + id=_nanoid(), + type="semantic", + content=deidentified_content, + content_hash=_content_hash(deidentified_content), + title=topic[:120] or "KB Note", + summary=_truncate(note.get("content", ""), 280), + tags=note.get("tags", [])[:10], + domain=config.default_domain, + language=config.default_language, + author=config.author_info, + access=config.default_access, + created_at=now, + updated_at=now, + _cleo_source_type="kb_note", + _cleo_source_id=note.get("slug", ""), + ) + + +# ══════════════════════════════════════════════════════════════════════════════ +# Content builder dispatch +# ══════════════════════════════════════════════════════════════════════════════ + +CONTENT_BUILDERS: dict[str, callable] = { + "episode": _build_episode_content, + "summary": _build_summary_content, + "case": _build_case_content, + "pattern": _build_pattern_content, + "kb_note": _build_kb_note_content, +} + +CONVERTERS: dict[str, callable] = { + "episode": episode_to_memo, + "summary": summary_to_memo, + "case": case_to_memo, + "pattern": pattern_to_memo, + "kb_note": kb_note_to_memo, +} diff --git a/adapters/memory/consolidator.py b/adapters/memory/consolidator.py new file mode 100644 index 0000000..bc66620 --- /dev/null +++ b/adapters/memory/consolidator.py @@ -0,0 +1,337 @@ +""" +adapters/memory/consolidator.py — V0.02 改进 7: MemoryConsolidator + +Three-phase pipeline for episodic memory lifecycle management: + + Phase 1 — Cluster: Group old episodes (>7 days) by tag overlap + Phase 2 — Compress: Merge each cluster into a single SummaryEpisode + Phase 3 — Promote: High-value summaries (source_count ≥ 3) → KB atomic notes + +Safety rules: + - Only processes episodes older than 7 days + - Original episodes are marked 'archived' (not deleted) + - Each compression records provenance (source task IDs) + - Consolidation log written to memory/consolidation_log.jsonl +""" + +from __future__ import annotations + +import json +import logging +import os +import time +from collections import defaultdict +from datetime import datetime, timezone +from typing import Optional + +logger = logging.getLogger(__name__) + +_PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname( + os.path.abspath(__file__)))) # → project root +CONSOLIDATION_LOG = os.path.join(_PROJECT_ROOT, "memory", "consolidation_log.jsonl") +_MIN_AGE_DAYS = 3 # was 7 — allow faster consolidation +_MIN_PROMOTE_SOURCES = 2 # was 3 — lower bar for KB promotion + + +class MemoryConsolidator: + """ + Periodic consolidation pipeline for episodic memories. + + Designed to run as a background task in the orchestrator + (non-blocking, failure-tolerant). + """ + + def __init__(self, episodic_memory, knowledge_base=None): + """ + Args: + episodic_memory: EpisodicMemory instance for the agent. + knowledge_base: Optional KnowledgeBase instance for KB promotion. + """ + self.episodic = episodic_memory + self.kb = knowledge_base + self._last_run: float = 0.0 + + def should_run(self, interval_seconds: int = 86400) -> bool: + """Check if enough time has passed since the last run.""" + return (time.time() - self._last_run) >= interval_seconds + + def run(self) -> dict: + """Execute the full consolidation pipeline (sync — use asyncio.to_thread). + + Returns: + Stats dict: {clustered, compressed, promoted, errors} + """ + self._last_run = time.time() + stats = {"clustered": 0, "compressed": 0, "promoted": 0, "errors": 0} + + try: + # Phase 1: Cluster old episodes + clusters = self._cluster_episodes() + stats["clustered"] = len(clusters) + + if not clusters: + self._log_consolidation(stats) + return stats + + # Phase 2: Compress each cluster into a summary + summaries = [] + for cluster in clusters: + try: + summary = self._compress_cluster(cluster) + if summary: + summaries.append(summary) + stats["compressed"] += 1 + except Exception as e: + logger.debug("Compress failed for cluster: %s", e) + stats["errors"] += 1 + + # Phase 3: Promote high-value summaries to KB + for summary in summaries: + source_count = summary.get("source_count", 0) + if source_count >= _MIN_PROMOTE_SOURCES and self.kb: + try: + self._promote_to_kb(summary) + stats["promoted"] += 1 + except Exception as e: + logger.debug("Promote failed: %s", e) + stats["errors"] += 1 + + # Phase 4: Dedup insights feed + if self.kb: + try: + dedup_stats = self.kb.dedup_insights() + removed = dedup_stats.get("removed", 0) + if removed > 0: + logger.debug("[%s] insight dedup: %s", + self.episodic.agent_id, dedup_stats) + stats["insights_deduped"] = removed + except Exception as e: + logger.debug("Insight dedup failed: %s", e) + + # Log consolidation run + self._log_consolidation(stats) + + except Exception as e: + logger.debug("Consolidation pipeline error: %s", e) + stats["errors"] += 1 + + return stats + + # ── Phase 1: Cluster ───────────────────────────────────────────────── + + def _cluster_episodes(self) -> list[list[dict]]: + """Group old episodes by tag overlap. + + Returns list of clusters, each cluster is a list of episodes. + """ + cutoff_ts = time.time() - (_MIN_AGE_DAYS * 86400) + old_episodes = [] + + # Collect episodes older than 7 days + for date_str in self.episodic._list_dates(): + try: + day_ts = datetime.strptime(date_str, "%Y-%m-%d").replace( + tzinfo=timezone.utc).timestamp() + except ValueError: + continue + + if day_ts >= cutoff_ts: + continue # Too recent + + day_dir = os.path.join(self.episodic.episodes_dir, date_str) + if not os.path.isdir(day_dir): + continue + + for fname in os.listdir(day_dir): + if not fname.endswith(".json") or fname.startswith("."): + continue + path = os.path.join(day_dir, fname) + try: + with open(path) as f: + ep = json.load(f) + # Skip already archived episodes + if ep.get("archived"): + continue + ep["_path"] = path + old_episodes.append(ep) + except (json.JSONDecodeError, OSError): + continue + + if not old_episodes: + return [] + + # Cluster by tag overlap (simple greedy approach) + clusters: list[list[dict]] = [] + used = set() + + for i, ep in enumerate(old_episodes): + if i in used: + continue + cluster = [ep] + used.add(i) + ep_tags = set(ep.get("tags", [])) + + for j, other in enumerate(old_episodes): + if j in used: + continue + other_tags = set(other.get("tags", [])) + # Require at least 1 tag overlap, or same date + if (ep_tags & other_tags + or ep.get("date") == other.get("date")): + cluster.append(other) + used.add(j) + + clusters.append(cluster) + + return clusters + + # ── Phase 2: Compress ──────────────────────────────────────────────── + + def _compress_cluster(self, cluster: list[dict]) -> dict | None: + """Merge a cluster of episodes into a single SummaryEpisode. + + Returns the summary dict, or None if cluster is too small. + """ + if len(cluster) < 2: + # Single episode — just mark as archived, no summary needed + if cluster: + self._archive_episode(cluster[0]) + return None + + # Build summary from cluster + all_tags: set[str] = set() + all_titles: list[str] = [] + all_task_ids: list[str] = [] + total_score = 0 + score_count = 0 + outcomes: dict[str, int] = defaultdict(int) + earliest_ts = float("inf") + latest_ts = 0.0 + content_pieces: list[str] = [] + + for ep in cluster: + all_tags.update(ep.get("tags", [])) + title = ep.get("title", "") + if title: + all_titles.append(title) + task_id = ep.get("task_id", "") + if task_id: + all_task_ids.append(task_id) + score = ep.get("score") + if score is not None: + total_score += score + score_count += 1 + outcome = ep.get("outcome", "unknown") + outcomes[outcome] += 1 + ts = ep.get("ts", 0) + earliest_ts = min(earliest_ts, ts) + latest_ts = max(latest_ts, ts) + # Collect brief result previews for summary content + preview = ep.get("result_preview", "")[:200] + if preview: + content_pieces.append(f"- {title}: {preview}") + + avg_score = (total_score / score_count) if score_count else None + dominant_outcome = max(outcomes, key=outcomes.get) if outcomes else "unknown" + + summary = { + "type": "summary_episode", + "agent_id": self.episodic.agent_id, + "source_task_ids": all_task_ids, + "source_count": len(cluster), + "tags": sorted(all_tags), + "titles": all_titles[:10], # Keep up to 10 titles + "avg_score": avg_score, + "dominant_outcome": dominant_outcome, + "outcome_distribution": dict(outcomes), + "earliest_ts": earliest_ts, + "latest_ts": latest_ts, + "content_summary": "\n".join(content_pieces[:10]), + "created_at": time.time(), + } + + # Save summary as a special episode + summary_id = f"summary_{int(earliest_ts)}_{int(latest_ts)}" + summary["task_id"] = summary_id + + # Use earliest date for directory + summary_date = datetime.fromtimestamp( + earliest_ts, tz=timezone.utc).strftime("%Y-%m-%d") + summary["date"] = summary_date + + self.episodic.save_episode(summary) + + # Archive original episodes + for ep in cluster: + self._archive_episode(ep) + + logger.debug("[%s] compressed %d episodes → %s", + self.episodic.agent_id, len(cluster), summary_id) + return summary + + def _archive_episode(self, episode: dict): + """Mark an episode as archived (don't delete).""" + path = episode.get("_path") + if not path or not os.path.exists(path): + return + try: + with open(path) as f: + data = json.load(f) + data["archived"] = True + data["archived_at"] = time.time() + with open(path, "w") as f: + json.dump(data, f, ensure_ascii=False, indent=2) + except (json.JSONDecodeError, OSError) as e: + logger.debug("Archive failed for %s: %s", path, e) + + # ── Phase 3: Promote to KB ─────────────────────────────────────────── + + def _promote_to_kb(self, summary: dict): + """Promote a high-value summary to a KB atomic note. + + Only promotes summaries with source_count ≥ 3. + Tags the note with density=HIGH. + """ + if not self.kb: + return + + topic = f"[{summary['agent_id']}] " + ", ".join( + summary.get("titles", [])[:3]) + if not topic.strip("[] "): + topic = f"[{summary['agent_id']}] consolidated episodes" + + content_parts = [ + f"**Sources:** {summary.get('source_count', 0)} episodes", + f"**Period:** {summary.get('date', 'unknown')}", + f"**Avg Score:** {summary.get('avg_score', 'N/A')}", + f"**Outcome:** {summary.get('dominant_outcome', 'unknown')}", + "", + summary.get("content_summary", ""), + ] + + self.kb.create_note( + topic=topic[:120], + content="\n".join(content_parts), + tags=summary.get("tags", []) + ["consolidated", "auto-promoted"], + author=summary.get("agent_id", "system"), + density="HIGH", + ) + + logger.debug("[%s] promoted summary to KB: %s", + summary.get("agent_id"), topic[:60]) + + # ── Logging ────────────────────────────────────────────────────────── + + def _log_consolidation(self, stats: dict): + """Append consolidation stats to the log file.""" + entry = { + "ts": time.time(), + "agent_id": self.episodic.agent_id, + **stats, + } + try: + os.makedirs(os.path.dirname(CONSOLIDATION_LOG) or ".", exist_ok=True) + with open(CONSOLIDATION_LOG, "a") as f: + f.write(json.dumps(entry, ensure_ascii=False) + "\n") + except OSError as e: + logger.debug("Failed to write consolidation log: %s", e) diff --git a/adapters/memory/episodic.py b/adapters/memory/episodic.py index 2d809ab..a7fc9c1 100644 --- a/adapters/memory/episodic.py +++ b/adapters/memory/episodic.py @@ -119,9 +119,14 @@ class EpisodicMemory: Integrates with the existing HybridMemory for vector+keyword retrieval. """ - def __init__(self, agent_id: str, base_dir: str = "memory/agents"): + _DEFAULT_BASE = os.path.join( + os.path.dirname(os.path.dirname(os.path.dirname( + os.path.abspath(__file__)))), # → project root + "memory", "agents") + + def __init__(self, agent_id: str, base_dir: str = ""): self.agent_id = agent_id - self.base = os.path.join(base_dir, agent_id) + self.base = os.path.join(base_dir or self._DEFAULT_BASE, agent_id) self.episodes_dir = os.path.join(self.base, "episodes") self.daily_dir = os.path.join(self.base, "daily") self.cases_dir = os.path.join(self.base, "cases") @@ -145,6 +150,28 @@ def save_episode(self, episode: dict) -> str: logger.debug("[%s] saved episode %s", self.agent_id, task_id) return path + def update_episode_score(self, task_id: str, score: int, + date: Optional[str] = None) -> bool: + """Retroactively set/overwrite the score field of an episode.""" + dates_to_check = [date] if date else sorted( + self._list_dates(), reverse=True) + for d in dates_to_check: + path = os.path.join(self.episodes_dir, d, f"{task_id}.json") + if os.path.exists(path): + try: + with open(path) as f: + ep = json.load(f) + ep["score"] = score + with open(path, "w") as f: + json.dump(ep, f, ensure_ascii=False, indent=2) + logger.debug("[%s] backfilled score=%s for %s", + self.agent_id, score, task_id) + return True + except (json.JSONDecodeError, OSError) as e: + logger.debug("[%s] update_episode_score failed: %s", + self.agent_id, e) + return False + def load_episode(self, task_id: str, date: Optional[str] = None, level: int = 1) -> Optional[dict]: """ @@ -668,8 +695,15 @@ def generate_memory_md(self, max_lines: int = 200) -> str: continue # Sort by use_count descending, take top entries + # V0.03+: Filter out cases whose problem is conversation history + # metadata (injected by channel managers), not actual task content + _HISTORY_MARKERS = ("对话历史", "Conversation History", + "[source:telegram]", "[source:dashboard]") cases.sort(key=lambda c: c.get("use_count", 0), reverse=True) - p0_cases = [c for c in cases if c.get("use_count", 0) >= 1][:10] + p0_cases = [c for c in cases + if c.get("use_count", 0) >= 1 + and not any(m in c.get("problem", "") + for m in _HISTORY_MARKERS)][:10] if p0_cases: for c in p0_cases: problem = c.get("problem", "")[:150] diff --git a/adapters/memory/extractor.py b/adapters/memory/extractor.py index 38a2d64..eed173d 100644 --- a/adapters/memory/extractor.py +++ b/adapters/memory/extractor.py @@ -138,6 +138,17 @@ def extract_insight(task_description: str, result: str, except (json.JSONDecodeError, ValueError): pass + # Fallback: for substantial results, use the first meaningful line + if len(result.strip()) > 300: + for line in result.strip().split('\n'): + line = line.strip() + # Skip markdown headers, code fences, short lines + if (line and len(line) > 30 + and not line.startswith('#') + and not line.startswith('```') + and not line.startswith('|')): + return f"{line[:150]} (task: {task_description[:80]})" + return None diff --git a/adapters/memory/knowledge_base.py b/adapters/memory/knowledge_base.py index f16bbbc..25a5a91 100644 --- a/adapters/memory/knowledge_base.py +++ b/adapters/memory/knowledge_base.py @@ -34,7 +34,9 @@ logger = logging.getLogger(__name__) -SHARED_DIR = os.path.join("memory", "shared") +_PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname( + os.path.abspath(__file__)))) # adapters/memory/ → adapters/ → project root +SHARED_DIR = os.path.join(_PROJECT_ROOT, "memory", "shared") ATOMIC_DIR = os.path.join(SHARED_DIR, "atomic") MOC_PATH = os.path.join(SHARED_DIR, "moc.md") INSIGHTS_PATH = os.path.join(SHARED_DIR, "insights.jsonl") @@ -287,14 +289,29 @@ def add_insight(self, agent_id: str, insight: str, """ Publish a cross-agent insight. Other agents can read the feed for collective learning. + Skips exact duplicates (same insight text already recorded). """ - entry = { - "agent_id": agent_id, - "insight": insight, - "tags": tags or [], - "ts": time.time(), - } with self.lock: + # Dedup: skip if identical insight text already exists + if os.path.exists(self.insights_path): + try: + with open(self.insights_path) as f: + for line in f: + if line.strip(): + try: + if json.loads(line).get("insight") == insight: + return # exact duplicate — skip + except json.JSONDecodeError: + continue + except OSError: + pass + + entry = { + "agent_id": agent_id, + "insight": insight, + "tags": tags or [], + "ts": time.time(), + } with open(self.insights_path, "a") as f: f.write(json.dumps(entry, ensure_ascii=False) + "\n") @@ -325,6 +342,46 @@ def recent_insights(self, limit: int = 20, # Return most recent return entries[-limit:] + def dedup_insights(self) -> dict: + """Remove duplicate insights, keeping earliest occurrence. + + Called periodically by MemoryConsolidator. + Returns stats: {before, after, removed}. + """ + with self.lock: + if not os.path.exists(self.insights_path): + return {"before": 0, "after": 0, "removed": 0} + entries = [] + try: + with open(self.insights_path) as f: + for line in f: + if line.strip(): + try: + entries.append(json.loads(line)) + except json.JSONDecodeError: + continue + except OSError: + return {"before": 0, "after": 0, "removed": 0} + + before = len(entries) + seen: set[str] = set() + unique: list[dict] = [] + for e in entries: + key = e.get("insight", "") + if key not in seen: + seen.add(key) + unique.append(e) + + if len(unique) < before: + with open(self.insights_path, "w") as f: + for e in unique: + f.write(json.dumps(e, ensure_ascii=False) + "\n") + logger.debug("Insight dedup: %d → %d (removed %d)", + before, len(unique), before - len(unique)) + + return {"before": before, "after": len(unique), + "removed": before - len(unique)} + # ── Recall for System Prompt Injection ──────────────────────────────── def recall(self, query: str, agent_id: str, diff --git a/cli/memo_cmd.py b/cli/memo_cmd.py new file mode 100644 index 0000000..53b21a0 --- /dev/null +++ b/cli/memo_cmd.py @@ -0,0 +1,334 @@ +""" +cli/memo_cmd.py — ``cleo memo`` command group. + +Actions: + status — show Memo integration status + tracking stats + export — batch export memories to Memo format (JSON) + search — search the Memo platform + skills — sync purchased skills from Memo + tracking — show export tracking records +""" + +from __future__ import annotations + +import asyncio +import json +import os +import sys +import logging +from typing import Optional + +logger = logging.getLogger(__name__) + +# ── theme (reuse memory_cmd's pattern) ──────────────────────────────────── + +try: + from cli.memory_cmd import _theme +except ImportError: + class _theme: # type: ignore + heading = "bold cyan" + success = "bold green" + warning = "bold yellow" + error = "bold red" + muted = "dim" + + +def _get_console(): + try: + from rich.console import Console + return Console() + except ImportError: + return None + + +def _load_config() -> dict: + """Load agents.yaml config.""" + for path in ("config/agents.yaml", "agents.yaml"): + if os.path.exists(path): + try: + import yaml + with open(path) as f: + return yaml.safe_load(f) or {} + except Exception: + pass + return {} + + +def _get_memo_config(): + """Get MemoConfig from agents.yaml.""" + from adapters.memo.config import MemoConfig + return MemoConfig.from_yaml(_load_config()) + + +# ══════════════════════════════════════════════════════════════════════════════ +# Main dispatcher +# ══════════════════════════════════════════════════════════════════════════════ + +def cmd_memo( + action: str = "status", + query: Optional[str] = None, + agent: Optional[str] = None, + memo_type: Optional[str] = None, + since: Optional[str] = None, + until: Optional[str] = None, + min_quality: float = 0.6, + min_score: int = 7, + output: Optional[str] = None, + upload: bool = False, + dry_run: bool = False, +): + """Dispatch ``cleo memo ``.""" + console = _get_console() + + if action == "status": + _memo_status(console) + elif action == "export": + _memo_export(console, agent=agent, memo_type=memo_type, + since=since, until=until, + min_quality=min_quality, min_score=min_score, + output=output, upload=upload, dry_run=dry_run) + elif action == "search": + _memo_search(console, query=query, memo_type=memo_type, + min_quality=min_quality) + elif action == "skills": + _memo_skills(console) + elif action == "tracking": + _memo_tracking(console) + else: + if console: + console.print(f"[{_theme.error}]Unknown action: {action}[/{_theme.error}]") + else: + print(f"Unknown action: {action}") + + +# ══════════════════════════════════════════════════════════════════════════════ +# Actions +# ══════════════════════════════════════════════════════════════════════════════ + +def _memo_status(console): + """Show Memo integration status.""" + config = _get_memo_config() + + if console: + console.print(f"\n[{_theme.heading}]🧠 Memo Protocol Integration[/{_theme.heading}]\n") + console.print(f" Enabled: {'✓ Yes' if config.enabled else '✗ No'}") + console.print(f" API Base: {config.api_base_url}") + console.print(f" Agent ID: {config.erc8004_agent_id or '(not set)'}") + console.print(f" Wallet: {config.wallet_address[:10] + '...' if config.wallet_address else '(not set)'}") + console.print(f" Auto Upload: {'✓' if config.auto_upload_enabled else '✗'}") + console.print(f" Skill Sync: {'✓' if config.skill_sync_enabled else '✗'}") + console.print(f" LLM Deident: {'✓' if config.deidentification_use_llm else '✗'}") + console.print(f" Domain: {config.default_domain}") + console.print(f" Language: {config.default_language}") + + # Tracking stats + try: + from adapters.memo.tracking import ExportTracker + tracker = ExportTracker() + stats = tracker.stats() + console.print(f"\n [{_theme.heading}]Export Tracking[/{_theme.heading}]") + console.print(f" Total Exported: {stats['total']}") + if stats["by_type"]: + for t, c in stats["by_type"].items(): + console.print(f" {t}: {c}") + except Exception: + pass + + # Local skills + try: + from adapters.memo.importer import MemoImporter + skills_dir = os.path.join("skills", "memo") + if os.path.isdir(skills_dir): + count = len([f for f in os.listdir(skills_dir) + if f.endswith(".md")]) + console.print(f"\n [{_theme.heading}]Local Memo Skills[/{_theme.heading}]") + console.print(f" Files: {count}") + except Exception: + pass + else: + print(f"Memo enabled: {config.enabled}") + print(f"API: {config.api_base_url}") + + +def _memo_export(console, *, agent, memo_type, since, until, + min_quality, min_score, output, upload, dry_run): + """Export memories to Memo format.""" + config = _get_memo_config() + + from adapters.memo.exporter import MemoExporter, ExportFilter + + filt = ExportFilter( + agents=[agent] if agent else [], + types=[memo_type] if memo_type else [], + date_from=since or "", + date_to=until or "", + min_score=min_score, + min_quality=min_quality, + ) + + output_dir = output or "memo_export" + mode = "DRY RUN" if dry_run else ("EXPORT + UPLOAD" if upload else "EXPORT") + + if console: + console.print(f"\n[{_theme.heading}]🧠 Memo Export ({mode})[/{_theme.heading}]\n") + if agent: + console.print(f" Agent: {agent}") + if memo_type: + console.print(f" Type: {memo_type}") + if since or until: + console.print(f" Date range: {since or '...'} → {until or '...'}") + console.print(f" Min quality: {min_quality}") + console.print(f" Min score: {min_score}") + console.print(f" Output: {output_dir}") + console.print() + + exporter = MemoExporter(config) + + try: + result = asyncio.run( + exporter.export_batch(filt, output_dir=output_dir, + upload=upload, dry_run=dry_run)) + except Exception as e: + if console: + console.print(f"[{_theme.error}]Export failed: {e}[/{_theme.error}]") + else: + print(f"Export failed: {e}") + return + + if console: + console.print(f"[{_theme.success}]✓[/{_theme.success}] Export complete\n") + console.print(f" Scanned: {result.total_scanned}") + console.print(f" Eligible: {result.total_eligible}") + console.print(f" Exported: {result.total_exported}") + console.print(f" Skipped (quality): {result.skipped_quality}") + console.print(f" Skipped (dup): {result.skipped_duplicate}") + if result.skipped_error: + console.print(f" Skipped (error): {result.skipped_error}") + if result.by_type: + console.print(f"\n [{_theme.heading}]By Type[/{_theme.heading}]") + for t, c in result.by_type.items(): + console.print(f" {t}: {c}") + console.print(f"\n Duration: {result.duration_seconds}s") + if not dry_run: + console.print(f" Output: {result.output_path}/") + if result.errors: + console.print(f"\n [{_theme.warning}]Errors ({len(result.errors)}):[/{_theme.warning}]") + for err in result.errors[:5]: + console.print(f" {err}") + else: + print(f"Exported {result.total_exported}/{result.total_scanned} " + f"({result.skipped_quality} skipped quality, " + f"{result.skipped_duplicate} duplicates)") + + +def _memo_search(console, *, query, memo_type, min_quality): + """Search the Memo platform.""" + if not query: + if console: + console.print(f"[{_theme.error}]Usage: cleo memo search [/{_theme.error}]") + return + + config = _get_memo_config() + if not config.enabled: + if console: + console.print(f"[{_theme.warning}]Memo integration is disabled. " + f"Enable it in config/agents.yaml[/{_theme.warning}]") + return + + from adapters.memo.client import MemoClient + client = MemoClient(config) + + try: + results = asyncio.run( + client.search_memories(query, type=memo_type or "", + min_quality=min_quality)) + except Exception as e: + if console: + console.print(f"[{_theme.error}]Search failed: {e}[/{_theme.error}]") + return + + if console: + console.print(f"\n[{_theme.heading}]🔍 Memo Search: \"{query}\"[/{_theme.heading}]\n") + if not results: + console.print(f" [{_theme.muted}]No results found[/{_theme.muted}]") + for i, r in enumerate(results[:10], 1): + title = r.get("title", "Untitled") + rtype = r.get("type", "?") + score = r.get("quality_score", 0) + mid = r.get("id", "?") + console.print(f" {i}. [{_theme.heading}]{title}[/{_theme.heading}]") + console.print(f" Type: {rtype} Quality: {score:.2f} ID: {mid}") + summary = r.get("summary", "") + if summary: + console.print(f" {summary[:120]}") + console.print() + + +def _memo_skills(console): + """Sync purchased skills from Memo.""" + config = _get_memo_config() + if not config.enabled: + if console: + console.print(f"[{_theme.warning}]Memo integration is disabled[/{_theme.warning}]") + return + + from adapters.memo.client import MemoClient + from adapters.memo.importer import MemoImporter + + client = MemoClient(config) + importer = MemoImporter(config, client) + + if console: + console.print(f"\n[{_theme.heading}]🧠 Memo Skill Sync[/{_theme.heading}]\n") + + try: + stats = asyncio.run(importer.sync_skills()) + except Exception as e: + if console: + console.print(f"[{_theme.error}]Sync failed: {e}[/{_theme.error}]") + return + + if console: + console.print(f" Fetched: {stats['fetched']}") + console.print(f" Written: {stats['written']}") + console.print(f" Updated: {stats['updated']}") + if stats["errors"]: + console.print(f" Errors: {stats['errors']}") + + # List local skills + local = importer.list_local_skills() + if console and local: + console.print(f"\n [{_theme.heading}]Local Memo Skills ({len(local)})[/{_theme.heading}]") + for s in local[:10]: + console.print(f" • {s.get('name', s.get('filename', '?'))}") + + +def _memo_tracking(console): + """Show export tracking records.""" + from adapters.memo.tracking import ExportTracker + + tracker = ExportTracker() + stats = tracker.stats() + + if console: + console.print(f"\n[{_theme.heading}]📋 Memo Export Tracking[/{_theme.heading}]\n") + console.print(f" Total Exports: {stats['total']}") + if stats["by_type"]: + console.print(f"\n [{_theme.heading}]By Source Type[/{_theme.heading}]") + for t, c in stats["by_type"].items(): + console.print(f" {t}: {c}") + + if stats.get("created_at"): + from datetime import datetime + created = datetime.fromtimestamp(stats["created_at"]) + console.print(f"\n Created: {created.strftime('%Y-%m-%d %H:%M')}") + if stats.get("updated_at"): + from datetime import datetime + updated = datetime.fromtimestamp(stats["updated_at"]) + console.print(f" Updated: {updated.strftime('%Y-%m-%d %H:%M')}") + + console.print(f"\n Tracking file: {tracker.path}") + else: + print(f"Total exports: {stats['total']}") + for t, c in stats.get("by_type", {}).items(): + print(f" {t}: {c}") diff --git a/core/agent.py b/core/agent.py index 044ec3b..754a854 100644 --- a/core/agent.py +++ b/core/agent.py @@ -438,6 +438,16 @@ async def run(self, task: "Task", bus: "ContextBus", # 6b. Strip ... blocks from model output result = _strip_think(result) + # 6c. Guard: empty LLM response (content filter, all retries exhausted) + if not result.strip(): + logger.warning("[%s] LLM returned empty result — injecting fallback", + self.cfg.agent_id) + result = ( + "I'm sorry, I wasn't able to generate a response for this " + "request. This may be due to content filtering. " + "Please try rephrasing your question." + ) + # 7. Tool execution loop — parse tool calls, execute, feed back results if tools_cfg: result = await self._tool_loop(messages, task, result, @@ -551,20 +561,37 @@ async def _call_llm_streaming(self, messages: list[dict], task: "Task", from core.task_board import TaskBoard board = TaskBoard() chunks: list[str] = [] + seq = 0 update_interval = 0 async for chunk in self.llm.chat_stream( messages, self.cfg.model, **llm_kwargs ): chunks.append(chunk) + seq += 1 + # Append chunk to per-task stream file (lockless, fast) + try: + TaskBoard.append_stream_chunk( + task.task_id, chunk, seq) + except Exception: + pass # non-critical: SSE just won't get this chunk update_interval += 1 - # Write partial result every 5 chunks to avoid excessive I/O - if update_interval >= 5: - board.update_partial(task.task_id, "".join(chunks)) + # Still update TaskBoard partial_result periodically + # for backward compat (WS gateway, HTTP polling) + if update_interval >= 20: + board.update_partial( + task.task_id, "".join(chunks)) update_interval = 0 result = "".join(chunks) # Final partial update (will be cleared when task completes) if chunks: board.update_partial(task.task_id, result) + # Detect empty streaming result (content filter / API issue) + if not result.strip(): + logger.warning( + "[%s] streaming returned empty result, " + "falling back to non-streaming", + self.cfg.agent_id) + raise RuntimeError("Empty streaming result") return result except Exception as e: logger.warning("[%s] streaming failed, falling back to blocking: %s", @@ -984,11 +1011,14 @@ def _store_to_memory(self, task: "Task", result: str, if self.episodic: try: from adapters.memory.episodic import make_episode + # Derive baseline score from outcome when no explicit score + _score = {"success": 8, "partial": 5}.get(outcome, 2) episode = make_episode( agent_id=self.cfg.agent_id, task_id=task.task_id, task_description=task.description, result=result, + score=_score, outcome=outcome, error_type=error_type, model=getattr(self.cfg, "model", None), diff --git a/core/cron.py b/core/cron.py index e36e9f1..68f02b8 100644 --- a/core/cron.py +++ b/core/cron.py @@ -251,9 +251,10 @@ def _run(): elif action == "exec": # Run shell command import subprocess + cmd_timeout = job.get("timeout", DEFAULT_JOB_TIMEOUT) result = subprocess.run( payload, shell=True, capture_output=True, - text=True, timeout=300) + text=True, timeout=cmd_timeout) if result.returncode == 0: return True, result.stdout[:500] or "(no output)" else: @@ -312,14 +313,21 @@ def _scheduler_tick(): prev_thread = _running_jobs.get(jid) if prev_thread and prev_thread.is_alive(): timeout = job.get("timeout", DEFAULT_JOB_TIMEOUT) - # Check watchdog — if running longer than timeout, log warning + # Check watchdog — if running longer than timeout, force-kill started = getattr(prev_thread, "_cron_started", 0) elapsed = time.time() - started if started else 0 if elapsed > timeout: logger.warning( "Cron job %s (%s) exceeded timeout (%.0fs > %ds), " - "skipping new run", + "force-terminating stale thread", jid, job["name"], elapsed, timeout) + # Remove from running jobs so next tick can start fresh + _running_jobs.pop(jid, None) + job["last_error"] = ( + f"terminated: exceeded {timeout}s timeout " + f"(ran {elapsed:.0f}s)") + changed = True + # Don't re-execute this tick — let next tick handle it else: logger.debug("Cron job %s still running (%.0fs), skipping", jid, elapsed) diff --git a/core/dashboard.html b/core/dashboard.html index 6dbb0b9..7203ae7 100644 --- a/core/dashboard.html +++ b/core/dashboard.html @@ -111,6 +111,23 @@ gap:8px; } .sidebar-logo span{color:var(--accent)} +/* ── Session Sidebar ── */ +.session-section{border-bottom:1px solid rgba(0,0,0,0.06);padding:4px 0 6px;max-height:260px;display:flex;flex-direction:column} +.session-header{display:flex;align-items:center;justify-content:space-between;padding:4px 14px 4px} +.session-header-title{font-size:10px;font-weight:700;text-transform:uppercase;letter-spacing:1px;color:var(--fg3)} +.session-new-btn{background:none;border:1px solid rgba(0,0,0,0.1);border-radius:6px;padding:3px 6px;cursor:pointer;color:var(--fg2);display:flex;align-items:center;transition:all .15s} +.session-new-btn:hover{background:rgba(0,122,255,0.08);border-color:var(--accent);color:var(--accent)} +.session-list{overflow-y:auto;flex:1;padding:0 6px} +.session-item{padding:7px 10px;border-radius:8px;cursor:pointer;font-size:12px;color:var(--fg2);display:flex;align-items:center;gap:6px;transition:background .15s;position:relative} +.session-item:hover{background:rgba(0,0,0,0.04)} +.session-item.active{background:rgba(0,122,255,0.08);color:var(--accent);font-weight:600} +.session-title{flex:1;overflow:hidden;text-overflow:ellipsis;white-space:nowrap} +.session-time{font-size:10px;color:var(--fg3);flex-shrink:0} +.session-actions{display:none;gap:4px;flex-shrink:0} +.session-item:hover .session-actions{display:flex} +.session-item:hover .session-time{display:none} +.session-action-btn{background:none;border:none;cursor:pointer;padding:2px;font-size:11px;color:var(--fg3);border-radius:4px} +.session-action-btn:hover{background:rgba(0,0,0,0.08);color:var(--fg)} .status-dot{width:8px;height:8px;border-radius:50%;background:var(--fg3);display:inline-block} .status-dot.online{background:var(--green);box-shadow:0 0 8px var(--green),0 0 16px rgba(52,199,89,0.25)} .status-dot.offline{background:var(--red);box-shadow:0 0 8px var(--red),0 0 16px rgba(255,59,48,0.20)} @@ -499,6 +516,13 @@ } .modal-tab-content{display:none} .modal-tab-content.active{display:block} +.tab-badge{ + display:inline-block;font-size:10px;font-weight:600; + background:var(--accent);color:#fff;border-radius:8px; + padding:1px 6px;margin-left:4px;min-width:16px;text-align:center; + font-family:var(--mono);vertical-align:middle;line-height:1.4; +} +.tab-badge-secondary{background:var(--bg4);color:var(--fg2)} /* ── Overview tab ── */ .overview-section{margin-bottom:14px} @@ -826,6 +850,102 @@ /* ── Empty State ──────────────────────────────────────────────── */ .empty{text-align:center;color:var(--fg3);padding:30px 0;font-size:13px} +/* ── Knowledge Base Notes ─────────────────────────────────────── */ +.kb-grid{display:grid;grid-template-columns:repeat(2,1fr);gap:14px} +.kb-note{ + background:rgba(255,255,255,0.55); + border:1px solid rgba(0,0,0,0.06); + border-radius:14px;padding:16px 18px; + transition:all .2s ease;position:relative;overflow:hidden; +} +.kb-note:hover{ + border-color:rgba(0,0,0,0.10); + box-shadow:0 4px 16px rgba(0,0,0,0.06); + transform:translateY(-1px); +} +.kb-note-head{display:flex;align-items:flex-start;gap:10px;margin-bottom:10px} +.kb-note-icon{ + width:32px;height:32px;border-radius:8px; + display:flex;align-items:center;justify-content:center; + font-size:15px;flex-shrink:0; + background:linear-gradient(135deg,var(--accent),#7c3aed); + color:#fff; +} +.kb-note-title{font-size:14px;font-weight:600;color:var(--fg1);line-height:1.35;flex:1} +.kb-note-tags{display:flex;flex-wrap:wrap;gap:5px;margin-bottom:10px} +.kb-tag{ + padding:2px 8px;border-radius:6px;font-size:10px;font-weight:500; + letter-spacing:.3px; +} +.kb-tag-blue{background:rgba(59,130,246,0.12);color:#3b82f6} +.kb-tag-purple{background:rgba(139,92,246,0.12);color:#8b5cf6} +.kb-tag-green{background:rgba(34,197,94,0.12);color:#16a34a} +.kb-tag-orange{background:rgba(249,115,22,0.12);color:#ea580c} +.kb-tag-pink{background:rgba(236,72,153,0.12);color:#db2777} +.kb-tag-cyan{background:rgba(6,182,212,0.12);color:#0891b2} +.kb-note-body{ + font-size:12.5px;line-height:1.65;color:var(--fg2); + max-height:200px;overflow-y:auto;position:relative; + scrollbar-width:thin;scrollbar-color:var(--bg4) transparent; +} +.kb-note-body::-webkit-scrollbar{width:5px} +.kb-note-body::-webkit-scrollbar-track{background:transparent} +.kb-note-body::-webkit-scrollbar-thumb{background:var(--bg4);border-radius:3px} +.kb-note-body h3{font-size:13px;font-weight:600;color:var(--fg1);margin:10px 0 4px} +.kb-note-body h4{font-size:12px;font-weight:600;color:var(--fg2);margin:8px 0 3px} +.kb-note-body pre{ + background:var(--bg3);border-radius:8px;padding:10px 12px; + font-size:11px;font-family:var(--mono);overflow-x:auto; + margin:6px 0;line-height:1.5; +} +.kb-note-body code{ + background:var(--bg3);padding:1px 5px;border-radius:4px; + font-size:11px;font-family:var(--mono); +} +.kb-note-body ul,.kb-note-body ol{margin:4px 0;padding-left:18px} +.kb-note-body li{margin:2px 0} +.kb-note-body p{margin:4px 0} +.kb-note-body strong{color:var(--fg1)} +.kb-note-body hr{border:none;border-top:1px solid var(--bg4);margin:10px 0} +.kb-note-footer{ + display:flex;justify-content:space-between;align-items:center; + margin-top:10px;padding-top:8px;border-top:1px solid rgba(0,0,0,0.05); + font-size:10px;color:var(--fg3); +} +.kb-note-footer .contributor{color:var(--accent);font-weight:500} +.kb-density{ + padding:1px 6px;border-radius:4px;font-size:9px; + font-weight:600;letter-spacing:.3px;text-transform:uppercase; +} +.kb-density-high{background:rgba(239,68,68,0.10);color:#ef4444} +.kb-density-normal{background:rgba(59,130,246,0.10);color:#3b82f6} +.kb-density-low{background:rgba(34,197,94,0.10);color:#22c55e} + +/* ── Daily Log Content ─────────────────────────────────────────── */ +.daily-log-content{font-size:13px;line-height:1.7;color:var(--fg2);max-height:500px;overflow-y:auto;scrollbar-width:thin;scrollbar-color:var(--bg4) transparent} +.daily-log-content::-webkit-scrollbar{width:5px} +.daily-log-content::-webkit-scrollbar-track{background:transparent} +.daily-log-content::-webkit-scrollbar-thumb{background:var(--bg4);border-radius:3px} +.daily-log-content h3{font-size:14px;font-weight:600;color:var(--fg1);margin:12px 0 4px} +.daily-log-content h4{font-size:12.5px;font-weight:600;color:var(--fg2);margin:8px 0 3px} +.daily-log-content p{margin:4px 0} +.daily-log-content strong{color:var(--fg1)} +.daily-log-content hr{border:none;border-top:1px solid var(--bg4);margin:10px 0} +.daily-log-content pre{background:var(--bg3);border-radius:8px;padding:10px 12px;font-size:11px;font-family:var(--mono);overflow-x:auto;margin:6px 0;line-height:1.5} +.daily-log-content code{background:var(--bg3);padding:1px 5px;border-radius:4px;font-size:11px;font-family:var(--mono)} +.daily-log-content ul,.daily-log-content ol{margin:4px 0;padding-left:18px} +.daily-log-content li{margin:2px 0} + +/* Dark mode overrides */ +[data-theme="dark"] .kb-note{background:rgba(255,255,255,0.04);border-color:rgba(255,255,255,0.06)} +[data-theme="dark"] .kb-note:hover{border-color:rgba(255,255,255,0.12);box-shadow:0 4px 16px rgba(0,0,0,0.3)} +[data-theme="dark"] .kb-note-body pre{background:rgba(0,0,0,0.3)} +[data-theme="dark"] .kb-note-body code{background:rgba(0,0,0,0.25)} +[data-theme="dark"] .kb-note-body::-webkit-scrollbar-thumb{background:rgba(255,255,255,0.15)} +[data-theme="dark"] .daily-log-content pre{background:rgba(0,0,0,0.3)} +[data-theme="dark"] .daily-log-content code{background:rgba(0,0,0,0.25)} +[data-theme="dark"] .daily-log-content::-webkit-scrollbar-thumb{background:rgba(255,255,255,0.15)} + /* ── Toast ────────────────────────────────────────────────────── */ .toast{ position:fixed;bottom:24px;right:24px; @@ -869,7 +989,7 @@ .dialog .actions{display:flex;gap:8px;justify-content:flex-end;margin-top:16px} /* ── Overview: Split Layout (top orchestration + bottom chat) ── */ -#panel-overview.active{display:flex!important;flex-direction:column;height:calc(100vh - 44px);gap:0} +#panel-overview.active{display:flex!important;flex-direction:column;height:calc(100vh - 44px);gap:0;position:relative} /* ── Overview: Header Bar (statuses left, agents right) ────── */ .ov-header{ @@ -1267,6 +1387,21 @@ .stream-meta .stream-agent-status{color:var(--cyan);font-weight:600;display:flex;align-items:center;gap:4px} .stream-meta .stream-agent-status::before{content:'';width:6px;height:6px;border-radius:50%;background:var(--cyan);animation:statusPulse 1.5s ease-in-out infinite} @keyframes statusPulse{0%,100%{opacity:1}50%{opacity:.3}} +/* ── Artifacts sidebar ── */ +.artifacts-sidebar{position:absolute;right:0;top:0;bottom:0;width:400px;max-width:50%;background:var(--bg);border-left:1px solid var(--border);display:flex;flex-direction:column;z-index:50;box-shadow:-4px 0 20px rgba(0,0,0,0.15)} +.artifacts-header{padding:12px 16px;font-size:14px;font-weight:700;color:var(--fg);border-bottom:1px solid var(--border);display:flex;align-items:center;justify-content:space-between} +.artifacts-close{background:none;border:none;color:var(--fg3);cursor:pointer;font-size:20px;padding:0 4px;line-height:1} +.artifacts-close:hover{color:var(--fg)} +.artifacts-list{flex:1;overflow-y:auto;padding:12px} +.artifact-card{background:var(--bg1);border:1px solid var(--border);border-radius:10px;margin-bottom:12px;overflow:hidden} +.artifact-card-header{padding:10px 14px;font-size:12px;font-weight:600;color:var(--fg2);display:flex;align-items:center;gap:8px;border-bottom:1px solid var(--border);cursor:pointer} +.artifact-card-header:hover{background:rgba(0,0,0,0.03)} +.artifact-card-body{padding:12px 14px;font-size:12px;line-height:1.6} +.artifact-card-body pre{margin:0;padding:10px;background:var(--bg2);border-radius:6px;overflow-x:auto;font-size:11px;line-height:1.5} +.artifact-card-body code{font-family:var(--mono)} +/* ── Think toggle ── */ +.think-toggle{cursor:pointer;text-decoration:underline;text-decoration-style:dotted} +.think-toggle:hover{color:var(--accent)} .chat-bubble.assistant-bubble .asst-body{word-break:break-word;color:var(--fg)} .chat-bubble.assistant-bubble .asst-body p{margin:6px 0} .chat-bubble.assistant-bubble .asst-body h1,.chat-bubble.assistant-bubble .asst-body h2,.chat-bubble.assistant-bubble .asst-body h3,.chat-bubble.assistant-bubble .asst-body h4,.chat-bubble.assistant-bubble .asst-body h5,.chat-bubble.assistant-bubble .asst-body h6{color:var(--fg);margin:16px 0 8px;font-weight:700} @@ -1516,6 +1651,7 @@ /* ── Responsive ───────────────────────────────────────────────── */ @media(max-width:800px){ .sidebar{width:56px;min-width:56px} + .session-section{display:none} .nav-section-title,.nav-item span:not(.icon),.sidebar-footer,.sidebar-logo span,.sidebar-logo div:not(:first-child){display:none} .nav-item{justify-content:center;padding:10px 0} .nav-item .icon{margin:0} @@ -1530,6 +1666,7 @@ .chat-bubble.assistant-bubble .asst-body table{font-size:11px} .chat-bubble.assistant-bubble .asst-body th,.chat-bubble.assistant-bubble .asst-body td{padding:4px 6px} .chat-msg{max-width:92%} + .kb-grid{grid-template-columns:1fr} } /* ── Responsive: MacBook 14" (~1440-1512px) ──────────────────── */ @@ -1573,6 +1710,23 @@ + +
+
+ Sessions + +
+
+
Loading…
+
+
+
Overview @@ -1700,6 +1854,15 @@
+ + + @@ -1944,20 +2107,12 @@ - -
-
—
Episodes
-
—
Cases
-
—
Patterns
-
—
KB Notes
-
- - + @@ -2202,6 +2357,10 @@ let chatMsgCounter = 0; // unique msg id counter let dashboardUser = 'createpjf'; // default submitter name for dashboard tasks +// ── Session state ── +let currentSessionId = null; +let dashboardSessions = []; + // ══════════════════════════════════════════════════════════════════ // AUTH & API HELPERS // ══════════════════════════════════════════════════════════════════ @@ -2795,6 +2954,8 @@ text = text.replace(/[\s\S]*?(<\/tool_result>|$)/g, ''); // Strip Minimax-specific tool call XML (complete or streaming) text = text.replace(/[\s\S]*?(<\/minimax:tool_call>|$)/g, ''); + // Strip ... blocks (routed to sidebar) + text = text.replace(/]*)?>[\s\S]*?(<\/artifact>|$)/g, ''); // Strip generic XML tool patterns: , , text = text.replace(/|$)/g, ''); text = text.replace(/|$)/g, ''); @@ -3312,6 +3473,17 @@ return; } + // ── Auto-create session if none selected ── + if (!currentSessionId) { + const autoTitle = desc.length > 50 ? desc.substring(0, 47) + '…' : desc; + const ns = await apiPost('/v1/sessions', {title: autoTitle}); + if (ns && ns.session_id) { + currentSessionId = ns.session_id; + localStorage.setItem('cleo_current_session', currentSessionId); + loadSessions(); + } + } + // ── Normal mode ── let taskDesc = desc; if (files.length) { @@ -3325,11 +3497,15 @@ taskDesc += '\n\n[' + images.length + ' image(s) attached]'; } - const result = await apiPost('/v1/task', {description: taskDesc}); + const payload = {description: taskDesc}; + if (currentSessionId) payload.session_id = currentSessionId; + const result = await apiPost('/v1/task', payload); if (result && result.task_id) { // Task creation goes to TOP dispatch addDispatch('📤', 'system', 'Task Submitted: ' + result.task_id.slice(0,8) + '…', {taskId: result.task_id}); chatTaskMap[result.task_id] = new Set(['created']); + // Start SSE token-level streaming for this task + startTaskStream(result.task_id); } else { addChatMsg('system', '✗ Failed to submit task'); renderChat(); @@ -3339,6 +3515,134 @@ setTimeout(pollChatUpdates, 500); } +// ══════════════════════════════════════════════════════════════════ +// DASHBOARD SESSIONS +// ══════════════════════════════════════════════════════════════════ + +async function loadSessions() { + const data = await api('/v1/sessions'); + if (!data || !data.sessions) return; + dashboardSessions = data.sessions; + renderSessionList(); +} + +function renderSessionList() { + const container = document.getElementById('sessionList'); + if (!container) return; + if (!dashboardSessions.length) { + container.innerHTML = '
No sessions yet
'; + return; + } + const sorted = [...dashboardSessions].sort((a, b) => { + if (a.pinned !== b.pinned) return b.pinned ? 1 : -1; + return b.last_active - a.last_active; + }); + container.innerHTML = sorted.map(s => { + const isActive = s.session_id === currentSessionId; + const timeAgo = fmtTimeAgo(s.last_active); + const title = escHtml(s.title || 'New Chat'); + const sid = escHtml(s.session_id); + return '
' + + '' + (s.pinned ? '📌 ' : '') + title + '' + + '' + timeAgo + '' + + '' + + '' + + '' + + '
'; + }).join(''); +} + +async function createNewSession() { + const data = await apiPost('/v1/sessions', {title: ''}); + if (!data || !data.session_id) return; + currentSessionId = data.session_id; + localStorage.setItem('cleo_current_session', currentSessionId); + resetChatState(); + await loadSessions(); + switchPanel('overview'); +} + +async function switchSession(sessionId) { + if (sessionId === currentSessionId) return; + currentSessionId = sessionId; + localStorage.setItem('cleo_current_session', sessionId); + + // Load session messages from backend + const data = await api('/v1/sessions/' + encodeURIComponent(sessionId)); + if (!data) return; + + resetChatState(); + if (data.messages && data.messages.length) { + for (const msg of data.messages) { + addChatMsg(msg.role === 'user' ? 'user' : 'assistant', msg.content, { + ts: msg.ts ? msg.ts * 1000 : Date.now(), + agent: msg.role === 'assistant' ? 'cleo' : undefined + }); + } + } + renderChat(); + renderSessionList(); + switchPanel('overview'); +} + +function resetChatState() { + stopAllTaskStreams(); + clearArtifacts(); + chatMsgs = []; + chatPrevSnapshot = {}; + chatTaskMap = {}; + chatMsgCounter = 0; + dispatchEntries = []; + dispatchLastIdx = 0; + const chatEl = document.getElementById('chatMessages'); + if (chatEl) { + chatEl.innerHTML = '
⬡
' + + '
What can I help you with?
' + + '
Your task will be planned, executed, and reviewed by the Cleo agent team
'; + } + const dispEl = document.getElementById('dispatchLog'); + if (dispEl) dispEl.innerHTML = ''; +} + +async function renameSession(sessionId) { + const current = dashboardSessions.find(s => s.session_id === sessionId); + const newTitle = prompt('Rename session:', current?.title || 'New Chat'); + if (newTitle === null) return; + await apiPut('/v1/sessions/' + encodeURIComponent(sessionId), {title: newTitle}); + await loadSessions(); +} + +async function deleteSession(sessionId) { + if (!confirm('Delete this session?')) return; + await apiDelete('/v1/sessions/' + encodeURIComponent(sessionId)); + if (currentSessionId === sessionId) { + currentSessionId = null; + localStorage.removeItem('cleo_current_session'); + resetChatState(); + } + await loadSessions(); +} + +function fmtTimeAgo(ts) { + if (!ts) return ''; + const diff = (Date.now() / 1000) - ts; + if (diff < 60) return 'now'; + if (diff < 3600) return Math.floor(diff / 60) + 'm'; + if (diff < 86400) return Math.floor(diff / 3600) + 'h'; + if (diff < 604800) return Math.floor(diff / 86400) + 'd'; + return new Date(ts * 1000).toLocaleDateString(); +} + +// Save assistant message to session when task completes +function persistAssistantMessage(content) { + if (!currentSessionId || !content) return; + apiPost('/v1/sessions/' + encodeURIComponent(currentSessionId) + '/message', { + role: 'assistant', content: content + }); +} + // ══════════════════════════════════════════════════════════════════ // POLL & DIFF — routes to dual panels // ══════════════════════════════════════════════════════════════════ @@ -3436,6 +3740,71 @@ container.scrollTop = container.scrollHeight; } +// ══════════════════════════════════════════════════════════════════ +// SSE TOKEN-LEVEL STREAMING (EventSource per task) +// ══════════════════════════════════════════════════════════════════ + +const _taskStreams = {}; // task_id → EventSource +const _taskStreamText = {}; // task_id → accumulated text + +function startTaskStream(taskId) { + if (_taskStreams[taskId]) return; + const url = API + '/v1/stream/' + taskId + + (TOKEN ? '?token=' + encodeURIComponent(TOKEN) : ''); + const es = new EventSource(url); + _taskStreamText[taskId] = ''; + + es.addEventListener('chunk', function(e) { + try { + const data = JSON.parse(e.data); + _taskStreamText[taskId] += data.c; + // Determine agent for this task from cached allTasks + const agent = (allTasks[taskId] && allTasks[taskId].agent_id) || 'system'; + updateStreamingBubble(taskId, agent, _taskStreamText[taskId]); + } catch(err) { /* skip malformed chunk */ } + }); + + es.addEventListener('done', function() { + es.close(); + delete _taskStreams[taskId]; + delete _taskStreamText[taskId]; + }); + + es.addEventListener('timeout', function() { + es.close(); + delete _taskStreams[taskId]; + delete _taskStreamText[taskId]; + }); + + es.onerror = function() { + es.close(); + delete _taskStreams[taskId]; + // Reconnect after 2 s if task is still active + setTimeout(function() { + if (!_taskStreams[taskId] && allTasks[taskId] + && !['completed','failed','cancelled'].includes(allTasks[taskId].status)) { + startTaskStream(taskId); + } + }, 2000); + }; + + _taskStreams[taskId] = es; +} + +function stopTaskStream(taskId) { + if (_taskStreams[taskId]) { + _taskStreams[taskId].close(); + delete _taskStreams[taskId]; + } + delete _taskStreamText[taskId]; +} + +function stopAllTaskStreams() { + for (const tid of Object.keys(_taskStreams)) { + stopTaskStream(tid); + } +} + // ══════════════════════════════════════════════════════════════════ // STREAMING — live partial results in ChatBox // ══════════════════════════════════════════════════════════════════ @@ -3493,7 +3862,8 @@ for (const t of thinkComplete) { const content = t.replace(/<\/?think>/g, '').trim(); const preview = content.length > 80 ? content.slice(0, 80) + '…' : content; - steps.push({type: 'think', status: 'done', label: 'Thought', preview}); + steps.push({type: 'think', status: 'done', label: 'Thought', preview, + fullContent: content}); } } @@ -3537,9 +3907,43 @@ const toolName = nameMatch ? titleCase(nameMatch[1].replace(/[_-]/g, ' ')) : 'Tool'; steps.push({type: 'tool', status: 'active', label: 'Calling ' + toolName + '…'}); } + + // ── Completed ... ── + const artifactComplete = partial.match(/]*)?>[\s\S]*?<\/artifact>/g); + if (artifactComplete) { + for (const a of artifactComplete) { + const typeMatch = a.match(/type="([^"]+)"/); + const titleMatch = a.match(/title="([^"]+)"/); + const aType = typeMatch ? typeMatch[1] : 'code'; + const aTitle = titleMatch ? titleMatch[1] : 'Artifact'; + const inner = a.replace(/]*>/, '').replace(/<\/artifact>/, '').trim(); + steps.push({type: 'artifact', status: 'done', label: aTitle, + artifactType: aType, content: inner}); + } + } + // ── Streaming (open, not closed) ── + const hasOpenArtifact = /]*)?>(?![\s\S]*<\/artifact>)[\s\S]*$/.test(partial); + if (hasOpenArtifact) { + steps.push({type: 'artifact', status: 'active', label: 'Creating artifact…'}); + } + return steps; } +// ── Sanitize partial markdown for safe rendering during streaming ── +function _sanitizePartialMarkdown(text) { + // Close unclosed code fences + const fenceCount = (text.match(/^```/gm) || []).length; + if (fenceCount % 2 !== 0) text += '\n```'; + // Close unclosed inline code backticks + const backtickCount = (text.match(/`/g) || []).length; + if (backtickCount % 2 !== 0) text += '`'; + // Close unclosed bold markers + const boldCount = (text.match(/\*\*/g) || []).length; + if (boldCount % 2 !== 0) text += '**'; + return text; +} + function _renderStreamContent(taskId) { const bubble = document.getElementById('stream-' + taskId); if (!bubble) return; @@ -3548,18 +3952,39 @@ const statusEl = bubble.querySelector('.stream-agent-status'); const partial = _streamingFullText[taskId] || ''; if (body) { - const cleaned = htmlToMd(stripLlmTagsCached(partial)); + const cleaned = _sanitizePartialMarkdown(htmlToMd(stripLlmTagsCached(partial))); const steps = _parseStreamSteps(partial); let stepsHtml = ''; if (steps.length) { stepsHtml = '
'; - for (const s of steps) { - const icon = s.status === 'done' - ? (s.type === 'think' ? '💭' : '✓') - : (s.type === 'think' ? '💭' : '⚡'); - const cls = s.status === 'done' ? 'step-done' : 'step-active'; - const previewHtml = (s.preview && s.status === 'done') - ? '' + escHtml(s.preview) + '' : ''; + for (let si = 0; si < steps.length; si++) { + const s = steps[si]; + let icon, cls; + if (s.type === 'artifact') { + icon = s.status === 'done' ? '📎' : '📝'; + } else if (s.type === 'think') { + icon = '💭'; + } else { + icon = s.status === 'done' ? '✓' : '⚡'; + } + cls = s.status === 'done' ? 'step-done' : 'step-active'; + let previewHtml = ''; + if (s.preview && s.status === 'done') { + if (s.type === 'think' && s.fullContent) { + // Expandable think block + const thinkId = 'think-' + taskId + '-' + si; + previewHtml = '' + + escHtml(s.preview) + ' ▸' + + ''; + } else { + previewHtml = '' + escHtml(s.preview) + ''; + } + } + // Route completed artifacts to sidebar + if (s.type === 'artifact' && s.status === 'done' && s.content) { + showArtifact(s.artifactType || 'code', s.label, s.content); + } stepsHtml += '
' + '' + icon + '' + '' + escHtml(s.label) + '' @@ -3622,6 +4047,77 @@ } } +// ── Think block expand/collapse ── +function toggleThinkContent(thinkId) { + const el = document.getElementById(thinkId); + if (!el) return; + el.style.display = el.style.display === 'none' ? 'block' : 'none'; +} + +// ── Artifacts sidebar ── +let _artifactCounter = 0; + +function showArtifact(type, title, content) { + const sidebar = document.getElementById('artifactsSidebar'); + const list = document.getElementById('artifactsList'); + if (!sidebar || !list) return; + + _artifactCounter++; + const cardId = 'artifact-' + _artifactCounter; + + // Check if an artifact with same title already exists (update it) + const existing = list.querySelector('[data-artifact-title="' + escHtml(title) + '"]'); + if (existing) { + const body = existing.querySelector('.artifact-card-body'); + if (body) { + body.innerHTML = (type === 'code' || type === 'application/code') + ? '
' + escHtml(content) + '
' + : renderMarkdown(content); + } + sidebar.style.display = 'flex'; + return; + } + + const card = document.createElement('div'); + card.className = 'artifact-card'; + card.setAttribute('data-artifact-title', title); + + const icon = type === 'code' || type === 'application/code' ? '💻' + : type === 'markdown' || type === 'text/markdown' ? '📄' + : type === 'html' || type === 'text/html' ? '🌐' : '📎'; + + let bodyContent; + if (type === 'code' || type === 'application/code') { + bodyContent = '
' + escHtml(content) + '
'; + } else { + bodyContent = renderMarkdown(content); + } + + card.innerHTML = '
' + + '' + icon + '' + + '' + escHtml(title) + '' + + '' + escHtml(type) + '' + + '
' + + '
' + bodyContent + '
'; + + list.appendChild(card); + sidebar.style.display = 'flex'; +} + +function toggleArtifacts() { + const sidebar = document.getElementById('artifactsSidebar'); + if (!sidebar) return; + sidebar.style.display = sidebar.style.display === 'none' ? 'flex' : 'none'; +} + +function clearArtifacts() { + const list = document.getElementById('artifactsList'); + if (list) list.innerHTML = ''; + _artifactCounter = 0; + const sidebar = document.getElementById('artifactsSidebar'); + if (sidebar) sidebar.style.display = 'none'; +} + function updateCostBadge(taskId, costUsd) { // Find the completed message bubble for this task and add cost const msgs = document.querySelectorAll('[data-task-id="' + taskId + '"]'); @@ -3805,9 +4301,10 @@ if (oldStatus === newStatus && old) continue; - // Remove streaming bubble when task completes + // Remove streaming bubble + close SSE when task completes if (newStatus === 'completed' || newStatus === 'failed') { removeStreamingBubble(taskId); + stopTaskStream(taskId); } if (!chatTaskMap[taskId]) chatTaskMap[taskId] = new Set(); @@ -3907,6 +4404,8 @@ promptTokens: totalIn || null, completionTokens: totalOut || null, }); + // Persist to session history + persistAssistantMessage(task.result); } } @@ -6100,12 +6599,12 @@ results.forEach((data, i) => { (data?.episodes || []).forEach(ep => { ep._agent = agents[i]; allEps.push(ep); }); }); - allEps.sort((a, b) => (b.timestamp || 0) - (a.timestamp || 0)); + allEps.sort((a, b) => (b.ts || 0) - (a.ts || 0)); renderEpisodesTable(area, allEps, true); } else { const data = await api('/v1/memory/episodes/' + agent); const eps = (data?.episodes || []).map(ep => { ep._agent = agent; return ep; }); - eps.sort((a, b) => (b.timestamp || 0) - (a.timestamp || 0)); + eps.sort((a, b) => (b.ts || 0) - (a.ts || 0)); renderEpisodesTable(area, eps, false); } } @@ -6117,11 +6616,12 @@ } let html = '
'; if (showAgent) html += ''; - html += ''; + html += ''; html += ''; episodes.slice(0, 50).forEach((ep, i) => { const desc = escHtml((ep.task_description || ep.description || ep.task_id || '—').substring(0, 80)); - const date = ep.timestamp ? new Date(ep.timestamp * 1000).toLocaleDateString() : '—'; + const date = ep.ts ? new Date(ep.ts * 1000).toLocaleDateString() : '—'; + const timeStr = ep.ts ? new Date(ep.ts * 1000).toLocaleTimeString([], {hour:'2-digit',minute:'2-digit'}) : '—'; const outcome = ep.outcome || ep.status || '—'; const score = ep.score != null ? ep.score : '—'; const outcomeColor = outcome === 'success' ? 'var(--green)' : outcome === 'failure' ? 'var(--red)' : 'var(--fg2)'; @@ -6129,12 +6629,13 @@ if (showAgent) html += ''; html += ''; html += ''; + html += ''; html += ''; html += ''; html += ''; html += ''; // Expandable detail row - html += '
AgentTaskDateOutcomeScoreTaskDateTimeOutcomeScore
' + escHtml(capitalize(ep._agent)) + '' + desc + '' + date + '' + timeStr + '' + escHtml(outcome) + '' + score + '