admin/Context-Lifecycle-Management-f-r-Agents

0 Fork 0

Files

T

grok-bot 39f61621dd docs: add CLM design, architecture, and metrics definitions

2026-08-04 20:43:54 +02:00

45 KiB

Raw Permalink Blame History

Automatisiertes Context-Lifecycle-Management (CLM) — Umsetzungsplan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Ein framework-agnostisches Python-System (Library + CLI), das als Outer Control Loop über Agents Token/Context überwacht, zwischen Compact + Hint, Handoff + neue Session und Weiter entscheidet, datei-basiertes Memory pflegt und strukturierte Handoff-Manifeste erzeugt — zunächst mit Simulated Agent, später mit echten Agent-Adaptern.

Architecture: Der OuterControlLoop liest Metriken (Tokens, Steps, Task-Status), ruft die DecisionEngine auf (regelbasiert + optional LLM-Prompt), orchestriert CompactionPipeline bzw. HandoffPipeline und speichert kritische Fakten über ein MemoryStore-Interface (MVP: Dateisystem). Die WorkingSession abstrahiert den ausführenden Agenten; MVP implementiert SimulatedAgent mit deterministischen Tool-Outputs und Token-Zähler. Artefakte (Manifest, MEMORY, Compact-Hint, Metrics-Log) sind versionierbare Dateien unter .clm/.

Tech Stack: Python ≥ 3.12 · pydantic v2 · typer · rich · pytest · ruff · mypy · hatchling/uv · JSON Schema (Manifest) · Markdown/JSON-Dateien als Memory

Projektroot: /Volumes/SSD2TB/books/clm-agents (aktuell leer)

Global Constraints

Nicht-Ziele: Kein Ersatz des Agent-Reasoning-Kerns; kein eigenes Foundation Model; Human-in-the-Loop bleibt optional möglich.
MVP-Grenzen: Kein Mem0/Letta, kein Traycer, keine adaptiven Schwellen, kein Multi-Agent, kein Live-LLM im Decision-Path (regelbasiert genügt für P1; LLM-Decision optional hinter Flag).
Memory-Default: Datei-basiert (MEMORY.md, Topics, Logs); Interface muss späteren Mem0/Letta-Adapter erlauben.
Integrations-Target P1: Framework-agnostisch + SimulatedAgent — echte Vendor-Hooks erst ab P2+.
Sprache: Code/Identifiers Englisch; Docs (README, HANDOFF, Plan-Artefakte) Deutsch oder bilingual OK.
Qualität: TDD wo sinnvoll; pytest grün; ruff check + mypy clean vor Phase-Exit.
Schwellen-Startwerte (konfigurierbar): siehe Spec §7; Default-Profil coding: soft_compact=250_000, hard_handoff=350_000.
Handoff-Manifest: Felder aus Spec §5.5 sind Pflichtschema (pydantic + JSON Schema).
Package-Name: clm (CLI-Entry clm), Import-Package clm.
Workspace-Layout: src-layout unter src/clm/.

Entscheidungen (festgelegt 2026-08-04)

Thema	Wahl
Stack	Python-Library + CLI
Memory MVP	Datei-basiert (+ Adapter-Interface)
Erster Agent	Simulated Agent (deterministisch)
Plantiefe	Roadmap P0–P4; detaillierte Tasks nur P0–P1
Decision Engine P1	Regelbasiert (Token/Steps/Task-Events); LLM-Prompt optional hinter `--llm-decide` (nicht im MVP-Must)

Ziel-Dateistruktur (nach Phase 1)

clm-agents/
├── pyproject.toml
├── README.md
├── LICENSE
├── .gitignore
├── .clm/                          # Runtime-Artefakte (gitignore partial)
│   ├── config.yaml                # optional local override
│   ├── sessions/<id>/
│   │   ├── metrics.jsonl
│   │   ├── decisions.jsonl
│   │   ├── compact-hints/
│   │   └── handoffs/
│   ├── memory/
│   │   ├── MEMORY.md
│   │   ├── topics/
│   │   └── logs/
│   └── manifests/
├── docs/
│   ├── architecture.md
│   ├── metrics.md
│   ├── manifest-schema.md
│   └── superpowers/
│       ├── specs/2026-08-04-clm-design.md   # spez-Extrakt aus Projektbeschreibung
│       └── plans/2026-08-04-clm-mvp.md      # Kopie dieses Plans (bei Execute)
├── schemas/
│   └── handoff_manifest.schema.json
├── src/clm/
│   ├── __init__.py
│   ├── __main__.py
│   ├── py.typed
│   ├── models/
│   │   ├── __init__.py
│   │   ├── metrics.py             # TokenSnapshot, StepMetrics
│   │   ├── decision.py            # Decision, DecisionAction, Rationale
│   │   ├── manifest.py            # HandoffManifest
│   │   ├── memory.py              # MemoryEntry, MemoryKey
│   │   ├── session.py             # SessionState, TaskStatus
│   │   └── config.py              # Thresholds, TaskProfile, ClmConfig
│   ├── memory/
│   │   ├── __init__.py
│   │   ├── base.py                # MemoryStore Protocol
│   │   ├── file_store.py          # FileMemoryStore
│   │   └── paths.py
│   ├── metrics/
│   │   ├── __init__.py
│   │   ├── monitor.py             # MetricsMonitor
│   │   └── estimators.py          # rough token estimate (chars/4 fallback)
│   ├── decision/
│   │   ├── __init__.py
│   │   ├── engine.py              # DecisionEngine
│   │   └── rules.py               # threshold + event rules
│   ├── compaction/
│   │   ├── __init__.py
│   │   ├── pipeline.py            # CompactionPipeline (P1: structured summary + hint; no verify yet)
│   │   └── summary.py
│   ├── handoff/
│   │   ├── __init__.py
│   │   ├── pipeline.py            # HandoffPipeline
│   │   └── seed.py                # SeedContext for new session
│   ├── session/
│   │   ├── __init__.py
│   │   ├── working.py             # WorkingSession
│   │   ├── agent.py               # Agent Protocol
│   │   └── simulated.py           # SimulatedAgent
│   ├── loop/
│   │   ├── __init__.py
│   │   └── control.py             # OuterControlLoop
│   ├── logging/
│   │   ├── __init__.py
│   │   └── audit.py               # decisions.jsonl / metrics.jsonl writers
│   └── cli/
│       ├── __init__.py
│       ├── app.py                 # typer app
│       ├── run.py
│       ├── compact.py
│       ├── handoff.py
│       └── status.py
└── tests/
    ├── conftest.py
    ├── unit/
    │   ├── test_models_manifest.py
    │   ├── test_config_thresholds.py
    │   ├── test_decision_rules.py
    │   ├── test_file_memory.py
    │   ├── test_metrics_monitor.py
    │   ├── test_compaction_summary.py
    │   ├── test_handoff_pipeline.py
    │   └── test_simulated_agent.py
    ├── integration/
    │   ├── test_outer_loop_continue.py
    │   ├── test_outer_loop_compact.py
    │   └── test_outer_loop_handoff.py
    └── fixtures/
        ├── sample_transcript.json
        └── sample_manifest.json

