feat(phase9-A): interaction capture loop foundation

Phase 9 Commit A from the agreed plan: turn AtoCore from a stateless
context enhancer into a system that records what it actually fed to an
LLM and what came back. This is the audit trail Reflection (Commit B)
and Extraction (Commit C) will be layered on top of.

The interactions table existed in the schema since the original PoC
but nothing wrote to it. This change makes it real:

Schema migration (additive only):
- response                full LLM response (caller decides how much)
- memories_used           JSON list of memory ids in the context pack
- chunks_used             JSON list of chunk ids in the context pack
- client                  identifier of the calling system
                          (openclaw, claude-code, manual, ...)
- session_id              groups multi-turn conversations
- project                 project name (mirrors the memory module pattern,
                          no FK so capture stays cheap)
- indexes on session_id, project, created_at

The created_at column is now written explicitly with a SQLite-compatible
'YYYY-MM-DD HH:MM:SS' format so the same string lives in the DB and the
returned dataclass. Without this the `since` filter on list_interactions
would silently fail because CURRENT_TIMESTAMP and isoformat use different
shapes that do not compare cleanly as strings.

New module src/atocore/interactions/:
- Interaction dataclass
- record_interaction()    persists one round-trip (prompt required;
                          everything else optional). Refuses empty prompts.
- list_interactions()     filters by project / session_id / client / since,
                          newest-first, hard-capped at 500
- get_interaction()       fetch by id, full response + context pack

API endpoints:
- POST   /interactions                capture one interaction
- GET    /interactions                list with summaries (no full response)
- GET    /interactions/{id}           full record incl. response + pack

Trust model:
- Capture is read-only with respect to memories, project state, and
  source chunks. Nothing here promotes anything into trusted state.
- The audit trail becomes the dataset Commit B (reinforcement) and
  Commit C (extraction + review queue) will operate on.

Tests (13 new, all green):
- service: persist + roundtrip every field
- service: minimum-fields path (prompt only)
- service: empty / whitespace prompt rejected
- service: get by id returns None for missing
- service: filter by project, session, client
- service: ordering newest-first with limit
- service: since filter inclusive on cutoff (the bug the timestamp
  fix above caught)
- service: limit=0 returns empty
- API: POST records and round-trips through GET /interactions/{id}
- API: empty prompt returns 400
- API: missing id returns 404
- API: list filter returns summaries (not full response bodies)

Full suite: 118 passing (was 105).

master-plan-status.md updated to move Phase 9 from "not started" to
"started" with the explicit note that Commit A is in and Commits B/C
remain.
This commit is contained in:
2026-04-06 19:31:43 -04:00
parent c9b9eede25
commit ea3fed3d44
6 changed files with 603 additions and 1 deletions

View File

