Initial public release

.env.example

@@ -0,0 +1,11 @@
+OPENCLAW_AGENTS_ROOT=~/.openclaw/agents
+OPENCLAW_STATE_DIR=~/.openclaw/workspace/ops/state
+SESSION_AMPLIFIER_DB_PATH=~/.openclaw/workspace/ops/state/session_amplifier.sqlite
+AGENTS_ALLOWLIST=
+SESSION_GLOB_EXCLUDE=
+MAX_TOOLRESULT_CHARS=2000
+SPOOLER_BATCH_SIZE=100
+SPOOLER_REDACT_PATTERNS=api_key,path,base64
+TOOL_NOISE_PATTERNS=ENOENT,no output,command exited
+REVIEW_CONFIDENCE_THRESHOLD=0.5
+API_PORT=8477

.gitignore

@@ -0,0 +1,14 @@
+.env
+.env.*
+!.env.example
+__pycache__/
+*.py[cod]
+.pytest_cache/
+.mypy_cache/
+.ruff_cache/
+.venv/
+venv/
+*.db
+*.sqlite
+*.log
+/state/

Dockerfile

@@ -0,0 +1,19 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+ENV OPENCLAW_AGENTS_ROOT=/openclaw/agents
+ENV OPENCLAW_STATE_DIR=/data/session-amplifier
+ENV MAX_TOOLRESULT_CHARS=2000
+ENV SPOOLER_BATCH_SIZE=100
+ENV REVIEW_CONFIDENCE_THRESHOLD=0.5
+ENV API_PORT=8477
+
+EXPOSE 8477
+
+ENTRYPOINT ["python", "main.py"]

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Patrick
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,128 @@
+# Session Amplifier
+
+Lightweight OpenClaw sidecar for transcript spooling and session review.
+
+## What it does
+
+- **Spooler** — walks `~/.openclaw/agents/<agent>/sessions/*.jsonl`, cleans/redacts tool results, stores summaries to SQLite
+- **Reviewer** — scores sessions for quality/failure patterns, detects unused tools/skills, surfaces recommendations
+- **Snapshot adapter** — exposes canonical `openclaw.session.v1` session snapshots for dashboards/reviewers without gateway core patches
+- **API** — FastAPI server on port 8477 with 4 endpoints
+
+## Quick start
+
+```bash
+# Install deps
+pip install -r requirements.txt
+
+# Init DB
+python main.py init
+
+# Run once (spool or review)
+python main.py spool
+python main.py review
+
+# Start API server
+python main.py serve
+
+# Watch transcript files and spool on changes
+python main.py watch --interval 5
+```
+
+## Running Tests
+
+The application uses `pytest` for testing. The tests use an in-memory SQLite database and do not require filesystem access.
+
+```bash
+# Run all tests
+pytest tests/
+```
+
+## Docker / container run path
+
+```bash
+cd sidecar/session-amplifier
+docker compose up -d --build
+```
+
+Manual equivalent:
+
+```bash
+docker build -t session-amplifier sidecar/session-amplifier/
+docker run -p 8477:8477 \
+  -v ~/.openclaw:/openclaw:ro \
+  -v session_amplifier_state:/data/session-amplifier \
+  -e OPENCLAW_AGENTS_ROOT=/openclaw/agents \
+  -e OPENCLAW_STATE_DIR=/data/session-amplifier \
+  session-amplifier serve
+```
+
+Host note: the bare host Python environment may not have `uvicorn` / `fastapi` installed. The intended live-service path is the container.
+
+Gateway integration note: when OpenClaw runs in Docker, gateway-side scripts should target the sidecar by container hostname, not `localhost`. Default wrapper target is `SESSION_AMPLIFIER_BASE_URL=http://session-amplifier:8477`. The sidecar compose file joins the external `librechat_default` network so the gateway container can resolve it.
+
+## API endpoints
+
+| Method | Path | Description |
+|--------|------|-------------|
+| GET | `/health` | Container health + version |
+| POST | `/spool` | Trigger incremental spooling |
+| GET | `/review/report` | Fetch latest review report |
+| GET | `/review/skills` | Fetch skill/MCP coverage report |
+| GET | `/sessions/recent` | Recent sessions with activity/error counts |
+| GET | `/sessions/snapshots` | Recent sessions as canonical `openclaw.session.v1` snapshots |
+| GET | `/session/{id}/snapshot` | One canonical `openclaw.session.v1` snapshot |
+| GET | `/session/{id}/activity` | Normalized per-session activity feed |
+
+## Config (env)
+
+| Var | Default | Description |
+|-----|---------|-------------|
+| `OPENCLAW_AGENTS_ROOT` | `~/.openclaw/agents` | Transcript source |
+| `OPENCLAW_STATE_DIR` | `~/.openclaw/workspace/ops/state` | SQLite + artifacts output |
+| `MAX_TOOLRESULT_CHARS` | `2000` | Truncate threshold |
+| `SPOOLER_BATCH_SIZE` | `100` | DB insert batch size |
+| `REVIEW_CONFIDENCE_THRESHOLD` | `0.5` | Min confidence for recommendations |
+| `API_PORT` | `8477` | HTTP server port |
+
+## Package layout
+
+```
+session-amplifier/
+├── config.py            # Env/config loading
+├── main.py              # CLI entrypoint (init/serve/serve-watch/spool/review/watch)
+├── requirements.txt
+├── Dockerfile
+├── spooler/
+│   ├── processor.py     # JSONL → spooled rows
+│   ├── redaction.py     # API key / path / base64 redaction
+│   ├── noise_filter.py  # Drop known-noise tool output
+│   └── store.py         # SQLite read/write
+├── reviewer/
+│   ├── scorer.py        # Session quality scoring
+│   ├── pattern_detector.py  # Recurring failure detection
+│   ├── skill_analyzer.py    # MCP/skill coverage
+│   └── report.py        # Report generation + persistence
+├── api/
+│   └── routes.py        # FastAPI route handlers
+└── tests/
+    └── ...              # pytest suite
+```
+
+## Architecture notes
+
+- Reads-only from `OPENCLAW_AGENTS_ROOT`; never modifies transcripts
+- Idempotent spooling via `UNIQUE(session_id, entry_idx)` constraint
+- **Incremental spooling**: Uses tracked `last_entry_idx` file state to only parse lines appending to existing files, drastically reducing processing time overhead.
+- Reviewer is deterministic (no LLM required in v1); recommendations scored by confidence threshold
+- Watch mode is polling-based in v1 for simplicity; `python main.py watch --interval 5` or `serve-watch`
+- By default `serve-watch` / `watch` only triggers spooling. To optionally trigger the reviewer append `--review-every <N>`.
+- Manual trigger remains available for cron/recovery: `POST /spool` → wait → `GET /review/report`
+- A simple CLI-style live monitor is available at `/home/node/.openclaw/workspace/ops/scripts/session_amplifier_live_monitor.py`
+- Snapshot endpoints are read-only and sidecar-local. Rollback is removing `reviewer/session_snapshot.py`, the two route handlers/imports in `api/routes.py`, and the snapshot smoke test; no gateway config change is required unless the service was rebuilt/redeployed.
+
+## Troubleshooting
+
+- **API keys exposed in tool outputs?** Check `SPOOLER_REDACT_PATTERNS`.
+- **Database lock errors?** Multiple cron instances might be racing. Restart with a clean volume or disable concurrent spool calls.
+- **Reporting "No data"?** Ensure `OPENCLAW_AGENTS_ROOT` path exists inside the container and matches host volume bindings.

SECURITY.md

@@ -0,0 +1,5 @@
+# Security
+
+Session transcripts can contain secrets, personal data, and operational details. Use synthetic fixtures for public examples.
+
+The sidecar redacts common API-key, path, and base64 patterns, but operators should still treat source transcript directories and generated SQLite databases as private.

api/__init__.py

@@ -0,0 +1 @@
+# api package

api/routes.py

