Forge
search
pythonf8555a15
1"""Contract: CASA agent_journal SQL on witdb (ADR-0004 gate)."""
2
3from __future__ import annotations
4
5import json
6import tempfile
7from pathlib import Path
8
9import pytest
10
11from pywitdb._bridge.client import bridge_available
12from pywitdb.sql import connect
13
14pytestmark = pytest.mark.skipif(
15 not bridge_available(), reason="witdb-bridge not built"
16)
17
18# Mirror of casa/domain/agent_journal.py _CREATE_SQL (schema v1).
19JOURNAL_CREATE_SQL = """
20CREATE TABLE IF NOT EXISTS skill_trace (
21 id INTEGER PRIMARY KEY AUTOINCREMENT,
22 at TEXT NOT NULL,
23 tick INTEGER,
24 goal_id TEXT,
25 skill TEXT,
26 plan_step INTEGER,
27 outcome TEXT,
28 payload TEXT NOT NULL
29);
30CREATE INDEX IF NOT EXISTS idx_skill_trace_goal ON skill_trace(goal_id);
31CREATE INDEX IF NOT EXISTS idx_skill_trace_skill ON skill_trace(skill);
32CREATE INDEX IF NOT EXISTS idx_skill_trace_outcome ON skill_trace(outcome);
33
34CREATE TABLE IF NOT EXISTS tick_log (
35 id INTEGER PRIMARY KEY AUTOINCREMENT,
36 at TEXT NOT NULL,
37 tick INTEGER,
38 query TEXT,
39 goal_id TEXT,
40 step_capability TEXT,
41 srl TEXT,
42 outcome TEXT,
43 check_passed INTEGER,
44 surprise INTEGER,
45 payload TEXT NOT NULL
46);
47CREATE INDEX IF NOT EXISTS idx_tick_log_tick ON tick_log(tick);
48
49CREATE TABLE IF NOT EXISTS experience (
50 id INTEGER PRIMARY KEY AUTOINCREMENT,
51 at TEXT NOT NULL,
52 tick INTEGER,
53 name TEXT,
54 title TEXT,
55 outcome TEXT,
56 payload TEXT NOT NULL
57);
58
59CREATE TABLE IF NOT EXISTS retrospective (
60 id INTEGER PRIMARY KEY AUTOINCREMENT,
61 at TEXT NOT NULL,
62 window INTEGER,
63 entries_analyzed INTEGER,
64 payload TEXT NOT NULL
65);
66"""
67
68
69def _json_dumps(obj: object) -> str:
70 return json.dumps(obj, ensure_ascii=False, separators=(",", ":"))
71
72
73def _insert_skill_trace(conn, entry: dict, *, prune_max: int = 0) -> int:
74 sit = entry.get("situation") if isinstance(entry.get("situation"), dict) else {}
75 cur = conn.execute(
76 """
77 INSERT INTO skill_trace (at, tick, goal_id, skill, plan_step, outcome, payload)
78 VALUES (?, ?, ?, ?, ?, ?, ?)
79 """,
80 (
81 str(entry.get("at") or "2026-01-01T00:00:00Z"),
82 entry.get("tick"),
83 str(sit.get("goal_id") or ""),
84 str(entry.get("skill") or ""),
85 int(entry.get("plan_step") or 0),
86 str(entry.get("outcome") or ""),
87 _json_dumps(entry),
88 ),
89 )
90 if prune_max:
91 _prune(conn, "skill_trace", max_rows=prune_max)
92 return int(cur.lastrowid or 0)
93
94
95def _prune(conn, table: str, *, max_rows: int) -> None:
96 """Same one-liner as casa/domain/agent_journal.py (_prune)."""
97 conn.execute(
98 f"""
99 DELETE FROM {table}
100 WHERE id NOT IN (
101 SELECT id FROM {table} ORDER BY id DESC LIMIT ?
102 )
103 """,
104 (max(1, max_rows),),
105 )
106
107
108def test_journal_schema_bootstrap() -> None:
109 with tempfile.TemporaryDirectory() as tmp:
110 path = str(Path(tmp) / "agent_journal.witdb")
111 with connect(path, backend="bridge", create=True) as conn:
112 conn.executescript(JOURNAL_CREATE_SQL)
113 conn.commit()
114 row = conn.execute("SELECT COUNT(*) AS c FROM skill_trace").fetchone()
115 assert row is not None
116 assert int(row["c"]) == 0
117
118
119def test_journal_insert_prune_and_aggregate() -> None:
120 with tempfile.TemporaryDirectory() as tmp:
121 path = str(Path(tmp) / "journal_ops.witdb")
122 with connect(path, backend="bridge", create=True) as conn:
123 conn.executescript(JOURNAL_CREATE_SQL)
124 conn.commit()
125
126 for i in range(5):
127 _insert_skill_trace(
128 conn,
129 {
130 "at": f"2026-01-01T00:00:{i:02d}Z",
131 "tick": i,
132 "skill": "nav" if i % 2 == 0 else "plan",
133 "outcome": "ok" if i < 4 else "fail",
134 "situation": {"goal_id": f"g{i % 2}"},
135 "plan_step": i,
136 },
137 prune_max=0,
138 )
139 conn.commit()
140
141 total = conn.execute("SELECT COUNT(*) AS c FROM skill_trace").fetchone()
142 assert total is not None
143 assert int(total["c"]) == 5
144
145 _prune(conn, "skill_trace", max_rows=3)
146 conn.commit()
147
148 after_prune = conn.execute("SELECT COUNT(*) AS c FROM skill_trace").fetchone()
149 assert after_prune is not None
150 assert int(after_prune["c"]) == 3
151
152 rows = conn.execute(
153 "SELECT skill, outcome, COUNT(*) AS c FROM skill_trace GROUP BY skill, outcome"
154 ).fetchall()
155 assert rows
156
157 goals = conn.execute(
158 "SELECT goal_id, COUNT(*) AS c FROM skill_trace GROUP BY goal_id"
159 ).fetchall()
160 assert goals
161
162
163def test_journal_query_skill_trace_pattern() -> None:
164 with tempfile.TemporaryDirectory() as tmp:
165 path = str(Path(tmp) / "journal_query.witdb")
166 with connect(path, backend="bridge", create=True) as conn:
167 conn.executescript(JOURNAL_CREATE_SQL)
168 conn.commit()
169
170 _insert_skill_trace(
171 conn,
172 {
173 "skill": "alpha",
174 "outcome": "ok",
175 "situation": {"goal_id": "target-42"},
176 "payload": {"x": 1},
177 },
178 )
179 _insert_skill_trace(
180 conn,
181 {
182 "skill": "beta",
183 "outcome": "ok",
184 "situation": {"goal_id": "other"},
185 },
186 )
187 conn.commit()
188
189 rows = conn.execute(
190 "SELECT payload FROM skill_trace WHERE goal_id LIKE ? ORDER BY id DESC LIMIT ?",
191 ("%target%", 10),
192 ).fetchall()
193 assert len(rows) == 1
194 payload = json.loads(rows[0]["payload"])
195 assert payload.get("skill") == "alpha"
196
View only · write via MCP/CIDE