@@ -0,0 +1,219 @@
"""Interaction capture service.
An *interaction* is one round-trip of:
- a user prompt
- the AtoCore context pack that was assembled for it
- the LLM response (full text or a summary, caller's choice)
- which memories and chunks were actually used in the pack
- a client identifier (e.g. ``openclaw``, ``claude-code``, ``manual``)
- an optional session identifier so multi-turn conversations can be
reconstructed later
The capture is intentionally additive: it never modifies memories,
project state, or chunks. Reflection (Phase 9 Commit B/C) and
write-back (Phase 10) are layered on top of this audit trail without
violating the AtoCore trust hierarchy.
"""
from __future__ import annotations
import json
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from atocore.models.database import get_connection
from atocore.observability.logger import get_logger
log = get_logger("interactions")
@dataclass
class Interaction:
id: str
prompt: str
response: str
response_summary: str
project: str
client: str
session_id: str
memories_used: list[str] = field(default_factory=list)
chunks_used: list[str] = field(default_factory=list)
context_pack: dict = field(default_factory=dict)
created_at: str = ""
def record_interaction(
prompt: str,
response: str = "",
response_summary: str = "",
project: str = "",
client: str = "",
session_id: str = "",
memories_used: list[str] | None = None,
chunks_used: list[str] | None = None,
context_pack: dict | None = None,
) -> Interaction:
"""Persist a single interaction to the audit trail.
The only required field is ``prompt`` so this can be called even when
the caller is in the middle of a partial turn (for example to record
that AtoCore was queried even before the LLM response is back).
"""
if not prompt or not prompt.strip():
raise ValueError("Interaction prompt must be non-empty")
interaction_id = str(uuid.uuid4())
# Store created_at explicitly so the same string lives in both the DB
# column and the returned dataclass. SQLite's CURRENT_TIMESTAMP uses
# 'YYYY-MM-DD HH:MM:SS' which would not compare cleanly against ISO
# timestamps with 'T' and tz offset, breaking the `since` filter on
# list_interactions.
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
memories_used = list(memories_used or [])
chunks_used = list(chunks_used or [])
context_pack_payload = context_pack or {}
with get_connection() as conn:
conn.execute(
"""
INSERT INTO interactions (
id, prompt, context_pack, response_summary, response,
memories_used, chunks_used, client, session_id, project,
created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
interaction_id,
prompt,
json.dumps(context_pack_payload, ensure_ascii=True),
response_summary,
response,
json.dumps(memories_used, ensure_ascii=True),
json.dumps(chunks_used, ensure_ascii=True),
client,
session_id,
project,
now,
),
)
log.info(
"interaction_recorded",
interaction_id=interaction_id,
project=project,
client=client,
session_id=session_id,
memories_used=len(memories_used),
chunks_used=len(chunks_used),
response_chars=len(response),
)
return Interaction(
id=interaction_id,
prompt=prompt,
response=response,
response_summary=response_summary,
project=project,
client=client,
session_id=session_id,
memories_used=memories_used,
chunks_used=chunks_used,
context_pack=context_pack_payload,
created_at=now,
)
def list_interactions(
project: str | None = None,
session_id: str | None = None,
client: str | None = None,
since: str | None = None,
limit: int = 50,
) -> list[Interaction]:
"""List captured interactions, optionally filtered.
``since`` is an ISO timestamp string; only interactions created at or
after that time are returned. ``limit`` is hard-capped at 500 to keep
casual API listings cheap.
"""
if limit <= 0:
return []
limit = min(limit, 500)
query = "SELECT * FROM interactions WHERE 1=1"
params: list = []
if project:
query += " AND project = ?"
params.append(project)
if session_id:
query += " AND session_id = ?"
params.append(session_id)
if client:
query += " AND client = ?"
params.append(client)
if since:
query += " AND created_at >= ?"
params.append(since)
query += " ORDER BY created_at DESC LIMIT ?"
params.append(limit)
with get_connection() as conn:
rows = conn.execute(query, params).fetchall()
return [_row_to_interaction(row) for row in rows]
def get_interaction(interaction_id: str) -> Interaction | None:
"""Fetch one interaction by id, or return None if it does not exist."""
if not interaction_id:
return None
with get_connection() as conn:
row = conn.execute(
"SELECT * FROM interactions WHERE id = ?", (interaction_id,)
).fetchone()
if row is None:
return None
return _row_to_interaction(row)
def _row_to_interaction(row) -> Interaction:
return Interaction(
id=row["id"],
prompt=row["prompt"],
response=row["response"] or "",
response_summary=row["response_summary"] or "",
project=row["project"] or "",
client=row["client"] or "",
session_id=row["session_id"] or "",
memories_used=_safe_json_list(row["memories_used"]),
chunks_used=_safe_json_list(row["chunks_used"]),
context_pack=_safe_json_dict(row["context_pack"]),
created_at=row["created_at"] or "",
)
def _safe_json_list(raw: str | None) -> list[str]:
if not raw:
return []
try:
value = json.loads(raw)
except json.JSONDecodeError:
return []
if not isinstance(value, list):
return []
return [str(item) for item in value]
def _safe_json_dict(raw: str | None) -> dict:
if not raw:
return {}
try:
value = json.loads(raw)
except json.JSONDecodeError:
return {}
if not isinstance(value, dict):
return {}
return value