From 6a93539a9764923200ff32219b798e2a91fb0cf0 Mon Sep 17 00:00:00 2001 From: Moritz Graf Date: Fri, 9 Oct 2026 10:28:11 +0200 Subject: [PATCH] feat(datastore): support --kind flag and add 15m duplicate suppression --- skills/datastore/scripts/datastore.py | 120 +++++++++++++- skills/datastore/tests/test_datastore.py | 190 +++++++++++++++++++++++ 2 files changed, 303 insertions(+), 7 deletions(-) diff --git a/skills/datastore/scripts/datastore.py b/skills/datastore/scripts/datastore.py index 13e339a..6fb4a83 100755 --- a/skills/datastore/scripts/datastore.py +++ b/skills/datastore/scripts/datastore.py @@ -81,8 +81,9 @@ def parse_duration(val: str) -> datetime.timedelta: def find_data_dir() -> Path: """Discovers the knowledge-wiki data directory.""" - if os.environ.get("DATASTORE_DIR"): - return Path(os.environ["DATASTORE_DIR"]).resolve() + for env_var in ["DATASTORE_DIR", "OPENCLAW_DATA_DIR"]: + if os.environ.get(env_var): + return Path(os.environ[env_var]).resolve() candidates = [ Path.cwd() / "knowledge-wiki" / "data", @@ -303,6 +304,86 @@ class DatastoreEngine: # Ignore comment lines or malformed lines with warning pass + def _matches_payload(self, kind: str, p1: Dict[str, Any], p2: Dict[str, Any]) -> bool: + """Determines if two event payloads represent the same intake/session.""" + if p1 == p2: + return True + if kind.startswith("nutrition."): + food1 = p1.get("food_id") + food2 = p2.get("food_id") + item1 = str(p1.get("item", "")).strip().lower() + item2 = str(p2.get("item", "")).strip().lower() + same_item = ( + (food1 and food2 and food1 == food2) + or (item1 and item2 and item1 == item2) + or (food1 and item2 and food1 == item2) + or (food2 and item1 and food2 == item1) + ) + if same_item: + qty1 = p1.get("qty") + qty2 = p2.get("qty") + if qty1 is not None and qty2 is not None: + try: + return abs(float(qty1) - float(qty2)) < 0.01 + except (ValueError, TypeError): + return qty1 == qty2 + return True + elif kind.startswith("feeding."): + amt1 = p1.get("amount_ml") or p1.get("consumed_ml") + amt2 = p2.get("amount_ml") or p2.get("consumed_ml") + if amt1 is not None and amt2 is not None: + return amt1 == amt2 + elif kind.startswith("diaper."): + if p1.get("pee") == p2.get("pee") and p1.get("poo") == p2.get("poo"): + return True + elif kind.startswith("sleep."): + if p1.get("type") == p2.get("type"): + return True + return False + + def _is_duplicate_event( + self, + kind: str, + subject: str, + payload: Dict[str, Any], + start_iso: str, + tolerance_seconds: int = 900, + ) -> Optional[Dict[str, Any]]: + """Checks if a matching event already exists within tolerance_seconds (default 15m).""" + rows = self.con.execute( + """ + SELECT id, kind, subject, actor, start, "end", estimated, meal, note, payload, source_ref, created_at, updated_at + FROM events + WHERE kind = ? + AND subject = ? + AND deleted_at IS NULL + AND abs(epoch(start) - epoch(?::TIMESTAMPTZ)) <= ? + ORDER BY start DESC + """, + [kind, subject, start_iso, tolerance_seconds], + ).fetchall() + + for r in rows: + existing_payload = json.loads(r[9]) if isinstance(r[9], str) else (r[9] or {}) + if self._matches_payload(kind, payload, existing_payload): + return { + "id": r[0], + "kind": r[1], + "subject": r[2], + "actor": r[3], + "start": r[4].isoformat() if hasattr(r[4], "isoformat") else str(r[4]), + "end": r[5].isoformat() if hasattr(r[5], "isoformat") and r[5] else (str(r[5]) if r[5] else None), + "estimated": r[6], + "meal": r[7], + "note": r[8], + "payload": existing_payload, + "source_ref": r[10], + "created_at": r[11].isoformat() if hasattr(r[11], "isoformat") else str(r[11]), + "updated_at": r[12].isoformat() if hasattr(r[12], "isoformat") and r[12] else (str(r[12]) if r[12] else None), + "deleted_at": None, + } + return None + # -------------------------------------------------------------------------- # Event Operations # -------------------------------------------------------------------------- @@ -319,6 +400,7 @@ class DatastoreEngine: meal: Optional[str] = None, note: Optional[str] = None, source_ref: Optional[str] = None, + allow_duplicate: bool = False, ) -> Dict[str, Any]: """Adds a time-series event, saves to JSONL partition, and returns confirmation.""" event_id = generate_ulid() @@ -351,6 +433,19 @@ class DatastoreEngine: raise ValueError(f"End time {end} cannot be earlier than start time {start}") end_iso = end_dt.isoformat() + # Check for duplicate event within 15 minutes unless explicitly allowed + if not allow_duplicate: + existing_event = self._is_duplicate_event(kind, subject, payload, start_iso) + if existing_event: + feedback = self._get_event_feedback(kind, subject, start_dt.date()) + return { + "status": "success", + "duplicate": True, + "message": f"Duplicate event detected within 15 minutes (existing event ID: {existing_event['id']}). Event retained without duplicate insertion.", + "event": existing_event, + "feedback": feedback, + } + # Validation for open intervals in sleep if kind == "sleep.session" and end_iso is None: # Check if there is already an open session for this subject @@ -760,6 +855,7 @@ def run_mcp_server(): "estimated": {"type": "boolean", "description": "True if portions or macros are estimated"}, "meal": {"type": "string", "description": "Optional meal grouping (e.g. 'Frühstück', 'Mittagessen')"}, "note": {"type": "string", "description": "Free text notes or context"}, + "allow_duplicate": {"type": "boolean", "description": "Set to true to force adding duplicate event within 15m"}, }, "required": ["kind", "subject", "actor", "payload"], }, @@ -881,6 +977,7 @@ def run_mcp_server(): meal=args.get("meal"), note=args.get("note"), source_ref=args.get("source_ref"), + allow_duplicate=args.get("allow_duplicate", False), ) elif name == "datastore_stop_event": data = engine.stop_event( @@ -945,7 +1042,8 @@ def run_cli(): # Add Event p_add = subparsers.add_parser("add", help="Add an event") - p_add.add_argument("kind", help="e.g. nutrition.intake, sleep.session") + p_add.add_argument("kind", nargs="?", default=None, help="e.g. nutrition.intake, sleep.session") + p_add.add_argument("--kind", dest="kind_flag", help="e.g. nutrition.intake, sleep.session") p_add.add_argument("--subject", required=True, help="moritz, mila, amelie") p_add.add_argument("--actor", default="cli", help="Reporter identifier") p_add.add_argument("--payload", required=True, help="JSON payload string") @@ -955,10 +1053,12 @@ def run_cli(): p_add.add_argument("--meal", help="Meal grouping") p_add.add_argument("--note", help="Free-form note") p_add.add_argument("--estimated", action="store_true", help="Flag as estimated") + p_add.add_argument("--allow-duplicate", action="store_true", help="Allow duplicate event within 15 minutes") # Stop Event p_stop = subparsers.add_parser("stop", help="Stop an open interval event") - p_stop.add_argument("kind", help="e.g. sleep.session") + p_stop.add_argument("kind", nargs="?", default=None, help="e.g. sleep.session") + p_stop.add_argument("--kind", dest="kind_flag", help="e.g. sleep.session") p_stop.add_argument("--subject", required=True, help="mila, moritz") p_stop.add_argument("--end", help="ISO end time (defaults to now)") @@ -982,9 +1082,12 @@ def run_cli(): engine = DatastoreEngine() if args.command == "add": + kind = args.kind or args.kind_flag + if not kind: + parser.error("the 'add' command requires 'kind' (positional or via --kind)") payload = json.loads(args.payload) res = engine.add_event( - kind=args.kind, + kind=kind, subject=args.subject, actor=args.actor, payload=payload, @@ -994,13 +1097,16 @@ def run_cli(): meal=args.meal, note=args.note, estimated=args.estimated, + allow_duplicate=args.allow_duplicate, ) print(json.dumps(res, indent=2, ensure_ascii=False)) elif args.command == "stop": - res = engine.stop_event(kind=args.kind, subject=args.subject, end_time=args.end) + kind = args.kind or args.kind_flag + if not kind: + parser.error("the 'stop' command requires 'kind' (positional or via --kind)") + res = engine.stop_event(kind=kind, subject=args.subject, end_time=args.end) print(json.dumps(res, indent=2, ensure_ascii=False)) - elif args.command == "query": res = engine.query(args.sql) print(json.dumps(res, indent=2, ensure_ascii=False)) diff --git a/skills/datastore/tests/test_datastore.py b/skills/datastore/tests/test_datastore.py index f8d1d17..7d39730 100644 --- a/skills/datastore/tests/test_datastore.py +++ b/skills/datastore/tests/test_datastore.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 import datetime import json +import os import shutil import tempfile from pathlib import Path @@ -21,6 +22,7 @@ def temp_datastore(): if schema_src.is_dir(): shutil.copytree(schema_src, data_dir / "_schema") + (data_dir / "_meta.json").write_text("{}") engine = DatastoreEngine(data_dir=data_dir) yield engine shutil.rmtree(temp_dir) @@ -384,3 +386,191 @@ def test_sleep_over_midnight_attribution(temp_datastore: DatastoreEngine): assert "sleep" not in thu_summary or thu_summary["sleep"] is None +def test_duplicate_suppression_nutrition(temp_datastore: DatastoreEngine): + """Verifies that duplicate intake events within 15 minutes are suppressed idempotently.""" + start_time = "2026-10-09T08:00:00+02:00" + payload = { + "item": "Protein Porridge", + "qty": 300, + "unit": "g", + "protein_g": 35.0, + "kcal": 400.0, + } + + # 1. First logging + res1 = temp_datastore.add_event( + kind="nutrition.intake", + subject="moritz", + actor="agent:koch", + start=start_time, + payload=payload, + ) + assert res1["status"] == "success" + assert res1.get("duplicate") is not True + first_id = res1["event"]["id"] + + # 2. Duplicate logging 5 minutes later + res2 = temp_datastore.add_event( + kind="nutrition.intake", + subject="moritz", + actor="agent:koch", + start="2026-10-09T08:05:00+02:00", + payload=payload, + ) + assert res2["status"] == "success" + assert res2.get("duplicate") is True + assert res2["event"]["id"] == first_id + + # Verify DuckDB summary: only 1 item counted + sum1 = temp_datastore.get_summary(subject="moritz", date_str="2026-10-09") + assert sum1["nutrition"]["item_count"] == 1 + assert sum1["nutrition"]["total_kcal"] == 400.0 + + # 3. Explicit bypass with allow_duplicate=True + res3 = temp_datastore.add_event( + kind="nutrition.intake", + subject="moritz", + actor="agent:koch", + start="2026-10-09T08:06:00+02:00", + payload=payload, + allow_duplicate=True, + ) + assert res3["status"] == "success" + assert res3.get("duplicate") is not True + assert res3["event"]["id"] != first_id + + # Verify DuckDB summary: 2 items counted + sum2 = temp_datastore.get_summary(subject="moritz", date_str="2026-10-09") + assert sum2["nutrition"]["item_count"] == 2 + assert sum2["nutrition"]["total_kcal"] == 800.0 + + # 4. Different item within 15 minutes is NOT suppressed + res4 = temp_datastore.add_event( + kind="nutrition.intake", + subject="moritz", + actor="agent:koch", + start="2026-10-09T08:07:00+02:00", + payload={"item": "Apfel", "qty": 150, "kcal": 80.0}, + ) + assert res4.get("duplicate") is not True + + # 5. Same item outside 15 minutes window is NOT suppressed + res5 = temp_datastore.add_event( + kind="nutrition.intake", + subject="moritz", + actor="agent:koch", + start="2026-10-09T09:30:00+02:00", + payload=payload, + ) + assert res5.get("duplicate") is not True + + +def test_duplicate_suppression_sleep(temp_datastore: DatastoreEngine): + """Verifies that starting the same sleep session within 15 minutes returns existing open session.""" + res1 = temp_datastore.add_event( + kind="sleep.session", + subject="mila", + actor="agent:hebamme", + start="2026-10-09T13:00:00+02:00", + payload={"type": "nap", "location": "bett"}, + ) + assert res1["status"] == "success" + first_id = res1["event"]["id"] + + # Repeat 3 minutes later: should return existing open session without throwing ValueError + res2 = temp_datastore.add_event( + kind="sleep.session", + subject="mila", + actor="agent:hebamme", + start="2026-10-09T13:03:00+02:00", + payload={"type": "nap", "location": "bett"}, + ) + assert res2["status"] == "success" + assert res2.get("duplicate") is True + assert res2["event"]["id"] == first_id + + # Stop session + temp_datastore.stop_event( + kind="sleep.session", + subject="mila", + end_time="2026-10-09T14:00:00+02:00", + ) + + +def test_cli_kind_flag_and_positional(temp_datastore: DatastoreEngine): + """Verifies CLI accepts both positional and --kind flag for add and stop commands.""" + import subprocess + import sys + from pathlib import Path + + script_path = Path(__file__).resolve().parent.parent / "scripts" / "datastore.py" + data_dir = str(temp_datastore.data_dir) + + # 1. Add using --kind flag + proc_add = subprocess.run( + [ + sys.executable, + str(script_path), + "add", + "--kind", + "nutrition.intake", + "--subject", + "moritz", + "--payload", + '{"item": "CLI Snack", "qty": 50, "kcal": 100}', + "--start", + "2026-10-09T15:00:00+02:00", + ], + env=dict(os.environ, OPENCLAW_DATA_DIR=data_dir), + capture_output=True, + text=True, + ) + assert proc_add.returncode == 0 + data = json.loads(proc_add.stdout) + assert data["status"] == "success" + assert data["event"]["kind"] == "nutrition.intake" + + # 2. Add open sleep using positional kind + proc_sleep_start = subprocess.run( + [ + sys.executable, + str(script_path), + "add", + "sleep.session", + "--subject", + "mila", + "--payload", + '{"type": "nap"}', + "--start", + "2026-10-09T15:00:00+02:00", + ], + env=dict(os.environ, OPENCLAW_DATA_DIR=data_dir), + capture_output=True, + text=True, + ) + assert proc_sleep_start.returncode == 0 + + # 3. Stop sleep using --kind flag + proc_sleep_stop = subprocess.run( + [ + sys.executable, + str(script_path), + "stop", + "--kind", + "sleep.session", + "--subject", + "mila", + "--end", + "2026-10-09T16:00:00+02:00", + ], + env=dict(os.environ, OPENCLAW_DATA_DIR=data_dir), + capture_output=True, + text=True, + ) + assert proc_sleep_stop.returncode == 0 + stop_data = json.loads(proc_sleep_stop.stdout) + assert stop_data["status"] == "success" + assert stop_data["duration_minutes"] == 60 + + +