feat: add reversible human review

src/rowbridge/models.py

@@ -18,6 +18,9 @@ class ReviewAction(StrEnum):
     ACCEPT = "accept"
     REJECT = "reject"
     MANUAL_LINK = "manual_link"
+    REOPEN = "reopen"
+    UNLINK = "unlink"
+    RESTORE_REJECTED = "restore_rejected"
 
 
 class RuleKind(StrEnum):

src/rowbridge/static/app.css

@@ -354,6 +354,7 @@ select:focus, input[type="number"]:focus { outline: 2px solid #bfd0e9; outline-o
 .history-item p { margin-top: 5px; font-size: 13px; line-height: 1.4; }
 .history-item small { display: block; margin-top: 7px; color: #8b948f; font-size: 11px; line-height: 1.4; }
 .history-detail { color: #6f7873; }
+.history-action { margin-top: 9px; }
 .panel-empty { margin-top: 13px; font-size: 13px; line-height: 1.45; }
 
 @media (max-width: 1050px) {

src/rowbridge/storage.py

@@ -143,6 +143,7 @@ def initialize(self) -> None:
                     previous_status TEXT,
                     resulting_status TEXT NOT NULL,
                     score REAL,
+                    evidence_json TEXT,
                     detail TEXT NOT NULL
                 );
 
@@ -155,6 +156,14 @@ def initialize(self) -> None:
                     ON matches(run_id, b_row_id) WHERE b_row_id IS NOT NULL;
                 """
             )
+            review_event_columns = {
+                str(row["name"])
+                for row in connection.execute("PRAGMA table_info(review_events)").fetchall()
+            }
+            if "evidence_json" not in review_event_columns:
+                connection.execute(
+                    "ALTER TABLE review_events ADD COLUMN evidence_json TEXT"
+                )
 
     def save_run(
         self,
@@ -581,6 +590,7 @@ def reject_review(self, run_id: str, match_id: int) -> None:
                 resulting_status=MatchStatus.UNMATCHED.value,
                 score=float(row["score"]),
                 detail="Rejected proposed match; both rows returned to unmatched.",
+                evidence_json=str(row["evidence_json"]),
             )
             self._refresh_run_counts(connection, run_id)
 
@@ -646,6 +656,202 @@ def create_manual_link(self, run_id: str, a_match_id: int, b_match_id: int) -> N
             )
             self._refresh_run_counts(connection, run_id)
 
+    def reopen_confirmed(self, run_id: str, match_id: int) -> None:
+        with self._connect() as connection:
+            row = self._get_match_row(connection, run_id, match_id)
+            self._require_status(row, MatchStatus.CONFIRMED)
+            connection.execute(
+                "UPDATE matches SET status = ? WHERE id = ?",
+                (MatchStatus.REVIEW.value, match_id),
+            )
+            self._record_event(
+                connection=connection,
+                run_id=run_id,
+                action=ReviewAction.REOPEN,
+                source_match_id=match_id,
+                a_row_id=self._optional_int(row["a_row_id"]),
+                b_row_id=self._optional_int(row["b_row_id"]),
+                previous_status=MatchStatus.CONFIRMED.value,
+                resulting_status=MatchStatus.REVIEW.value,
+                score=float(row["score"]),
+                detail="Reopened confirmed match for review.",
+                evidence_json=str(row["evidence_json"]),
+            )
+            self._refresh_run_counts(connection, run_id)
+
+    def unlink_manual(self, run_id: str, match_id: int) -> None:
+        with self._connect() as connection:
+            row = self._get_match_row(connection, run_id, match_id)
+            self._require_status(row, MatchStatus.MANUAL_MATCHED)
+            a_row_id = self._optional_int(row["a_row_id"])
+            b_row_id = self._optional_int(row["b_row_id"])
+            if a_row_id is None or b_row_id is None:
+                raise ValueError("A manual match must contain one row from each side")
+
+            connection.execute("DELETE FROM matches WHERE id = ?", (match_id,))
+            self._insert_unmatched(connection, run_id, a_row_id, side="a")
+            self._insert_unmatched(connection, run_id, b_row_id, side="b")
+            self._record_event(
+                connection=connection,
+                run_id=run_id,
+                action=ReviewAction.UNLINK,
+                source_match_id=match_id,
+                a_row_id=a_row_id,
+                b_row_id=b_row_id,
+                previous_status=MatchStatus.MANUAL_MATCHED.value,
+                resulting_status=MatchStatus.UNMATCHED.value,
+                score=None,
+                detail="Unlinked manual match; both rows returned to unmatched.",
+            )
+            self._refresh_run_counts(connection, run_id)
+
+    def restore_rejected(self, run_id: str, event_id: int) -> None:
+        with self._connect() as connection:
+            event = connection.execute(
+                """
+                SELECT *
+                FROM review_events
+                WHERE id = ? AND run_id = ? AND action = ?
+                """,
+                (event_id, run_id, ReviewAction.REJECT.value),
+            ).fetchone()
+            if event is None:
+                raise ValueError("Rejected review event not found for this run")
+
+            evidence_json = event["evidence_json"]
+            if evidence_json is None:
+                raise ValueError(
+                    "This rejected proposal predates reversible review and has no saved evidence"
+                )
+
+            a_row_id = self._optional_int(event["a_row_id"])
+            b_row_id = self._optional_int(event["b_row_id"])
+            if a_row_id is None or b_row_id is None:
+                raise ValueError("Rejected proposal is missing one of its source rows")
+
+            newer_rejection = connection.execute(
+                """
+                SELECT 1
+                FROM review_events
+                WHERE run_id = ? AND action = ?
+                  AND a_row_id = ? AND b_row_id = ?
+                  AND id > ?
+                LIMIT 1
+                """,
+                (
+                    run_id,
+                    ReviewAction.REJECT.value,
+                    a_row_id,
+                    b_row_id,
+                    event_id,
+                ),
+            ).fetchone()
+            if newer_rejection is not None:
+                raise ValueError("A newer rejection exists for this proposal")
+
+            a_unmatched = connection.execute(
+                """
+                SELECT id
+                FROM matches
+                WHERE run_id = ? AND status = ?
+                  AND a_row_id = ? AND b_row_id IS NULL
+                """,
+                (run_id, MatchStatus.UNMATCHED.value, a_row_id),
+            ).fetchone()
+            b_unmatched = connection.execute(
+                """
+                SELECT id
+                FROM matches
+                WHERE run_id = ? AND status = ?
+                  AND b_row_id = ? AND a_row_id IS NULL
+                """,
+                (run_id, MatchStatus.UNMATCHED.value, b_row_id),
+            ).fetchone()
+            if a_unmatched is None or b_unmatched is None:
+                raise ValueError(
+                    "Rejected proposal can only be restored while both rows are unmatched"
+                )
+
+            score = event["score"]
+            if score is None:
+                raise ValueError("Rejected proposal is missing its original score")
+
+            connection.execute(
+                "DELETE FROM matches WHERE id IN (?, ?)",
+                (int(a_unmatched["id"]), int(b_unmatched["id"])),
+            )
+            cursor = connection.execute(
+                """
+                INSERT INTO matches (
+                    run_id, a_row_id, b_row_id, score, status, evidence_json
+                ) VALUES (?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    run_id,
+                    a_row_id,
+                    b_row_id,
+                    float(score),
+                    MatchStatus.REVIEW.value,
+                    str(evidence_json),
+                ),
+            )
+            if cursor.lastrowid is None:
+                raise RuntimeError("SQLite did not return a match id")
+
+            self._record_event(
+                connection=connection,
+                run_id=run_id,
+                action=ReviewAction.RESTORE_REJECTED,
+                source_match_id=cursor.lastrowid,
+                a_row_id=a_row_id,
+                b_row_id=b_row_id,
+                previous_status=MatchStatus.UNMATCHED.value,
+                resulting_status=MatchStatus.REVIEW.value,
+                score=float(score),
+                detail="Restored rejected proposal to review.",
+                evidence_json=str(evidence_json),
+            )
+            self._refresh_run_counts(connection, run_id)
+
+    def restorable_rejected_event_ids(self, run_id: str) -> frozenset[int]:
+        with self._connect() as connection:
+            rows = connection.execute(
+                """
+                SELECT e.id
+                FROM review_events e
+                JOIN matches a_match
+                  ON a_match.run_id = e.run_id
+                 AND a_match.status = ?
+                 AND a_match.a_row_id = e.a_row_id
+                 AND a_match.b_row_id IS NULL
+                JOIN matches b_match
+                  ON b_match.run_id = e.run_id
+                 AND b_match.status = ?
+                 AND b_match.b_row_id = e.b_row_id
+                 AND b_match.a_row_id IS NULL
+                WHERE e.run_id = ?
+                  AND e.action = ?
+                  AND e.evidence_json IS NOT NULL
+                  AND NOT EXISTS (
+                      SELECT 1
+                      FROM review_events newer
+                      WHERE newer.run_id = e.run_id
+                        AND newer.action = ?
+                        AND newer.a_row_id = e.a_row_id
+                        AND newer.b_row_id = e.b_row_id
+                        AND newer.id > e.id
+                  )
+                """,
+                (
+                    MatchStatus.UNMATCHED.value,
+                    MatchStatus.UNMATCHED.value,
+                    run_id,
+                    ReviewAction.REJECT.value,
+                    ReviewAction.REJECT.value,
+                ),
+            ).fetchall()
+        return frozenset(int(row["id"]) for row in rows)
+
     def list_review_events(
         self,
         run_id: str,
@@ -801,13 +1007,14 @@ def _record_event(
         resulting_status: str,
         score: float | None,
         detail: str,
+        evidence_json: str | None = None,
     ) -> None:
         connection.execute(
             """
             INSERT INTO review_events (
                 run_id, created_at, action, source_match_id, a_row_id, b_row_id,
-                previous_status, resulting_status, score, detail
-            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                previous_status, resulting_status, score, evidence_json, detail
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
             """,
             (
                 run_id,
@@ -819,6 +1026,7 @@ def _record_event(
                 previous_status,
                 resulting_status,
                 score,
+                evidence_json,
                 detail,
             ),
         )

src/rowbridge/templates/results.html

@@ -45,7 +45,20 @@ <h1>Reconciliation results</h1>
               <td class="record">{% if match.a_payload %}<strong>{{ display_value(match, 'a', run.mapping.primary_a) }}</strong><small>row {{ match.a_row_number }}</small>{% else %}<span class="muted">No row</span>{% endif %}</td>
               <td class="record">{% if match.b_payload %}<strong>{{ display_value(match, 'b', run.mapping.primary_b) }}</strong><small>row {{ match.b_row_number }}</small>{% else %}<span class="muted">No row</span>{% endif %}</td>
               <td>{% if match.evidence %}<dl class="evidence-list">{% for item in match.evidence %}<div><dt>{{ item.field|replace('_', ' ') }}</dt><dd>{{ item.detail }}{% if item.score %}<em>+{{ "%.0f"|format(item.score * 100) }}</em>{% endif %}</dd></div>{% endfor %}</dl>{% else %}<span class="muted">No candidate above review threshold.</span>{% endif %}</td>
-              <td>{% if match.status.value == 'review' %}<div class="row-actions"><form method="post" action="/runs/{{ run.id }}/matches/{{ match.id }}/accept"><button class="mini-button mini-primary" type="submit">Accept</button></form><form method="post" action="/runs/{{ run.id }}/matches/{{ match.id }}/reject"><button class="mini-button mini-danger" type="submit">Reject</button></form></div>{% else %}<span class="muted">No action</span>{% endif %}</td>
+              <td>
+                {% if match.status.value == 'review' %}
+                <div class="row-actions">
+                  <form method="post" action="/runs/{{ run.id }}/matches/{{ match.id }}/accept"><button class="mini-button mini-primary" type="submit">Accept</button></form>
+                  <form method="post" action="/runs/{{ run.id }}/matches/{{ match.id }}/reject"><button class="mini-button mini-danger" type="submit">Reject</button></form>
+                </div>
+                {% elif match.status.value == 'confirmed' %}
+                <form method="post" action="/runs/{{ run.id }}/matches/{{ match.id }}/reopen"><button class="mini-button" type="submit">Reopen review</button></form>
+                {% elif match.status.value == 'manual_matched' %}
+                <form method="post" action="/runs/{{ run.id }}/matches/{{ match.id }}/unlink"><button class="mini-button" type="submit">Unlink</button></form>
+                {% else %}
+                <span class="muted">No action</span>
+                {% endif %}
+              </td>
             </tr>
           {% else %}<tr><td colspan="6" class="empty-state">No rows in this view.</td></tr>{% endfor %}
           </tbody>
@@ -96,6 +109,11 @@ <h2>Review history</h2>
               <p>{% if event.a_payload %}{{ payload_value(event.a_payload, run.mapping.primary_a) }}{% else %}No Side A row{% endif %} ↔ {% if event.b_payload %}{{ payload_value(event.b_payload, run.mapping.primary_b) }}{% else %}No Side B row{% endif %}</p>
               <p class="history-detail">{{ event.detail }}</p>
               <small>{{ event.previous_status or 'none' }} → {{ event.resulting_status }} · {{ event.created_at }}</small>
+              {% if event.id in restorable_rejected_event_ids %}
+              <form class="history-action" method="post" action="/runs/{{ run.id }}/review-events/{{ event.id }}/restore">
+                <button class="mini-button" type="submit">Restore proposal</button>
+              </form>
+              {% endif %}
             </article>
           {% endfor %}{% else %}<p class="muted panel-empty">No human decisions yet.</p>{% endif %}
         </div>

src/rowbridge/web.py

@@ -397,6 +397,9 @@ def results(
                 "summary": repository.get_summary(run_id),
                 "matches": visible_matches,
                 "events": repository.list_review_events(run_id, limit=100),
+                "restorable_rejected_event_ids": repository.restorable_rejected_event_ids(
+                    run_id
+                ),
                 "unmatched_a": unmatched_a,
                 "unmatched_b": unmatched_b,
                 "unmatched_a_count": unmatched_a_count,
@@ -432,6 +435,36 @@ def reject_match(run_id: str, match_id: int) -> RedirectResponse:
             raise HTTPException(status_code=409, detail=str(exc)) from exc
         return RedirectResponse(url=f"/runs/{run_id}#manual-link", status_code=303)
 
+    @app.post("/runs/{run_id}/matches/{match_id}/reopen")
+    def reopen_match(run_id: str, match_id: int) -> RedirectResponse:
+        if repository.get_run(run_id) is None:
+            raise HTTPException(status_code=404, detail="Run not found")
+        try:
+            repository.reopen_confirmed(run_id, match_id)
+        except ValueError as exc:
+            raise HTTPException(status_code=409, detail=str(exc)) from exc
+        return RedirectResponse(url=f"/runs/{run_id}#review-history", status_code=303)
+
+    @app.post("/runs/{run_id}/matches/{match_id}/unlink")
+    def unlink_match(run_id: str, match_id: int) -> RedirectResponse:
+        if repository.get_run(run_id) is None:
+            raise HTTPException(status_code=404, detail="Run not found")
+        try:
+            repository.unlink_manual(run_id, match_id)
+        except ValueError as exc:
+            raise HTTPException(status_code=409, detail=str(exc)) from exc
+        return RedirectResponse(url=f"/runs/{run_id}#manual-link", status_code=303)
+
+    @app.post("/runs/{run_id}/review-events/{event_id}/restore")
+    def restore_rejected(run_id: str, event_id: int) -> RedirectResponse:
+        if repository.get_run(run_id) is None:
+            raise HTTPException(status_code=404, detail="Run not found")
+        try:
+            repository.restore_rejected(run_id, event_id)
+        except ValueError as exc:
+            raise HTTPException(status_code=409, detail=str(exc)) from exc
+        return RedirectResponse(url=f"/runs/{run_id}?view=review#review-history", status_code=303)
+
     @app.post("/runs/{run_id}/manual-links")
     def manual_link(
         run_id: str,