Kern-Interfaces (verbindlich für P0–P1)

# clm.session.agent
from typing import Protocol, Sequence
from clm.models.session import Message, AgentStepResult

class Agent(Protocol):
    def step(self, messages: Sequence[Message]) -> AgentStepResult: ...
    def inject_context(self, system_addon: str) -> None: ...
    def reset_with_seed(self, seed_messages: Sequence[Message]) -> None: ...

# clm.memory.base
class MemoryStore(Protocol):
    def write(self, key: str, content: str, *, tags: list[str] | None = None) -> None: ...
    def read(self, key: str) -> str | None: ...
    def search(self, query: str, *, limit: int = 10) -> list[MemoryEntry]: ...
    def list_keys(self) -> list[str]: ...

# clm.decision.engine
class DecisionEngine:
    def decide(self, state: SessionState, metrics: TokenSnapshot) -> Decision: ...
    # Decision.action in {CONTINUE, COMPACT, HANDOFF}

# clm.loop.control
class OuterControlLoop:
    def run(self, max_steps: int | None = None) -> LoopResult: ...
    def on_step_end(self, step: AgentStepResult) -> Decision: ...

Decision-Priorität (P1-Regeln):

task_event == PHASE_BOUNDARY und prefer_handoff_on_phase=True → HANDOFF
tokens >= hard_handoff → HANDOFF
tokens >= soft_compact und seit letztem Compact ≥ min_steps_between_compacts → COMPACT
degradation_signal (optional, P1 stub: False) → COMPACT
sonst CONTINUE

Verified Compaction: erst Phase 2 (hier nur Stub/Interface + simple Summary).

Phasen-Überblick

Phase	Dauer (Richt)	Deliverable	Detaillierung
P0 Vorbereitung	~1 Woche	Repo, Spec-Extrakt, Schema, Metriken-Definition, Config	Tasks unten
P1 MVP Outer Loop	2–3 Wochen	Loop + Rules + File Memory + Manifest + CLI + SimAgent	Tasks unten
P2 Verified Compaction + Memory-Adapters	2–4 Wochen	Verify-Pipeline, Recall-Score, Mem0-optional	Meilensteine
P3 Vollautomatisierung & Feintuning	2–3 Wochen	Adaptive thresholds, Dashboard-Export, Phase-Hooks	Meilensteine
P4 Produktion & Multi-Agent	ongoing	A/B, Monitoring, Multi-Agent-Handoffs	Meilensteine

Phase 0 — Vorbereitung

Task 0.1: Repository-Scaffold und Tooling

Files:

Create: pyproject.toml
Create: .gitignore
Create: README.md
Create: src/clm/__init__.py
Create: src/clm/py.typed
Create: tests/conftest.py

Interfaces:

Produces: installierbares Package clm Version 0.1.0a1, pytest lauffähig
Step 1: pyproject.toml anlegen

[project]
name = "clm"
version = "0.1.0a1"
description = "Automated Context Lifecycle Management for LLM Agents"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
  "pydantic>=2.7",
  "pydantic-settings>=2.3",
  "typer>=0.12",
  "rich>=13.7",
  "pyyaml>=6.0",
]

[project.optional-dependencies]
dev = [
  "pytest>=8.2",
  "pytest-cov>=5.0",
  "ruff>=0.5",
  "mypy>=1.10",
  "types-PyYAML>=6.0",
]

[project.scripts]
clm = "clm.cli.app:app"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/clm"]