@@ -0,0 +1,355 @@
+from fastapi import FastAPI, Query, HTTPException
+from fastapi.responses import StreamingResponse
+from datetime import datetime, timezone
+from pydantic import BaseModel
+import asyncio
+import json
+import sqlite3
+
+from spooler.processor import run_spool
+from spooler.store import get_recent_sessions, get_session_activity, get_conn, get_session_summary
+from reviewer.report import generate_report, generate_skills_report
+from reviewer.skill_analyzer import find_skill_candidates
+from reviewer.intelligence import (
+    generate_context_pressure_report,
+    generate_failure_mode_report,
+    generate_intelligence_bundle,
+    generate_session_sprawl_report,
+)
+from reviewer.session_snapshot import build_recent_session_snapshots, build_session_snapshot
+from api.scheduler import claim_lock, release_lock, get_job_status
+from config import settings
+
+app = FastAPI(title="Session Amplifier", version="0.1.0")
+
+
+@app.get("/health")
+def health():
+    conn = get_conn()
+    stats = conn.execute(
+        "SELECT COUNT(*) AS entries, COUNT(DISTINCT session_id) AS sessions FROM spooled_entries"
+    ).fetchone()
+    conn.close()
+    entries = int(stats["entries"]) if stats else 0
+    sessions = int(stats["sessions"]) if stats else 0
+    return {
+        "status": "ok",
+        "version": "0.1.0",
+        "db_path": str(settings.db_path),
+        "agents_root": str(settings.openclaw_agents_root),
+        "spool_ready": entries > 0,
+        "entries": entries,
+        "sessions": sessions,
+    }
+
+
+@app.post("/spool")
+def spool():
+    count, sessions = run_spool()
+    return {"entries_spooled": count, "sessions_updated": sessions}
+
+
+@app.get("/review/report")
+def review_report(since: str | None = Query(None)):
+    try:
+        report = generate_report(since=since)
+        return report
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=str(exc))
+
+
+@app.get("/review/skills")
+def review_skills():
+    return generate_skills_report()
+
+
+@app.get("/review/skills/candidates")
+def review_skill_candidates(
+    query: str = Query(..., min_length=3, max_length=500),
+    agent_id: str | None = Query(None, max_length=100),
+    limit: int = Query(8, ge=1, le=25),
+):
+    return find_skill_candidates(query=query, agent_id=agent_id, limit=limit)
+
+
+@app.get("/sessions/recent")
+def sessions_recent(limit: int = Query(25, ge=1, le=200)):
+    return {
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "sessions": get_recent_sessions(limit),
+    }
+
+
+@app.get("/sessions/active-bulk")
+def sessions_active_bulk(
+    limit: int = Query(40, ge=1, le=200),
+    activity_limit: int = Query(200, ge=1, le=1000),
+):
+    sessions = get_recent_sessions(limit)
+    return {
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "sessions": sessions,
+        "activity": {
+            row["session_id"]: _normalize_activity_rows(get_session_activity(row["session_id"], activity_limit))
+            for row in sessions
+        },
+    }
+
+
+@app.get("/sessions/snapshots")
+def sessions_snapshots(
+    limit: int = Query(40, ge=1, le=200),
+    activity_limit: int = Query(80, ge=1, le=500),
+):
+    """Return canonical OpenClaw session snapshots for recent sessions."""
+    return build_recent_session_snapshots(limit=limit, activity_limit=activity_limit)
+
+
+@app.get("/session/{session_id}/snapshot")
+def session_snapshot(session_id: str, activity_limit: int = Query(80, ge=1, le=500)):
+    """Return a canonical OpenClaw session snapshot for one session."""
+    summary = get_session_summary(session_id)
+    if not summary:
+        raise HTTPException(status_code=404, detail="session not found")
+    return build_session_snapshot(summary, activity_limit=activity_limit)
+
+
+@app.get("/reports/session-sprawl")
+def report_session_sprawl(
+    limit: int = Query(500, ge=1, le=5000),
+    stale_days: int = Query(30, ge=1, le=3650),
+):
+    return generate_session_sprawl_report(limit=limit, stale_days=stale_days)
+
+
+@app.get("/reports/context-pressure")
+def report_context_pressure(limit: int = Query(200, ge=1, le=5000)):
+    return generate_context_pressure_report(limit=limit)
+
+
+@app.get("/reports/failure-modes")
+def report_failure_modes(limit: int = Query(200, ge=1, le=1000)):
+    return generate_failure_mode_report(limit=limit)
+
+
+@app.post("/review/run")
+def review_run(kind: str = Query("light", pattern="^(light|deep)$")):
+    return generate_intelligence_bundle(kind=kind)
+
+
+def _normalize_activity_rows(rows: list[dict]) -> list[dict]:
+    normalized = []
+    for row in rows:
+        role = row.get("role") or ""
+        tool_name = row.get("tool_name") or ""
+        clean_text = row.get("clean_text") or ""
+        is_error = bool(row.get("is_error"))
+        preview = row.get("preview") or ""
+
+        # Classify event type
+        if role == "toolResult":
+            if is_error:
+                event_type = "tool_error"
+                summary = f"✗ {tool_name or 'tool'}"
+            else:
+                event_type = "tool_result"
+                summary = f"✓ {tool_name}" if tool_name else (preview[:80] or "tool result")
+        elif tool_name and role in ("assistant", "user"):
+            event_type = "tool_call"
+            summary = f"→ {tool_name}"
+        elif role == "assistant":
+            lower = clean_text.strip().lower()
+            if lower.startswith(("using", "i'll use", "i will use")) or lower.startswith(("tool call", "calling")):
+                event_type = "assistant_meta"
+                summary = preview[:120] or "assistant planning"
+            elif any(kw in lower[:100] for kw in ("thinking", "reasoning", "analyzing")):
+                event_type = "assistant_thinking"
+                summary = preview[:120] or "thinking"
+            else:
+                event_type = "assistant_text"
+                summary = preview[:120] or "assistant"
+        elif role == "user":
+            event_type = "user_message"
+            summary = preview[:120] or "user"
+        elif role == "system":
+            event_type = "system"
+            summary = preview[:80] or "system"
+        else:
+            event_type = "event"
+            summary = preview[:80] or str(role) or "event"
+
+        normalized.append(
+            {
+                "timestamp": row.get("timestamp") or row.get("indexed_at"),
+                "session_id": row.get("session_id"),
+                "agent_id": row.get("agent_id"),
+                "event_type": event_type,
+                "role": role,
+                "tool_name": tool_name,
+                "summary": summary,
+                "details": clean_text[:500] if clean_text else "",
+                "is_error": is_error,
+                "entry_idx": row.get("entry_idx"),
+            }
+        )
+    return normalized
+
+
+@app.get("/session/{session_id}/activity")
+def session_activity(session_id: str, limit: int = Query(200, ge=1, le=1000)):
+    rows = get_session_activity(session_id, limit)
+    return {
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "session_id": session_id,
+        "activity": _normalize_activity_rows(rows),
+    }
+
+
+# Pricing per 1M tokens (input, output). Mirror of session_context_report.py MODEL_PRICING.
+_API_MODEL_PRICING = {
+    "gpt-5": (2.5, 10.0), "gpt-5-4o": (2.5, 10.0), "gpt-4o": (2.5, 10.0),
+    "gpt-4o-mini": (0.15, 0.6), "gpt-4.1": (2.0, 8.0),
+    "gpt-4-turbo": (10.0, 30.0), "gpt-4": (30.0, 60.0),
+    "claude-opus-4-6": (3.0, 15.0), "claude-sonnet-4-6": (3.0, 15.0),
+    "claude-haiku-4-6": (0.8, 4.0), "claude-3-5-sonnet": (3.0, 15.0),
+    "claude-3-opus": (15.0, 75.0), "claude-3-sonnet": (3.0, 15.0),
+    "deepseek-chat": (0.14, 0.28), "deepseek-reasoner": (0.55, 2.19),
+    "gemini-2.5-pro": (1.25, 5.0), "gemini-2.5-flash": (0.075, 0.30),
+    "gemini-2.5-flash-lite": (0.075, 0.15),
+    "mistral-large": (2.0, 6.0), "mistral-small": (0.15, 0.6),
+    "minimax-m2.7": (0.099, 0.396), "minimax-m2": (0.099, 0.396),
+    "qwen": (0.5, 2.0), "moonshotai/kimi-k2": (0.5, 1.5),
+    "default": (0.1, 0.4),
+}
+
+
+def _infer_pricing(model: str) -> tuple[float, float]:
+    lowered = model.lower()
+    for key, price in _API_MODEL_PRICING.items():
+        if key.lower() in lowered or lowered in key.lower():
+            return price
+    return (0.1, 0.4)
+
+
+@app.get("/session/{session_id}/cost_summary")
+def session_cost_summary(session_id: str):
+    """Return token count and cost estimate for a session."""
+    import math
+    conn = get_conn()
+    conn.row_factory = sqlite3.Row  # return Row objects for dict-like access
+    rows = conn.execute(
+        """
+        SELECT role, tool_name, clean_text, is_error, entry_type
+        FROM spooled_entries
+        WHERE session_id = ? OR session_id LIKE ?
+        ORDER BY entry_idx ASC
+        """,
+        (session_id, session_id + "%"),
+    ).fetchall()
+
+    if not rows:
+        raise HTTPException(status_code=404, detail="session not found")
+
+    # Get model from model_change entries
+    model = "default"
+    for row in reversed(rows):
+        if row["entry_type"] == "model_change" or row["role"] == "model_change":
+            try:
+                obj = json.loads(row["clean_text"] or "{}")
+                model = obj.get("modelId", obj.get("provider", "default"))
+            except:
+                pass
+            break
+
+    inp_price, out_price = _infer_pricing(model)
+
+    user_tokens = assistant_tokens = tool_result_tokens = error_count = 0
+    tool_usage = {}
+
+    for row in rows:
+        role = row["role"] or ""
+        text = row["clean_text"] or ""
+        tool_name = row["tool_name"] or ""
+        is_error = bool(row["is_error"])
+        tokens = max(1, math.ceil(len(text) / 4))
+
+        if role in ("user", "system"):
+            user_tokens += tokens
+        elif role == "assistant":
+            assistant_tokens += tokens
+        elif role == "toolResult":
+            tool_result_tokens += tokens
+            if is_error:
+                error_count += 1
+            if tool_name:
+                if tool_name not in tool_usage:
+                    tool_usage[tool_name] = {"calls": 0, "result_tokens": 0, "errors": 0}
+                tool_usage[tool_name]["result_tokens"] += tokens
+                if is_error:
+                    tool_usage[tool_name]["errors"] += 1
+
+    input_cost = (user_tokens / 1_000_000) * inp_price
+    output_cost = ((assistant_tokens + tool_result_tokens) / 1_000_000) * out_price
+
+    return {
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "session_id": session_id,
+        "model": model,
+        "pricing_per_1m": {"input": inp_price, "output": out_price},
+        "tokens": {
+            "user_input": user_tokens,
+            "assistant_output": assistant_tokens,
+            "tool_results": tool_result_tokens,
+            "total": user_tokens + assistant_tokens + tool_result_tokens,
+        },
+        "cost_usd": {
+            "input": round(input_cost, 4),
+            "output": round(output_cost, 4),
+            "total": round(input_cost + output_cost, 4),
+        },
+        "error_count": error_count,
+        "tool_usage": tool_usage,
+    }
+
+
+@app.get("/session/{session_id}/stream")
+async def session_stream(session_id: str):
+    async def event_generator():
+        last_idx = -1
+        while True:
+            data = session_activity(session_id, limit=200)
+            act = data.get("activity", [])
+            new_events = [e for e in act if e.get("entry_idx") is not None and e["entry_idx"] > last_idx]
+            if new_events:
+                new_events.sort(key=lambda x: x["entry_idx"])
+                for evt in new_events:
+                    yield f"data: {json.dumps(evt)}\n\n"
+                    last_idx = evt["entry_idx"]
+            await asyncio.sleep(1.0)
+            
+    return StreamingResponse(event_generator(), media_type="text/event-stream")
+
+
+class LockRequest(BaseModel):
+    owner: str
+    ttl_minutes: int = 60
+
+class ReleaseRequest(BaseModel):
+    owner: str
+    status_msg: str = "completed"
+
+@app.post("/jobs/{job_name}/lock")
+def api_lock_job(job_name: str, req: LockRequest):
+    if claim_lock(job_name, req.owner, req.ttl_minutes):
+        return {"status": "ok", "message": "lock acquired"}
+    raise HTTPException(status_code=409, detail="lock held by another owner")
+
+@app.post("/jobs/{job_name}/release")
+def api_release_job(job_name: str, req: ReleaseRequest):
+    if release_lock(job_name, req.owner, req.status_msg):
+        return {"status": "ok", "message": "lock released"}
+    raise HTTPException(status_code=403, detail="lock not held by owner")
+
+@app.get("/jobs")
+def api_list_jobs():
+    return {"jobs": get_job_status()}

api/scheduler.py

@@ -0,0 +1,78 @@
+import time
+from datetime import datetime, timezone, timedelta
+from typing import Optional
+
+from spooler.store import get_conn
+
+def claim_lock(job_name: str, owner: str, ttl_minutes: int = 60) -> bool:
+    """Attempt to claim a distributed lock for a job name. Returns True if claimed."""
+    conn = get_conn()
+    now_iso = datetime.now(timezone.utc).isoformat()
+    until_iso = (datetime.now(timezone.utc) + timedelta(minutes=ttl_minutes)).isoformat()
+    
+    # Check if a valid lock exists
+    row = conn.execute(
+        "SELECT locked_until FROM scheduler_jobs WHERE job_name = ?",
+        (job_name,)
+    ).fetchone()
+    
+    if row and row["locked_until"] and row["locked_until"] > now_iso:
+        # Lock is currently held and not expired
+        conn.close()
+        return False
+        
+    # Claim the lock (insert if new, overwrite if expired)
+    conn.execute(
+        """
+        INSERT OR REPLACE INTO scheduler_jobs 
+        (job_name, last_run_at, locked_until, locked_by, status) 
+        VALUES (?, COALESCE((SELECT last_run_at FROM scheduler_jobs WHERE job_name = ?), ?), ?, ?, 'locked')
+        """,
+        (job_name, job_name, now_iso, until_iso, owner)
+    )
+    conn.commit()
+    conn.close()
+    return True
+
+def release_lock(job_name: str, owner: str, status_msg: str = "completed") -> bool:
+    """Release a held lock, providing a status message."""
+    conn = get_conn()
+    now_iso = datetime.now(timezone.utc).isoformat()
+    
+    # Ensure this owner actually holds the lock
+    row = conn.execute(
+        "SELECT locked_by FROM scheduler_jobs WHERE job_name = ?",
+        (job_name,)
+    ).fetchone()
+    
+    if not row or row["locked_by"] != owner:
+        conn.close()
+        return False
+        
+    conn.execute(
+        """
+        UPDATE scheduler_jobs 
+        SET locked_until = NULL,
+            locked_by = NULL,
+            last_run_at = ?,
+            status = ?
+        WHERE job_name = ? AND locked_by = ?
+        """,
+        (now_iso, status_msg, job_name, owner)
+    )
+    conn.commit()
+    conn.close()
+    return True
+
+def get_job_status(job_name: Optional[str] = None) -> list[dict]:
+    """Get status of all jobs or a specific job."""
+    conn = get_conn()
+    if job_name:
+        rows = conn.execute(
+            "SELECT * FROM scheduler_jobs WHERE job_name = ?",
+            (job_name,)
+        ).fetchall()
+    else:
+        rows = conn.execute("SELECT * FROM scheduler_jobs").fetchall()
+    conn.close()
+    return [dict(row) for row in rows]

config.py

@@ -0,0 +1,58 @@
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+
+@dataclass
+class Settings:
+    # Paths
+    openclaw_agents_root: Path
+    openclaw_state_dir: Path
+    db_path: Path
+
+    # Agent allowlist (comma-separated; empty = watch all)
+    agents_allowlist: list[str]
+
+    # Glob exclusion patterns for session files (comma-separated; e.g. "*.trajectory.jsonl,*.lock")
+    session_glob_exclude: list[str]
+
+    # Spooler
+    max_toolresult_chars: int
+    spooler_batch_size: int
+    spooler_redact_patterns: str
+    tool_noise_patterns: str
+
+    # Reviewer
+    review_confidence_threshold: float
+
+    # API
+    api_port: int
+
+
+def _env_path(name: str, default: str) -> Path:
+    return Path(os.environ.get(name, default)).expanduser()
+
+
+def load_settings() -> Settings:
+    state_dir = _env_path("OPENCLAW_STATE_DIR", "~/.openclaw/workspace/ops/state")
+    state_dir.mkdir(parents=True, exist_ok=True)
+    raw_allowlist = os.environ.get("AGENTS_ALLOWLIST", "")
+    parsed_allowlist = [a.strip() for a in raw_allowlist.split(",") if a.strip()] if raw_allowlist else []
+    raw_exclude = os.environ.get("SESSION_GLOB_EXCLUDE", "")
+    parsed_exclude = [e.strip() for e in raw_exclude.split(",") if e.strip()] if raw_exclude else []
+    return Settings(
+        agents_allowlist=parsed_allowlist,
+        session_glob_exclude=parsed_exclude,
+        openclaw_agents_root=_env_path("OPENCLAW_AGENTS_ROOT", "~/.openclaw/agents"),
+        openclaw_state_dir=state_dir,
+        db_path=_env_path("SESSION_AMPLIFIER_DB_PATH", str(state_dir / "session_amplifier.sqlite")),
+        max_toolresult_chars=int(os.environ.get("MAX_TOOLRESULT_CHARS", "2000")),
+        spooler_batch_size=int(os.environ.get("SPOOLER_BATCH_SIZE", "100")),
+        spooler_redact_patterns=os.environ.get("SPOOLER_REDACT_PATTERNS", "api_key,path,base64"),
+        tool_noise_patterns=os.environ.get("TOOL_NOISE_PATTERNS", "ENOENT,no output,command exited"),
+        review_confidence_threshold=float(os.environ.get("REVIEW_CONFIDENCE_THRESHOLD", "0.5")),
+        api_port=int(os.environ.get("API_PORT", "8477")),
+    )
+
+
+settings = load_settings()

docker-compose.yml

