From 09ada28c6b2964caeaf277b4ca0580b188372092 Mon Sep 17 00:00:00 2001 From: fy59 Date: Thu, 17 Sep 2026 17:05:40 +0200 Subject: [PATCH] fix(sync): compare feedback values across row modes --- tests/test_training_feedback_sync.py | 7 +++++++ tools/import_training_feedback.py | 9 ++++++--- 2 files changed, 13 insertions(+), 3 deletions(-) diff --git a/tests/test_training_feedback_sync.py b/tests/test_training_feedback_sync.py index b80c365..035af91 100644 --- a/tests/test_training_feedback_sync.py +++ b/tests/test_training_feedback_sync.py @@ -27,6 +27,13 @@ def main(): out=d/"out.json";run("python",ROOT/"tools/export_training_feedback.py",out,"--database",db);x=json.loads(out.read_text());assert x["version"]==2 and len(x["exercise_feedback"])==1 and len(x["session_followups"])==2 revision={"revision_id":"fr_55555555-5555-4555-8555-555555555555","created_at":"2026-09-03T10:00:00+02:00","raw_text":"ressenti corrigé"};x["exercise_feedback"][0]["revisions"].append(revision);artifact.write_text(json.dumps(x),encoding="utf8") run("python",ROOT/"tools/import_training_feedback.py",artifact,"--database",db);again=run("python",ROOT/"tools/import_training_feedback.py",artifact,"--database",db).stdout;assert "revisions_added=0" in again + # The generation worker imports history first, leaving sqlite3.Row enabled + # on the shared connection. Identical feedback must remain idempotent there. + import sys + sys.path.insert(0,str(ROOT/"tools")) + from import_training_feedback import apply_feedback,load + con=sqlite3.connect(db);con.row_factory=sqlite3.Row + con.execute("BEGIN IMMEDIATE");assert apply_feedback(con,load(artifact))==(0,3,0,4);con.rollback();con.close() con=sqlite3.connect(db);assert con.execute("select count(*) from exercise_feedback_revisions").fetchone()[0]==2;assert con.execute("select raw_text from exercise_feedback").fetchone()[0]=="ressenti corrigé";con.close() x["exercise_feedback"][0]["revisions"][-1]["raw_text"]="conflit";artifact.write_text(json.dumps(x),encoding="utf8");bad=subprocess.run(["python",str(ROOT/"tools/import_training_feedback.py"),str(artifact),"--database",str(db)],capture_output=True);assert bad.returncode!=0 print("TRAINING_FEEDBACK_SYNC_TEST=PASS") diff --git a/tools/import_training_feedback.py b/tools/import_training_feedback.py index 1e4194d..02f09ae 100644 --- a/tools/import_training_feedback.py +++ b/tools/import_training_feedback.py @@ -42,7 +42,10 @@ def merge_revisions(db,table,parent_column,parent_id,revisions): for revision in revisions: old=db.execute(f"SELECT {parent_column},created_at,raw_text FROM {table} WHERE revision_id=?",(revision[0],)).fetchone(); expected=(parent_id,revision[1],revision[2]) if old is None:db.execute(f"INSERT INTO {table}(revision_id,{parent_column},created_at,raw_text) VALUES(?,?,?,?)",(revision[0],*expected));added+=1 - elif old==expected:skipped+=1 + # INVARIANT: generation workers reuse the history importer connection, + # which installs sqlite3.Row. Idempotence compares persisted values, + # never the process-local row representation selected by a prior step. + elif tuple(old)==expected:skipped+=1 else:fail("HARD CONFLICT "+revision[0]) current=max(db.execute(f"SELECT revision_id,created_at,raw_text FROM {table} WHERE {parent_column}=?",(parent_id,)).fetchall(),key=lambda r:(instant(r[1]),r[0].encode())) return added,skipped,current[2] @@ -63,7 +66,7 @@ def apply_feedback(db,root,complete_causal_envelope=False): if not canonical or canonical[0]!=occurrence[1]:fail("exercise_id incompatible avec l'occurrence") old=db.execute("SELECT session_exercise_row_id,observed_at FROM exercise_feedback WHERE feedback_id=?",(fid,)).fetchone();expected=(occurrence[0],item["observed_at"]);revisions=revision_list(item,fid,root["version"]) if old is None:db.execute("INSERT INTO exercise_feedback(feedback_id,session_exercise_row_id,observed_at,raw_text) VALUES(?,?,?,?)",(fid,*expected,revisions[0][2]));roots_added+=1 - elif old==expected:roots_skipped+=1 + elif tuple(old)==expected:roots_skipped+=1 else:fail("HARD CONFLICT "+fid) added,skipped,current=merge_revisions(db,"exercise_feedback_revisions","feedback_id",fid,revisions);revisions_added+=added;revisions_skipped+=skipped;db.execute("UPDATE exercise_feedback SET raw_text=? WHERE feedback_id=?",(current,fid)) seen=set() @@ -76,7 +79,7 @@ def apply_feedback(db,root,complete_causal_envelope=False): if not session:fail("session introuvable") old=db.execute("SELECT session_row_id,observed_at FROM session_followups WHERE followup_id=?",(fid,)).fetchone();expected=(session[0],item["observed_at"]);revisions=revision_list(item,fid,root["version"]) if old is None:db.execute("INSERT INTO session_followups(followup_id,session_row_id,observed_at,raw_text) VALUES(?,?,?,?)",(fid,*expected,revisions[0][2]));roots_added+=1 - elif old==expected:roots_skipped+=1 + elif tuple(old)==expected:roots_skipped+=1 else:fail("HARD CONFLICT "+fid) added,skipped,current=merge_revisions(db,"session_followup_revisions","followup_id",fid,revisions);revisions_added+=added;revisions_skipped+=skipped;db.execute("UPDATE session_followups SET raw_text=? WHERE followup_id=?",(current,fid)) return roots_added,roots_skipped,revisions_added,revisions_skipped