[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["src"]

[tool.ruff]
line-length = 100
target-version = "py312"

[tool.mypy]
python_version = "3.12"
strict = true
packages = ["clm"]
mypy_path = "src"

Step 2: .gitignore

__pycache__/
*.py[cod]
.venv/
dist/
*.egg-info/
.pytest_cache/
.mypy_cache/
.ruff_cache/
.coverage
htmlcov/
.clm/sessions/
.clm/manifests/
*.jsonl
.DS_Store

Hinweis: .clm/memory/ und Beispiel-Configs nicht pauschal ignorieren, wenn sie als Template committed werden; Runtime-Sessions ja.

Step 3: Minimal src/clm/__init__.py

"""CLM — Context Lifecycle Management for LLM Agents."""

__version__ = "0.1.0a1"

Step 4: venv + install + smoke test

cd /Volumes/SSD2TB/books/clm-agents
uv venv && source .venv/bin/activate
uv pip install -e ".[dev]"
python -c "import clm; print(clm.__version__)"

Expected: 0.1.0a1

Step 5: leeren Test + pytest

# tests/conftest.py
from pathlib import Path
import pytest

@pytest.fixture
def tmp_clm_home(tmp_path: Path) -> Path:
    home = tmp_path / ".clm"
    (home / "memory" / "topics").mkdir(parents=True)
    (home / "memory" / "logs").mkdir(parents=True)
    (home / "sessions").mkdir(parents=True)
    (home / "manifests").mkdir(parents=True)
    return home

# tests/unit/test_version.py
import clm

def test_version():
    assert clm.__version__ == "0.1.0a1"

Run: pytest tests/unit/test_version.py -v → PASS

Step 6: Commit

git init  # falls noch kein Repo
git add pyproject.toml .gitignore README.md src/clm tests
git commit -m "chore: scaffold clm package and tooling"

Task 0.2: Spezifikation und Metriken-Dokumentation im Repo verankern

Files:

Create: docs/superpowers/specs/2026-08-04-clm-design.md
Create: docs/architecture.md
Create: docs/metrics.md
Create: docs/manifest-schema.md

Interfaces:

Produces: kanonische Design-Zusammenfassung (aus User-Projektbeschreibung §1–13), Metrik-Definitionen für Erfolgsmessung
Step 1: Design-Spec extrahieren

Datei docs/superpowers/specs/2026-08-04-clm-design.md mit:

Problem, Ziele, Nicht-Ziele
4-Schichten-Architektur
Decision-Optionen A/B/C
Verified Compaction Pipeline (Zielbild)
Schwellen-Tabelle
Erfolgsmetriken
Explizite MVP-Abgrenzung (P1 = ohne Verify, ohne externe Memory-SDKs)

Inhaltlich 1:1 aus der gelieferten Projektbeschreibung verdichten (kein neues Design).

Step 2: docs/metrics.md — messbare Definitionen

Mindestens definieren:

Metrik	Formel / Erfassung
`task_completion_rate`	erfolgreiche Tasks / gestartete Tasks
`tokens_per_successful_task`	Summe input+output Tokens / erfolgreiche Tasks
`human_interventions_per_session`	manuelle Overrides + explizite User-Prompts (Counter)
`compaction_recall_score`	bestanden Verify-Fragen / gestellte (P2)
`session_length_until_handoff`	Tokens und Steps bis HANDOFF
`cost_per_task`	tokens × price_table (config)
`e2e_latency_s`	wall clock session start→end

P1 loggt Rohdaten in metrics.jsonl (auch wenn Dashboard erst P3 kommt).

Step 3: docs/architecture.md — Komponenten + Datenfluss

Sequenz: Agent.step → MetricsMonitor.update → DecisionEngine.decide → (CONTINUE | CompactionPipeline | HandoffPipeline) → MemoryStore / Manifest → ggf. Session reset.

Step 4: Commit

git add docs
git commit -m "docs: add CLM design, architecture, and metrics definitions"

Task 0.3: Handoff-Manifest Schema + Pydantic-Modell

Files:

Create: schemas/handoff_manifest.schema.json
Create: src/clm/models/__init__.py
Create: src/clm/models/manifest.py
Create: tests/unit/test_models_manifest.py
Create: tests/fixtures/sample_manifest.json
Create: docs/manifest-schema.md

Interfaces:

Produces: HandoffManifest mit Validierung; JSON Schema Draft 2020-12
Step 1: Failing test schreiben

# tests/unit/test_models_manifest.py
from pathlib import Path
import json
from clm.models.manifest import HandoffManifest, ManifestStatus

FIXTURE = Path(__file__).parents[1] / "fixtures" / "sample_manifest.json"

def test_manifest_roundtrip_from_fixture():
    raw = json.loads(FIXTURE.read_text())
    m = HandoffManifest.model_validate(raw)
    assert m.status == ManifestStatus.IN_PROGRESS
    assert m.next_safe_action
    assert m.token_snapshot > 0
    assert m.goal
    dumped = json.loads(m.model_dump_json())
    m2 = HandoffManifest.model_validate(dumped)
    assert m2 == m

def test_manifest_requires_goal():
    import pytest
    from pydantic import ValidationError
    with pytest.raises(ValidationError):
        HandoffManifest(
            goal="",
            status=ManifestStatus.IN_PROGRESS,
            source_of_truth=[],
            changed_artifacts=[],
            commands_and_results=[],
            verified=[],
            not_verified=[],
            assumptions_risks=[],
            blockers=[],
            next_safe_action="continue tests",
            memory_keys=[],
            token_snapshot=1000,
            timestamp="2026-08-04T12:00:00Z",
            decision_rationale="threshold",
        )

Step 2: Run → FAIL (HandoffManifest missing)

Run: pytest tests/unit/test_models_manifest.py -v

Step 3: Modell + Fixture + Schema implementieren

# src/clm/models/manifest.py
from enum import Enum
from typing import Annotated
from pydantic import BaseModel, Field, field_validator

class ManifestStatus(str, Enum):
    IN_PROGRESS = "in_progress"
    BLOCKED = "blocked"
    PARTIAL_COMPLETE = "partial_complete"

class HandoffManifest(BaseModel):
    goal: Annotated[str, Field(min_length=1)]
    status: ManifestStatus
    source_of_truth: list[str]
    changed_artifacts: list[str]
    commands_and_results: list[str]
    verified: list[str]
    not_verified: list[str]
    assumptions_risks: list[str]
    blockers: list[str]
    next_safe_action: Annotated[str, Field(min_length=1)]
    memory_keys: list[str]
    token_snapshot: Annotated[int, Field(ge=0)]
    timestamp: str  # ISO-8601
    decision_rationale: Annotated[str, Field(min_length=1)]

    @field_validator("goal", "next_safe_action", "decision_rationale")
    @classmethod
    def strip_nonempty(cls, v: str) -> str:
        v = v.strip()
        if not v:
            raise ValueError("must not be empty")
        return v

tests/fixtures/sample_manifest.json — vollständiges Beispiel analog Spec §5.5.

schemas/handoff_manifest.schema.json — JSON Schema mit denselben required fields und enum für status.

Step 4: Tests PASS

Run: pytest tests/unit/test_models_manifest.py -v → PASS

Step 5: Commit

git add schemas src/clm/models tests/unit/test_models_manifest.py tests/fixtures docs/manifest-schema.md
git commit -m "feat: add HandoffManifest model and JSON schema"

Task 0.4: Config, Task-Profile und Schwellen

Files:

Create: src/clm/models/config.py
Create: src/clm/models/metrics.py
Create: src/clm/models/session.py
Create: src/clm/models/decision.py
Create: tests/unit/test_config_thresholds.py
Create: src/clm/default_config.yaml (optional packaged)

Interfaces:

Produces: ClmConfig, TaskProfile, Thresholds, DecisionAction, SessionState, TokenSnapshot
Step 1: Failing tests

# tests/unit/test_config_thresholds.py
from clm.models.config import ClmConfig, TaskType

def test_coding_defaults():
    cfg = ClmConfig.for_task(TaskType.CODING)
    assert cfg.thresholds.soft_compact == 250_000
    assert cfg.thresholds.hard_handoff == 350_000
    assert cfg.thresholds.soft_compact < cfg.thresholds.hard_handoff

def test_tool_heavy_defaults():
    cfg = ClmConfig.for_task(TaskType.TOOL_HEAVY)
    assert cfg.thresholds.soft_compact == 225_000
    assert cfg.thresholds.hard_handoff == 320_000

def test_invalid_thresholds_rejected():
    import pytest
    from pydantic import ValidationError
    from clm.models.config import Thresholds
    with pytest.raises(ValidationError):
        Thresholds(soft_compact=400_000, hard_handoff=300_000)

Startwerte (Mitte der Spec-Bänder):

TaskType	soft_compact	hard_handoff
CODING	250_000	350_000
RESEARCH	285_000	400_000
EXPLORATION	340_000	450_000
TOOL_HEAVY	225_000	320_000
MULTI_AGENT	210_000	300_000

Weitere Config-Felder: min_steps_between_compacts: int = 5, prefer_handoff_on_phase: bool = True, clm_home: Path, price_per_1k_input/output (optional für cost metric).

Step 2: Implement models

# src/clm/models/decision.py
from enum import Enum
from pydantic import BaseModel, Field

class DecisionAction(str, Enum):
    CONTINUE = "continue"
    COMPACT = "compact"
    HANDOFF = "handoff"

class Decision(BaseModel):
    action: DecisionAction
    rationale: str = Field(min_length=1)
    trigger: str  # e.g. "hard_handoff", "soft_compact", "phase_boundary", "manual"

# src/clm/models/metrics.py
from pydantic import BaseModel, Field

class TokenSnapshot(BaseModel):
    input_tokens: int = Field(ge=0)
    output_tokens: int = Field(ge=0)
    total_tokens: int = Field(ge=0)
    estimated: bool = False  # True wenn char/4 Heuristik

class StepMetrics(BaseModel):
    step_index: int = Field(ge=0)
    tokens: TokenSnapshot
    tool_calls: int = Field(ge=0)

# src/clm/models/session.py
from enum import Enum
from pydantic import BaseModel, Field
from clm.models.metrics import TokenSnapshot

class TaskStatus(str, Enum):
    RUNNING = "running"
    WAITING = "waiting"
    PHASE_BOUNDARY = "phase_boundary"
    COMPLETED = "completed"
    FAILED = "failed"
    BLOCKED = "blocked"

class MessageRole(str, Enum):
    SYSTEM = "system"
    USER = "user"
    ASSISTANT = "assistant"
    TOOL = "tool"

class Message(BaseModel):
    role: MessageRole
    content: str
    name: str | None = None

class AgentStepResult(BaseModel):
    messages_added: list[Message]
    tool_calls: int = 0
    input_tokens: int = 0
    output_tokens: int = 0
    task_status: TaskStatus = TaskStatus.RUNNING
    notes: str = ""

class SessionState(BaseModel):
    session_id: str
    goal: str
    status: TaskStatus = TaskStatus.RUNNING
    step_index: int = 0
    messages: list[Message] = Field(default_factory=list)
    last_compact_step: int | None = None
    tokens: TokenSnapshot = Field(
        default_factory=lambda: TokenSnapshot(input_tokens=0, output_tokens=0, total_tokens=0)
    )
    human_interventions: int = 0
    changed_artifacts: list[str] = Field(default_factory=list)
    commands_and_results: list[str] = Field(default_factory=list)
    verified: list[str] = Field(default_factory=list)
    not_verified: list[str] = Field(default_factory=list)
    assumptions_risks: list[str] = Field(default_factory=list)
    blockers: list[str] = Field(default_factory=list)
    source_of_truth: list[str] = Field(default_factory=list)
    memory_keys: list[str] = Field(default_factory=list)

ClmConfig / Thresholds / TaskType in config.py mit model_validator, der soft_compact < hard_handoff erzwingt.

Step 3: Tests PASS + Commit

pytest tests/unit/test_config_thresholds.py -v
git add src/clm/models tests/unit/test_config_thresholds.py
git commit -m "feat: add config profiles, session and decision models"

Phase-0 Exit-Kriterien

Package installierbar
Manifest-Schema + Modell grün
Config-Profile mit Spec-Schwellen
Docs: design, architecture, metrics, manifest
Kein Outer Loop noch nötig (kommt P1)

Phase 1 — MVP Outer Loop

Task 1.1: Token/Metrics Monitor + Audit-Log

Files:

Create: src/clm/metrics/estimators.py
Create: src/clm/metrics/monitor.py
Create: src/clm/logging/audit.py
Create: tests/unit/test_metrics_monitor.py

Interfaces:

Consumes: TokenSnapshot, AgentStepResult, SessionState
Produces: MetricsMonitor.update(state, step) -> TokenSnapshot; append-only metrics.jsonl
Step 1: Tests

def test_estimate_tokens_from_chars():
    from clm.metrics.estimators import estimate_tokens
    assert estimate_tokens("abcd" * 100) == 100  # 400 chars / 4

def test_monitor_accumulates_and_logs(tmp_clm_home):
    from clm.metrics.monitor import MetricsMonitor
    from clm.models.session import AgentStepResult, Message, MessageRole, SessionState, TaskStatus
    mon = MetricsMonitor(log_path=tmp_clm_home / "sessions" / "s1" / "metrics.jsonl")
    state = SessionState(session_id="s1", goal="demo")
    step = AgentStepResult(
        messages_added=[Message(role=MessageRole.ASSISTANT, content="hello world")],
        input_tokens=100,
        output_tokens=50,
        tool_calls=1,
        task_status=TaskStatus.RUNNING,
    )
    snap = mon.update(state, step)
    assert snap.total_tokens == 150
    assert mon.log_path.exists()
    assert mon.log_path.read_text().count("\n") == 1

Step 2: Implement

# estimators.py
def estimate_tokens(text: str) -> int:
    return max(0, len(text) // 4)

# monitor.py
class MetricsMonitor:
    def __init__(self, log_path: Path) -> None: ...
    def update(self, state: SessionState, step: AgentStepResult) -> TokenSnapshot:
        # add step tokens to state.tokens; if step tokens are 0, estimate from messages_added
        # append JSON line via AuditLogger
        ...

# logging/audit.py
class AuditLogger:
    def __init__(self, path: Path) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        self.path = path
    def append(self, record: dict) -> None:
        with self.path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")

Step 3: pytest PASS + Commit feat: metrics monitor and audit logging

Task 1.2: FileMemoryStore

Files:

Create: src/clm/memory/base.py
Create: src/clm/memory/paths.py
Create: src/clm/memory/file_store.py
Create: src/clm/models/memory.py
Create: tests/unit/test_file_memory.py

Interfaces:

Produces: FileMemoryStore implementiert MemoryStore
Layout:
- {clm_home}/memory/MEMORY.md — Index/Übersicht
- {clm_home}/memory/topics/{slug}.md — Topic-Dateien
- {clm_home}/memory/logs/YYYYMMDD.jsonl — Append-Log jeder write
Step 1: Tests

def test_write_read_search(tmp_clm_home):
    from clm.memory.file_store import FileMemoryStore
    store = FileMemoryStore(tmp_clm_home / "memory")
    store.write("decisions/auth", "Use JWT with 15m expiry", tags=["security"])
    assert "JWT" in (store.read("decisions/auth") or "")
    hits = store.search("JWT")
    assert any("JWT" in h.content for h in hits)
    assert "decisions/auth" in store.list_keys()
    assert (tmp_clm_home / "memory" / "MEMORY.md").exists()

Step 2: Implement MemoryEntry + FileMemoryStore

class MemoryEntry(BaseModel):
    key: str
    content: str
    tags: list[str] = Field(default_factory=list)
    updated_at: str

class FileMemoryStore:
    def write(self, key: str, content: str, *, tags: list[str] | None = None) -> None:
        # sanitize key → path under topics/; update MEMORY.md index; append log
        ...
    def read(self, key: str) -> str | None: ...
    def search(self, query: str, *, limit: int = 10) -> list[MemoryEntry]:
        # case-insensitive substring over topic files (MVP; no embeddings)
        ...
    def list_keys(self) -> list[str]: ...

Key-Sanitization: nur [a-zA-Z0-9/_-], keine ...

Step 3: PASS + Commit feat: file-based memory store

Task 1.3: DecisionEngine (regelbasiert)

Files:

Create: src/clm/decision/rules.py
Create: src/clm/decision/engine.py
Create: tests/unit/test_decision_rules.py

Interfaces:

Consumes: SessionState, TokenSnapshot, ClmConfig
Produces: Decision
Step 1: Tests (Tabelle der Priorität)

import pytest
from clm.decision.engine import DecisionEngine
from clm.models.config import ClmConfig, TaskType
from clm.models.decision import DecisionAction
from clm.models.metrics import TokenSnapshot
from clm.models.session import SessionState, TaskStatus

@pytest.fixture
def engine():
    return DecisionEngine(ClmConfig.for_task(TaskType.CODING))

def _state(**kwargs) -> SessionState:
    base = dict(session_id="s", goal="g", step_index=10)
    base.update(kwargs)
    return SessionState(**base)

def test_continue_below_soft(engine):
    d = engine.decide(_state(), TokenSnapshot(0, 0, 100_000))
    assert d.action == DecisionAction.CONTINUE

def test_compact_at_soft(engine):
    d = engine.decide(_state(last_compact_step=0), TokenSnapshot(0, 0, 250_000))
    assert d.action == DecisionAction.COMPACT
    assert d.trigger == "soft_compact"

def test_handoff_at_hard(engine):
    d = engine.decide(_state(), TokenSnapshot(0, 0, 350_000))
    assert d.action == DecisionAction.HANDOFF
    assert d.trigger == "hard_handoff"

def test_phase_boundary_prefers_handoff(engine):
    d = engine.decide(
        _state(status=TaskStatus.PHASE_BOUNDARY),
        TokenSnapshot(0, 0, 50_000),
    )
    assert d.action == DecisionAction.HANDOFF
    assert d.trigger == "phase_boundary"

def test_min_steps_between_compacts(engine):
    # last compact was step 9, current step 10, min_steps=5 → no compact yet
    d = engine.decide(
        _state(step_index=10, last_compact_step=9),
        TokenSnapshot(0, 0, 260_000),
    )
    assert d.action == DecisionAction.CONTINUE

Step 2: Implement DecisionEngine.decide exakt nach Priorität in „Kern-Interfaces“
Step 3: PASS + Commit feat: rule-based decision engine

Task 1.4: CompactionPipeline (MVP ohne Verify)

Files:

Create: src/clm/compaction/summary.py
Create: src/clm/compaction/pipeline.py
Create: tests/unit/test_compaction_summary.py

Interfaces:

Consumes: SessionState, MemoryStore
Produces: CompactResult(hint: str, summary: str, memory_keys_written: list[str], new_messages: list[Message])

MVP-Verhalten (kein LLM nötig):

Deterministische strukturierte Summary aus SessionState (goal, last N tool/commands, decisions in memory_keys, blockers).
Schreibe Snapshot nach Memory: Key sessions/{session_id}/compact-{step}.
Baue Compact-Hint (Markdown) mit: Goal, Constraints, Next action, Memory keys, Verified/Not verified.
Ersetze state.messages durch Seed: [system_hint_message] + last_k_user_assistant (config retain_recent_messages: int = 6).
Setze state.last_compact_step = state.step_index.

Step 1: Tests

def test_summary_includes_goal_and_blockers():
    from clm.compaction.summary import build_structured_summary
    from clm.models.session import SessionState
    state = SessionState(session_id="s", goal="Ship auth", blockers=["missing secret"])
    text = build_structured_summary(state)
    assert "Ship auth" in text
    assert "missing secret" in text

def test_pipeline_writes_memory_and_shrinks_messages(tmp_clm_home):
    from clm.compaction.pipeline import CompactionPipeline
    from clm.memory.file_store import FileMemoryStore
    from clm.models.config import ClmConfig, TaskType
    from clm.models.session import Message, MessageRole, SessionState
    store = FileMemoryStore(tmp_clm_home / "memory")
    cfg = ClmConfig.for_task(TaskType.CODING)
    pipe = CompactionPipeline(store=store, config=cfg)
    msgs = [Message(role=MessageRole.USER, content=f"m{i}") for i in range(20)]
    state = SessionState(session_id="s1", goal="g", messages=msgs, step_index=12)
    result = pipe.run(state)
    assert len(state.messages) < 20
    assert result.memory_keys_written
    assert "g" in result.hint
    assert state.last_compact_step == 12

Step 2: Implement + PASS + Commit feat: deterministic compaction pipeline (MVP)

Task 1.5: HandoffPipeline + SeedContext

Files:

Create: src/clm/handoff/pipeline.py
Create: src/clm/handoff/seed.py
Create: tests/unit/test_handoff_pipeline.py

Interfaces:

Consumes: SessionState, Decision, MemoryStore, ClmConfig
Produces: HandoffResult(manifest: HandoffManifest, manifest_path: Path, seed_messages: list[Message])
Step 1: Tests

def test_handoff_writes_valid_manifest(tmp_clm_home):
    from clm.handoff.pipeline import HandoffPipeline
    from clm.memory.file_store import FileMemoryStore
    from clm.models.config import ClmConfig, TaskType
    from clm.models.decision import Decision, DecisionAction
    from clm.models.metrics import TokenSnapshot
    from clm.models.session import SessionState
    store = FileMemoryStore(tmp_clm_home / "memory")
    pipe = HandoffPipeline(
        store=store,
        config=ClmConfig.for_task(TaskType.CODING),
        manifests_dir=tmp_clm_home / "manifests",
    )
    state = SessionState(
        session_id="s1",
        goal="Implement login",
        tokens=TokenSnapshot(input_tokens=100, output_tokens=50, total_tokens=150),
        source_of_truth=["src/auth.py"],
        blockers=[],
    )
    # next_safe_action wird in der Pipeline aus state abgeleitet (Default-Resume-Text)
    decision = Decision(
        action=DecisionAction.HANDOFF,
        rationale="hard limit",
        trigger="hard_handoff",
    )
    result = pipe.run(state, decision)
    assert result.manifest_path.exists()
    assert result.manifest.goal == "Implement login"
    assert result.manifest.token_snapshot == 150
    assert result.seed_messages  # system seed includes manifest summary + memory pulls

Beim Implementieren: next_safe_action aus state.blockers ableiten oder Default "Resume from manifest next_safe_action / re-read source_of_truth". Pipeline muss HandoffManifest strikt validieren bevor Write.

Manifest-Dateiname: {timestamp}_{session_id}.json unter manifests/ und Kopie unter sessions/{id}/handoffs/.

Seed ( seed.py ):

System message mit Manifest-JSON (kompakt) + Goal + next_safe_action
Top memory hits für state.memory_keys und search(goal)
Keine volle Transcript-Historie

Step 2: Implement + PASS + Commit feat: handoff pipeline and seed context

Task 1.6: SimulatedAgent + WorkingSession

Files:

Create: src/clm/session/agent.py
Create: src/clm/session/simulated.py
Create: src/clm/session/working.py
Create: tests/unit/test_simulated_agent.py
Create: tests/fixtures/sample_transcript.json

Interfaces:

SimulatedAgent(script: list[AgentStepResult] | callable) — liefert vordefinierte Steps, zählt Tokens
WorkingSession(agent, state) — step() merged messages in state
Step 1: Tests

def test_simulated_agent_plays_script():
    from clm.session.simulated import SimulatedAgent
    from clm.models.session import AgentStepResult, Message, MessageRole, TaskStatus
    script = [
        AgentStepResult(
            messages_added=[Message(role=MessageRole.ASSISTANT, content="step1")],
            input_tokens=10_000,
            output_tokens=2_000,
            tool_calls=2,
            task_status=TaskStatus.RUNNING,
        ),
        AgentStepResult(
            messages_added=[Message(role=MessageRole.ASSISTANT, content="done")],
            input_tokens=12_000,
            output_tokens=1_000,
            tool_calls=0,
            task_status=TaskStatus.COMPLETED,
        ),
    ]
    agent = SimulatedAgent(script)
    r1 = agent.step([])
    assert r1.messages_added[0].content == "step1"
    r2 = agent.step([])
    assert r2.task_status.value == "completed"

def test_inject_and_reset():
    from clm.session.simulated import SimulatedAgent
    agent = SimulatedAgent([])
    agent.inject_context("HINT")
    assert "HINT" in agent.system_addons
    agent.reset_with_seed([])
    assert agent.system_addons == []

Step 2: Implement Protocol + SimulatedAgent + WorkingSession

class WorkingSession:
    def __init__(self, agent: Agent, state: SessionState) -> None: ...
    def step(self) -> AgentStepResult:
        result = self.agent.step(self.state.messages)
        self.state.messages.extend(result.messages_added)
        self.state.step_index += 1
        self.state.status = result.task_status
        # track changed_artifacts if result.notes contains paths (optional heuristic)
        return result

Step 3: PASS + Commit feat: simulated agent and working session

Task 1.7: OuterControlLoop

Files:

Create: src/clm/loop/control.py
Create: tests/integration/test_outer_loop_continue.py
Create: tests/integration/test_outer_loop_compact.py
Create: tests/integration/test_outer_loop_handoff.py

Interfaces:

Consumes: alle P1-Komponenten
Produces: LoopResult(final_state, decisions: list[Decision], handoffs: list[Path], compact_count: int)
Step 1: Integrationstest CONTINUE

Script mit niedrigen Tokens → nur CONTINUE, status COMPLETED, compact_count=0.

Step 2: Integrationstest COMPACT

Config mit abgesenkten Schwellen für Test: soft=1_000, hard=1_000_000.
Script kumuliert >1000 Tokens, min_steps_between_compacts=1 → mindestens 1 COMPACT; nach Compact len(messages) kleiner; Memory-Key geschrieben; decisions.jsonl enthält compact.

Step 3: Integrationstest HANDOFF

hard=500; Script liefert Tokens über hard → HANDOFF, Manifest existiert, agent.reset_with_seed wurde aufgerufen (flag am SimulatedAgent), neue Session-Messages = seed, Loop endet oder setzt session_id fort (continue_after_handoff: bool = False im MVP → Loop stoppt nach Handoff mit LoopResult.stopped_reason="handoff").

class OuterControlLoop:
    def __init__(
        self,
        session: WorkingSession,
        monitor: MetricsMonitor,
        engine: DecisionEngine,
        compaction: CompactionPipeline,
        handoff: HandoffPipeline,
        audit: AuditLogger,  # decisions.jsonl
        config: ClmConfig,
    ) -> None: ...

    def run(self, max_steps: int | None = None) -> LoopResult:
        steps = 0
        while True:
            if max_steps is not None and steps >= max_steps:
                return LoopResult(..., stopped_reason="max_steps")
            if self.session.state.status in {TaskStatus.COMPLETED, TaskStatus.FAILED}:
                return LoopResult(..., stopped_reason=self.session.state.status.value)
            step = self.session.step()
            snap = self.monitor.update(self.session.state, step)
            decision = self.engine.decide(self.session.state, snap)
            self.audit.append(decision.model_dump())
            if decision.action == DecisionAction.CONTINUE:
                pass
            elif decision.action == DecisionAction.COMPACT:
                result = self.compaction.run(self.session.state)
                self.session.agent.inject_context(result.hint)
                # messages already replaced by pipeline
            elif decision.action == DecisionAction.HANDOFF:
                result = self.handoff.run(self.session.state, decision)
                self.session.agent.reset_with_seed(result.seed_messages)
                self.session.state.messages = list(result.seed_messages)
                return LoopResult(..., stopped_reason="handoff", handoffs=[result.manifest_path])
            steps += 1

Step 4: Alle Integrationstests PASS + Commit feat: outer control loop with compact and handoff paths

Task 1.8: CLI (`clm`)

Files:

Create: src/clm/cli/app.py
Create: src/clm/cli/run.py
Create: src/clm/cli/compact.py
Create: src/clm/cli/handoff.py
Create: src/clm/cli/status.py
Create: tests/unit/test_cli_smoke.py (Typer CliRunner)

Commands:

Command	Zweck
`clm init`	legt `.clm/` Struktur + default MEMORY.md an
`clm run --goal "..." --script fixtures/...`	Outer Loop mit SimulatedAgent
`clm compact --session-id`	manuelles Compact auf gespeichertem State (JSON dump)
`clm handoff --session-id`	manuelles Handoff-Manifest
`clm status`	letzte Metriken/Decisions anzeigen
`clm decide --tokens N --profile coding`	DecisionEngine dry-run

Step 1: clm init + clm decide implementieren und testen

clm decide --tokens 260000 --profile coding
# → action=compact trigger=soft_compact
clm decide --tokens 360000 --profile coding
# → action=handoff trigger=hard_handoff

Step 2: clm run mit Fixture-Script end-to-end

clm init --path .clm
clm run --goal "demo" --script tests/fixtures/sample_transcript.json --profile coding \
  --soft-compact 1000 --hard-handoff 50000 --max-steps 20

Expected: Exit 0, schreibt metrics/decisions, ggf. compact/handoff Artefakte.

Step 3: README mit Quickstart
Step 4: Commit feat: CLI for init, run, decide, compact, handoff, status

Task 1.9: Phase-1 Quality Gate

Files:

Modify: README.md (Statusbadges optional)
Create: docs/superpowers/plans/2026-08-04-clm-mvp.md (Plan-Kopie ins Repo beim Execute)
Step 1: Full test suite

pytest --cov=clm --cov-report=term-missing
ruff check src tests
mypy

Expected: alle Tests grün; Coverage Core-Module (decision, loop, handoff, memory) ≥ 80%.

Step 2: Manuelles Demo-Skript in README

Szenario A: nur CONTINUE
Szenario B: erzwungenes COMPACT
Szenario C: erzwungenes HANDOFF + Manifest-Validierung gegen Schema

Step 3: Tag / Commit chore: phase1 mvp quality gate

Phase-1 Exit-Kriterien

Outer Loop steuert SimulatedAgent vollständig (A/B/C)
File Memory schreibt/liest/sucht
Manifest schema-valide bei jedem Handoff
CLI: init, run, decide, compact, handoff, status
Audit: metrics.jsonl + decisions.jsonl
Kein externer LLM/Memory-Service erforderlich
Docs aktuell

Phase 2 — Meilensteine (nicht bite-sized; nach P1 planen)

M2.1 Verified Compaction

CompactionPipeline erweitert: VerificationSuite mit Fragen aus summary/decisions/constraints
RecallScore = passed/total; Threshold default 0.8
Score < threshold → automatischer Fallback HANDOFF (Decision override)
Optional: LLM-Summarizer Adapter (Summarizer Protocol; default bleibt deterministic; OpenAISummarizer optional extra)

M2.2 Memory Adapter Layer

MemoryStore stabil halten
Mem0Store optional-extra dependency
Feature-Flag memory.backend: file | mem0
Parallel-Write-Mode (file + mem0) für Migrationstests

M2.3 Session Persistence

SessionState serialisieren nach .clm/sessions/<id>/state.json
Resume: clm run --resume <id>
Compact/Handoff manuell auf persisted state

M2.4 Erster Real-Agent Adapter (nach Sim-Stabilität)

OpenAICompatibleAgent (messages API + tool stubs) oder
Hook-Dokumentation für Grok/Claude (stdin/stdout JSON Protocol)
Adapter-Conformance-Tests gegen Agent Protocol

P2 Exit: Verify-Score geloggt; mind. 1 realer Adapter PoC; Mem0 optional grün in CI (skip wenn kein API-Key).

Phase 3 — Meilensteine

M3.1 Adaptive Thresholds

Aus metrics.jsonl gleitende Perzentile / Fehlerraten
Vorschlag clm calibrate --profile coding schreibt empfohlene soft/hard
Manueller Override bleibt führend

M3.2 Observability Export

clm export metrics --format csv|json
Einfaches HTML-Report (token over steps, decision timeline) — kein schweres Dashboard-Framework

M3.3 Phase-Orchestration Hooks

Event PHASE_BOUNDARY von außen (clm signal --phase-done)
Optionale Traycer-Export-Notiz in docs (Integration, kein Vendor-Lock im Core)

M3.4 Human-in-the-Loop Policy

Config: require_confirm_on: [handoff] | [compact,handoff] | []
CLI Prompt / Exit-Code für externe Orchestratoren

P3 Exit: Kalibrierungsbefehl, Export, confirm policy, Phase-Signal.

Phase 4 — Meilensteine

M4.1 Produktion

Structured logging, Versioning der Manifeste, Backup von .clm/memory
Cost model pro Provider in config

M4.2 A/B Memory-Strategien

Experiment-Runner: file vs mem0 vs hybrid auf gleichen Simulated scripts
Report: completion, tokens, recall_score, interventions

M4.3 Multi-Agent Handoffs

Shared memory namespace
Manifest memory_keys + agent_id
Keine Annahmen über spezielles Multi-Agent-Framework — nur Artefakt-Konventionen

M4.4 Research-Integrationen (optional)

Evaluieren ARC/ACM-Papers als Policy-Plugins hinter DecisionEngine (nicht hardcode)

Risiken → Plan-Mapping

Risiko	Mitigation im Plan
Informationsverlust Summary	P2 Verify + Handoff-Fallback; P1 speichert Vollstate in Memory-Snapshot
Over-Engineering	P1 ohne LLM/Mem0/Traycer; klare Exit-Kriterien
Falsche Schwellen	Config + CLI Overrides + später calibrate
Memory-Qualität	File-first, Interface für Wechsel
Vendor-Lock-in	`Agent`, `MemoryStore`, `Summarizer` Protocols

Erfolgsmetriken — wann messen?

Metrik	Ab wann
tokens_per_task, session_length_until_handoff	P1 (Sim-Scripts)
human_interventions	P1 Counter + P3 confirm policy
compaction_recall_score	P2
task_completion_rate, cost, latency	P2+ mit realem Adapter
Qualitäts-Delta +15–35%	P4 A/B, nicht P1-Claim

Empfohlene Ausführungsreihenfolge (P0–P1)

0.1 Scaffold → 0.2 Docs → 0.3 Manifest → 0.4 Config/Models
→ 1.1 Metrics → 1.2 FileMemory → 1.3 Decision
→ 1.4 Compaction → 1.5 Handoff → 1.6 SimAgent
→ 1.7 OuterLoop → 1.8 CLI → 1.9 Quality Gate

Geschätzter Aufwand P0–P1: 3–4 Wochen (1 Person, TDD).

Spec-Coverage Check (Self-Review)

Spec-Abschnitt	Abgedeckt durch
§3 Hauptziele Automation Compact/Handoff	P1 Loop + P2 Verify
§4 4 Schichten	architecture + Module
§5.1 Outer Control Loop	Task 1.7
§5.2 Working Session	Task 1.6
§5.3 Memory Optionen	P1 File; P2 Mem0; Interface für Rest
§5.4 Phase-Orchestration	P3 M3.3 (optional)
§5.5 Handoff Manifest	Task 0.3, 1.5
§6 Decision + Verified Compaction	Task 1.3; Verify P2
§7 Schwellen	Task 0.4
§8 Erwartete Impacts	docs/metrics; messen ab P2/P4 — keine Fake-Claims in P1
§9 Roadmap	dieses Dokument
§10 Risiken	Risiko-Tabelle
§11 Erfolgsmetriken	docs/metrics + logging P1
§12 Erweiterungen	P3–P4 Meilensteine
Nicht-Ziele	Global Constraints

Placeholder-Scan: Keine TBD in P0–P1-Tasks. P2–P4 bewusst Meilensteine (nächster Plan nach P1-Exit).

Type-Konsistenz: DecisionAction, HandoffManifest, SessionState, TokenSnapshot, MemoryStore, Agent durchgängig verwendet.

Execution Handoff

Plan liegt in der Session-Plan-Datei. Bei Umsetzung ins Repo kopieren nach:

docs/superpowers/plans/2026-08-04-clm-mvp.md

Zwei Ausführungsoptionen:

Subagent-Driven (empfohlen) — frischer Subagent pro Task, Review zwischen Tasks
Inline Execution — Tasks in einer Session mit Checkpoints

Nach Plan-Freigabe: Phase 0.1 starten (Scaffold).

45 KiB Raw Permalink Blame History Unescape Escape

Automatisiertes Context-Lifecycle-Management (CLM) — Umsetzungsplan

Global Constraints

Entscheidungen (festgelegt 2026-08-04)

Ziel-Dateistruktur (nach Phase 1)

Kern-Interfaces (verbindlich für P0–P1)

Phasen-Überblick

Phase 0 — Vorbereitung

Task 0.1: Repository-Scaffold und Tooling

Task 0.2: Spezifikation und Metriken-Dokumentation im Repo verankern

Task 0.3: Handoff-Manifest Schema + Pydantic-Modell

Task 0.4: Config, Task-Profile und Schwellen

Phase 1 — MVP Outer Loop

Task 1.1: Token/Metrics Monitor + Audit-Log

Task 1.2: FileMemoryStore

Task 1.3: DecisionEngine (regelbasiert)

Task 1.4: CompactionPipeline (MVP ohne Verify)

Task 1.5: HandoffPipeline + SeedContext

Task 1.6: SimulatedAgent + WorkingSession

Task 1.7: OuterControlLoop

Task 1.8: CLI (clm)

Task 1.9: Phase-1 Quality Gate

Phase 2 — Meilensteine (nicht bite-sized; nach P1 planen)

M2.1 Verified Compaction

M2.2 Memory Adapter Layer

M2.3 Session Persistence

M2.4 Erster Real-Agent Adapter (nach Sim-Stabilität)

Phase 3 — Meilensteine

M3.1 Adaptive Thresholds

M3.2 Observability Export

M3.3 Phase-Orchestration Hooks

M3.4 Human-in-the-Loop Policy

Phase 4 — Meilensteine

M4.1 Produktion

M4.2 A/B Memory-Strategien

M4.3 Multi-Agent Handoffs

M4.4 Research-Integrationen (optional)

Risiken → Plan-Mapping

Erfolgsmetriken — wann messen?

Empfohlene Ausführungsreihenfolge (P0–P1)

Spec-Coverage Check (Self-Review)

Execution Handoff

45 KiB

Raw Permalink Blame History

Task 1.8: CLI (`clm`)