@@ -0,0 +1,32 @@
+services:
+  session-amplifier:
+    build:
+      context: .
+    image: session-amplifier:local
+    container_name: session-amplifier
+    restart: unless-stopped
+    command: ["serve"]
+    ports:
+      - "8477:8477"
+    environment:
+      OPENCLAW_AGENTS_ROOT: /openclaw/agents
+      OPENCLAW_STATE_DIR: /data/session-amplifier
+      SESSION_AMPLIFIER_DB_PATH: /data/session-amplifier/session_amplifier.sqlite
+      MAX_TOOLRESULT_CHARS: "2000"
+      SPOOLER_BATCH_SIZE: "100"
+      REVIEW_CONFIDENCE_THRESHOLD: "0.5"
+      API_PORT: "8477"
+    volumes:
+      - session_amplifier_state:/data/session-amplifier
+      # Mount your OpenClaw home read-only so session and skill data are visible.
+      - ${OPENCLAW_HOME:-~/.openclaw}:/openclaw:ro
+    networks:
+      - librechat_default
+
+networks:
+  librechat_default:
+    external: true
+    name: librechat_default
+
+volumes:
+  session_amplifier_state:

main.py

@@ -0,0 +1,196 @@
+"""
+Session Amplifier — OpenClaw sidecar
+Lightweight transcript spooler + review worker.
+"""
+import argparse
+import os
+import threading
+import time
+from pathlib import Path
+
+from config import settings
+
+
+def init_db():
+    """Create SQLite schema if not present."""
+    import sqlite3
+
+    conn = sqlite3.connect(settings.db_path)
+    conn.executescript("""
+        CREATE TABLE IF NOT EXISTS spooled_entries (
+            id INTEGER PRIMARY KEY,
+            session_id TEXT NOT NULL,
+            agent_id TEXT NOT NULL,
+            entry_idx INTEGER NOT NULL,
+            entry_type TEXT,
+            role TEXT,
+            timestamp TEXT,
+            tool_name TEXT,
+            clean_text TEXT,
+            original_length INTEGER,
+            preview TEXT,
+            is_error INTEGER DEFAULT 0,
+            indexed_at TEXT DEFAULT (datetime('now')),
+            UNIQUE(session_id, entry_idx)
+        );
+
+        CREATE TABLE IF NOT EXISTS spool_state (
+            key TEXT PRIMARY KEY,
+            value TEXT
+        );
+
+        CREATE TABLE IF NOT EXISTS review_reports (
+            id INTEGER PRIMARY KEY,
+            review_id TEXT UNIQUE,
+            generated_at TEXT,
+            period_from TEXT,
+            period_to TEXT,
+            report_json TEXT
+        );
+
+        CREATE TABLE IF NOT EXISTS scheduler_jobs (
+            job_name TEXT PRIMARY KEY,
+            last_run_at TEXT,
+            locked_until TEXT,
+            locked_by TEXT,
+            status TEXT
+        );
+    """)
+    conn.commit()
+    conn.close()
+    print(f"[init] DB ready at {settings.db_path}")
+
+
+def run_server():
+    import uvicorn
+    from api.routes import app
+
+    uvicorn.run(app, host="0.0.0.0", port=settings.api_port)
+
+
+def run_spooler():
+    from spooler.processor import run_spool
+
+    count, sessions = run_spool()
+    print(f"[spool] entries={count} sessions={sessions}")
+
+
+def run_reviewer():
+    from reviewer.report import generate_report
+
+    report = generate_report()
+    print(f"[review] id={report.get('review_id','?')} sessions={report.get('sessions_reviewed',0)}")
+
+
+def run_watcher(poll_interval: int = 5, review_every: int = 0):
+    """Poll transcript files and run spooler on changes.
+
+    Args:
+        poll_interval: seconds between filesystem polls.
+        review_every: if > 0, run reviewer every N spool runs. 0 = disabled.
+    """
+    from spooler.processor import run_spool
+
+    print(f"[watch] watching {settings.openclaw_agents_root} every {poll_interval}s"
+          f" (review_every={review_every})")
+    seen: dict[str, tuple[int, int]] = {}
+    spool_run_count = 0
+
+    while True:
+        changed = False
+        for transcript_file in settings.openclaw_agents_root.glob("*/sessions/*.jsonl"):
+            try:
+                # Skip excluded glob patterns
+                import fnmatch
+                fname = transcript_file.name
+                if any(fnmatch.fnmatch(fname, pat) for pat in settings.session_glob_exclude):
+                    continue
+                # Also keep hard exclude for trajectory files
+                if ".trajectory.jsonl" in fname:
+                    continue
+                parts = transcript_file.relative_to(settings.openclaw_agents_root).parts
+                if len(parts) >= 2 and settings.agents_allowlist:
+                    agent_part = parts[0]  # agent id is first segment after root
+                    if agent_part not in settings.agents_allowlist:
+                        continue
+                stat = transcript_file.stat()
+                fingerprint = (int(stat.st_mtime), stat.st_size)
+            except (FileNotFoundError, ValueError):
+                continue
+            key = str(transcript_file)
+            if seen.get(key) != fingerprint:
+                seen[key] = fingerprint
+                changed = True
+
+        if changed:
+            count, sessions = run_spool()
+            print(f"[watch] spool run entries={count} sessions={sessions}")
+            spool_run_count += 1
+
+            if review_every > 0 and spool_run_count % review_every == 0:
+                try:
+                    from reviewer.report import generate_report
+                    report = generate_report()
+                    print(f"[watch] review run id={report.get('review_id','?')}"
+                          f" sessions={report.get('sessions_reviewed',0)}")
+                except Exception as exc:
+                    print(f"[watch] review failed: {exc}")
+
+        time.sleep(poll_interval)
+
+
+def run_serve_watch(poll_interval: int = 5, review_every: int = 0):
+    """Start API server and run file watcher in a background thread."""
+    watcher_thread = threading.Thread(
+        target=run_watcher,
+        args=(poll_interval, review_every),
+        daemon=True,
+    )
+    watcher_thread.start()
+    print(f"[serve-watch] watcher thread started (interval={poll_interval}s)")
+    run_server()
+
+
+def main():
+    parser = argparse.ArgumentParser(prog="session-amplifier")
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    sub.add_parser("init", help="Initialize SQLite schema")
+    sub.add_parser("serve", help="Run FastAPI server")
+    sub.add_parser("spool", help="Run spooler once")
+    sub.add_parser("review", help="Run reviewer once")
+
+    watch_parser = sub.add_parser("watch", help="Poll transcript files and run spooler on changes")
+    watch_parser.add_argument("--interval", type=int, default=5, help="Polling interval in seconds")
+    watch_parser.add_argument("--review-every", type=int, default=0,
+                              help="Run reviewer every N spool runs (0=disabled, default=0)")
+
+    sw_parser = sub.add_parser("serve-watch",
+                               help="Run API server + file watcher in background thread")
+    sw_parser.add_argument("--interval", type=int, default=5, help="Polling interval in seconds")
+    sw_parser.add_argument("--review-every", type=int, default=0,
+                           help="Run reviewer every N spool runs (0=disabled, default=0)")
+
+    args = parser.parse_args()
+
+    if args.command == "init":
+        init_db()
+    elif args.command == "serve":
+        init_db()
+        run_server()
+    elif args.command == "spool":
+        init_db()
+        run_spooler()
+    elif args.command == "review":
+        init_db()
+        run_reviewer()
+    elif args.command == "watch":
+        init_db()
+        run_watcher(args.interval, args.review_every)
+    elif args.command == "serve-watch":
+        init_db()
+        run_serve_watch(args.interval, args.review_every)
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,3 @@
+fastapi>=0.115.0
+uvicorn[standard]>=0.30.0
+pydantic>=2.0.0

reviewer/__init__.py

@@ -0,0 +1 @@
+# reviewer package

reviewer/intelligence.py

