original-label-> information

copienator/annotation_data.py

@@ -10,6 +10,7 @@ from PIL import Image
 from .json_io import read_json
 from .feedback_boxes import valid_feedback_box
 from .workspace import EvaluationWorkspace
+from .answer_origins import origins_for
 
 AnnotationData = dict[str, dict[str, dict[str, Any]]]
 RefaireList = list[list[Any]]
@@ -208,4 +209,9 @@ def load_annotation_data(
         data = {copy_id: data[copy_id]} if copy_id in data else {}
         if not data:
             warnings.append(f"Copy id {copy_id} was not found in correction.json")
+    for student_id, answers in data.items():
+        for label, answer in answers.items():
+            answer["result"]["content_origins"] = origins_for(
+                workspace.root, student_id, label, answer["pdf_path"]
+            )
     return AnnotationLoadResult(data, warnings)

copienator/commands/annotating.py

@@ -324,6 +324,10 @@ def compose_label_image(base_img, label, result, hmin,
     if error == "empty-answer" and not with_empty:
         return None, 0
 
+    origins = result.get("content_origins", [])
+    if origins:
+        error = " ".join([error, *(f"{origin}->" for origin in origins)]).strip()
+
 
     # Filter deleted items (used by reading_annotations.py)
     feedbacks = [f for f in feedbacks if "to_delete" not in f]

copienator/commands/correction.py

@@ -18,6 +18,7 @@ from google import genai
 
 from copienator import configuration as config
 from copienator.commands import grouping
+from copienator.answer_origins import save_origins
 from copienator import prompting
 from copienator.feedback_boxes import valid_feedback_box
 from copienator import (
@@ -435,6 +436,7 @@ def handle_label_errors(pid, label, res, pdf_path):
 
             if not base_add_pdf_path.exists() and not add_pdf_path.exists():
                 shutil.copy(str(pdf_path), str(add_pdf_path))
+                save_origins(INPUT_DIR, pid, add_label, add_pdf_path, [label])
                 tprint(f"\t\tCopying Copie{pid} : {label} -> {add_label}")
                 idx = reserve_next_group_idx(add_label)
                 tprint(f"\t\tMaking {add_label} group {idx+1}")
@@ -704,6 +706,7 @@ def resolve_delayed_moves():
                                 resolved_any = True
 
                                 shutil.copy(str(pdf_path), str(add_pdf_path))
+                                save_origins(INPUT_DIR, pid, target_label, add_pdf_path, [label])
                                 idx = reserve_next_group_idx(target_label)
                                 height = grouping.get_pdf_height(str(add_pdf_path))
                                 grouping.create_jpg(target_label, idx, [(pid, str(add_pdf_path), height)], GROUPS_DIR)

copienator/commands/resolve_manual.py

@@ -10,6 +10,7 @@ from typing import Any
 
 from pypdf import PdfWriter
 from copienator.pdf_cut import split_pdf
+from copienator.answer_origins import origins_for, save_origins
 
 from copienator import (
     CliError,
@@ -204,6 +205,8 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode:
     _validate_pdf_inputs(instructions, initial_paths)
 
     files_to_old: set[Path] = set()
+    origins = {key: origins_for(workspace.root, *key, path)
+               for key, path in initial_paths.items()}
     temp_files: list[Path] = []
     try:
         for instruction in instructions:
@@ -212,6 +215,8 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode:
             source = initial_paths[key_old]
             destination = current_paths[key_new]
             if instruction.cut:
+                origins[key_old] = []
+                origins[key_new] = []
                 percent, keep = instruction.cut
                 first = source.parent / f"temp_{len(temp_files)}.pdf"
                 second = source.parent / f"temp_{len(temp_files) + 1}.pdf"
@@ -232,6 +237,9 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode:
                 files_to_old.add(initial_paths[key_new])
 
             if instruction.should_merge:
+                if not instruction.cut and instruction.operator == "->":
+                    origins[key_new] = list(dict.fromkeys([
+                        *origins[key_new], instruction.old_label]))
                 writer = PdfWriter()
                 try:
                     if instruction.pipe_first:
@@ -252,6 +260,10 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode:
                 temp_files.append(temp_output)
                 files_to_old.add(initial_paths[key_new])
             elif instruction.should_copy:
+                origins[key_new] = (
+                    [instruction.old_label] if instruction.operator == "-x"
+                    and not instruction.cut else []
+                )
                 shutil.copy(source, temp_output)
                 current_paths[key_new] = temp_output
                 temp_files.append(temp_output)
@@ -287,6 +299,10 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode:
         final_name = workspace.copies_dir / f"Copie{copy_id}" / f"{label}_new.pdf"
         final_name.unlink(missing_ok=True)
         shutil.move(str(current_path), str(final_name))
+        retained_origins = [origin for origin in origins[(copy_id, label)]
+                            if (copy_id, origin) not in initial_paths
+                            or initial_paths[(copy_id, origin)] not in files_to_old]
+        save_origins(workspace.root, copy_id, label, final_name, retained_origins)
         set_suffix_and_clean_error(results, copy_id, label, "_new")
         labels = refaire_by_copy.setdefault(f"Copie{copy_id}", [])
         if label not in labels:

tests/test_manual_resolution_state.py

@@ -8,10 +8,62 @@ import pymupdf
 
 from copienator import CliError, EvaluationWorkspace, atomic_write_json, read_json
 from copienator.commands import correction, resolve_manual
+from copienator.answer_origins import origins_for, save_origins
+from copienator.annotation_data import load_annotation_data
 from copienator_gui.app import has_manual_conflicts
 
 
 class ManualResolutionStateTests(unittest.TestCase):
+    def test_automatic_additional_content_records_origin_but_wrong_label_does_not(self):
+        for kind in ("additional-answer", "wrong-label"):
+            with self.subTest(kind=kind), tempfile.TemporaryDirectory() as temporary:
+                workspace = self.fixture(Path(temporary), "->")
+                copies = workspace.copies_dir / "Copie01"
+                (copies / "B.pdf").unlink()
+                (workspace.root / "labels").write_text("A\nB\n")
+                with (
+                    patch.object(correction, "INPUT_DIR", workspace.root, create=True),
+                    patch.object(correction, "COPIES_DIR", workspace.copies_dir, create=True),
+                    patch.object(correction, "GROUPS_DIR", workspace.groups_dir, create=True),
+                    patch.object(correction, "enonce_total", return_value=""),
+                    patch.object(correction.prompting, "request_for_additional_answer", return_value=([], {})),
+                    patch.object(correction.prompting, "request_for_wrong_label", return_value=([], {})),
+                    patch.object(correction, "call_gemini_with_retries", return_value='["B"]' if kind == "additional-answer" else '"B"'),
+                    patch.object(correction, "reserve_next_group_idx", return_value=0),
+                    patch.object(correction.grouping, "get_pdf_height", return_value=200),
+                    patch.object(correction.grouping, "create_jpg"),
+                ):
+                    correction.handle_label_errors("01", "A", {"error": kind}, copies / "A.pdf")
+                self.assertEqual(
+                    origins_for(workspace.root, "01", "B", copies / "B_new.pdf"),
+                    ["A"] if kind == "additional-answer" else [],
+                )
+
+    def test_target_provenance_only_for_uncut_retained_sources(self):
+        for operator in (*resolve_manual.OPERATORS, "c{50}1>", "c{50}2x"):
+            with self.subTest(operator=operator), tempfile.TemporaryDirectory() as temporary:
+                workspace = self.fixture(Path(temporary), operator)
+                resolve_manual.resolve_manual(workspace)
+                target = resolve_manual.get_actual_pdf(workspace.copies_dir, "01", "B")
+                self.assertEqual(
+                    origins_for(workspace.root, "01", "B", target),
+                    ["A"] if operator in ("->", "-x") else [],
+                )
+
+    def test_provenance_survives_new_correction_but_not_replaced_pdf(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            workspace = self.fixture(Path(temporary), "->")
+            target = workspace.copies_dir / "Copie01" / "B.pdf"
+            save_origins(workspace.root, "01", "B", target, ["A", "A"])
+            atomic_write_json(workspace.correction_file, {
+                "B": [[{"id": "01", "result": {"score": 4, "error": "", "feedback": []}}]]
+            })
+            self.assertEqual(origins_for(workspace.root, "01", "B", target), ["A"])
+            loaded = load_annotation_data(workspace)
+            self.assertEqual(loaded.data["01"]["B"]["result"]["content_origins"], ["A"])
+            target.write_bytes(b"replacement")
+            self.assertEqual(origins_for(workspace.root, "01", "B", target), [])
+
     def test_merge_into_missing_target_uses_source_and_schedules_redo(self):
         for operator in ("->", "x>"):
             for target in ("B|", "|B"):