[abrichr]

Size: about 6,500 added code lines (generator 5,000, verifier 80, tests 1,400), over the 1,000-line bar. Merge by hand.

What this is

Until now the contract generator (ensure_contract) only worked on FHIR, so contracts for other apps were written by hand, with 10 or 11 person decisions per app.

python -m openadapt_flow.generate.verify_contract writes openadapt.verify/v1 contracts from:

• the store's schema (MySQL, MariaDB, PostgreSQL), read with the contracts' own read-only account;
• the kinds of write with their parameter names and example values;
• one demonstration write per example, snapshotted before and after (snapshot, then record).

It finds where each value lands by exact rules only: equal text; email by case folding; phone by digits, but only for a parameter the person confirms; numbers, dates and date-times by value in SQL; a display column behind a key; first and last name joined by a space; minutes between two date-times; rows the write added beside the record. It then writes:

• one contract per field set the kind's tasks can use (an update gets one for each set of fields it may change), list parameters by length, a parameter that names the field to change by its value, and a found twin when the record already holds the values;
• the whole-database frame (frame.database, with the store's dialect, PostgreSQL included) and recheck_after_s from the person's answer;
• frame.versions for updates, from a version column that steps by one per save or from the app's audit trail of saves, narrowed to write events;
• rows the app adds on its own, declared by count and constant columns; logs only as the person names them;
• manifest.yaml and choose(), which picks the contract for a task's values. No contract fits: GenerateError, so the write is stopped, never sent unchecked.

Every place it can't decide is one numbered question in QUESTIONS.md; answers go in answers.yaml. Every contract is loaded and dry-read against the live store before it is written. Setup steps are applied with a flag through an admin connection, or the kind is refused with the statement to run (--grant-trigger, --create-views).

Docs: packages/flow/docs/GENERATE_CONTRACTS.md, a pointer from VERIFY_WRITES.md, --help on both generators, the CAPABILITIES row, llms-full.txt.

Round 3 review: what changed

• SF1. The bookkeeping proposal names only the columns that vary or are time stamps. Free text every demonstration left empty stays pinned empty unless the person adds it to unchecked. On a row the app adds beside the write, such a column must hold the exact empty value the demonstrations stored (NULL or ''), which also covers the NIT on '' against a NULL default.
• SF3. The versions read from an audit trail now uses a deny-list. Every entry counts as a save except an event code shown to be a read: one a demonstration logged on a record it didn't add or change (in any table), and never on one it did. An unseen code (a bulk edit, an import) or a NULL code counts as a save.
• SF4. PostgreSQL check_read_only checks every role the account can act as, including roles reachable only with SET ROLE (a NOINHERIT membership). It also checks column-level INSERT, UPDATE and REFERENCES grants.
• SF6.1. The audit table's own primary key is never a candidate record column.
• SF2 (docs). A version column is trusted to step from the stored value on every save. An app with an audit trail can read it as well.
• SF5, the rest of SF6 and the nits are in Contract generator: two false stops from the #1672 round-2 review #1675.

Each item has the reviewer's attack as a test in tests/test_generate_verify_contract_attacks_r3.py.

Round 2 review: what changed

• Lost update with no earlier version (blocker 1). A target is new only when the read before the write shows it wasn't there. For a surfaces verifier that is the effect's own surface read; with no read, the target is treated as existing. With no version before the write, the target stops as missing. The verify_run code comes from Catch a lost update by reading the target's versions #1662. frame.versions.target_surface makes only the main record a target; rows the app touches beside it are not.

• Version source. A version column that steps by one comes first, because every existing record has a version there. The audit trail is the fallback. Its write-event filter now compares update rows (one code across every update kind, so it can't vary with the fields saved) only with rows from create kinds. It keeps both codes (action IN (create, update)), so a record the app created has a first version.

• Per-surface target fields (blocker 2). frame.target_fields_by_surface lists, for each surface, the fields its declared rows may change. A table counts as the surface that database.views maps it to. A touched parent row can't take the target's fields, and the target can't take the parent's.

• A key back to the main table (SF1). A parent issue named by its subject must name exactly one row too. Only a child row's link to its own record skips the check.

• Answers (SF3).

  • bookkeeping accepts accept or a mapping with only accept, logs_add, columns_add, columns_remove and unchecked; any other key is refused.
  • A mapping takes the proposed columns only with accept: true.
  • logs_add, columns_add and unchecked can name only what the question listed. A log applies only to the kinds whose demonstrations added rows there.

• New-record columns (SF4). Every column that varies, is free text left empty, or is the write's time stamp must be named under unchecked, or else:

  • free text not named is pinned empty on the main record and its child rows, and holds its default on rows the app adds;
  • any other column not named leaves the kind without a contract.

  The proposal names every listed column, so accepting it is explicit.

• PostgreSQL read-only check (SF5). The check now uses effective privileges, including inherited and predefined roles: INSERT, UPDATE, DELETE, TRUNCATE, TRIGGER and REFERENCES on tables; USAGE and UPDATE on sequences; and CREATE on schemas and the database.

• SF2 (a row the demonstrations happened to share becomes a literal-key excuse) and SF6 (a per-field audit table refused) are in Contract generator: two false stops from the #1672 round-2 review #1675.

Each item has the reviewer's attack as a test in tests/test_generate_verify_contract_attacks_r2.py. The tracker there audits one row per field and touches the project on each save; the wrong writes stop and the correct writes are CONFIRMED.

Round 1 review: what changed

• B1 phone. The phone rule applies only to a parameter the person confirms (phones:). At run time a value compares as a phone only when it has the same digits and at least 7 of them; any other value is compared exactly.
• B2 bookkeeping. A column is excused per (table, column), and only on the rows the demonstrations showed changing: the target, a row with the same literal key (for every kind), or a row the task's values name (only in field sets that give those values). Any other change stops the write.
• B3 found twin. found pins the same constants as create. A display lookup behind a key skips soft-deleted rows, and all the values the task reads through one key must together name exactly one row.
• B4 child rows. Child surfaces link to the main record by identity only, so a correct update of a record with earlier history no longer reads as collateral.
• Logs. There is no name rule. The bookkeeping question lists the tables the app added rows to, and a table is a log only if the person names it (logs_add). A stream-note table is not a log by default.
• New-record columns. Every column of a new record is pinned (new_row_fields) or listed in the bookkeeping question.
• frame.versions sources. The read is limited to write events: a letters-only code that every update has and no create has, never an id column. If the agent's own save adds no version (the setting is off), the write is stopped.
• --grant-trigger. The account is quoted by the driver. check_read_only reports TRIGGER.
• Docs. The false claims are fixed and the CAPABILITIES status no longer says "On main".
• PostgreSQL (after Whole-database frame and recheck on PostgreSQL #1673). It now gets frame.database with dialect: postgresql and the recheck question, and the PostgreSQL limit is removed from the docs. The reviewer's PostgreSQL log-rewrite attack is now a plain test, and the write is stopped.

Each reviewer attack is a test in tests/test_generate_verify_contract_attacks.py.

Real-app results

The generator was run on instance copies of three apps from the first-contact study, as a new customer would run it: only the read-only account, the kinds, parameter names and examples, and demonstrations made through each app's REST API. The study's tasks, oracle and planted faults were used only afterwards, to judge the contracts.

	Dolibarr	Redmine	EspoCRM
Database	MariaDB 11.4	PostgreSQL	MariaDB
Questions for a person	4	2	3
Person decisions with hand-written contracts	11	11	11
Correct writes confirmed	32/43	32/38	27/40
In-scope planted faults stopped	22/22	19/19	21/21
Out-of-scope late writes stopped	2/2	1/2	1/2

Measured after the round-3 fixes: Redmine and Dolibarr at abaa14c2f, EspoCRM at 19d74134a. The commit between them only narrows which codes count as reads, which can't change the other two apps' reads. Dolibarr's 8 invoice creates now stop, because the reference write fills each invoice line's description, which every demonstration left empty and the proposal now keeps pinned empty. A person can add that column under unchecked. The evidence (scripts, per-app questions, answers, reports, example contracts and the per-case results) is in the private study repository, because the study's fault sets stay private.

Tests

• tests/test_generate_verify_contract.py (18): a throwaway PostgreSQL (initdb) with a toy issue-tracker schema and two demonstrations per kind. Correct writes are CONFIRMED, and the toy's faults are stopped.
• tests/test_generate_verify_contract_attacks.py (10): the round-1 reviewer attacks on the same engine.
• tests/test_generate_verify_contract_attacks_r2.py (16): the round-2 attacks, answer validation, and the PostgreSQL privilege check.
• tests/test_generate_verify_contract_attacks_r3.py (12): the round-3 attacks on an audit-trail tracker, and SET ROLE and column grants.

Both are skipped where PostgreSQL isn't installed. tests/test_generate_ensure_contract.py still passes.

Not done

• REST metadata (OpenAPI) is not read. FHIR stays on ensure_contract.
• The recheck doesn't re-read the main record's table, because the contract reads it through its own surface.
• There is no unit test on a local MySQL engine. MySQL and MariaDB are covered by the real-app runs.
• The branch was rebased onto main for Whole-database frame and recheck on PostgreSQL #1673. Because force-push is blocked here, the old tip was merged back with -s ours; the diff against main is unaffected.

[abrichr]

Adversarial review (round 2)

Verdict: BLOCKED (2 blockers, plus public text to clean before merge)

Reviewed at cd4d48176. The PR's tests pass locally against a throwaway PostgreSQL (initdb): test_generate_verify_contract.py, test_generate_verify_contract_attacks.py and test_generate_ensure_contract.py. All outcomes below come from runs on that engine. The attack files are in the collapsed sections at the end.

Round 1, rechecked

I reran the round-1 attack file unchanged. The 9 attacks that were CONFIRMED in round 1 are now stopped (8 REFUTED, 1 UNCERTAIN). The colleague-history update (A5) is now CONFIRMED, and the deleted control (B4) is still stopped.

Round-1 item	Now
B1 phone	Fixed. The phone rule is proposed only when the app stored other text with the same digits, and it applies only to parameters listed under phones:. choose() compares a value with fewer than 7 digits, or a value not shaped like a phone number, exactly. (NIT: that check is in _load_for, so a contract loaded without choose() skips it.)
B2 bookkeeping	Fixed for other rows and other tables (B1 and B2 attacks stopped). New hole: see blocker 2.
B3 found and soft-delete	Fixed. found pins the same constants. A key hop filters deleted, is_deleted and deleted_at, and a display value must name exactly one live row. Gap: see should-fix 1.
B4 child-surface false stop	Fixed. main__ uses identity parameters only.
SF1 unique display	Fixed for keys into other tables, including first and last name together. Not for a key back to the main table (should-fix 1).
SF2 PostgreSQL logs	Fixed. frame.database is emitted for PostgreSQL. A rewrite of a log row is stopped, and so are an unrelated insert, a rewrite in another schema, a renamed status, and an edited journal (see the frame test file).
SF3 log classification	Fixed. There is no name regex and no count rule. A table is a log only when the person names it (should-fix 3 covers what the answer can name).
SF4 new-record columns	Partly fixed. The columns are listed, but no answer turns them on (should-fix 4).
SF5 frame.versions sources	The filter is new and can hide a lost update (blocker 1).
SF6 --grant-trigger	Fixed: the driver escapes the account name. TRIGGER appears as a note in read_account, not as a check_read_only problem. PostgreSQL has a gap (should-fix 5).

The four real-app fixes: the description-text key match (by_key now requires literal selectors), per-field-set rows (verify_contract.py:1884), and logs_add for every kind all look correct, apart from should-fix 3. The write-event filter is blocker 1.

Blockers

1. With a version read from an audit table, a lost update is CONFIRMED when the target has no earlier row in that read. The new write-event filter makes this the usual case.

• _judge_versions (verify_run.py:500-518) decides whether a target existed from before.records. A surfaces verifier always returns records=[] (surfaces.py:119), and every generated contract is a surfaces verifier. So existed is always False. A target with no row in the baseline version read is treated as "created in the window" (versions.py:272), and its versions are not checked.
• The filter added at verify_contract.py:1186-1210 keeps only rows holding the one code every update row had. A record that never had such an event before (never edited since it was created, or never edited in that field) has no baseline version. Its first edit is then unprotected. Spied targets: [('18', False, False), ...] for an issue that existed before the window.
• Fixture (audits table, one row per changed field; edit_description reads audits WHERE field_name = 'description'):
  • Another user sets the description inside the window, and the agent's save overwrites it: CONFIRMED (test_v5).
  • Another user sets the due date, and the agent's stale save puts the old due date back: CONFIRMED (test_v4).
  • Control: set_status, whose versions come from lock_version, stops the same lost update (test_v2, test_v3).
• The filter also compares against rows from every other kind's demonstrations (for d in all_demos if d not in plan.demos), not only creates, as the docstring and PR body say. That is how field_name = 'description' was chosen: the other update kinds wrote other field names.
• Fix: work out existed from the per-surface baseline in before.detail. A target that existed and has no version should give not_checked. Compare the filter only against rows from create kinds, and drop a column whose values in update rows vary with the field set.

2. A row the app touches beside the write may also take the target's changed fields, because frame.target_fields is one flat set.

• build_contract adds the touched row's excused columns and the target's changed columns to the same frame.target_fields (verify_contract.py:1875-1905, :1957). The frame applies may_change to every declared row of every surface (frame.py:841, :1545). A touched row is declared through its raw_<table> effect.
• Fixture: the app touches projects.updated_on on each issue save, as a parent with touch: true does. Both projects and issues have a description column. Task edit_description: a correct save, plus UPDATE projects SET description = 'Overwritten by the agent' on the project the task names: CONFIRMED (test_t2). Control: a different stray column on the target is stopped (test_t3).
• The reverse also holds: a column excused on the touched row may change on the target row. Parent and child tables often share column names (description, name, status, notes), so this is the round-1 B2 class again, by column name across rows.
• Fix: per-surface (or per-effect) target fields, or refuse a kind whose touched table shares a column name with the target's change set.

SHOULD-FIX

1. A display value that leads back to the main table skips the exactly-one-row check. verify_contract.py:1453 skips the check when hop.table == self.anchor, which is meant for a child's link to its own record. A parameter that names another record of the main table, such as a parent issue named by its subject, is also skipped. Task parent: "Refund button missing" (two issues in two projects have that subject), and the agent links the other one: CONFIRMED (test_p1). Apply the check when the hop isn't a child's link to the target.
2. A row two demonstrations happened to share becomes a literal-key excuse (false stop). Both set_status demonstrations were in one project, so the touched project row became projects row id=3 (by key, app-wide). A correct set_status in another project is REFUTED "1 record(s) outside the declared write moved" (test_k1). Prefer the selector built from the task's values when one exists, and use a literal key only when no located value names the row.
3. An answer can excuse more than its question offers. logs_add is checked against log candidates and counted side tables (verify_contract.py:2818). A table the question lists as "expected exactly that many times" can be made an append-anything log for every kind, and its count and pins are dropped. Any mapping under bookkeeping also counts as accept (:2772): {logs_add: [x]} silently accepts every proposed column. The PR's own fixture answers this way. Accept logs_add only from the listed candidates, and require accept: true before taking the proposed columns.
4. New-record columns marked "varies", "free" or "timestamp" stay unchecked whatever the answer. unchecked is used only in the question text. "Unanswered, nothing is excused" is not true for these columns, and no answer turns them on. Either pin them when the person doesn't accept, or reword the question.
5. PostgreSQL check_read_only misses inherited and non-DML privileges. It filters on grantee = current_user, so a reader that is a member of a role with INSERT passes, and its INSERT then succeeds. It doesn't check TRIGGER or CREATE on the schema either. Reader ro1 (a member of writers, which has INSERT, UPDATE and DELETE) and reader ro2 (TRIGGER on the table, CREATE on public) both return [] (test_r1_pg...). Use has_table_privilege(...) over all tables in the read schemas, and has_schema_privilege(..., 'CREATE').
6. A multi-field kind with a per-field audit table is refused. edit_plan (one demonstration changed two fields, the other one) was analyzed as a create on audits and refused: "the main record is issues (update), but other demonstrations write audits (create)". This fails closed, but it is common in apps that audit each field.

NIT

• _judge_versions targets include the touched rows' keys (a project id), which are looked up in the issue-keyed audit read. That can only stop a write, never pass one.
• The recheck still doesn't cover mode: skip tables. The PR body discloses this.

Public text

Several public places name third-party apps used in the private validation and quote its private per-app counts: the PR body, the CAPABILITIES.md row, llms-full.txt, GENERATE_CONTRACTS.md (around the "Lost updates" and limits sections), comments and docstrings in verify_contract.py, the docstring of test_generate_verify_contract.py, and several commit subjects on this branch. Replace them with neutral examples ("an issue tracker", "a CRM") and keep the results in the private repository. The squash commit message should not carry them either.

Round-2 attack tests (tests/test_review_1672_r2_attacks.py; run alone, it creates its own database)

# ruff: noqa: F811
"""Adversarial review of #1672, round 2 (not for merge). Real PostgreSQL."""

from __future__ import annotations

import json
import os
from typing import Any

import pytest
import yaml

psycopg2 = pytest.importorskip("psycopg2")

from openadapt_flow.generate import verify_contract as gen  # noqa: E402
from tests.test_generate_verify_contract import (  # noqa: E402,F401
    BOT,
    SCHEMA,
    App,
    _run,
    pg,
)


def _outcome(o: str) -> str:
    print("OUTCOME:", o)
    return o


SCHEMA_V = (
    SCHEMA.replace(
        "CREATE TABLE projects (id serial PRIMARY KEY, name text NOT NULL);",
        "CREATE TABLE projects (id serial PRIMARY KEY, name text NOT NULL, "
        "description text, updated_on timestamp);",
    )
    .replace(
        "lock_version integer NOT NULL DEFAULT 0);",
        "lock_version integer NOT NULL DEFAULT 0, description text, parent_id integer);",
    )
    .replace(
        "CREATE TABLE settings",
        "CREATE TABLE audits (id serial PRIMARY KEY, issue_id integer, user_id integer, "
        "action text, field_name text, old_value text, new_value text);\nCREATE TABLE settings",
    )
    .replace("CREATE ROLE oa_reader LOGIN PASSWORD 'reader-pw';", "")
)


class AppV(App):
    """Audits one row per changed field; touches the project on every save
    (a parent ``touch``); writes no journals."""

    def create_issue(self, v: dict[str, Any], *, project_id: Any = None) -> int:
        pid = project_id or self.one("SELECT id FROM projects WHERE name=%s", (v["project"],))
        sid = self.one("SELECT id FROM statuses WHERE name=%s", (v["status"],))
        parent = None
        if v.get("parent"):
            parent = v.get("parent_id") or self.one(
                "SELECT i.id FROM issues i WHERE i.subject=%s ORDER BY id LIMIT 1", (v["parent"],))
        iid = self.one(
            "INSERT INTO issues (project_id, subject, status_id, author_id, created_on, updated_on, parent_id) "
            "VALUES (%s,%s,%s,%s,now(),now(),%s) RETURNING id", (pid, v["subject"], sid, BOT, parent))
        self.q("INSERT INTO audits (issue_id, user_id, action) VALUES (%s,%s,'create')", (iid, BOT))
        self.q("UPDATE projects SET updated_on=now() WHERE id=%s", (pid,))
        return iid

    def save(self, iid: int, sets: dict[str, Any], *, user: int = BOT) -> None:
        for col, val in sets.items():
            if col == "status":
                col, val = "status_id", self.one("SELECT id FROM statuses WHERE name=%s", (val,))
            old = self.one(f"SELECT {col}::text FROM issues WHERE id=%s", (iid,))
            self.q(f"UPDATE issues SET {col}=%s WHERE id=%s", (val, iid))
            self.q("INSERT INTO audits (issue_id, user_id, action, field_name, old_value, new_value) "
                   "VALUES (%s,%s,'update',%s,%s,%s)", (iid, user, col, old, None if val is None else f"v{val}"))
        self.q("UPDATE issues SET updated_on=now(), lock_version=lock_version+1 WHERE id=%s", (iid,))
        self.q("UPDATE projects SET updated_on=now() WHERE id=(SELECT project_id FROM issues WHERE id=%s)", (iid,))


KINDS_V = {
    "kinds": [
        {"kind": "create_issue", "parameters": ["project", "subject", "status", "parent"],
         "examples": [
             {"values": {"project": "Help Desk", "subject": "Toner low", "status": "New", "parent": "Printer jams"}},
             {"values": {"project": "Billing Portal", "subject": "VPN cert", "status": "New", "parent": "VPN drops"}},
         ]},
        {"kind": "set_status", "parameters": ["project", "subject", "status"],
         "examples": [
             {"values": {"project": "Help Desk", "subject": "Printer jams", "status": "In Progress"}},
             {"values": {"project": "Help Desk", "subject": "VPN drops", "status": "Resolved"}},
         ]},
        {"kind": "edit_plan", "parameters": ["project", "subject", "done_ratio", "due_date"],
         "examples": [
             {"values": {"project": "Billing Portal", "subject": "Refund button missing",
                         "done_ratio": 30, "due_date": "2026-11-30"}},
             {"values": {"project": "Help Desk", "subject": "Printer jams", "done_ratio": 50, "due_date": None}},
         ]},
        {"kind": "edit_description", "parameters": ["project", "subject", "description"],
         "examples": [
             {"values": {"project": "Help Desk", "subject": "VPN drops", "description": "Drops every hour"}},
             {"values": {"project": "Billing Portal", "subject": "Refund button missing",
                         "description": "Only on Safari"}},
         ]},
    ]
}


@pytest.fixture(scope="module")
def gv(pg: Any) -> dict[str, Any]:
    root, port = pg["root"], pg["port"]
    admin = psycopg2.connect(host="127.0.0.1", port=port, user="admin", dbname="postgres")
    admin.autocommit = True
    admin.cursor().execute("CREATE DATABASE trackerv")
    admin.close()
    app = AppV(port, "trackerv")
    with app.con() as c, c.cursor() as cur:
        cur.execute("DO $$BEGIN IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname='oa_reader') "
                    "THEN CREATE ROLE oa_reader LOGIN PASSWORD 'reader-pw'; END IF; END$$")
        cur.execute(SCHEMA_V)
    os.environ["OA_GEN_TEST_PW"] = "reader-pw"
    work = root / "workv"
    work.mkdir()
    read = {"sql_driver": "psycopg2",
            "sql_connect_args": {"host": "127.0.0.1", "port": port, "user": "oa_reader", "dbname": "trackerv"},
            "sql_password_env": "OA_GEN_TEST_PW"}
    (work / "read.yaml").write_text(yaml.safe_dump(read))
    (work / "kinds.yaml").write_text(yaml.safe_dump(KINDS_V, allow_unicode=True))
    common = ["--read", str(work / "read.yaml")]

    def demo(kind: str, values: dict[str, Any], write: Any) -> None:
        assert gen.main(["snapshot", *common, "--out", str(work / "before.json")]) == 0
        write()
        assert gen.main(["record", *common, "--before", str(work / "before.json"), "--kind", kind,
                         "--values", json.dumps(values), "--demos", str(work / "demos")]) == 0

    def find(v: dict[str, Any]) -> int:
        return app.find(v["project"], v["subject"])

    for k in KINDS_V["kinds"]:
        for ex in k["examples"]:
            v = ex["values"]
            if k["kind"] == "create_issue":
                demo("create_issue", v, lambda v=v: app.create_issue(v))
            else:
                sets = {p: v[p] for p in k["parameters"][2:] if v.get(p) is not None}
                demo(k["kind"], v, lambda v=v, s=sets: app.save(find(v), s))
    argv = ["propose", *common, "--kinds", str(work / "kinds.yaml"), "--demos", str(work / "demos"),
            "--out", str(work / "out"), "--actor-param", f"bot_user_id={BOT}"]
    assert gen.main(argv) == 0
    report = json.loads((work / "out" / "report.json").read_text())
    print("QUESTIONS:", (work / "out" / "QUESTIONS.md").read_text())
    answers = {q["key"]: q["proposal"] for q in report["questions"]}
    answers["bookkeeping"] = {"logs_add": ["audits"]}
    (work / "answers.yaml").write_text(yaml.safe_dump(answers))
    assert gen.main([*argv, "--answers", str(work / "answers.yaml")]) == 0
    report = json.loads((work / "out" / "report.json").read_text())
    if os.environ.get("OA_GEN_TEST_KEEP"):
        import shutil
        shutil.copytree(work, os.environ["OA_GEN_TEST_KEEP"] + "-v", dirs_exist_ok=True)
    return {"app": app, "out": work / "out", "report": report, "read": read}


def _fresh(app: AppV, subject: str) -> int:
    return app.create_issue({"project": "Help Desk", "subject": subject, "status": "New"})


def test_v0_report(gv: dict[str, Any]) -> None:
    for k in ("set_status", "edit_plan", "edit_description"):
        print(k, json.dumps(gv["report"]["kinds"][k].get("versions")),
              gv["report"]["kinds"][k].get("touched_rows"))
    print("create", gv["report"]["kinds"]["create_issue"].get("touched_rows"))


def test_v1_control_correct_status_update_confirmed(gv: dict[str, Any]) -> None:
    app = gv["app"]
    iid = _fresh(app, "V1 control")
    v = {"project": "Help Desk", "subject": "V1 control", "status": "Resolved"}
    out = _outcome(_run(gv, "set_status", v, lambda: app.save(iid, {"status": "Resolved"})))
    assert out.startswith("CONFIRMED"), out


def test_v2_control_step_versions_catch_a_lost_update(gv: dict[str, Any]) -> None:
    """Control: set_status reads its versions from lock_version (a step
    column), and the same lost update is stopped."""
    app = gv["app"]
    iid = _fresh(app, "V2 lost update")
    v = {"project": "Help Desk", "subject": "V2 lost update", "status": "Resolved"}

    def write() -> None:
        app.save(iid, {"due_date": "2026-12-31"}, user=2)
        app.save(iid, {"status": "Resolved", "due_date": None})

    out = _outcome(_run(gv, "set_status", v, write))
    assert not out.startswith("CONFIRMED"), out


def test_v3_control_other_user_same_field(gv: dict[str, Any]) -> None:
    app = gv["app"]
    iid = _fresh(app, "V3 same field")
    v = {"project": "Help Desk", "subject": "V3 same field", "status": "Resolved"}

    def write() -> None:
        app.save(iid, {"status": "In Progress"}, user=2)
        app.save(iid, {"status": "Resolved"})

    out = _outcome(_run(gv, "set_status", v, write))
    assert not out.startswith("CONFIRMED"), out


def test_t1_control_correct_description_edit(gv: dict[str, Any]) -> None:
    app = gv["app"]
    iid = _fresh(app, "T1 control")
    v = {"project": "Help Desk", "subject": "T1 control", "description": "Fixed"}
    out = _outcome(_run(gv, "edit_description", v, lambda: app.save(iid, {"description": "Fixed"})))
    assert out.startswith("CONFIRMED"), out


def test_t2_touched_parent_row_takes_the_targets_field(gv: dict[str, Any]) -> None:
    """The project row (touched by the app, named by the task's project) is
    a declared row; frame.target_fields is one flat set, so the issue's
    changed field 'description' may change on the project row too."""
    app = gv["app"]
    iid = _fresh(app, "T2 leak")
    v = {"project": "Help Desk", "subject": "T2 leak", "description": "Fixed"}

    def write() -> None:
        app.save(iid, {"description": "Fixed"})
        app.q("UPDATE projects SET description='Overwritten by the agent' WHERE name='Help Desk'")

    try:
        out = _outcome(_run(gv, "edit_description", v, write))
    finally:
        app.q("UPDATE projects SET description=NULL WHERE name='Help Desk'")
    assert out.startswith("CONFIRMED"), out


def test_t3_control_touched_parent_other_column(gv: dict[str, Any]) -> None:
    app = gv["app"]
    iid = _fresh(app, "T3")
    v = {"project": "Help Desk", "subject": "T3", "description": "Fixed"}

    def write() -> None:
        app.save(iid, {"description": "Fixed"})
        app.q("UPDATE projects SET name='Help Desk' WHERE name='Help Desk'")
        app.q("UPDATE issues SET is_private=true WHERE id=%s", (iid,))

    out = _outcome(_run(gv, "edit_description", v, write))
    assert not out.startswith("CONFIRMED"), out


def test_p0_control_correct_create_with_parent(gv: dict[str, Any]) -> None:
    app = gv["app"]
    v = {"project": "Help Desk", "subject": "P0 child", "status": "New", "parent": "VPN drops"}
    out = _outcome(_run(gv, "create_issue", v, lambda: app.create_issue(v)))
    assert out.startswith("CONFIRMED"), out


def test_p1_parent_named_by_a_subject_two_issues_share(gv: dict[str, Any]) -> None:
    """'Refund button missing' names two issues (two projects). A key back to
    the main table skips the exactly-one-row check, so either passes."""
    app = gv["app"]
    legacy = app.find("Billing Portal Legacy", "Refund button missing")
    v = {"project": "Help Desk", "subject": "P1 child", "status": "New", "parent": "Refund button missing"}
    out = _outcome(_run(gv, "create_issue", v, lambda: app.create_issue({**v, "parent_id": legacy})))
    assert out.startswith("CONFIRMED"), out


def test_v4_lost_update_on_the_audit_filtered_kind(gv: dict[str, Any]) -> None:
    """edit_description reads its versions from audits WHERE
    field_name='description' (a value no other kind's rows had). Another
    user's save of the due date inside the window is filtered out; the
    agent's stale save puts the old due date back."""
    app = gv["app"]
    iid = _fresh(app, "V4 lost update")
    v = {"project": "Help Desk", "subject": "V4 lost update", "description": "Fixed"}

    def write() -> None:
        app.save(iid, {"due_date": "2026-12-31"}, user=2)
        app.save(iid, {"description": "Fixed", "due_date": None})

    out = _outcome(_run(gv, "edit_description", v, write))
    assert out.startswith("CONFIRMED"), out


def test_v5_same_field_lost_update_must_stop_FAILS(gv: dict[str, Any]) -> None:
    app = gv["app"]
    iid = _fresh(app, "V5")
    v = {"project": "Help Desk", "subject": "V5", "description": "Fixed"}

    def write() -> None:
        app.save(iid, {"description": "Theirs"}, user=2)
        app.save(iid, {"description": "Fixed"})

    out = _outcome(_run(gv, "edit_description", v, write))
    assert not out.startswith("CONFIRMED"), out


def test_k1_correct_status_update_in_another_project(gv: dict[str, Any]) -> None:
    """Both set_status demonstrations were in one project, so the project row
    the app touches became a literal key (id=3). A correct update of an
    issue in another project touches another project row."""
    app = gv["app"]
    iid = app.create_issue({"project": "Billing Portal", "subject": "K1", "status": "New"})
    v = {"project": "Billing Portal", "subject": "K1", "status": "Resolved"}
    out = _outcome(_run(gv, "set_status", v, lambda: app.save(iid, {"status": "Resolved"})))
    assert not out.startswith("CONFIRMED"), out


def test_dbg_v5_targets(gv: dict[str, Any], monkeypatch: Any) -> None:
    import openadapt_flow.runtime.effects.versions as vm
    real = vm.judge_versions

    def spy(spec, stored, rows, targets, key, params={}):
        r = real(spec, stored, rows, targets, key, params)
        print("DBG targets", targets, "stored n", len((stored or {}).get("records") or {}), "rows", rows, "->", r.status, r.problem if hasattr(r, "problem") else "")
        return r

    monkeypatch.setattr(vm, "judge_versions", spy)
    app = gv["app"]
    iid = _fresh(app, "V6")
    v = {"project": "Help Desk", "subject": "V6", "description": "Fixed"}

    def write() -> None:
        app.save(iid, {"description": "Theirs"}, user=2)
        app.save(iid, {"description": "Fixed"})

    print(_outcome(_run(gv, "edit_description", v, write)))


def test_r1_pg_read_only_check_misses_inherited_and_trigger(pg: Any) -> None:
    port = pg["port"]
    admin = psycopg2.connect(host="127.0.0.1", port=port, user="admin", dbname="postgres")
    admin.autocommit = True
    cur = admin.cursor()
    cur.execute("CREATE DATABASE rotest")
    admin.close()
    a = psycopg2.connect(host="127.0.0.1", port=port, user="admin", dbname="rotest")
    a.autocommit = True
    c = a.cursor()
    c.execute("CREATE TABLE t (id serial primary key, v text); CREATE ROLE writers; "
              "GRANT INSERT, UPDATE, DELETE ON t TO writers; "
              "CREATE ROLE ro1 LOGIN; GRANT SELECT ON t TO ro1; GRANT writers TO ro1; "
              "CREATE ROLE ro2 LOGIN; GRANT SELECT, TRIGGER ON t TO ro2; GRANT CREATE ON SCHEMA public TO ro2;")
    for role in ("ro1", "ro2"):
        probs = gen.check_read_only(
            lambda r=role: psycopg2.connect(host="127.0.0.1", port=port, user=r, dbname="rotest"), "postgresql")
        print("RO", role, probs)
    w = psycopg2.connect(host="127.0.0.1", port=port, user="ro1", dbname="rotest")
    w.autocommit = True
    w.cursor().execute("INSERT INTO t (id, v) VALUES (99, 'written by the read account')")
    print("RO ro1 inserted a row")

PostgreSQL frame tests (tests/test_review_1672_r2_pgframe.py; run in its own session)

# ruff: noqa: F811
"""Adversarial review of #1672, round 2 (not for merge): the PostgreSQL
whole-database frame on the PR's own generated contracts."""

from __future__ import annotations

from typing import Any

import pytest

psycopg2 = pytest.importorskip("psycopg2")

from tests.test_generate_verify_contract import _run, pg  # noqa: E402,F401


def _outcome(o: str) -> str:
    print("OUTCOME:", o)
    return o


from tests.test_generate_verify_contract import CREATE, generated  # noqa: E402,F401


def test_g1_pg_frame_unrelated_insert(generated: dict[str, Any]) -> None:
    app = generated["app"]
    v = {**CREATE, "subject": "G1"}
    try:
        out = _outcome(_run(generated, "create_issue", v, lambda: (
            app.create_issue(v), app.q("INSERT INTO users (login) VALUES ('g1x')"))))
    finally:
        app.q("DELETE FROM users WHERE login='g1x'")
    assert not out.startswith("CONFIRMED"), out


def test_g2_pg_frame_other_schema(generated: dict[str, Any]) -> None:
    app = generated["app"]
    app.q("CREATE SCHEMA IF NOT EXISTS side; CREATE TABLE IF NOT EXISTS side.t (id serial primary key, v text); "
          "GRANT USAGE ON SCHEMA side TO oa_reader; GRANT SELECT ON side.t, side.t_id_seq TO oa_reader")
    v = {**CREATE, "subject": "G2"}
    out = _outcome(_run(generated, "create_issue", v, lambda: (
        app.create_issue(v), app.q("INSERT INTO side.t (v) VALUES ('x')"))))
    assert not out.startswith("CONFIRMED"), out


def test_g3_pg_frame_update_plus_status_rename(generated: dict[str, Any]) -> None:
    app = generated["app"]
    iid = app.create_issue({"project": "Help Desk", "subject": "G3", "status": "New",
                            "assignee": None, "due_date": None, "customer_ref": None})
    v = {"project": "Help Desk", "subject": "G3", "status": "Resolved", "assignee": None,
         "due_date": None, "done_ratio": None, "notes": None}
    try:
        out = _outcome(_run(generated, "update_issue", v, lambda: (
            app.update_issue(iid, v), app.q("UPDATE statuses SET name='Closed' WHERE name='New'"))))
    finally:
        app.q("UPDATE statuses SET name='New' WHERE name='Closed'")
    assert not out.startswith("CONFIRMED"), out


def test_g4_pg_frame_create_rewrites_old_journal(generated: dict[str, Any]) -> None:
    """journals is not named by create_issue: any change stops."""
    app = generated["app"]
    app.q("INSERT INTO journals (journalized_id, journalized_type, user_id, notes) VALUES (1,'Issue',2,'old')")
    v = {**CREATE, "subject": "G4"}
    out = _outcome(_run(generated, "create_issue", v, lambda: (
        app.create_issue(v), app.q("UPDATE journals SET notes='rewritten' WHERE notes='old'"))))
    assert not out.startswith("CONFIRMED"), out

[abrichr]

Adversarial review (round 3)

Verdict: APPROVE (0 blockers, 6 should-fix, 3 nits)

Reviewed at 2bc534fcb. I ran everything against a throwaway PostgreSQL (initdb). The PR's tests pass: test_generate_verify_contract.py, _attacks.py, _attacks_r2.py and test_generate_ensure_contract.py (73 passed). So do the versions, ensure-record, frame and verify-run suites (337 passed). The new attack files are in the collapsed sections at the end.

Every round-2 blocker is fixed. Each wrong write I could get CONFIRMED needs one of four things: an answer the person accepted, an app that skips its own version check, an app that logs an edit under an event code no demonstration showed, or a coincidence of ids in both demonstrations. Each one is listed as a should-fix below.

Round 1 and round 2, rerun

• Round-1 file, unchanged: the same as round 2. A1, A1b, A2, A3, A4, B1 and B2 are REFUTED, and B3 is UNCERTAIN. The correct colleague-history update (A5) is CONFIRMED, and the deleted control (B4) is still stopped.
• Round-2 file, unchanged: edit_description and create_issue get no contract. The file answers bookkeeping: {logs_add: [audits]}, which no longer accepts the proposed columns, so both kinds are refused ("new-row columns ... not named under bookkeeping.unchecked"). The strict answers work as intended. The PostgreSQL frame file still stops all 4 of its attacks.
• Round-2 attacks with answers in the new form (the proposal plus logs_add: [audits]):

Attack	Round 2	Now
T2: the touched project row takes the issue's description	CONFIRMED	REFUTED
P1: parent named by a subject two issues share	CONFIRMED	REFUTED
V4 / V5: lost update, version column	REFUTED	REFUTED
V4 / V5: lost update, audit trail (no version column)	CONFIRMED	REFUTED (another_author)
V6: no earlier audit row for the target (the blocker-1 shape)	CONFIRMED	UNCERTAIN (missing)
Controls V1, T1, P0	CONFIRMED	CONFIRMED
K1: correct status update in another project (false stop)	REFUTED	still REFUTED (should-fix 6)

The round-2 fixes, checked

1. Target existence from the surface's own read (verify_run.py:509-544). Fixed. The record surface reads the whole table with no row cap (fetchall), so an existing target is always in the read. The change can't loosen the check for other contract types:
   • An effect with no surface probe still uses before.records. The one difference is a baseline read that failed: that used to count the target as new and skip it, and now counts it as existing, which can only stop a write.
   • An effect whose surface has no read in before.detail is treated as existing, which also only stops.
   • None of the repository's hand-written contracts pairs a surfaces verifier with frame.versions, so no merged contract changes verdict.
2. frame.versions.target_surface. Fixed. The generator always sets record, and the touched rows' keys are no longer looked up in the versions read (round-2 NIT). See NIT 1 for an unchecked name.
3. target_fields_by_surface. Fixed for rows of another table (T2 is stopped). A touched row in the target's own table is folded into record (verify_contract.py:1947), which could let the target take a parent row's columns. In practice that kind always false-stops instead (should-fix 5), so it doesn't pass a wrong write today.
4. Exactly one row on a key back to the main table (:1464). Fixed for a self-reference (P1). A child table's second key into the main table is still treated as the child's link and skips the check. I couldn't make a contract for it: a relation row's "related issue" parameter isn't located, so the kind is refused. Worth one test once that case is supported.
5. Strict answers. Fixed. Unknown keys, a non-boolean accept, and tables or columns the question didn't list all raise an error. Only accept: true takes the proposed columns. logs_add applies per kind, to the tables that kind's demonstrations added rows to.
6. Every new-row column pinned or named unchecked. Fixed in the mechanism. The default proposal undoes it (should-fix 1).
7. PostgreSQL effective privileges. Fixed for inherited roles, pg_write_all_data, TRIGGER, schema CREATE and sequences: ro1, ro2 and ro5 are all reported. Two gaps remain (should-fix 4).

The three attacks asked for

• (a) unchecked is per column. I couldn't make a wrong write pass through a side row's unchecked column:

  • The side row's count is pinned.
  • The main record is checked on its own.
  • To change a side row on another record, the agent would also have to move that record, and the frame stops that.
  • The realistic case (a relation row whose link columns vary) is refused, because the parameter isn't located.

  The per-table naming does leak across kinds. A column that varies in kind A and was left empty in kind B becomes unchecked in B too once A's answer names it. Together with the default proposal, that gives should-fix 1.

• (b) A step version column and a stale save. If the app writes the form's version plus one and doesn't check it, another user's save plus the agent's stale save is one step. The verdict is CONFIRMED, and the other user's due date is lost (test_gl_b_stale_version_written_back). This needs an app that skips the check its own version column exists for. An optimistic-lock app rejects the stale save, and an app that increments the stored value gives two steps, which is stopped (V4, V5). Should-fix 2 (document it, and read the audit trail too when the app has one).

• (c) Create and update codes both kept. Keeping the create codes only adds rows, so it can only stop a write: another user's row under a create code is REFUTED (test_ga2_c_create_code_other_user). The list of kept codes is the real issue (should-fix 3).

SHOULD-FIX

1. The default bookkeeping proposal names every open column, including free text that every demonstration left empty (verify_contract.py:3080). Accepting it undoes "free text stays empty". Fixture: create_issue's demonstrations left issues.description empty. The agent creates the right issue and also fills the description with "Refund approved, wire to acct 99": CONFIRMED (test_gl_a1). Leave the free-text columns out of the proposal, so they stay pinned empty unless the person adds them. The same applies across kinds: a column named for one kind is unchecked in every kind that leaves it open.
2. A version column is trusted to step on every save from the stored value (_version_column, preferred at :2778). One step can hide a lost update when the app writes the form's version back (finding (b)). Say this in GENERATE_CONTRACTS.md. When the app also has an audit trail of saves, read it as well: authors catch what a step count can't.
3. The write-event filter is an allow-list (:1215, action IN ('create', 'update')). Another user's save logged under any other write code (a bulk edit, an import, an API path) is dropped from the read. The agent's own save then makes a fresh version, so the no_new_version guard doesn't fire. Other user sets the due date as bulk_update, and the agent's stale save puts the old one back: CONFIRMED. The same with the same field: CONFIRMED (test_ga2_c_*). Prefer leaving out only codes the person confirms are reads, or list the codes the table already holds in a question.
4. PostgreSQL check_read_only (:2531) still misses two cases:
   • A NOINHERIT member of a writing role returns [] but can SET ROLE and write (ro3). Check every role the account can SET ROLE to (pg_has_role(..., 'MEMBER')).
   • A column-level UPDATE grant returns [] but can update that column (ro4). Use has_any_column_privilege.
5. A row in the target's own table that the app touches (a parent issue recalculated when a subtask is saved) stops every write of the kind. The record and raw_issues surfaces read the same table, so each declared row counts as collateral on the other surface. A correct set_status on a subtask is REFUTED "2 record(s) outside the declared write moved" (test_gp2_control). This fails closed and also happens at the round-2 head, but parent and child rows in one table are common in trackers. Declare touched rows of the main table on record, by key, rather than as a second surface over the same table.
6. Earlier items still open, all fail closed except (a):
   • (a) Dropped lost-update check: _audit_source counts the audit table's own primary key as a candidate record column (:1146). When an audit row's id equals the record's id in both demonstrations, two columns tie and the source is dropped. With no version column, set_status then has no versions check at all, and a lost update, even of the same field, is CONFIRMED (test_ga_s1, test_ga_s2). Skip the primary key. versions: null appears only in report.json, so state it in QUESTIONS.md as well.
   • (b) Literal-key false stop: a row two demonstrations happened to share still becomes a literal-key excuse (round-2 should-fix 2): K1 is REFUTED.
   • (c) Multi-field kind refused: a multi-field kind with a per-field audit table is still refused as a create (round-2 should-fix 6). It also counts as a create kind when the code filter is chosen, which empties the filter. That fails closed.

NIT

1. A hand-written target_surface that no effect uses gives zero targets and held. Validate the name against the verifier's surfaces, or treat zero targets as not_checked.
2. A row the app adds keeps its column default when a free column isn't named. When the default is NULL and the app writes '', a correct write is stopped.
3. The PR body says the real-app counts were rerun "after the round-2 fixes". The last three commits change what the proposal names, so state the commit the counts were measured at.

Public text

I found no third-party app beyond the three already named, no fault ids and no fault descriptions in the PR body, the commit messages, the docs, CAPABILITIES.md, llms-full.txt or the code.

Round-3 attack tests (tests/test_review_1672_r3_attacks.py; needs the round-2 attack file beside it)

# ruff: noqa: F811
"""Adversarial review of #1672, round 3 (not for merge). Real PostgreSQL.

Three databases:
- ``gl``: the round-2 tracker (per-field audit rows, lock_version, the
  project touched on every save). Round-2 attacks rerun with answers in the
  new strict form (the proposal, plus ``logs_add: [audits]``).
- ``ga``: the same tracker with no version column, so update kinds read
  their versions from the audit trail.
- ``gp``: subtasks: saving a child recalculates its parent issue (same
  table): the parent's done_ratio, updated_on and lock_version move.
"""

from __future__ import annotations

import json
import os
from typing import Any, Callable

import pytest
import yaml

psycopg2 = pytest.importorskip("psycopg2")

from openadapt_flow.generate import verify_contract as gen  # noqa: E402
from tests.test_generate_verify_contract import (  # noqa: E402,F401
    BOT,
    _run,
    pg,
)
from tests.test_review_1672_r2_attacks import KINDS_V, SCHEMA_V, AppV  # noqa: E402


def _outcome(o: str) -> str:
    print("OUTCOME:", o)
    return o


SCHEMA_A = SCHEMA_V.replace("lock_version integer NOT NULL DEFAULT 0, ", "")


class AppA(AppV):
    """No version column."""

    def save(self, iid: int, sets: dict[str, Any], *, user: int = BOT, action: str = "update") -> None:
        for col, val in sets.items():
            if col == "status":
                col, val = "status_id", self.one("SELECT id FROM statuses WHERE name=%s", (val,))
            old = self.one(f"SELECT {col}::text FROM issues WHERE id=%s", (iid,))
            self.q(f"UPDATE issues SET {col}=%s WHERE id=%s", (val, iid))
            self.q("INSERT INTO audits (issue_id, user_id, action, field_name, old_value, new_value) "
                   "VALUES (%s,%s,%s,%s,%s,%s)", (iid, user, action, col, old, None if val is None else f"v{val}"))
        self.q("UPDATE issues SET updated_on=now() WHERE id=%s", (iid,))
        self.q("UPDATE projects SET updated_on=now() WHERE id=(SELECT project_id FROM issues WHERE id=%s)", (iid,))


class AppL(AppV):
    def save_stale(self, iid: int, sets: dict[str, Any], stale: int) -> None:
        """An app that writes the version the form carried, plus one, with no
        check (the agent's form was opened before another user's save)."""
        for col, val in sets.items():
            if col == "status":
                col, val = "status_id", self.one("SELECT id FROM statuses WHERE name=%s", (val,))
            old = self.one(f"SELECT {col}::text FROM issues WHERE id=%s", (iid,))
            self.q(f"UPDATE issues SET {col}=%s WHERE id=%s", (val, iid))
            self.q("INSERT INTO audits (issue_id, user_id, action, field_name, old_value, new_value) "
                   "VALUES (%s,%s,'update',%s,%s,%s)", (iid, BOT, col, old, None if val is None else f"v{val}"))
        self.q("UPDATE issues SET updated_on=now(), lock_version=%s WHERE id=%s", (stale + 1, iid))
        self.q("UPDATE projects SET updated_on=now() WHERE id=(SELECT project_id FROM issues WHERE id=%s)", (iid,))


class AppP(AppV):
    """Saving a subtask recalculates its parent (Redmine's derived done ratio)."""

    def save(self, iid: int, sets: dict[str, Any], *, user: int = BOT) -> None:
        super().save(iid, sets, user=user)
        parent = self.one("SELECT parent_id FROM issues WHERE id=%s", (iid,))
        if parent is not None:
            self.q("UPDATE issues SET done_ratio=done_ratio+1, updated_on=now(), "
                   "lock_version=lock_version+1 WHERE id=%s", (parent,))


def _build(pg: Any, db: str, schema: str, app: Any, kinds: dict[str, Any],
           seed: Callable[[Any], None] = lambda a: None, logs: bool = True) -> dict[str, Any]:
    root, port = pg["root"], pg["port"]
    admin = psycopg2.connect(host="127.0.0.1", port=port, user="admin", dbname="postgres")
    admin.autocommit = True
    admin.cursor().execute(f"CREATE DATABASE {db}")
    admin.close()
    app.port, app.db = port, db
    with app.con() as c, c.cursor() as cur:
        cur.execute("DO $$BEGIN IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname='oa_reader') "
                    "THEN CREATE ROLE oa_reader LOGIN PASSWORD 'reader-pw'; END IF; END$$")
        cur.execute(schema)
    seed(app)
    os.environ["OA_GEN_TEST_PW"] = "reader-pw"
    work = root / f"work_{db}"
    work.mkdir()
    read = {"sql_driver": "psycopg2",
            "sql_connect_args": {"host": "127.0.0.1", "port": port, "user": "oa_reader", "dbname": db},
            "sql_password_env": "OA_GEN_TEST_PW"}
    (work / "read.yaml").write_text(yaml.safe_dump(read))
    (work / "kinds.yaml").write_text(yaml.safe_dump(kinds, allow_unicode=True))
    common = ["--read", str(work / "read.yaml")]

    def demo(kind: str, values: dict[str, Any], write: Any) -> None:
        assert gen.main(["snapshot", *common, "--out", str(work / "before.json")]) == 0
        write()
        assert gen.main(["record", *common, "--before", str(work / "before.json"), "--kind", kind,
                         "--values", json.dumps(values), "--demos", str(work / "demos")]) == 0

    for k in kinds["kinds"]:
        for ex in k["examples"]:
            v = ex["values"]
            if k["kind"] == "create_issue":
                demo("create_issue", v, lambda v=v: app.create_issue(v))
            else:
                sets = {p: v[p] for p in k["parameters"][2:] if v.get(p) is not None}
                demo(k["kind"], v, lambda v=v, s=sets: app.save(app.find(v["project"], v["subject"]), s))
    argv = ["propose", *common, "--kinds", str(work / "kinds.yaml"), "--demos", str(work / "demos"),
            "--out", str(work / "out"), "--actor-param", f"bot_user_id={BOT}"]
    assert gen.main(argv) == 0
    report = json.loads((work / "out" / "report.json").read_text())
    print("QUESTIONS:", (work / "out" / "QUESTIONS.md").read_text())
    answers = {q["key"]: q["proposal"] for q in report["questions"]}
    bk = answers.get("bookkeeping")
    if logs and bk is not None:
        bk = {"accept": True} if bk == "accept" else dict(bk)
        bk["logs_add"] = ["audits"]
        answers["bookkeeping"] = bk
    print("ANSWERS:", json.dumps(answers))
    (work / "answers.yaml").write_text(yaml.safe_dump(answers))
    assert gen.main([*argv, "--answers", str(work / "answers.yaml")]) == 0
    report = json.loads((work / "out" / "report.json").read_text())
    for k, v in report["kinds"].items():
        print("KIND", k, "refused", v.get("refused"), "versions", json.dumps(v.get("versions")),
              "touched", v.get("touched_rows"))
    keep = os.environ.get("OA_GEN_TEST_KEEP")
    if keep:
        import shutil
        shutil.copytree(work, f"{keep}-{db}", dirs_exist_ok=True)
    return {"app": app, "out": work / "out", "report": report, "read": read}


@pytest.fixture(scope="module")
def gl(pg: Any) -> dict[str, Any]:
    return _build(pg, "trk_l", SCHEMA_V, AppL(0, ""), KINDS_V)


@pytest.fixture(scope="module")
def ga(pg: Any) -> dict[str, Any]:
    return _build(pg, "trk_a", SCHEMA_A, AppA(0, ""), KINDS_V)


def _seed_parents(app: Any) -> None:
    for proj, parent, kids in (("Help Desk", "Office move", ["Desks", "Phones", "Cables"]),
                               ("Billing Portal", "Tax rules", ["EU VAT", "US sales tax", "GST"])):
        app.create_issue({"project": proj, "subject": parent, "status": "New"})
        for k in kids:
            app.create_issue({"project": proj, "subject": k, "status": "New", "parent": parent})


KINDS_P = {"kinds": [
    {"kind": "set_status", "parameters": ["project", "subject", "status"],
     "examples": [
         {"values": {"project": "Help Desk", "subject": "Desks", "status": "Resolved"}},
         {"values": {"project": "Billing Portal", "subject": "EU VAT", "status": "In Progress"}},
     ]},
]}


@pytest.fixture(scope="module")
def gp(pg: Any) -> dict[str, Any]:
    return _build(pg, "trk_p", SCHEMA_V, AppP(0, ""), KINDS_P, seed=_seed_parents)


def _fresh(app: Any, subject: str, project: str = "Help Desk") -> int:
    return app.create_issue({"project": project, "subject": subject, "status": "New"})


# ---------- gl: round-2 attacks, rerun ----------

def test_gl_v1_control(gl: dict[str, Any]) -> None:
    app = gl["app"]
    iid = _fresh(app, "V1 control")
    v = {"project": "Help Desk", "subject": "V1 control", "status": "Resolved"}
    out = _outcome(_run(gl, "set_status", v, lambda: app.save(iid, {"status": "Resolved"})))
    assert out.startswith("CONFIRMED"), out


def test_gl_t1_control_description(gl: dict[str, Any]) -> None:
    app = gl["app"]
    iid = _fresh(app, "T1 control")
    v = {"project": "Help Desk", "subject": "T1 control", "description": "Fixed"}
    out = _outcome(_run(gl, "edit_description", v, lambda: app.save(iid, {"description": "Fixed"})))
    assert out.startswith("CONFIRMED"), out


def test_gl_t2_touched_parent_takes_target_field(gl: dict[str, Any]) -> None:
    app = gl["app"]
    iid = _fresh(app, "T2 leak")
    v = {"project": "Help Desk", "subject": "T2 leak", "description": "Fixed"}

    def write() -> None:
        app.save(iid, {"description": "Fixed"})
        app.q("UPDATE projects SET description='Overwritten by the agent' WHERE name='Help Desk'")

    try:
        out = _outcome(_run(gl, "edit_description", v, write))
    finally:
        app.q("UPDATE projects SET description=NULL WHERE name='Help Desk'")
    assert not out.startswith("CONFIRMED"), out


def test_gl_p0_control_create_with_parent(gl: dict[str, Any]) -> None:
    app = gl["app"]
    v = {"project": "Help Desk", "subject": "P0 child", "status": "New", "parent": "VPN drops"}
    out = _outcome(_run(gl, "create_issue", v, lambda: app.create_issue(v)))
    assert out.startswith("CONFIRMED"), out


def test_gl_p1_parent_subject_shared(gl: dict[str, Any]) -> None:
    app = gl["app"]
    legacy = app.find("Billing Portal Legacy", "Refund button missing")
    v = {"project": "Help Desk", "subject": "P1 child", "status": "New", "parent": "Refund button missing"}
    out = _outcome(_run(gl, "create_issue", v, lambda: app.create_issue({**v, "parent_id": legacy})))
    assert not out.startswith("CONFIRMED"), out


def test_gl_v2_v4_lost_update_other_field(gl: dict[str, Any]) -> None:
    app = gl["app"]
    iid = _fresh(app, "V4 lost update")
    v = {"project": "Help Desk", "subject": "V4 lost update", "description": "Fixed"}

    def write() -> None:
        app.save(iid, {"due_date": "2026-12-31"}, user=2)
        app.save(iid, {"description": "Fixed", "due_date": None})

    out = _outcome(_run(gl, "edit_description", v, write))
    assert not out.startswith("CONFIRMED"), out


def test_gl_v5_same_field(gl: dict[str, Any]) -> None:
    app = gl["app"]
    iid = _fresh(app, "V5")
    v = {"project": "Help Desk", "subject": "V5", "description": "Fixed"}

    def write() -> None:
        app.save(iid, {"description": "Theirs"}, user=2)
        app.save(iid, {"description": "Fixed"})

    out = _outcome(_run(gl, "edit_description", v, write))
    assert not out.startswith("CONFIRMED"), out


def test_gl_k1_status_in_another_project(gl: dict[str, Any]) -> None:
    """Round-2 SF2 (false stop): correct, so CONFIRMED is the right answer."""
    app = gl["app"]
    iid = _fresh(app, "K1", project="Billing Portal")
    v = {"project": "Billing Portal", "subject": "K1", "status": "Resolved"}
    print(_outcome(_run(gl, "set_status", v, lambda: app.save(iid, {"status": "Resolved"}))))


def test_gl_b_stale_version_written_back(gl: dict[str, Any]) -> None:
    """(b) Another user saves (lock 0->1); the app writes the agent's stale
    form version + 1 (=1) with no check. One step in the window."""
    app = gl["app"]
    iid = _fresh(app, "B stale")
    v = {"project": "Help Desk", "subject": "B stale", "description": "Fixed"}

    def write() -> None:
        stale = app.one("SELECT lock_version FROM issues WHERE id=%s", (iid,))
        app.save(iid, {"due_date": "2026-12-31"}, user=2)
        app.save_stale(iid, {"description": "Fixed", "due_date": None}, stale)

    out = _outcome(_run(gl, "edit_description", v, write))
    print("lock now", app.one("SELECT lock_version FROM issues WHERE id=%s", (iid,)))
    assert out.startswith("CONFIRMED"), out  # documents the outcome


# ---------- ga: versions from the audit trail ----------

def test_ga_control_edit_created_issue(ga: dict[str, Any]) -> None:
    app = ga["app"]
    iid = _fresh(app, "A ctrl")
    v = {"project": "Help Desk", "subject": "A ctrl", "description": "Fixed"}
    out = _outcome(_run(ga, "edit_description", v, lambda: app.save(iid, {"description": "Fixed"})))
    assert out.startswith("CONFIRMED"), out


def test_ga_control_edit_seed_issue_no_audit(ga: dict[str, Any]) -> None:
    """A record saved before the audit trail existed: fail closed expected."""
    app = ga["app"]
    v = {"project": "Billing Portal Legacy", "subject": "Refund button missing", "description": "Fixed"}
    iid = app.find("Billing Portal Legacy", "Refund button missing")
    print(_outcome(_run(ga, "edit_description", v, lambda: app.save(iid, {"description": "Fixed"}))))


def test_ga_v4_lost_update_other_field(ga: dict[str, Any]) -> None:
    app = ga["app"]
    iid = _fresh(app, "AV4")
    v = {"project": "Help Desk", "subject": "AV4", "description": "Fixed"}

    def write() -> None:
        app.save(iid, {"due_date": "2026-12-31"}, user=2)
        app.save(iid, {"description": "Fixed", "due_date": None})

    out = _outcome(_run(ga, "edit_description", v, write))
    assert not out.startswith("CONFIRMED"), out


def test_ga_v5_same_field(ga: dict[str, Any]) -> None:
    app = ga["app"]
    iid = _fresh(app, "AV5")
    v = {"project": "Help Desk", "subject": "AV5", "description": "Fixed"}

    def write() -> None:
        app.save(iid, {"description": "Theirs"}, user=2)
        app.save(iid, {"description": "Fixed"})

    out = _outcome(_run(ga, "edit_description", v, write))
    assert not out.startswith("CONFIRMED"), out


def test_ga_v6_never_edited_seed_lost_update(ga: dict[str, Any]) -> None:
    """Blocker-1 shape: a target with no earlier row in the version read."""
    app = ga["app"]
    iid = app.find("Help Desk", "VPN drops")
    app.q("DELETE FROM audits WHERE issue_id=%s", (iid,))
    v = {"project": "Help Desk", "subject": "VPN drops", "description": "Fixed"}

    def write() -> None:
        app.save(iid, {"description": "Theirs"}, user=2)
        app.save(iid, {"description": "Fixed"})

    out = _outcome(_run(ga, "edit_description", v, write))
    assert not out.startswith("CONFIRMED"), out


def test_ga_c_other_write_code(ga: dict[str, Any]) -> None:
    """(c) Another user's save is logged under a write code no demonstration
    showed (a bulk edit); the agent's stale save puts the old due date back."""
    app = ga["app"]
    iid = _fresh(app, "AC1")
    v = {"project": "Help Desk", "subject": "AC1", "description": "Fixed"}

    def write() -> None:
        app.save(iid, {"due_date": "2026-12-31"}, user=2, action="bulk_update")
        app.save(iid, {"description": "Fixed", "due_date": None})

    out = _outcome(_run(ga, "edit_description", v, write))
    assert out.startswith("CONFIRMED"), out  # documents the outcome


def test_ga_c2_other_user_create_code(ga: dict[str, Any]) -> None:
    """(c) A 'create' row by another user on an existing target is kept."""
    app = ga["app"]
    iid = _fresh(app, "AC2")
    v = {"project": "Help Desk", "subject": "AC2", "description": "Fixed"}

    def write() -> None:
        app.save(iid, {"due_date": "2026-12-31"}, user=2, action="create")
        app.save(iid, {"description": "Fixed", "due_date": None})

    out = _outcome(_run(ga, "edit_description", v, write))
    assert not out.startswith("CONFIRMED"), out


# ---------- gp: parent recalculated in the same table ----------

def test_gp_control(gp: dict[str, Any]) -> None:
    app = gp["app"]
    iid = app.find("Help Desk", "Phones")
    v = {"project": "Help Desk", "subject": "Phones", "status": "Resolved"}
    out = _outcome(_run(gp, "set_status", v, lambda: app.save(iid, {"status": "Resolved"})))
    assert out.startswith("CONFIRMED"), out


def test_gp_target_takes_parent_touched_column(gp: dict[str, Any]) -> None:
    """The parent's recalculated done_ratio is excused on the `record`
    surface; the agent's set_status also sets the target's done ratio."""
    app = gp["app"]
    iid = app.find("Billing Portal", "GST")
    v = {"project": "Billing Portal", "subject": "GST", "status": "Resolved"}

    def write() -> None:
        app.q("UPDATE issues SET done_ratio=90 WHERE id=%s", (iid,))
        app.save(iid, {"status": "Resolved"})

    out = _outcome(_run(gp, "set_status", v, write))
    assert not out.startswith("CONFIRMED"), out


def test_ga_s1_set_status_lost_update(ga: dict[str, Any]) -> None:
    """set_status has no versions source in the audit-only tracker."""
    app = ga["app"]
    iid = _fresh(app, "AS1")
    v = {"project": "Help Desk", "subject": "AS1", "status": "Resolved"}

    def write() -> None:
        app.save(iid, {"due_date": "2026-12-31"}, user=2)
        app.save(iid, {"status": "Resolved", "due_date": None})

    out = _outcome(_run(ga, "set_status", v, write))
    assert not out.startswith("CONFIRMED"), out


def test_ga_s2_set_status_same_field(ga: dict[str, Any]) -> None:
    app = ga["app"]
    iid = _fresh(app, "AS2")
    v = {"project": "Help Desk", "subject": "AS2", "status": "Resolved"}

    def write() -> None:
        app.save(iid, {"status": "In Progress"}, user=2)
        app.save(iid, {"status": "Resolved"})

    out = _outcome(_run(ga, "set_status", v, write))
    assert not out.startswith("CONFIRMED"), out


def test_gl_a1_create_with_unrequested_description(gl: dict[str, Any]) -> None:
    """(a) The accepted proposal names issues.description (empty in every
    create demonstration) as unchecked; the agent's create also fills it."""
    app = gl["app"]
    v = {"project": "Help Desk", "subject": "A1 child", "status": "New", "parent": "VPN drops"}

    def write() -> None:
        iid = app.create_issue(v)
        app.q("UPDATE issues SET description='Refund approved, wire to acct 99' WHERE id=%s", (iid,))

    out = _outcome(_run(gl, "create_issue", v, write))
    assert out.startswith("CONFIRMED"), out  # documents the outcome


KINDS_P2 = {"kinds": [
    {"kind": "set_status", "parameters": ["project", "subject", "status"],
     "examples": [
         {"values": {"project": "Help Desk", "subject": "Desks", "status": "Resolved"}},
         {"values": {"project": "Help Desk", "subject": "Phones", "status": "In Progress"}},
     ]},
]}


@pytest.fixture(scope="module")
def gp2(pg: Any) -> dict[str, Any]:
    return _build(pg, "trk_p2", SCHEMA_V, AppP(0, ""), KINDS_P2, seed=_seed_parents)


def test_gp2_control(gp2: dict[str, Any]) -> None:
    app = gp2["app"]
    iid = app.find("Help Desk", "Cables")
    v = {"project": "Help Desk", "subject": "Cables", "status": "Resolved"}
    out = _outcome(_run(gp2, "set_status", v, lambda: app.save(iid, {"status": "Resolved"})))
    assert out.startswith("CONFIRMED"), out


def test_gp2_target_takes_parent_touched_column(gp2: dict[str, Any]) -> None:
    app = gp2["app"]
    iid = app.find("Help Desk", "Cables")
    app.q("UPDATE issues SET status_id=1 WHERE id=%s", (iid,))
    v = {"project": "Help Desk", "subject": "Cables", "status": "Resolved"}

    def write() -> None:
        app.q("UPDATE issues SET done_ratio=90 WHERE id=%s", (iid,))
        app.save(iid, {"status": "Resolved"})

    out = _outcome(_run(gp2, "set_status", v, write))
    assert not out.startswith("CONFIRMED"), out


SCHEMA_R = SCHEMA_V.replace(
    "CREATE TABLE settings",
    "CREATE TABLE issue_relations (id serial PRIMARY KEY, issue_from_id integer, "
    "issue_to_id integer, relation_type text);\nCREATE TABLE settings",
)


class AppR(AppV):
    def save(self, iid: int, sets: dict[str, Any], *, user: int = BOT, to: Any = None) -> None:
        if "related" in sets:
            other = to or self.one("SELECT id FROM issues WHERE subject=%s", (sets["related"],))
            self.q("INSERT INTO issue_relations (issue_from_id, issue_to_id, relation_type) "
                   "VALUES (%s,%s,'relates')", (iid, other))
            self.q("INSERT INTO audits (issue_id, user_id, action, field_name, old_value, new_value) "
                   "VALUES (%s,%s,'update','relates',NULL,%s)", (iid, user, str(other)))
            self.q("UPDATE issues SET updated_on=now(), lock_version=lock_version+1 WHERE id=%s", (iid,))
            return
        super().save(iid, sets, user=user)


def _seed_rel(app: Any) -> None:
    for proj, subj in (("Help Desk", "Login slow"), ("Billing Portal", "Invoice PDF"),
                       ("Help Desk", "Mail bounce"), ("Billing Portal", "Card declined")):
        app.create_issue({"project": proj, "subject": subj, "status": "New"})


KINDS_R = {"kinds": [
    {"kind": "add_relation", "parameters": ["project", "subject", "related"],
     "examples": [
         {"values": {"project": "Help Desk", "subject": "Login slow", "related": "Invoice PDF"}},
         {"values": {"project": "Billing Portal", "subject": "Card declined", "related": "Mail bounce"}},
     ]},
]}


@pytest.fixture(scope="module")
def gr(pg: Any) -> dict[str, Any]:
    return _build(pg, "trk_r", SCHEMA_R, AppR(0, ""), KINDS_R, seed=_seed_rel)


def test_gr_control(gr: dict[str, Any]) -> None:
    app = gr["app"]
    iid = app.find("Help Desk", "Mail bounce")
    v = {"project": "Help Desk", "subject": "Mail bounce", "related": "Login slow"}
    out = _outcome(_run(gr, "add_relation", v, lambda: app.save(iid, {"related": "Login slow"})))
    assert out.startswith("CONFIRMED"), out


def test_gr_related_subject_shared(gr: dict[str, Any]) -> None:
    """'Refund button missing' names two issues; the agent relates the
    legacy project's one."""
    app = gr["app"]
    iid = app.find("Billing Portal", "Invoice PDF")
    legacy = app.find("Billing Portal Legacy", "Refund button missing")
    v = {"project": "Billing Portal", "subject": "Invoice PDF", "related": "Refund button missing"}
    out = _outcome(_run(gr, "add_relation", v,
                        lambda: app.save(iid, {"related": "x"}, to=legacy)))
    assert not out.startswith("CONFIRMED"), out


KINDS_A2 = {"kinds": [k for k in KINDS_V["kinds"] if k["kind"] != "edit_plan"]}


@pytest.fixture(scope="module")
def ga2(pg: Any) -> dict[str, Any]:
    return _build(pg, "trk_a2", SCHEMA_A, AppA(0, ""), KINDS_A2)


def test_ga2_control(ga2: dict[str, Any]) -> None:
    app = ga2["app"]
    iid = _fresh(app, "A2 ctrl")
    v = {"project": "Help Desk", "subject": "A2 ctrl", "description": "Fixed"}
    out = _outcome(_run(ga2, "edit_description", v, lambda: app.save(iid, {"description": "Fixed"})))
    assert out.startswith("CONFIRMED"), out


def test_ga2_c_other_write_code(ga2: dict[str, Any]) -> None:
    app = ga2["app"]
    iid = _fresh(app, "A2C1")
    v = {"project": "Help Desk", "subject": "A2C1", "description": "Fixed"}

    def write() -> None:
        app.save(iid, {"due_date": "2026-12-31"}, user=2, action="bulk_update")
        app.save(iid, {"description": "Fixed", "due_date": None})

    out = _outcome(_run(ga2, "edit_description", v, write))
    assert out.startswith("CONFIRMED"), out  # documents the outcome


def test_ga2_c_same_field_other_code(ga2: dict[str, Any]) -> None:
    app = ga2["app"]
    iid = _fresh(app, "A2C2")
    v = {"project": "Help Desk", "subject": "A2C2", "description": "Fixed"}

    def write() -> None:
        app.save(iid, {"description": "Theirs"}, user=2, action="bulk_update")
        app.save(iid, {"description": "Fixed"})

    out = _outcome(_run(ga2, "edit_description", v, write))
    assert out.startswith("CONFIRMED"), out  # documents the outcome


def test_ga2_c_create_code_other_user(ga2: dict[str, Any]) -> None:
    app = ga2["app"]
    iid = _fresh(app, "A2C3")
    v = {"project": "Help Desk", "subject": "A2C3", "description": "Fixed"}

    def write() -> None:
        app.save(iid, {"description": "Theirs"}, user=2, action="create")
        app.save(iid, {"description": "Fixed"})

    out = _outcome(_run(ga2, "edit_description", v, write))
    assert not out.startswith("CONFIRMED"), out

PostgreSQL privilege attacks (tests/test_review_1672_r3_pg.py)

# ruff: noqa
from typing import Any
import psycopg2
from tests.test_generate_verify_contract import pg  # noqa
from openadapt_flow.generate import verify_contract as gen

def test_ro_more(pg: Any) -> None:
    port = pg["port"]
    a = psycopg2.connect(host="127.0.0.1", port=port, user="admin", dbname="postgres"); a.autocommit = True
    a.cursor().execute("CREATE DATABASE rotest3"); a.close()
    a = psycopg2.connect(host="127.0.0.1", port=port, user="admin", dbname="rotest3"); a.autocommit = True
    c = a.cursor()
    c.execute("REVOKE CREATE ON SCHEMA public FROM PUBLIC; CREATE TABLE t (id serial primary key, v text); CREATE ROLE w3; GRANT INSERT, UPDATE ON t TO w3; "
              "CREATE ROLE ro3 LOGIN NOINHERIT; GRANT SELECT ON t TO ro3; GRANT w3 TO ro3; "
              "CREATE ROLE ro4 LOGIN; GRANT SELECT ON t TO ro4; GRANT UPDATE (v) ON t TO ro4; "
              "CREATE ROLE ro5 LOGIN; GRANT pg_write_all_data TO ro5; GRANT SELECT ON t TO ro5; "
              "CREATE ROLE ro6 LOGIN; GRANT SELECT ON t TO ro6; INSERT INTO t (v) VALUES ('a');")
    c.execute("REVOKE CREATE ON DATABASE rotest3 FROM PUBLIC; REVOKE TEMP ON DATABASE rotest3 FROM PUBLIC")
    for role in ("ro3", "ro4", "ro5", "ro6"):
        probs = gen.check_read_only(lambda r=role: psycopg2.connect(host="127.0.0.1", port=port, user=r, dbname="rotest3"), "postgresql")
        print("RO", role, probs)
    w = psycopg2.connect(host="127.0.0.1", port=port, user="ro3", dbname="rotest3"); w.autocommit = True
    w.cursor().execute("SET ROLE w3; UPDATE t SET v='by ro3'"); print("RO ro3 wrote via SET ROLE")
    w = psycopg2.connect(host="127.0.0.1", port=port, user="ro4", dbname="rotest3"); w.autocommit = True
    w.cursor().execute("UPDATE t SET v='by ro4'"); print("RO ro4 updated a column")

[abrichr]

Adversarial review (round 4): APPROVE

Scope: only the delta from the round-3-approved head 2bc534f to b1ce256. Round 3 approved with four should-fixes that each still ended a wrong write CONFIRMED in a realistic setup. I asked for exactly those four, plus one docs note.

• SF1: free text that every demonstration left empty is no longer in the default bookkeeping proposal. On the main record and its child rows, it is pinned to the empty value the demonstrations stored (a NULL pin or ''). On a row the app adds beside the write, it must hold that exact empty value. A mixed empty value falls back to the column's default. Only a person's explicit unchecked entry frees such a column. Cost: Dolibarr's invoice creates now stop (40/43 to 32/43 confirmed), because the reference write fills a line description. That is the intended fail-closed trade, and the PR body says so.
• SF3: the audit read is now a deny-list. A code is excluded only if the demonstrations logged it on records they never added or changed, and never on one they did, in any table. Unseen codes and NULL codes count as saves: (col IS NULL OR col NOT IN (...)). A code that appears on both sides is never excluded, and only plain word codes are ever excluded. I found no way for a real save to fall into the deny set except an app that logs saves and reads under the same code with no demonstration of the save. In that case the code lands in writes and is kept.
• SF4: check_read_only iterates every role with pg_has_role(current_user, oid, 'MEMBER'), which covers NOINHERIT memberships reachable with SET ROLE. It also checks column-level INSERT, UPDATE and REFERENCES with has_any_column_privilege.
• SF6.1: the audit table's own primary key is skipped as a candidate record column.
• SF2: docs only.

No new excuse path; the delta only adds pins, keeps more audit rows and reports more privileges. The real-app counts were rerun and are stated with their commits.