@@ -0,0 +1,285 @@
+from __future__ import annotations
+
+from collections import Counter, defaultdict
+from datetime import datetime, timezone
+from difflib import SequenceMatcher
+from pathlib import Path
+from typing import Iterable
+
+from config import settings
+from spooler.store import get_conn, get_recent_sessions, get_session_activity
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _write_artifact(relative_path: str, payload: dict) -> None:
+    base = settings.openclaw_state_dir / "session_amplifier"
+    target = base / relative_path
+    target.parent.mkdir(parents=True, exist_ok=True)
+    import json
+
+    target.write_text(json.dumps(payload, indent=2, sort_keys=True))
+
+
+def _session_transcript_files() -> dict[str, Path]:
+    root = settings.openclaw_agents_root
+    if not root.exists():
+        return {}
+    out: dict[str, Path] = {}
+    for agent_dir in root.iterdir():
+        sessions = agent_dir / "sessions"
+        if not sessions.exists():
+            continue
+        for path in sessions.glob("*.jsonl"):
+            out.setdefault(path.stem, path)
+    return out
+
+
+def generate_session_sprawl_report(limit: int = 500, stale_days: int = 30) -> dict:
+    """Return non-destructive session sprawl/archive candidates."""
+    conn = get_conn()
+    rows = conn.execute(
+        """
+        SELECT session_id, agent_id,
+               MAX(COALESCE(timestamp, indexed_at)) AS last_event_at,
+               COUNT(*) AS event_count,
+               SUM(COALESCE(original_length, 0)) AS original_chars,
+               SUM(CASE WHEN role = 'toolResult' THEN 1 ELSE 0 END) AS tool_result_count,
+               SUM(CASE WHEN is_error = 1 THEN 1 ELSE 0 END) AS error_count,
+               MAX(entry_idx) AS last_entry_idx
+        FROM spooled_entries
+        GROUP BY session_id, agent_id
+        ORDER BY event_count DESC
+        LIMIT ?
+        """,
+        (limit,),
+    ).fetchall()
+    conn.close()
+
+    files = _session_transcript_files()
+    candidates = []
+    now_ts = datetime.now(timezone.utc).timestamp()
+    stale_seconds = stale_days * 24 * 60 * 60
+    for row in rows:
+        rd = dict(row)
+        path = files.get(rd["session_id"])
+        size = path.stat().st_size if path and path.exists() else None
+        mtime = path.stat().st_mtime if path and path.exists() else None
+        reasons = []
+        if (rd.get("event_count") or 0) > 2000:
+            reasons.append("very_high_event_count")
+        if (rd.get("original_chars") or 0) > 1_000_000:
+            reasons.append("very_large_transcript_content")
+        if size and size > 5_000_000:
+            reasons.append("large_file")
+        if mtime and now_ts - mtime > stale_seconds:
+            reasons.append("stale_file")
+        if reasons:
+            candidates.append({
+                **rd,
+                "file_path": str(path) if path else None,
+                "file_size_bytes": size,
+                "file_mtime": datetime.fromtimestamp(mtime, timezone.utc).isoformat() if mtime else None,
+                "candidate_reasons": reasons,
+                "action": "review_then_archive_or_summarize",
+            })
+
+    report = {
+        "generated_at": _now(),
+        "policy": "non_destructive_candidates_only",
+        "stale_days": stale_days,
+        "sessions_scanned": len(rows),
+        "candidate_count": len(candidates),
+        "candidates": candidates[:200],
+    }
+    _write_artifact("reports/session-sprawl-latest.json", report)
+    return report
+
+
+def generate_context_pressure_report(limit: int = 200) -> dict:
+    """Find transcripts likely to bloat context or retrieval."""
+    conn = get_conn()
+    rows = conn.execute(
+        """
+        SELECT session_id, agent_id,
+               COUNT(*) AS event_count,
+               SUM(COALESCE(original_length, 0)) AS original_chars,
+               SUM(CASE WHEN role = 'toolResult' THEN COALESCE(original_length, 0) ELSE 0 END) AS tool_chars,
+               SUM(CASE WHEN role = 'assistant' THEN LENGTH(COALESCE(clean_text, '')) ELSE 0 END) AS assistant_chars,
+               SUM(CASE WHEN role = 'toolResult' AND COALESCE(original_length, 0) > 5000 THEN 1 ELSE 0 END) AS giant_tool_results,
+               SUM(CASE WHEN role = 'toolResult' THEN 1 ELSE 0 END) AS tool_result_count
+        FROM spooled_entries
+        GROUP BY session_id, agent_id
+        HAVING original_chars > 100000 OR giant_tool_results > 0 OR event_count > 1000
+        ORDER BY original_chars DESC
+        LIMIT ?
+        """,
+        (limit,),
+    ).fetchall()
+    conn.close()
+
+    sessions = []
+    for row in rows:
+        rd = dict(row)
+        flags = []
+        if (rd.get("giant_tool_results") or 0) > 0:
+            flags.append("giant_tool_results")
+        if (rd.get("event_count") or 0) > 1000:
+            flags.append("long_running_session")
+        if (rd.get("tool_chars") or 0) > max(1, (rd.get("assistant_chars") or 0)) * 3:
+            flags.append("tool_output_dominates")
+        rd["flags"] = flags
+        rd["recommendation"] = "summarize_before_reuse" if flags else "monitor"
+        sessions.append(rd)
+
+    report = {
+        "generated_at": _now(),
+        "sessions_scanned": len(rows),
+        "pressure_sessions": sessions,
+    }
+    _write_artifact("reports/context-pressure-latest.json", report)
+    return report
+
+
+def generate_failure_mode_report(limit: int = 200) -> dict:
+    """Mine repeated operational failure modes from spooled transcript rows."""
+    conn = get_conn()
+    rows = conn.execute(
+        """
+        SELECT session_id, agent_id, role, tool_name, clean_text, preview, is_error, timestamp, indexed_at
+        FROM spooled_entries
+        WHERE is_error = 1
+           OR LOWER(COALESCE(clean_text, '')) LIKE '%permission%'
+           OR LOWER(COALESCE(clean_text, '')) LIKE '%approve%'
+           OR LOWER(COALESCE(clean_text, '')) LIKE '%timeout%'
+           OR LOWER(COALESCE(clean_text, '')) LIKE '%failover%'
+           OR LOWER(COALESCE(clean_text, '')) LIKE '%fallback%'
+           OR LOWER(COALESCE(clean_text, '')) LIKE '%no session found%'
+           OR LOWER(COALESCE(clean_text, '')) LIKE '%context limit%'
+        ORDER BY COALESCE(timestamp, indexed_at) DESC
+        LIMIT 5000
+        """
+    ).fetchall()
+    conn.close()
+
+    buckets: dict[str, list[dict]] = defaultdict(list)
+    tool_errors: Counter[str] = Counter()
+    for r in rows:
+        rd = dict(r)
+        text = (rd.get("clean_text") or rd.get("preview") or "").lower()
+        key = None
+        if rd.get("is_error"):
+            tool = rd.get("tool_name") or "unknown_tool"
+            key = f"tool_error:{tool}"
+            tool_errors[tool] += 1
+        else:
+            key = _classify_failure_text(text)
+        if key:
+            buckets[key].append({
+                "session_id": rd.get("session_id"),
+                "agent_id": rd.get("agent_id"),
+                "tool_name": rd.get("tool_name"),
+                "timestamp": rd.get("timestamp") or rd.get("indexed_at"),
+                "preview": (rd.get("preview") or rd.get("clean_text") or "")[:240],
+            })
+
+    patterns = []
+    for key, hits in sorted(buckets.items(), key=lambda kv: len(kv[1]), reverse=True):
+        patterns.append({
+            "pattern": key,
+            "count": len(hits),
+            "sessions": sorted({h["session_id"] for h in hits if h.get("session_id")})[:20],
+            "examples": hits[:10],
+            "recommendation": _failure_recommendation(key),
+        })
+
+    report = {
+        "generated_at": _now(),
+        "patterns": patterns[:limit],
+        "top_error_tools": [{"tool": tool, "count": count} for tool, count in tool_errors.most_common(25)],
+    }
+    _write_artifact("reports/failure-modes-latest.json", report)
+    return report
+
+
+def _failure_recommendation(pattern: str) -> str:
+    if pattern.startswith("tool_error:"):
+        return "inspect repeated tool failures and add guardrails or repair wrapper"
+    if pattern == "model_failover_or_fallback":
+        return "audit model routing/fallback logs and expose failover in user-visible status"
+    if pattern == "approval_or_permission_stall":
+        return "improve approval prompts and stale approval recovery"
+    if pattern == "timeout":
+        return "identify timeout source and add bounded wait/retry or progress heartbeat"
+    if pattern == "stale_session_reference":
+        return "repair session lifecycle references and stale session cleanup"
+    if pattern == "context_limit":
+        return "summarize/archive before continuing session"
+    return "review clustered examples"
+
+
+def _classify_failure_text(text: str) -> str | None:
+    """Classify failure text while avoiding overly broad buckets.
+
+    The first intelligence pass bucketed any mention of "permission" or
+    "approve" as an approval stall. That was useful for discovery but too noisy
+    for ongoing degradation detection. Keep the public pattern names stable, but
+    require stronger textual evidence.
+    """
+    lowered = text.lower()
+    if "failover" in lowered or "fallback" in lowered:
+        return "model_failover_or_fallback"
+    if any(phrase in lowered for phrase in (
+        "approval pending",
+        "approval required",
+        "approve this",
+        "permission denied",
+        "requires permission",
+        "insufficient permission",
+        "not permitted",
+    )):
+        return "approval_or_permission_stall"
+    if any(phrase in lowered for phrase in (
+        "timed out",
+        "timeout",
+        "deadline exceeded",
+        "context deadline",
+    )):
+        return "timeout"
+    if "no session found" in lowered or "unknown session" in lowered:
+        return "stale_session_reference"
+    if "context limit" in lowered or "context length" in lowered:
+        return "context_limit"
+    return None
+
+
+def generate_active_sessions_bulk(limit: int = 40, activity_limit: int = 200) -> dict:
+    """Bulk endpoint for visual clients: recent sessions plus normalized activity."""
+    sessions = get_recent_sessions(limit)
+    activity = {
+        row["session_id"]: get_session_activity(row["session_id"], activity_limit)
+        for row in sessions
+    }
+    return {
+        "generated_at": _now(),
+        "sessions": sessions,
+        "activity": activity,
+    }
+
+
+def generate_intelligence_bundle(kind: str = "light") -> dict:
+    """Run the deterministic intelligence suite and write a compact bundle."""
+    sprawl_limit = 1000 if kind == "deep" else 300
+    context_limit = 500 if kind == "deep" else 150
+    failure_limit = 500 if kind == "deep" else 150
+    bundle = {
+        "generated_at": _now(),
+        "kind": kind,
+        "session_sprawl": generate_session_sprawl_report(limit=sprawl_limit),
+        "context_pressure": generate_context_pressure_report(limit=context_limit),
+        "failure_modes": generate_failure_mode_report(limit=failure_limit),
+    }
+    _write_artifact(f"reports/intelligence-{kind}-latest.json", bundle)
+    return bundle

reviewer/pattern_detector.py

@@ -0,0 +1,71 @@
+from collections import Counter, defaultdict
+
+
+def detect_failure_patterns(session_scores: list[dict], spooled_entries: list[dict]) -> list[dict]:
+    """
+    Scan session data for recurring failure patterns.
+    Returns a list of pattern objects with count + affected sessions.
+    """
+    patterns = []
+
+    tool_empty_count = defaultdict(list)
+    tool_error_count = defaultdict(list)
+    for entry in spooled_entries:
+        if entry.get("role") != "toolResult":
+            continue
+        tool_name = entry.get("tool_name", "unknown") or "unknown"
+        clean_text = entry.get("clean_text", "").strip()
+        if clean_text == "":
+            tool_empty_count[tool_name].append(entry["session_id"])
+        if entry.get("is_error"):
+            tool_error_count[tool_name].append(entry["session_id"])
+
+    for tool, sessions in tool_empty_count.items():
+        uniq = sorted(set(sessions))
+        if len(uniq) >= 3:
+            patterns.append({
+                "pattern": f"empty_tool_result:{tool}",
+                "count": len(uniq),
+                "sessions": uniq[:5],
+                "description": f"Tool '{tool}' returned empty output across {len(uniq)} sessions",
+            })
+
+    for tool, sessions in tool_error_count.items():
+        uniq = sorted(set(sessions))
+        if len(uniq) >= 3:
+            patterns.append({
+                "pattern": f"tool_errors:{tool}",
+                "count": len(uniq),
+                "sessions": uniq[:5],
+                "description": f"Tool '{tool}' showed error-like output across {len(uniq)} sessions",
+            })
+
+    flag_counts = Counter()
+    flag_sessions = defaultdict(list)
+    for sess in session_scores:
+        for flag in sess.get("flags", []):
+            flag_counts[flag] += 1
+            flag_sessions[flag].append(sess["session_id"])
+
+    for flag, count in flag_counts.items():
+        if count >= 5:
+            patterns.append({
+                "pattern": f"session_flag:{flag}",
+                "count": count,
+                "sessions": sorted(set(flag_sessions[flag]))[:5],
+                "description": f"Flag '{flag}' appeared in {count} sessions",
+            })
+
+    multi_flag_sessions = [
+        sess for sess in session_scores
+        if len(sess.get("flags", [])) >= 2
+    ]
+    if multi_flag_sessions:
+        patterns.append({
+            "pattern": "session_multiple_flags",
+            "count": len(multi_flag_sessions),
+            "sessions": [sess["session_id"] for sess in multi_flag_sessions[:5]],
+            "description": f"{len(multi_flag_sessions)} sessions triggered multiple review flags",
+        })
+
+    return patterns

reviewer/report.py

@@ -0,0 +1,133 @@
+import json
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+from config import settings
+from reviewer.scorer import score_session, get_all_spooled_entries
+from reviewer.pattern_detector import detect_failure_patterns
+from reviewer.skill_analyzer import analyze_skill_coverage
+
+
+def _group_by_session(rows: list[dict]) -> dict[str, list[dict]]:
+    grouped = {}
+    for row in rows:
+        grouped.setdefault(row["session_id"], []).append(row)
+    return grouped
+
+
+def generate_report(since: str | None = None) -> dict:
+    from spooler.store import get_conn
+
+    conn = get_conn()
+    rows = get_all_spooled_entries(conn, since)
+    conn.close()
+
+    # Group and score
+    grouped = _group_by_session(rows)
+    session_scores = []
+    for session_id, sess_rows in grouped.items():
+        agent_id = sess_rows[0].get("agent_id", "unknown")
+        scored = score_session(sess_rows)
+        scored["session_id"] = session_id
+        scored["agent_id"] = agent_id
+        session_scores.append(scored)
+
+    # Detect patterns
+    patterns = detect_failure_patterns(session_scores, rows)
+
+    # Skill coverage
+    skill_coverage = analyze_skill_coverage()
+
+    # Optimization recommendations (stub — deterministic only)
+    recommendations = []
+    for mcp in skill_coverage.get("mcps_missing_skill_surface", []):
+        recommendations.append({
+            "type": "skill_surface_missing",
+            "description": f"MCP '{mcp}' is registered but has no skill surface",
+            "target": mcp,
+            "confidence": 0.8,
+        })
+
+    # Unused skills (skills dir exists, no tool_name match in spool)
+    skill_names = set(skill_coverage.get("mcps_with_skill_surface", []))
+    used_tools = {r.get("tool_name") for r in rows if r.get("tool_name")}
+    unused_skills = [s for s in skill_names if s not in used_tools and s not in skill_coverage.get("mcps_missing_skill_surface", [])]
+    for skill in unused_skills:
+        recommendations.append({
+            "type": "skill_never_used",
+            "description": f"Skill '{skill}' exists but was not used in this period",
+            "target": skill,
+            "confidence": 0.7,
+        })
+
+    # Filter by confidence threshold
+    recommendations = [r for r in recommendations if r["confidence"] >= settings.review_confidence_threshold]
+
+    # Drift signals (stub — compare to previous report if available)
+    drift_signals = []
+
+    report = {
+        "review_id": str(uuid.uuid4()),
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "period": {
+            "from": since or "all",
+            "to": datetime.now(timezone.utc).isoformat(),
+        },
+        "sessions_reviewed": len(session_scores),
+        "session_quality_scores": session_scores,
+        "failure_patterns": patterns,
+        "unused_tools": [
+            {"tool": mcp, "registered": True, "used_this_week": False}
+            for mcp in skill_coverage.get("mcps_missing_skill_surface", [])
+        ],
+        "optimization_recommendations": recommendations,
+        "drift_signals": drift_signals,
+        "skill_coverage": skill_coverage,
+    }
+
+    # Persist to DB
+    _persist_report(report)
+
+    return report
+
+
+def generate_skills_report() -> dict:
+    coverage = analyze_skill_coverage()
+    report = {
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "period": "weekly",
+        **coverage,
+    }
+    _write_json_artifact("skills-latest.json", report)
+    return report
+
+
+def _persist_report(report: dict):
+    from spooler.store import get_conn
+
+    conn = get_conn()
+    conn.execute(
+        """
+        INSERT OR REPLACE INTO review_reports
+        (review_id, generated_at, period_from, period_to, report_json)
+        VALUES (?, ?, ?, ?, ?)
+        """,
+        (
+            report["review_id"],
+            report["generated_at"],
+            report["period"]["from"],
+            report["period"]["to"],
+            json.dumps(report),
+        ),
+    )
+    conn.commit()
+    conn.close()
+    _write_json_artifact("review-latest.json", report)
+    _write_json_artifact(f"history/review-{report['generated_at'].replace(':', '').replace('+00:00', 'Z')}.json", report)
+
+
+def _write_json_artifact(relative_path: str, payload: dict):
+    base = settings.openclaw_state_dir / "session_amplifier"
+    target = base / relative_path
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_text(json.dumps(payload, indent=2, sort_keys=True))

reviewer/scorer.py

