feat(datastore): support --kind flag and add 15m duplicate suppression

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"):
+    for env_var in ["DATASTORE_DIR", "OPENCLAW_DATA_DIR"]:
-        return Path(os.environ["DATASTORE_DIR"]).resolve()
+        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))

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
+
+
+