From 0686b93bebebe7b7e10f1a8740ce173ee03097bd Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 3 Nov 2026 20:51:06 -0500 Subject: [PATCH] fix(flow): count duplicate rows in keyless tables, key the AppWorld seed cache by version, accept tz-aware stamps --- .../verifier_audit/targets/agentdojo.py | 2 ++ .../verifier_audit/targets/appworld_worker.py | 32 ++++++++++++----- packages/flow/tests/test_verifier_audit.py | 35 +++++++++++++++++++ 3 files changed, 61 insertions(+), 8 deletions(-) diff --git a/packages/flow/openadapt_flow/verifier_audit/targets/agentdojo.py b/packages/flow/openadapt_flow/verifier_audit/targets/agentdojo.py index 54de2364c3..615f504fad 100644 --- a/packages/flow/openadapt_flow/verifier_audit/targets/agentdojo.py +++ b/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 "" return value diff --git a/packages/flow/openadapt_flow/verifier_audit/targets/appworld_worker.py b/packages/flow/openadapt_flow/verifier_audit/targets/appworld_worker.py index 5546cfa7a1..fad895c66c 100644 --- a/packages/flow/openadapt_flow/verifier_audit/targets/appworld_worker.py +++ b/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") diff --git a/packages/flow/tests/test_verifier_audit.py b/packages/flow/tests/test_verifier_audit.py index f6a1c940d8..1c3bb7f708 100644 --- a/packages/flow/tests/test_verifier_audit.py +++ b/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) == "" + assert _clock_free("2020-01-01T00:00:00+00:00", window) != ""