@@ -0,0 +1,93 @@
+import sqlite3
+from config import settings
+
+
+META_MARKERS = (
+    "using the",
+    "i'll use",
+    "i will use",
+    "tool call",
+    "internal process",
+    "routing this",
+)
+
+
+def score_session(rows: list[dict]) -> dict:
+    """
+    Compute a quality score 0.0-1.0 for a session based on spooled entries.
+    Flags are list of string tags.
+    """
+    if not rows:
+        return {"score": 0.0, "flags": ["no_data"]}
+
+    flags = []
+    tool_results = [r for r in rows if r.get("role") == "toolResult"]
+    assistant_msgs = [r for r in rows if r.get("role") == "assistant"]
+    user_msgs = [r for r in rows if r.get("role") == "user"]
+
+    score = 1.0
+
+    long_results = [
+        r for r in tool_results if r.get("original_length", 0) > settings.max_toolresult_chars * 2
+    ]
+    if long_results:
+        flags.append("excessively_long_tool_output")
+        score -= 0.1 * min(1.0, len(long_results) / max(1, len(tool_results)))
+
+    if assistant_msgs and len(tool_results) > len(assistant_msgs) * 5:
+        flags.append("high_tool_call_ratio")
+        score -= 0.05
+
+    empty_results = [r for r in tool_results if not r.get("clean_text") or len(r.get("clean_text", "")) < 5]
+    if empty_results:
+        flags.append("empty_tool_results")
+        score -= 0.05
+
+    if len(rows) < 3 and not tool_results:
+        flags.append("minimal_session")
+
+    assistant_text_total = sum(len(r.get("clean_text", "")) for r in assistant_msgs)
+    tool_text_total = sum(len(r.get("clean_text", "")) for r in tool_results)
+    if assistant_text_total and tool_text_total > assistant_text_total * 3:
+        flags.append("tool_output_dominates_context")
+        score -= 0.08
+
+    meta_msgs = [
+        r for r in assistant_msgs
+        if any(marker in r.get("clean_text", "").lower() for marker in META_MARKERS)
+    ]
+    if meta_msgs:
+        flags.append("meta_process_narration")
+        score -= 0.05
+        
+    fts_empty_queries = [
+        r for r in tool_results 
+        if r.get("tool_name", "") == "session_search" 
+        and ("0 results" in r.get("clean_text", "").lower() or "no results found" in r.get("clean_text", "").lower())
+    ]
+    if len(fts_empty_queries) > 2:
+        flags.append("fts_query_noise")
+        score -= 0.15
+
+    score = max(0.0, min(1.0, score))
+
+    return {
+        "score": round(score, 3),
+        "flags": sorted(set(flags)),
+        "tool_result_count": len(tool_results),
+        "assistant_msg_count": len(assistant_msgs),
+        "user_msg_count": len(user_msgs),
+        "assistant_text_total": assistant_text_total,
+        "tool_text_total": tool_text_total,
+    }
+
+
+def get_all_spooled_entries(conn: sqlite3.Connection, since: str | None = None) -> list[dict]:
+    if since:
+        rows = conn.execute(
+            "SELECT * FROM spooled_entries WHERE indexed_at > ? ORDER BY indexed_at",
+            (since,),
+        ).fetchall()
+    else:
+        rows = conn.execute("SELECT * FROM spooled_entries ORDER BY indexed_at").fetchall()
+    return [dict(r) for r in rows]

reviewer/session_snapshot.py

@@ -0,0 +1,178 @@
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+from typing import Iterable
+
+from spooler.store import get_recent_sessions, get_session_activity
+
+
+SNAPSHOT_SCHEMA = "openclaw.session.v1"
+log = logging.getLogger("session_snapshot")
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _parse_ts(value: str | None) -> datetime | None:
+    if not value:
+        return None
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except (TypeError, ValueError):
+        log.debug("invalid session timestamp", extra={"timestamp_value": value})
+        return None
+
+
+def _age_seconds(value: str | None) -> int | None:
+    ts = _parse_ts(value)
+    if not ts:
+        return None
+    if ts.tzinfo is None:
+        ts = ts.replace(tzinfo=timezone.utc)
+    return max(0, int((datetime.now(timezone.utc) - ts).total_seconds()))
+
+
+def _classify_activity(row: dict) -> dict:
+    role = row.get("role") or ""
+    tool_name = row.get("tool_name") or ""
+    clean_text = row.get("clean_text") or ""
+    preview = row.get("preview") or clean_text[:300]
+    is_error = bool(row.get("is_error"))
+
+    if role == "toolResult":
+        event_type = "tool_error" if is_error else "tool_result"
+        summary = f"✗ {tool_name or 'tool'}" if is_error else f"✓ {tool_name or 'tool'}"
+    elif role == "assistant":
+        event_type = "assistant_text"
+        summary = preview[:140] or "assistant"
+    elif role == "user":
+        event_type = "user_message"
+        summary = preview[:140] or "user"
+    elif role in ("system", "developer"):
+        event_type = role
+        summary = preview[:120] or role
+    else:
+        event_type = "event"
+        summary = preview[:120] or role or "event"
+
+    return {
+        "timestamp": row.get("timestamp") or row.get("indexed_at"),
+        "entry_idx": row.get("entry_idx"),
+        "event_type": event_type,
+        "role": role,
+        "tool_name": tool_name,
+        "summary": summary,
+        "is_error": is_error,
+    }
+
+
+def _risk_flags(activity: Iterable[dict]) -> list[str]:
+    flags: set[str] = set()
+    for row in activity:
+        text = f"{row.get('clean_text') or ''}\n{row.get('preview') or ''}".lower()
+        if any(token in text for token in ("approve", "approval", "permission")):
+            flags.add("approval_or_permission")
+        if any(token in text for token in ("delete", "rm -rf", "drop table", "destroy", "rollback")):
+            flags.add("destructive_or_rollback_language")
+        if any(token in text for token in ("public deploy", "dns", "domain", "cloud run", "gcloud")):
+            flags.add("external_infra")
+        if any(token in text for token in ("api key", "secret", "password", "token")):
+            flags.add("secret_sensitive")
+    return sorted(flags)
+
+
+def _health_from_summary(summary: dict, last_event_age_seconds: int | None) -> dict:
+    error_count = int(summary.get("error_count") or 0)
+    event_count = int(summary.get("event_count") or 0)
+    noisy = int(summary.get("noisy_tool_results") or 0)
+    state = "active"
+    reasons: list[str] = []
+
+    if last_event_age_seconds is not None and last_event_age_seconds > 30 * 24 * 60 * 60:
+        state = "stale"
+        reasons.append("no_recent_activity_30d")
+    elif error_count > 0:
+        state = "warning"
+        reasons.append("errors_present")
+    elif noisy > 2:
+        state = "warning"
+        reasons.append("noisy_tool_outputs")
+    elif event_count == 0:
+        state = "unknown"
+        reasons.append("no_spooled_events")
+
+    return {
+        "state": state,
+        "reasons": reasons,
+        "last_event_age_seconds": last_event_age_seconds,
+    }
+
+
+def build_session_snapshot(session_summary: dict, activity_limit: int = 80) -> dict:
+    """Build a canonical, backend-neutral OpenClaw session snapshot.
+
+    This is intentionally sidecar-local and read-only. It gives dashboards and
+    reviewers one stable shape without requiring gateway core changes.
+    """
+    session_id = session_summary.get("session_id")
+    activity = get_session_activity(session_id, activity_limit) if session_id else []
+    recent_events = [_classify_activity(row) for row in activity]
+    last_event_at = session_summary.get("last_event_at")
+    age = _age_seconds(last_event_at)
+    tool_result_count = int(session_summary.get("tool_result_count") or 0)
+    event_count = int(session_summary.get("event_count") or 0)
+    error_count = int(session_summary.get("error_count") or 0)
+    tool_ratio = round(tool_result_count / event_count, 3) if event_count else 0.0
+
+    pressure_flags = []
+    if event_count > 1000:
+        pressure_flags.append("long_running_session")
+    if tool_ratio > 0.7:
+        pressure_flags.append("tool_heavy")
+    if int(session_summary.get("noisy_tool_results") or 0) > 2:
+        pressure_flags.append("noisy_tool_outputs")
+
+    return {
+        "schema": SNAPSHOT_SCHEMA,
+        "generated_at": _now(),
+        "id": session_id,
+        "kind": "transcript_session",
+        "owner": {"agent_id": session_summary.get("agent_id")},
+        "state": _health_from_summary(session_summary, age),
+        "runtime": {
+            "adapter": "session-amplifier-spooler",
+            "source": "spooled_transcript",
+        },
+        "health": {
+            "event_count": event_count,
+            "tool_result_count": tool_result_count,
+            "error_count": error_count,
+            "tool_ratio": tool_ratio,
+            "pressure_flags": pressure_flags,
+            "hints": session_summary.get("hints") or [],
+        },
+        "risk": {
+            "flags": _risk_flags(activity),
+            "policy": "signal_only_non_destructive",
+        },
+        "outputs": {
+            "last_event_at": last_event_at,
+            "last_entry_idx": session_summary.get("last_entry_idx"),
+            "recent_events": recent_events,
+        },
+        "rollback": {
+            "note": "Snapshot generation is read-only; remove session_snapshot.py and route imports to roll back this sidecar feature.",
+        },
+    }
+
+
+def build_recent_session_snapshots(limit: int = 40, activity_limit: int = 80) -> dict:
+    sessions = get_recent_sessions(limit)
+    return {
+        "schema": f"{SNAPSHOT_SCHEMA}.collection",
+        "generated_at": _now(),
+        "count": len(sessions),
+        "snapshots": [build_session_snapshot(row, activity_limit=activity_limit) for row in sessions],
+    }

reviewer/skill_analyzer.py

