fix(flow): verifier-audit minor follow-ups (keyless-table diff, seed-cache version, tz-aware stamps)

packages/flow/openadapt_flow/verifier_audit/targets/agentdojo.py

@@ -56,6 +56,8 @@ def _clock_free(value: Any, window: tuple[datetime, datetime]) -> Any:
             stamp = datetime.fromisoformat(value)
         except ValueError:
             return value
+        if stamp.tzinfo is not None:  # the window is naive local time
+            stamp = stamp.astimezone().replace(tzinfo=None)
         if window[0] <= stamp <= window[1]:
             return "<run-clock>"
     return value

packages/flow/openadapt_flow/verifier_audit/targets/appworld_worker.py

@@ -59,6 +59,10 @@ def _tables(conn: sqlite3.Connection, schema: str) -> set[str]:
     return {r[0] for r in conn.execute(query)}
 
 
+def _has_primary_key(conn: sqlite3.Connection, schema: str, table: str) -> bool:
+    return any(row[5] for row in conn.execute(f"PRAGMA {schema}.table_info({table})"))
+
+
 def _select(conn: sqlite3.Connection, sql: str) -> list[str]:
     cur = conn.execute(sql)
     cols = [d[0] for d in cur.description]
@@ -69,9 +73,9 @@ def _select(conn: sqlite3.Connection, sql: str) -> list[str]:
 
 def _delta(seed_dir: Path, end_dir: Path) -> list[dict]:
     """Rows added or removed between two full dumps, computed inside SQLite
-    (``EXCEPT``) so a run never serializes the unchanged tables. Every
-    AppWorld table carries a primary key, so set difference is the row
-    multiset difference."""
+    (``EXCEPT``) so a run never serializes the unchanged tables. ``EXCEPT`` is
+    a set difference, so a table without a primary key is compared with its
+    rowid included: two identical inserts stay two rows."""
     rows = []
     names = sorted(
         {p.name for p in seed_dir.glob("*.db")} | {p.name for p in end_dir.glob("*.db")}
@@ -91,19 +95,28 @@ def _delta(seed_dir: Path, end_dir: Path) -> list[dict]:
             stem = name[:-3]
             for table in sorted(seed | end):
                 q = f'"{table}"'
+                cols = "*"
+                if not _has_primary_key(conn, "e" if table in end else "s", q):
+                    cols = "rowid AS _rowid, *"
                 if table in seed and table in end:
                     added = _select(
-                        conn, f"SELECT * FROM e.{q} EXCEPT SELECT * FROM s.{q}"
+                        conn,
+                        f"SELECT {cols} FROM e.{q} EXCEPT SELECT {cols} FROM s.{q}",
                     )
                     removed = _select(
-                        conn, f"SELECT * FROM s.{q} EXCEPT SELECT * FROM e.{q}"
+                        conn,
+                        f"SELECT {cols} FROM s.{q} EXCEPT SELECT {cols} FROM e.{q}",
                     )
                 else:
                     added = (
-                        _select(conn, f"SELECT * FROM e.{q}") if table in end else []
+                        _select(conn, f"SELECT {cols} FROM e.{q}")
+                        if table in end
+                        else []
                     )
                     removed = (
-                        _select(conn, f"SELECT * FROM s.{q}") if table in seed else []
+                        _select(conn, f"SELECT {cols} FROM s.{q}")
+                        if table in seed
+                        else []
                     )
                 rows += [
                     {"table": f"{stem}.{table}", "op": "added", "row": r} for r in added
@@ -128,7 +141,10 @@ def __init__(self, appworld_root: str, cache: Path) -> None:
         self.n = 0
 
     def _seed_dir(self, world, task: str) -> Path:
-        path = self.cache / task
+        from appworld import __version__
+
+        # Keyed by AppWorld version: another release may ship other seed data.
+        path = self.cache / f"appworld-{__version__}" / task
         if not (path / ".complete").is_file():
             _dump(world, path)
             (path / ".complete").write_text("full seed dump\n")

packages/flow/tests/test_verifier_audit.py

@@ -211,3 +211,38 @@ def test_family_spec_is_shareable_json():
     ids = {f["id"] for f in spec["families"]}
     assert set(spec["server_fault_modes"].values()) <= ids
     json.dumps(spec)
+
+
+def test_sqlite_delta_counts_duplicate_rows_in_a_table_without_a_key(tmp_path):
+    import sqlite3
+
+    from openadapt_flow.verifier_audit.targets.appworld_worker import _delta
+
+    for name in ("seed", "end"):
+        (tmp_path / name).mkdir()
+        conn = sqlite3.connect(tmp_path / name / "app.db")
+        conn.execute("CREATE TABLE queue (song INTEGER)")
+        conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
+        conn.execute("INSERT INTO queue VALUES (1)")
+        if name == "end":
+            conn.execute("INSERT INTO queue VALUES (7)")
+            conn.execute("INSERT INTO queue VALUES (7)")
+            conn.execute("INSERT INTO users VALUES (1, 'ann')")
+        conn.commit()
+        conn.close()
+    rows = _delta(tmp_path / "seed", tmp_path / "end")
+    assert sum(r["table"] == "app.queue" and r["op"] == "added" for r in rows) == 2
+    users = [json.loads(r["row"]) for r in rows if r["table"] == "app.users"]
+    assert users == [{"id": 1, "name": "ann"}]
+
+
+def test_clock_free_handles_timezone_aware_stamps():
+    from datetime import datetime, timedelta, timezone
+
+    from openadapt_flow.verifier_audit.targets.agentdojo import _clock_free
+
+    now = datetime.now()
+    window = (now - timedelta(seconds=5), now + timedelta(seconds=5))
+    aware = datetime.now(timezone.utc).isoformat()
+    assert _clock_free(aware, window) == "<run-clock>"
+    assert _clock_free("2020-01-01T00:00:00+00:00", window) != "<run-clock>"