@@ -0,0 +1,228 @@
+import json
+import logging
+import os
+import re
+from pathlib import Path
+from config import settings
+
+log = logging.getLogger("skill_analyzer")
+
+SKILLS_DIR = Path(os.environ.get("OPENCLAW_SKILLS_DIR", "/home/node/.openclaw/skills")).expanduser()
+MCPORTER_CONFIG = Path("/home/node/.openclaw/workspace/_shared/mcporter/config/mcporter.json")
+OPENCLAW_CONFIG = Path(os.environ.get("OPENCLAW_CONFIG", "/home/node/.openclaw/openclaw.json")).expanduser()
+STOP_WORDS = {
+    "the", "and", "for", "with", "from", "that", "this", "when", "into", "your", "you",
+    "are", "can", "use", "using", "skill", "skills", "agent", "agents", "tool", "tools",
+    "task", "tasks", "workflow", "workflows", "process", "openclaw",
+    "before", "after", "existing", "new", "review", "checks", "first", "instead", "only",
+    "improve", "simple", "appropriate", "repo", "updates", "adding", "draft", "recommend",
+    "requested", "evaluate",
+}
+REUSE_SCORE_THRESHOLD = 6
+
+
+def _list_mcp_skills() -> list[str]:
+    """Return skill names from the skills dir."""
+    if not SKILLS_DIR.exists():
+        return []
+    return sorted(d.name for d in SKILLS_DIR.iterdir() if d.is_dir() and (d / "SKILL.md").exists())
+
+
+def _tokens(text: str | None) -> set[str]:
+    if not text:
+        return set()
+    return {
+        tok
+        for tok in re.findall(r"[a-z][a-z0-9_-]{2,}", text.lower())
+        if tok not in STOP_WORDS
+    }
+
+
+def _frontmatter(text: str) -> dict[str, str]:
+    if not text.startswith("---"):
+        return {}
+    parts = text.split("---", 2)
+    if len(parts) < 3:
+        return {}
+    out: dict[str, str] = {}
+    for line in parts[1].splitlines():
+        if ":" not in line or line.startswith((" ", "\t")):
+            continue
+        key, value = line.split(":", 1)
+        out[key.strip()] = value.strip().strip('"\'')
+    return out
+
+
+def _skill_catalog() -> list[dict]:
+    if not SKILLS_DIR.exists():
+        return []
+    rows: list[dict] = []
+    for skill_md in sorted(SKILLS_DIR.glob("**/SKILL.md")):
+        if any(part in {"_archive", ".git", "node_modules", ".cache"} for part in skill_md.parts):
+            continue
+        try:
+            text = skill_md.read_text(encoding="utf-8", errors="replace")
+        except Exception:
+            continue
+        fm = _frontmatter(text)
+        rel = str(skill_md.parent.relative_to(SKILLS_DIR))
+        name = fm.get("name") or skill_md.parent.name
+        description = fm.get("description") or ""
+        body = text.split("---", 2)[2] if text.startswith("---") and len(text.split("---", 2)) > 2 else text
+        rows.append({
+            "name": name,
+            "path": str(skill_md),
+            "relativePath": rel,
+            "description": description[:500],
+            "tokens": sorted(_tokens(" ".join([name, rel, description, body[:4000]]))),
+        })
+    return rows
+
+
+def find_skill_candidates(query: str, agent_id: str | None = None, limit: int = 8) -> dict:
+    query_tokens = _tokens(query)
+    matches = []
+    for skill in _skill_catalog():
+        skill_tokens = set(skill["tokens"])
+        overlap = sorted(query_tokens & skill_tokens)
+        if not overlap:
+            continue
+        score = len(overlap)
+        score += 3 * len(query_tokens & _tokens(skill["name"]))
+        score += 2 * len(query_tokens & _tokens(skill["description"]))
+        matches.append({
+            "name": skill["name"],
+            "path": skill["path"],
+            "relativePath": skill["relativePath"],
+            "description": skill["description"],
+            "score": score,
+            "matchedTerms": overlap[:12],
+        })
+    matches = sorted(matches, key=lambda row: (-row["score"], row["name"]))[:limit]
+    top = matches[0] if matches and matches[0]["score"] >= REUSE_SCORE_THRESHOLD else None
+    agent_filter = load_agent_skill_filter(agent_id)
+    if top and isinstance(agent_filter.get("skills"), list):
+        present = top["name"] in agent_filter["skills"] or top["relativePath"] in agent_filter["skills"]
+        decision = "reuse-existing-skill"
+        allowlist_action = "none" if present else "add-existing-skill-to-agent-allowlist"
+    elif top:
+        decision = "reuse-existing-skill"
+        allowlist_action = "none-unrestricted-or-unknown"
+    else:
+        decision = "draft-new-skill-for-review"
+        allowlist_action = "n/a"
+    return {
+        "query": query,
+        "agentId": agent_id,
+        "decision": decision,
+        "allowlistAction": allowlist_action,
+        "allowlistValue": top["name"] if top else None,
+        "agentSkillFilter": agent_filter,
+        "matches": matches,
+    }
+
+
+def load_agent_skill_filter(agent_id: str | None) -> dict:
+    if not agent_id:
+        return {"agentId": None, "mode": "not-evaluated", "skills": None}
+    try:
+        cfg = json.loads(OPENCLAW_CONFIG.read_text(encoding="utf-8"))
+    except Exception as exc:
+        return {"agentId": agent_id, "mode": "config-unavailable", "error": str(exc), "skills": None}
+    defaults = (cfg.get("agents") or {}).get("defaults") or {}
+    for agent in (cfg.get("agents") or {}).get("list") or []:
+        if agent.get("id") == agent_id:
+            if "skills" in agent:
+                return {"agentId": agent_id, "mode": "explicit-agent-allowlist", "skills": agent.get("skills") or []}
+            if "skills" in defaults:
+                return {"agentId": agent_id, "mode": "inherits-default-allowlist", "skills": defaults.get("skills") or []}
+            return {"agentId": agent_id, "mode": "unrestricted", "skills": None}
+    return {"agentId": agent_id, "mode": "agent-not-found", "skills": None}
+
+
+def analyze_agent_skill_filters() -> dict:
+    try:
+        cfg = json.loads(OPENCLAW_CONFIG.read_text(encoding="utf-8"))
+    except Exception as exc:
+        return {"error": str(exc), "agents": []}
+    defaults = (cfg.get("agents") or {}).get("defaults") or {}
+    rows = []
+    for agent in (cfg.get("agents") or {}).get("list") or []:
+        if "skills" in agent:
+            mode = "explicit-agent-allowlist"
+            skills = agent.get("skills") or []
+        elif "skills" in defaults:
+            mode = "inherits-default-allowlist"
+            skills = defaults.get("skills") or []
+        else:
+            mode = "unrestricted"
+            skills = None
+        rows.append({"agentId": agent.get("id"), "mode": mode, "skillCount": len(skills) if isinstance(skills, list) else None})
+    return {"defaultSkillsSet": "skills" in defaults, "agents": rows}
+
+
+def _list_registered_mcps() -> list[dict]:
+    """Read registered MCP servers from shared mcporter config."""
+    if not MCPORTER_CONFIG.exists():
+        return []
+    try:
+        data = json.loads(MCPORTER_CONFIG.read_text())
+        servers = data.get("mcpServers", {})
+        if not isinstance(servers, dict):
+            return []
+        return [
+            {
+                "name": name,
+                "transport": "http" if isinstance(cfg, dict) and cfg.get("type") == "http" else "stdio",
+                "enabled": True,
+            }
+            for name, cfg in servers.items()
+            if isinstance(cfg, dict)
+        ]
+    except Exception as exc:
+        log.warning("Could not read mcporter config: %s", exc)
+        return []
+
+
+def _skill_aliases(name: str) -> set[str]:
+    aliases = {name, name.replace("_", "-"), name.replace("-", "_")}
+    if name.endswith("-mcp"):
+        aliases.add(name[:-4])
+        aliases.add(name[:-4].replace("-", "_"))
+    if name.endswith("_mcp"):
+        aliases.add(name[:-4].replace("_", "-"))
+        aliases.add(name[:-4])
+    return aliases
+
+
+def analyze_skill_coverage() -> dict:
+    """Compare registered MCPs vs skill surfaces."""
+    registered_mcps = _list_registered_mcps()
+    skill_names = set(_list_mcp_skills())
+
+    mcps_with_skill = []
+    mcps_missing_skill = []
+    for mcp in registered_mcps:
+        name = mcp["name"]
+        aliases = _skill_aliases(name)
+        if aliases & skill_names:
+            mcps_with_skill.append(name)
+        else:
+            mcps_missing_skill.append(name)
+
+    skills_without_registered_mcp = sorted(
+        skill for skill in skill_names
+        if not any(skill in _skill_aliases(mcp["name"]) for mcp in registered_mcps)
+    )
+
+    missing_sorted = sorted(mcps_missing_skill)
+    return {
+        "registered_mcps": registered_mcps,
+        "mcps_with_skill_surface": sorted(mcps_with_skill),
+        "mcps_missing_skill_surface": missing_sorted,
+        "skills_without_registered_mcp": skills_without_registered_mcp,
+        "skill_catalog_count": len(_skill_catalog()),
+        "agent_skill_filters": analyze_agent_skill_filters(),
+        # Deprecated alias — use mcps_missing_skill_surface instead
+        "mcps_missing_skill": missing_sorted,
+    }

spooler/__init__.py

@@ -0,0 +1 @@
+# spooler package

spooler/noise_filter.py

@@ -0,0 +1,17 @@
+from config import settings
+
+
+_NOISE_PATTERNS = [p.strip() for p in settings.tool_noise_patterns.split(",")]
+
+
+def is_noise(clean_text: str) -> bool:
+    """Return True if this tool result should be dropped entirely."""
+    if not clean_text:
+        return True
+    lower = clean_text.lower()
+    for pat in _NOISE_PATTERNS:
+        if pat.lower() in lower:
+            return True
+    if len(clean_text.strip()) < 2:
+        return True
+    return False

spooler/processor.py

@@ -0,0 +1,177 @@
+import json
+import logging
+from pathlib import Path
+
+from config import settings
+from spooler.redaction import redact
+from spooler.noise_filter import is_noise
+from spooler.store import get_file_state, set_file_state, spool_entries
+
+log = logging.getLogger("spooler")
+
+
+MESSAGE_TYPES_TO_SPOOL = {"message"}
+ROLES_TO_SPOOL = {"user", "assistant", "toolResult", "system", "developer"}
+BLOCK_TYPES_TO_KEEP = {"text", "input_text", "resource", "thinking"}
+
+
+def _extract_text_from_content(content) -> str:
+    text_parts: list[str] = []
+    if isinstance(content, str):
+        text_parts.append(content)
+    elif isinstance(content, list):
+        for block in content:
+            if not isinstance(block, dict):
+                text_parts.append(str(block))
+                continue
+            block_type = block.get("type")
+            if block_type in ("text", "input_text"):
+                text_parts.append(block.get("text", ""))
+            elif block_type == "thinking":
+                text_parts.append(block.get("thinking", ""))
+            elif block_type == "resource":
+                text_parts.append(str(block.get("resource", "")))
+    elif content is not None:
+        text_parts.append(str(content))
+    return "\n".join(part for part in text_parts if part)
+
+
+def _extract_message_payload(entry: dict) -> tuple[str, str, str, bool]:
+    message = entry.get("message") or {}
+    role = message.get("role", "")
+    tool_name = message.get("toolName", "") or entry.get("toolName", "")
+    is_error = bool(message.get("isError") or entry.get("isError"))
+    raw_text = _extract_text_from_content(message.get("content", ""))
+    return role, tool_name, raw_text, is_error
+
+
+def _process_entry(entry: dict, session_id: str, agent_id: str, entry_idx: int) -> dict | None:
+    """Transform a raw transcript entry into a spooled row."""
+    entry_type = entry.get("type", "")
+    if entry_type not in MESSAGE_TYPES_TO_SPOOL:
+        return None
+
+    role, tool_name, raw_text, is_error = _extract_message_payload(entry)
+    if role not in ROLES_TO_SPOOL:
+        return None
+
+    timestamp = entry.get("timestamp", "")
+    original_length = len(raw_text)
+    clean_text = redact(raw_text)
+
+    if len(clean_text) > settings.max_toolresult_chars:
+        clean_text = clean_text[: settings.max_toolresult_chars] + "\n... [truncated]"
+
+    if role == "toolResult" and is_noise(clean_text):
+        return None
+
+    preview = clean_text[:300]
+
+    return {
+        "session_id": session_id,
+        "agent_id": agent_id,
+        "entry_idx": entry_idx,
+        "entry_type": entry_type,
+        "role": role,
+        "timestamp": timestamp,
+        "tool_name": tool_name,
+        "clean_text": clean_text,
+        "original_length": original_length,
+        "preview": preview,
+        "is_error": is_error,
+    }
+
+
+def _process_transcript_file(path: Path, agent_id: str, *, skip_before: int = -1) -> list[dict]:
+    """Parse a transcript JSONL and return spooled rows.
+
+    Args:
+        skip_before: skip entries with idx <= this value (incremental ingestion).
+    """
+    rows = []
+    session_id = path.stem
+    try:
+        with open(path) as f:
+            for idx, line in enumerate(f):
+                if idx <= skip_before:
+                    continue
+                line = line.strip()
+                if not line:
+                    continue
+                try:
+                    entry = json.loads(line)
+                except json.JSONDecodeError:
+                    continue
+                if entry.get("type") == "session" and entry.get("id"):
+                    session_id = entry["id"]
+                    continue
+                row = _process_entry(entry, session_id, agent_id, idx)
+                if row:
+                    rows.append(row)
+    except Exception as exc:
+        log.warning("Failed to process %s: %s", path, exc)
+    return rows
+
+
+def run_spool() -> tuple[int, int]:
+    """Walk agent session dirs, process transcript messages into SQLite."""
+    agents_root = settings.openclaw_agents_root
+    if not agents_root.exists():
+        log.warning("Agents root not found: %s", agents_root)
+        return 0, 0
+
+    all_rows = []
+    sessions_updated = 0
+
+    for agent_dir in agents_root.iterdir():
+        if not agent_dir.is_dir():
+            continue
+        agent_id = agent_dir.name
+        # Skip if agent not in allowlist (allowlist empty = watch all)
+        if settings.agents_allowlist and agent_id not in settings.agents_allowlist:
+            continue
+        sessions_dir = agent_dir / "sessions"
+        if not sessions_dir.exists():
+            continue
+
+        for transcript_file in sessions_dir.glob("*.jsonl"):
+            # Skip excluded glob patterns and trajectory files
+            import fnmatch
+            fname = transcript_file.name
+            if any(fnmatch.fnmatch(fname, pat) for pat in settings.session_glob_exclude):
+                continue
+            if ".trajectory.jsonl" in fname:
+                continue
+            try:
+                stat = transcript_file.stat()
+                current_state = {"mtime": int(stat.st_mtime), "size": stat.st_size}
+            except FileNotFoundError:
+                continue
+
+            previous_state = get_file_state(str(transcript_file))
+            if previous_state and previous_state.get("mtime") == current_state["mtime"] \
+                    and previous_state.get("size") == current_state["size"]:
+                continue
+
+            skip_before = previous_state.get("last_entry_idx", -1) if previous_state else -1
+            rows = _process_transcript_file(transcript_file, agent_id, skip_before=skip_before)
+            max_idx = skip_before
+            if rows:
+                all_rows.extend(rows)
+                sessions_updated += 1
+                max_idx = max(r["entry_idx"] for r in rows)
+            set_file_state(str(transcript_file),
+                           mtime=current_state["mtime"],
+                           size=current_state["size"],
+                           last_entry_idx=max(max_idx, skip_before))
+
+    if not all_rows:
+        return 0, 0
+
+    batch_size = settings.spooler_batch_size
+    inserted = 0
+    for i in range(0, len(all_rows), batch_size):
+        batch = all_rows[i : i + batch_size]
+        inserted += spool_entries(batch)
+    return inserted, sessions_updated
+

spooler/redaction.py

@@ -0,0 +1,40 @@
+import re
+import os
+from pathlib import Path
+from config import settings
+
+
+# Patterns resolved from env
+_REDACT_PATTERNS = settings.spooler_redact_patterns.split(",")
+
+_REDACTORS = []
+
+
+def _build_redactors():
+    global _REDACTORS
+    for pat in _REDACT_PATTERNS:
+        pat = pat.strip()
+        if pat == "api_key":
+            # Matches common key formats
+            _REDACTORS.append(
+                (re.compile(r"(?i)(api[_-]?key|token|secret|password)\s*[:=]\s*['\"]?[\w\-]{8,}['\"]?"), "[REDACTED]")
+            )
+        elif pat == "path":
+            # Redact home dirs and usernames in paths
+            _REDACTORS.append(
+                (re.compile(r"/home/[^/]+|C:\\Users\\[^\\]+"), "[REDACTED_PATH]")
+            )
+        elif pat == "base64":
+            # Long base64-ish strings
+            _REDACTORS.append(
+                (re.compile(r"[A-Za-z0-9+/]{60,}={0,2}"), "[BASE64_REDACTED]")
+            )
+
+
+_build_redactors()
+
+
+def redact(text: str) -> str:
+    for pattern, replacement in _REDACTORS:
+        text = pattern.sub(replacement, text)
+    return text

spooler/store.py

@@ -0,0 +1,204 @@
+import sqlite3
+from config import settings
+
+
+def _compact_title(text: str, *, fallback: str = "") -> str:
+    title = " ".join((text or "").strip().split())
+    if not title:
+        return fallback
+    prefixes = (
+        "Conversation info",
+        "Current user request:",
+        "OpenClaw assembled context",
+        "Treat the conversation context",
+    )
+    for prefix in prefixes:
+        if title.startswith(prefix):
+            return fallback
+    return title[:96]
+
+
+def _derive_display_title(session_id: str, activity: list[dict]) -> str:
+    for row in activity:
+        role = row.get("role") or ""
+        if role not in {"user", "assistant"}:
+            continue
+        title = _compact_title(row.get("clean_text") or row.get("preview") or "")
+        if title:
+            return title
+    return session_id[:32]
+
+
+def get_conn() -> sqlite3.Connection:
+    conn = sqlite3.connect(settings.db_path)
+    conn.row_factory = sqlite3.Row
+    return conn
+
+
+def spool_entries(rows: list[dict]):
+    """Bulk-insert spooled entries. Skips duplicates via UNIQUE constraint."""
+    if not rows:
+        return 0
+    conn = get_conn()
+    inserted = 0
+    for row in rows:
+        try:
+            cur = conn.execute(
+                """
+                INSERT OR IGNORE INTO spooled_entries
+                (session_id, agent_id, entry_idx, entry_type, role, timestamp,
+                 tool_name, clean_text, original_length, preview, is_error)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    row["session_id"],
+                    row["agent_id"],
+                    row.get("entry_idx", 0),
+                    row.get("entry_type"),
+                    row.get("role"),
+                    row.get("timestamp"),
+                    row.get("tool_name"),
+                    row["clean_text"],
+                    row.get("original_length"),
+                    row["preview"],
+                    int(bool(row.get("is_error", False))),
+                ),
+            )
+            inserted += cur.rowcount > 0
+        except Exception:
+            pass
+    conn.commit()
+    conn.close()
+    return inserted
+
+
+def get_file_state(file_path: str) -> dict | None:
+    conn = get_conn()
+    row = conn.execute(
+        "SELECT value FROM spool_state WHERE key = ?",
+        (f"file::{file_path}",),
+    ).fetchone()
+    conn.close()
+    if not row:
+        return None
+    value = row["value"]
+    try:
+        parts = value.split(":")
+        state = {"mtime": int(parts[0]), "size": int(parts[1])}
+        if len(parts) > 2:
+            state["last_entry_idx"] = int(parts[2])
+        return state
+    except Exception:
+        return None
+
+
+def set_file_state(file_path: str, *, mtime: int, size: int, last_entry_idx: int = -1):
+    conn = get_conn()
+    conn.execute(
+        "INSERT OR REPLACE INTO spool_state (key, value) VALUES (?, ?)",
+        (f"file::{file_path}", f"{mtime}:{size}:{last_entry_idx}"),
+    )
+    conn.commit()
+    conn.close()
+
+
+def get_recent_sessions(limit: int = 25) -> list[dict]:
+    conn = get_conn()
+    rows = conn.execute(
+        """
+        SELECT session_id, agent_id,
+               MAX(COALESCE(timestamp, indexed_at)) AS last_event_at,
+               COUNT(*) AS event_count,
+               SUM(CASE WHEN role = 'toolResult' THEN 1 ELSE 0 END) AS tool_result_count,
+               SUM(CASE WHEN is_error = 1 THEN 1 ELSE 0 END) AS error_count,
+               SUM(CASE WHEN role = 'toolResult' AND original_length > 5000 THEN 1 ELSE 0 END) AS noisy_tool_results,
+               MAX(entry_idx) AS last_entry_idx
+        FROM spooled_entries
+        GROUP BY session_id, agent_id
+        ORDER BY MAX(COALESCE(timestamp, indexed_at)) DESC
+        LIMIT ?
+        """,
+        (limit,),
+    ).fetchall()
+    conn.close()
+
+    # Attach last summary text to each session
+    result = []
+    for r in rows:
+        rd = dict(r)
+        # Add health hints
+        tool_count = rd.get("tool_result_count", 0) or 0
+        error_count = rd.get("error_count", 0) or 0
+        noisy = rd.get("noisy_tool_results", 0) or 0
+        event_count = rd.get("event_count", 0) or 0
+        hints = []
+        if error_count > 0:
+            hints.append(f"{error_count} error(s)")
+        if noisy > 2:
+            hints.append(f"{noisy} noisy tool outputs")
+        if tool_count > 0 and event_count > 0 and tool_count / event_count > 0.7:
+            hints.append("tool-heavy")
+        rd["hints"] = hints
+        rd["health"] = "error" if error_count > 0 else "warning" if noisy > 2 else "ok"
+        rd["display_title"] = _derive_display_title(rd["session_id"], get_session_activity(rd["session_id"], limit=50))
+        result.append(rd)
+    return result
+
+
+def get_session_summary(session_id: str) -> dict | None:
+    """Return the same summary shape as get_recent_sessions for one session."""
+    conn = get_conn()
+    row = conn.execute(
+        """
+        SELECT session_id, agent_id,
+               MAX(COALESCE(timestamp, indexed_at)) AS last_event_at,
+               COUNT(*) AS event_count,
+               SUM(CASE WHEN role = 'toolResult' THEN 1 ELSE 0 END) AS tool_result_count,
+               SUM(CASE WHEN is_error = 1 THEN 1 ELSE 0 END) AS error_count,
+               SUM(CASE WHEN role = 'toolResult' AND original_length > 5000 THEN 1 ELSE 0 END) AS noisy_tool_results,
+               MAX(entry_idx) AS last_entry_idx
+        FROM spooled_entries
+        WHERE session_id = ?
+        GROUP BY session_id, agent_id
+        LIMIT 1
+        """,
+        (session_id,),
+    ).fetchone()
+    conn.close()
+    if not row:
+        return None
+
+    rd = dict(row)
+    tool_count = rd.get("tool_result_count", 0) or 0
+    error_count = rd.get("error_count", 0) or 0
+    noisy = rd.get("noisy_tool_results", 0) or 0
+    event_count = rd.get("event_count", 0) or 0
+    hints = []
+    if error_count > 0:
+        hints.append(f"{error_count} error(s)")
+    if noisy > 2:
+        hints.append(f"{noisy} noisy tool outputs")
+    if tool_count > 0 and event_count > 0 and tool_count / event_count > 0.7:
+        hints.append("tool-heavy")
+    rd["hints"] = hints
+    rd["health"] = "error" if error_count > 0 else "warning" if noisy > 2 else "ok"
+    rd["display_title"] = _derive_display_title(rd["session_id"], get_session_activity(rd["session_id"], limit=50))
+    return rd
+
+
+
+def get_session_activity(session_id: str, limit: int = 200) -> list[dict]:
+    conn = get_conn()
+    rows = conn.execute(
+        """
+        SELECT session_id, agent_id, entry_idx, role, entry_type, timestamp,
+               tool_name, clean_text, preview, original_length, is_error, indexed_at
+        FROM spooled_entries
+        WHERE session_id = ?
+        ORDER BY entry_idx DESC
+        LIMIT ?
+        """,
+        (session_id, limit),
+    ).fetchall()
+    conn.close()
+    return [dict(r) for r in reversed(rows)]

tests/__init__.py

@@ -0,0 +1 @@
+# tests package

tests/conftest.py

@@ -0,0 +1,163 @@
+"""Shared test fixtures for Session Amplifier tests."""
+import json
+import sqlite3
+import sys
+import os
+import pytest
+
+# Add project root to path so imports work as they do in production
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+# Override settings before importing anything that uses them
+os.environ.setdefault("OPENCLAW_AGENTS_ROOT", os.path.join(os.path.dirname(__file__), "_fake_agents"))
+os.environ.setdefault("OPENCLAW_STATE_DIR", os.path.join(os.path.dirname(__file__), "_fake_state"))
+os.environ.setdefault("SESSION_AMPLIFIER_DB_PATH", ":memory:")
+
+
+@pytest.fixture
+def in_memory_db():
+    """Create an in-memory SQLite DB with the session amplifier schema."""
+    conn = sqlite3.connect(":memory:")
+    conn.row_factory = sqlite3.Row
+    conn.executescript("""
+        CREATE TABLE IF NOT EXISTS spooled_entries (
+            id INTEGER PRIMARY KEY,
+            session_id TEXT NOT NULL,
+            agent_id TEXT NOT NULL,
+            entry_idx INTEGER NOT NULL,
+            entry_type TEXT,
+            role TEXT,
+            timestamp TEXT,
+            tool_name TEXT,
+            clean_text TEXT,
+            original_length INTEGER,
+            preview TEXT,
+            is_error INTEGER DEFAULT 0,
+            indexed_at TEXT DEFAULT (datetime('now')),
+            UNIQUE(session_id, entry_idx)
+        );
+        CREATE TABLE IF NOT EXISTS spool_state (
+            key TEXT PRIMARY KEY,
+            value TEXT
+        );
+        CREATE TABLE IF NOT EXISTS review_reports (
+            id INTEGER PRIMARY KEY,
+            review_id TEXT UNIQUE,
+            generated_at TEXT,
+            period_from TEXT,
+            period_to TEXT,
+            report_json TEXT
+        );
+    """)
+    yield conn
+    conn.close()
+
+
+@pytest.fixture
+def sample_transcript_lines():
+    """Sample JSONL transcript lines for testing."""
+    return [
+        json.dumps({"type": "session", "id": "test-session-001"}),
+        json.dumps({
+            "type": "message",
+            "timestamp": "2026-04-05T10:00:00Z",
+            "message": {
+                "role": "user",
+                "content": [{"type": "text", "text": "Hello, can you help me?"}],
+            },
+        }),
+        json.dumps({
+            "type": "message",
+            "timestamp": "2026-04-05T10:00:05Z",
+            "message": {
+                "role": "assistant",
+                "content": [{"type": "text", "text": "Of course! Let me look into that for you."}],
+            },
+        }),
+        json.dumps({
+            "type": "message",
+            "timestamp": "2026-04-05T10:00:10Z",
+            "message": {
+                "role": "toolResult",
+                "content": [{"type": "text", "text": "file contents: line1\nline2\nline3"}],
+                "toolName": "read_file",
+            },
+        }),
+        json.dumps({
+            "type": "message",
+            "timestamp": "2026-04-05T10:00:15Z",
+            "message": {
+                "role": "toolResult",
+                "content": [{"type": "text", "text": ""}],
+                "toolName": "write_file",
+                "isError": True,
+            },
+        }),
+        json.dumps({
+            "type": "message",
+            "timestamp": "2026-04-05T10:00:20Z",
+            "message": {
+                "role": "assistant",
+                "content": [{"type": "text", "text": "I'll use the read_file tool to check."}],
+            },
+        }),
+    ]
+
+
+@pytest.fixture
+def sample_spooled_entries():
+    """Pre-processed spooled entry dicts for reviewer tests."""
+    return [
+        {
+            "session_id": "s1",
+            "agent_id": "ops",
+            "entry_idx": 1,
+            "entry_type": "message",
+            "role": "user",
+            "timestamp": "2026-04-05T10:00:00Z",
+            "tool_name": "",
+            "clean_text": "Help me debug this issue",
+            "original_length": 25,
+            "preview": "Help me debug this issue",
+            "is_error": 0,
+        },
+        {
+            "session_id": "s1",
+            "agent_id": "ops",
+            "entry_idx": 2,
+            "entry_type": "message",
+            "role": "assistant",
+            "timestamp": "2026-04-05T10:00:05Z",
+            "tool_name": "",
+            "clean_text": "I'll use the read_file tool to inspect the code.",
+            "original_length": 48,
+            "preview": "I'll use the read_file tool to inspect the code.",
+            "is_error": 0,
+        },
+        {
+            "session_id": "s1",
+            "agent_id": "ops",
+            "entry_idx": 3,
+            "entry_type": "message",
+            "role": "toolResult",
+            "timestamp": "2026-04-05T10:00:10Z",
+            "tool_name": "read_file",
+            "clean_text": "x" * 5000,
+            "original_length": 5000,
+            "preview": "x" * 300,
+            "is_error": 0,
+        },
+        {
+            "session_id": "s1",
+            "agent_id": "ops",
+            "entry_idx": 4,
+            "entry_type": "message",
+            "role": "toolResult",
+            "timestamp": "2026-04-05T10:00:15Z",
+            "tool_name": "write_file",
+            "clean_text": "",
+            "original_length": 0,
+            "preview": "",
+            "is_error": 1,
+        },
+    ]

tests/failure_classifier_smoke.py

@@ -0,0 +1,25 @@
+#!/usr/bin/env python3
+"""Standalone smoke test for deterministic failure-mode classification."""
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from reviewer.intelligence import _classify_failure_text, _failure_recommendation
+
+
+def main() -> int:
+    assert _classify_failure_text("approval required before running this command") == "approval_or_permission_stall"
+    assert _classify_failure_text("Permission denied while opening file") == "approval_or_permission_stall"
+    assert _classify_failure_text("permission model docs mention safe defaults") is None
+    assert _classify_failure_text("request timed out after 30s") == "timeout"
+    assert _classify_failure_text("context deadline exceeded") == "timeout"
+    assert _classify_failure_text("model fallback activated") == "model_failover_or_fallback"
+    assert _classify_failure_text("no session found for id abc") == "stale_session_reference"
+    assert "heartbeat" in _failure_recommendation("timeout")
+    print("failure_classifier_smoke: ok")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/session_snapshot_smoke.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+"""Standalone smoke test for session snapshot generation.
+
+This intentionally avoids pytest so the sidecar can be checked in constrained
+gateway/runtime shells where pytest is not installed.
+"""
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from reviewer import session_snapshot as snap
+
+
+def main() -> int:
+    summary = {
+        "session_id": "s1",
+        "agent_id": "ops",
+        "last_event_at": "2026-05-12T12:00:00+00:00",
+        "event_count": 4,
+        "tool_result_count": 2,
+        "error_count": 1,
+        "noisy_tool_results": 0,
+        "last_entry_idx": 4,
+        "hints": ["1 error(s)"],
+    }
+    activity = [
+        {"session_id": "s1", "agent_id": "ops", "entry_idx": 1, "role": "user", "preview": "deploy to Cloud Run?", "clean_text": "deploy to Cloud Run?", "is_error": 0},
+        {"session_id": "s1", "agent_id": "ops", "entry_idx": 2, "role": "toolResult", "tool_name": "exec", "preview": "permission denied", "clean_text": "permission denied", "is_error": 1},
+    ]
+    original_activity = snap.get_session_activity
+    original_recent = snap.get_recent_sessions
+    try:
+        snap.get_session_activity = lambda session_id, limit: activity
+        result = snap.build_session_snapshot(summary, activity_limit=20)
+        assert result["schema"] == "openclaw.session.v1"
+        assert result["id"] == "s1"
+        assert result["state"]["state"] == "warning"
+        assert "approval_or_permission" in result["risk"]["flags"]
+        assert "external_infra" in result["risk"]["flags"]
+        assert result["outputs"]["recent_events"][1]["event_type"] == "tool_error"
+
+        snap.get_recent_sessions = lambda limit: [summary]
+        collection = snap.build_recent_session_snapshots(limit=1, activity_limit=1)
+        assert collection["schema"] == "openclaw.session.v1.collection"
+        assert collection["count"] == 1
+    finally:
+        snap.get_session_activity = original_activity
+        snap.get_recent_sessions = original_recent
+    print("session_snapshot_smoke: ok")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/skill_analyzer_smoke.py

@@ -0,0 +1,31 @@
+#!/usr/bin/env python3
+"""Standalone smoke test for skill coverage alias/path behavior."""
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from reviewer import skill_analyzer
+
+
+def main() -> int:
+    assert "google-workspace" in skill_analyzer._skill_aliases("google_workspace")
+    assert "arxiv" in skill_analyzer._skill_aliases("arxiv-mcp")
+    with tempfile.TemporaryDirectory() as td:
+        root = Path(td)
+        (root / "google-workspace").mkdir()
+        (root / "google-workspace" / "SKILL.md").write_text("---\nname: google-workspace\n---\n")
+        old = skill_analyzer.SKILLS_DIR
+        try:
+            skill_analyzer.SKILLS_DIR = root
+            assert skill_analyzer._list_mcp_skills() == ["google-workspace"]
+        finally:
+            skill_analyzer.SKILLS_DIR = old
+    print("skill_analyzer_smoke: ok")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_reviewer.py

@@ -0,0 +1,86 @@
+from reviewer.scorer import score_session
+from reviewer.pattern_detector import detect_failure_patterns
+from reviewer import skill_analyzer
+
+def test_score_session(sample_spooled_entries):
+    # Base case with provided sample entries
+    result = score_session(sample_spooled_entries)
+    
+    assert "score" in result
+    assert result["tool_result_count"] == 2
+    assert result["assistant_msg_count"] == 1
+    assert result["user_msg_count"] == 1
+    
+    # Check flags triggered by sample entries
+    flags = result["flags"]
+    assert "excessively_long_tool_output" in flags  # One output is 5000 chars length
+    assert "empty_tool_results" in flags # One output is empty
+    assert "meta_process_narration" in flags # Assistant uses "I'll use the..."
+    assert result["score"] < 1.0
+
+def test_score_session_fts_noise():
+    entries = [
+        {"session_id": "s1", "role": "toolResult", "tool_name": "session_search", "clean_text": "0 results found for query xyz", "is_error": 0},
+        {"session_id": "s1", "role": "toolResult", "tool_name": "session_search", "clean_text": "no results found. try different terms.", "is_error": 0},
+        {"session_id": "s1", "role": "toolResult", "tool_name": "session_search", "clean_text": "0 results", "is_error": 0},
+    ]
+    result = score_session(entries)
+    assert "fts_query_noise" in result["flags"]
+
+def test_score_session_empty():
+    result = score_session([])
+    assert result["score"] == 0.0
+    assert "no_data" in result["flags"]
+
+def test_detect_failure_patterns():
+    # Setup some dummy session scores and entries
+    session_scores = [
+        {"session_id": "s1", "flags": ["some_flag", "other_flag"]},
+        {"session_id": "s2", "flags": ["some_flag"]},
+        {"session_id": "s3", "flags": ["some_flag", "third_flag"]},
+        {"session_id": "s4", "flags": ["some_flag"]},
+        {"session_id": "s5", "flags": ["some_flag"]},
+    ]
+    spooled_entries = [
+        {"session_id": "s1", "role": "toolResult", "tool_name": "bad_tool", "clean_text": "", "is_error": 0},
+        {"session_id": "s2", "role": "toolResult", "tool_name": "bad_tool", "clean_text": "", "is_error": 0},
+        {"session_id": "s3", "role": "toolResult", "tool_name": "bad_tool", "clean_text": "", "is_error": 0},
+    ]
+    
+    patterns = detect_failure_patterns(session_scores, spooled_entries)
+    
+    # We should have recognized 'bad_tool' as having empty output in >= 3 sessions
+    assert any(p["pattern"] == "empty_tool_result:bad_tool" for p in patterns)
+    
+    # We should have recognized 'some_flag' triggered in 5 sessions
+    assert any(p["pattern"] == "session_flag:some_flag" for p in patterns)
+
+
+def test_find_skill_candidates_prefers_allowlist_update(tmp_path, monkeypatch):
+    skills_dir = tmp_path / "skills"
+    skill_dir = skills_dir / "ops-framework"
+    skill_dir.mkdir(parents=True)
+    (skill_dir / "SKILL.md").write_text(
+        "---\n"
+        "name: ops-framework\n"
+        "description: Use for OpenClaw gateway troubleshooting and deterministic ops checks\n"
+        "---\n"
+        "# Ops Framework\n"
+        "Run bounded diagnostics before changing gateway configuration.\n",
+        encoding="utf-8",
+    )
+    config = tmp_path / "openclaw.json"
+    config.write_text(
+        '{"agents":{"list":[{"id":"azoth","skills":["context7"]}]}}',
+        encoding="utf-8",
+    )
+    monkeypatch.setattr(skill_analyzer, "SKILLS_DIR", skills_dir)
+    monkeypatch.setattr(skill_analyzer, "OPENCLAW_CONFIG", config)
+
+    result = skill_analyzer.find_skill_candidates(
+        "gateway troubleshooting deterministic ops checks", agent_id="azoth"
+    )
+
+    assert result["decision"] == "reuse-existing-skill"
+    assert result["allowlistAction"] == "add-existing-skill-to-agent-allowlist"
+    assert result["allowlistValue"] == "ops-framework"

tests/test_session_snapshot.py

@@ -0,0 +1,43 @@
+from reviewer import session_snapshot as snap
+
+
+def test_build_session_snapshot(monkeypatch):
+    summary = {
+        "session_id": "s1",
+        "agent_id": "ops",
+        "last_event_at": "2026-05-12T12:00:00+00:00",
+        "event_count": 4,
+        "tool_result_count": 2,
+        "error_count": 1,
+        "noisy_tool_results": 0,
+        "last_entry_idx": 4,
+        "hints": ["1 error(s)"],
+    }
+    activity = [
+        {"session_id": "s1", "agent_id": "ops", "entry_idx": 1, "role": "user", "preview": "deploy to Cloud Run?", "clean_text": "deploy to Cloud Run?", "is_error": 0},
+        {"session_id": "s1", "agent_id": "ops", "entry_idx": 2, "role": "toolResult", "tool_name": "exec", "preview": "permission denied", "clean_text": "permission denied", "is_error": 1},
+    ]
+    monkeypatch.setattr(snap, "get_session_activity", lambda session_id, limit: activity)
+
+    result = snap.build_session_snapshot(summary, activity_limit=20)
+
+    assert result["schema"] == "openclaw.session.v1"
+    assert result["id"] == "s1"
+    assert result["owner"]["agent_id"] == "ops"
+    assert result["state"]["state"] == "warning"
+    assert "errors_present" in result["state"]["reasons"]
+    assert result["health"]["tool_ratio"] == 0.5
+    assert "approval_or_permission" in result["risk"]["flags"]
+    assert "external_infra" in result["risk"]["flags"]
+    assert result["outputs"]["recent_events"][1]["event_type"] == "tool_error"
+
+
+def test_build_recent_session_snapshots(monkeypatch):
+    monkeypatch.setattr(snap, "get_recent_sessions", lambda limit: [{"session_id": "s1", "agent_id": "ops"}])
+    monkeypatch.setattr(snap, "get_session_activity", lambda session_id, limit: [])
+
+    result = snap.build_recent_session_snapshots(limit=1, activity_limit=1)
+
+    assert result["schema"] == "openclaw.session.v1.collection"
+    assert result["count"] == 1
+    assert result["snapshots"][0]["id"] == "s1"

tests/test_spooler.py

@@ -0,0 +1,75 @@
+import json
+from spooler.processor import _extract_text_from_content, _process_entry
+from spooler.redaction import redact
+from spooler.noise_filter import is_noise
+from spooler.store import _derive_display_title
+
+def test_extract_text_from_content():
+    # String content
+    assert _extract_text_from_content("simple text") == "simple text"
+    
+    # List of blocks
+    blocks = [
+        {"type": "text", "text": "block 1"},
+        {"type": "thinking", "thinking": "thinking text"},
+        {"type": "resource", "resource": "res data"},
+        {"type": "image", "source": "ignore this"}
+    ]
+    extracted = _extract_text_from_content(blocks)
+    assert "block 1" in extracted
+    assert "thinking text" in extracted
+    assert "res data" in extracted
+    assert "ignore this" not in extracted
+
+def test_process_entry(sample_transcript_lines):
+    # Parse the user message
+    user_entry = json.loads(sample_transcript_lines[1])
+    row = _process_entry(user_entry, "sess_1", "agent_1", 1)
+    assert row is not None
+    assert row["role"] == "user"
+    assert row["clean_text"] == "Hello, can you help me?"
+    assert row["entry_idx"] == 1
+    
+    # Parse tool result
+    tool_entry = json.loads(sample_transcript_lines[3])
+    row2 = _process_entry(tool_entry, "sess_1", "agent_1", 3)
+    assert row2 is not None
+    assert row2["role"] == "toolResult"
+    assert row2["tool_name"] == "read_file"
+    assert "file contents" in row2["clean_text"]
+
+def test_redaction():
+    text = "Here is my key: api_key='sk-1234567890abcdef' inside a sentence."
+    redacted = redact(text)
+    assert "sk-1234567890abcdef" not in redacted
+    assert "[REDACTED]" in redacted
+    
+def test_noise_filter():
+    assert is_noise("ENOENT: no such file") == True
+    assert is_noise("no output") == True
+    assert is_noise("command exited with code 1") == True
+    assert is_noise("Valid output from a tool") == False
+    assert is_noise("x\n") == True  # Too short
+    assert is_noise("") == True
+
+
+def test_derive_display_title_prefers_conversation_text():
+    activity = [
+        {"role": "system", "clean_text": "ignore system scaffolding"},
+        {"role": "user", "clean_text": "Can you inspect the Hermes agent update and dashboard timeout?"},
+    ]
+
+    assert _derive_display_title("548670c7-e187-4960-a714-1f8e70957060", activity) == (
+        "Can you inspect the Hermes agent update and dashboard timeout?"
+    )
+
+
+def test_derive_display_title_skips_scaffold_context():
+    activity = [
+        {"role": "user", "clean_text": "Conversation info (untrusted metadata): long envelope"},
+        {"role": "assistant", "clean_text": "Dashboard harvest is complete."},
+    ]
+
+    assert _derive_display_title("548670c7-e187-4960-a714-1f8e70957060", activity) == (
+        "Dashboard harvest is complete."
+    )