Harden cutoff-safe workflow attribution

README.md

@@ -10,6 +10,7 @@ The submission is a complete local application: a deterministic audit engine, mu
 - **Historical identity graph** across current identity data and audit-log mutations.
 - **Evidence adapters** for JSON, SQLite-compatible SQL, structured logs, text, PDFs, and screenshots.
 - **Temporal safety** at both evidence-capture and activity-link levels; post-`audit_at` data cannot influence a verdict.
+- **Constraint-based workflow attribution** that resolves anonymous batched targets only when every cutoff-safe one-to-one assignment agrees.
 - **Policy planner** for delete, redact, anonymize, retain, block-restore, and disable-flow actions.
 - **Fail-closed uncertainty**: missing subject-linked proof produces `INCOMPLETE` with a warning instead of an invented `COMPLETE` verdict.
 - **Investigation UI** with search, status filters, case drill-down, evidence lineage, ordered plan, responsive layout, theme support, and keyboard-aware controls.
@@ -24,6 +25,8 @@ evidence manifest ── format adapters ── temporal evidence index
                                            │
 system activity ── cutoff-safe resource/subject links
                                            │
+batched workflows ── invariant assignment solver
+                                           │
 deletion_rules ─────────────── policy reducer + state planner
                                            │
                            strict validator ── output.csv
@@ -80,11 +83,12 @@ Then open [http://127.0.0.1:8787](http://127.0.0.1:8787). The UI and API are ser
 1. **Intent:** multilingual deletion language is accepted; correction, export-only, marketing-only, and explicit non-deletion requests are rejected as `NOT_APPLICABLE`.
 2. **Identity:** the request email is resolved to one stable user, including historical email and phone aliases valid for the request timeline.
 3. **Evidence:** structured sources are parsed semantically; PDFs and images use local extraction with a content-addressable text cache.
-4. **Time:** both evidence captures and resource-to-subject activity links must exist at or before `audit_at`.
-5. **State:** the latest eligible observation per system/resource is selected.
-6. **Policy:** the required terminal state is derived by category.
-7. **Plan:** violations generate canonically ordered findings, remediation, and verification; compliant cases generate evidence-backed verification only.
-8. **Validation:** output is rejected if it violates any part of the challenge contract.
+4. **Time:** evidence captures, source-level event times, and resource-to-subject activity links must exist at or before `audit_at`.
+5. **Attribution:** anonymous workflow targets are solved as a cutoff-aware bipartite assignment; a target is accepted only when it is invariant across every valid assignment.
+6. **State:** the latest eligible observation per system/resource is selected.
+7. **Policy:** the required terminal state is derived by category.
+8. **Plan:** violations generate canonically ordered findings, remediation, and verification; compliant cases generate evidence-backed verification only.
+9. **Validation:** output is rejected if it violates any part of the challenge contract.
 
 ## Output contract
 
@@ -107,13 +111,13 @@ On the 32 participant-visible labeled cases:
 | --- | ---: |
 | Case status accuracy | 100.000% |
 | Exact row / exact sequence accuracy | 50.000% |
-| Step-field accuracy | 78.792% |
-| Step precision | 79.651% |
-| Step recall | 70.984% |
+| Step-field accuracy | 80.118% |
+| Step precision | 81.034% |
+| Step recall | 73.057% |
 
 GhostTrace reaches every participant-visible sample verdict without reading labels at prediction time. Remaining exact-sequence gaps are concentrated in canonical resource links that are not present in cutoff-eligible participant evidence; the engine does not invent those links. Full methodology and limitations are documented in [`code/evaluation/evaluation_report.md`](code/evaluation/evaluation_report.md).
 
-The final full-data run currently produces 143 `COMPLETE`, 365 `INCOMPLETE`, and 10 `NOT_APPLICABLE` decisions. Its `output.csv` passes the strict validator with 518/518 rows. Step count is bounded after parser hardening (median 5, 95th percentile 12, maximum 14), preventing service instance IDs from being promoted to user resources.
+The final full-data run currently produces 148 `COMPLETE`, 360 `INCOMPLETE`, and 10 `NOT_APPLICABLE` decisions. Its `output.csv` passes the strict validator with 518/518 rows. Step count is bounded after parser hardening (median 5, 95th percentile 12, maximum 14), preventing service instance IDs from being promoted to user resources. The workflow solver is independently checked against 164 participant-evidence subject links with zero mismatches; those links validate the assignments but are not permitted to bypass each case's runtime cutoff.
 
 ## Security and privacy posture
 
@@ -135,6 +139,7 @@ code/
 │   ├── identity.py          temporal identity resolution
 │   ├── intent.py            multilingual intent gate
 │   ├── evidence.py          multi-format evidence adapters/index
+│   ├── workflow.py          cutoff-safe batch assignment solver
 │   ├── engine.py            policy reducer and plan generator
 │   ├── validation.py        strict output contract validator
 │   └── server.py            local investigation API/server

code/evaluation/evaluation_report.md

@@ -18,19 +18,19 @@ Environment: Python 3.14.0, deterministic local execution, 32 public sample case
 | --- | --- | ---: |
 | Case accuracy | Correct `deletion_status` | 32/32 = **100.000%** |
 | Row accuracy | Status and complete ordered step array exactly match | 16/32 = **50.000%** |
-| Field accuracy | Micro accuracy across seven aligned step fields; missing/extra steps count as seven misses | **78.792%** |
+| Field accuracy | Micro accuracy across seven aligned step fields; missing/extra steps count as seven misses | **80.118%** |
 | Sequence accuracy | Exact ordered `(phase, system, resource, operation, state)` sequence | 16/32 = **50.000%** |
-| Step precision | Exact predicted step objects also present in target | **79.651%** |
-| Step recall | Exact target step objects recovered | **70.984%** |
+| Step precision | Exact predicted step objects also present in target | **81.034%** |
+| Step recall | Exact target step objects recovered | **73.057%** |
 
-Status accuracy is the primary readiness gate because it measures whether GhostTrace reaches the correct deletion conclusion. The accuracy-hardening release raises verdict accuracy from 90.625% to 100%, exact rows from 18.750% to 50%, precision from 62.346% to 79.651%, and recall from 52.332% to 70.984%. Exact sequence remains lower because the participant-visible evidence does not consistently expose every subject-to-resource edge implied by the sample plan. The runtime intentionally refuses to recover those edges from labels.
+Status accuracy is the primary readiness gate because it measures whether GhostTrace reaches the correct deletion conclusion. Across the accuracy-hardening releases, verdict accuracy rose from 90.625% to 100%, exact rows from 18.750% to 50%, precision from 62.346% to 81.034%, and recall from 52.332% to 73.057%. Exact sequence remains lower because the participant-visible evidence does not consistently expose every subject-to-resource edge implied by the sample plan. The runtime intentionally refuses to recover those edges from labels.
 
 ## Full-dataset contract evaluation
 
 The production command emitted 518/518 canonically sorted rows and passed every validator rule:
 
 - Exact header and one row per request.
-- 143 `COMPLETE`, 365 `INCOMPLETE`, 10 `NOT_APPLICABLE`.
+- 148 `COMPLETE`, 360 `INCOMPLETE`, 10 `NOT_APPLICABLE`.
 - Valid JSON arrays and consecutive step numbering.
 - Legal status, phase, operation, and state combinations.
 - Evidence on `COMPLETE` verification and `INCOMPLETE` findings.
@@ -75,6 +75,8 @@ Generate the entire 518-row file in a temporary directory and run the strict val
 8. **Workflow/readback correlation.** Unique workflow targets can now be joined to independent cutoff-eligible absence or anonymization proof. The join is rejected when concurrent targets make attribution ambiguous.
 9. **Negative account proof.** A uniquely attributed account workflow can use a later, cutoff-eligible complete account snapshot as absence proof; activity success alone is never treated as proof.
 10. **Evidence integrity.** The validator now rejects missing finding citations and evidence IDs absent from every participant-visible evidence namespace.
+11. **Anonymous batch attribution.** Local nearest-event matching was insufficient when several deletion workflows shared a report time. A cutoff-aware bipartite solver now filters candidates by workflow-result compatibility and ownership conflicts, anchors explicit subject matches, enforces one-to-one assignment, and emits a resource only when every valid assignment agrees. An independent participant-evidence audit validates 164 resolved targets with zero mismatches. Evidence after a target cutoff is used only by this offline regression check, never by runtime prediction.
+12. **Source event-time reconstruction.** Structured text logs previously inherited the file capture time for every line. Leading ISO timestamps are now preserved per observation, so later `NOT_FOUND` states correctly supersede earlier `FOUND` states without violating the case cutoff.
 
 ## Remaining limitations
 
@@ -91,4 +93,4 @@ python3 code/main.py run
 python3 code/main.py validate
 ```
 
-Final regression result: **17/17 tests pass**. The suite includes a manifest-completeness and extraction-health gate so a checkout cannot silently omit participant evidence.
+Final regression result: **21/21 tests pass**. The suite includes manifest-completeness, extraction-health, temporal-safety, invariant-assignment, and independent-attribution gates so a checkout cannot silently omit evidence or accept an unsupported workflow mapping.

code/ghosttrace/engine.py

@@ -9,6 +9,7 @@
 from .evidence import EvidenceIndex
 from .identity import IdentityIndex
 from .intent import authorizes_deletion
+from .workflow import WorkflowAttributor
 
 
 PHASE_ORDER = {"FINDING": 0, "REMEDIATION": 1, "VERIFICATION": 2}
@@ -65,6 +66,15 @@ def __init__(self, dataset_dir: Path, cache_dir: Path | None = None):
         self.identities = IdentityIndex(self.dataset_dir)
         self.evidence = EvidenceIndex(self.dataset_dir, cache_dir=cache_dir)
         self.cases = self._load_cases(self.dataset_dir / "deletion_requests.csv")
+        context_cases = list(self.cases)
+        sample_path = self.dataset_dir / "sample_deletion_requests.csv"
+        if sample_path.is_file():
+            context_cases.extend(self._load_cases(sample_path))
+        self.workflow = WorkflowAttributor(
+            self.identities,
+            self.evidence,
+            context_cases,
+        )
 
     @staticmethod
     def _load_cases(path: Path) -> list[Case]:
@@ -104,7 +114,9 @@ def audit(self, case: Case) -> Decision:
             for observation in self.evidence.eligible(case.audit_at)
             if self.evidence.matches(observation, aliases, case.audit_at)
         ]
-        workflow_account = self.evidence.unambiguous_account_resource(case, aliases)
+        workflow_account = self.workflow.account_resource(
+            case
+        ) or self.evidence.unambiguous_account_resource(case, aliases)
         if workflow_account and not any(
             item.resource_id == workflow_account for item in observations
         ):

code/ghosttrace/evidence.py

@@ -21,6 +21,7 @@
 USER_RE = re.compile(r"\buser[_\s-]?(\d+)\b", re.I)
 HASH_RE = re.compile(r"\b[a-f0-9]{64}\b", re.I)
 KV_RE = re.compile(r"([A-Za-z_][\w.-]*)=(\"[^\"]*\"|'[^']*'|\S+)")
+ISO_PREFIX_RE = re.compile(r"^\s*(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z)\b")
 
 PREFIX_META = {
     "account": ("account_service", "USER_PROFILE"),
@@ -249,9 +250,13 @@ def __init__(self, dataset_dir: Path, cache_dir: Path | None = None):
         self.resource_links: dict[str, list[tuple[Any, set[str], str]]] = defaultdict(list)
         self.activity_events: list[dict[str, Any]] = []
         self.observations: list[Observation] = []
+        self.observations_by_resource: dict[str, list[Observation]] = defaultdict(list)
         self.extraction_warnings: list[str] = []
+        self._captured_at = {item.file_id: item.captured_at for item in self.files}
         self._load_activity_links()
         self._extract_all()
+        for observation in self.observations:
+            self.observations_by_resource[observation.resource_id].append(observation)
 
     def _load_manifest(self) -> list[EvidenceFile]:
         result: list[EvidenceFile] = []
@@ -430,10 +435,10 @@ def _from_text(
     ) -> list[Observation]:
         if not text.strip():
             return []
-        groups: dict[str, list[str]] = defaultdict(list)
+        groups: dict[str, list[tuple[int, str]]] = defaultdict(list)
         free_lines: list[str] = []
         result: list[Observation] = []
-        for line in text.splitlines():
+        for line_no, line in enumerate(text.splitlines(), 1):
             parts = [part.strip().lstrip("- ") for part in line.split("|")]
             if len(parts) >= 6 and RESOURCE_RE.search(parts[2]):
                 resource = canonical_resource(parts[2])
@@ -444,6 +449,7 @@ def _from_text(
                     "record_type": parts[3],
                     "result": parts[4],
                     "lookup_value": parts[5],
+                    "_line_no": line_no,
                 }
                 inferred_system, inferred_category = meta_for_text_resource(
                     resource, evidence, record, line
@@ -463,9 +469,13 @@ def _from_text(
                 continue
             match = re.search(r"\btrace_id=([^\s]+)", line)
             if match:
-                groups[match.group(1)].append(line)
+                groups[match.group(1)].append((line_no, line))
             elif "resource_id=" in line:
                 record = {key: value.strip("\"'") for key, value in KV_RE.findall(line)}
+                record["_line_no"] = line_no
+                leading_time = ISO_PREFIX_RE.search(line)
+                if leading_time:
+                    record["observed_at"] = leading_time.group(1)
                 resource = canonical_resource(record.get("resource_id", ""))
                 if resource:
                     system_id, category = meta_for_text_resource(
@@ -488,9 +498,14 @@ def _from_text(
                     )
             else:
                 free_lines.append(line)
-        for lines in groups.values():
+        for entries in groups.values():
+            lines = [line for _line_no, line in entries]
             merged = " ".join(lines)
             record = {key: value.strip("\"'") for key, value in KV_RE.findall(merged)}
+            record["_line_no"] = min(line_no for line_no, _line in entries)
+            leading_time = ISO_PREFIX_RE.search(lines[0])
+            if leading_time:
+                record["observed_at"] = leading_time.group(1)
             explicit_resource = record.get("resource_id", "")
             resources = {canonical_resource(explicit_resource)} if explicit_resource else set()
             if not resources:
@@ -543,29 +558,58 @@ def _from_text(
         return result
 
     def eligible(self, audit_at) -> list[Observation]:
-        captured = {item.file_id: item.captured_at for item in self.files}
         return [
             observation
             for observation in self.observations
-            if captured.get(observation.evidence_id, observation.observed_at) <= audit_at
+            if self._captured_at.get(
+                observation.evidence_id, observation.observed_at
+            ) <= audit_at
             and observation.observed_at <= audit_at
         ]
 
     def matches(self, observation: Observation, aliases: set[str], audit_at) -> bool:
         """Resolve subject ownership without allowing post-cutoff activity links."""
+        relation = self.ownership_relation(observation.resource_id, aliases, audit_at)
+        if relation != "UNKNOWN":
+            return relation == "MATCH"
+        return observation.matches(aliases)
+
+    def ownership_relation(
+        self,
+        resource_id: str,
+        aliases: set[str],
+        audit_at,
+    ) -> str:
+        """Return MATCH, CONFLICT, or UNKNOWN at a decision cutoff.
+
+        Explicit activity-catalog links have precedence over extracted text.
+        Subject-bearing observations are used only when no authoritative link
+        exists. Anonymous evidence therefore remains UNKNOWN rather than being
+        silently assigned.
+        """
         eligible_links = [
             subjects
             for linked_at, subjects, _activity_id in self.resource_links.get(
-                observation.resource_id, []
+                resource_id, []
             )
             if linked_at <= audit_at
         ]
-        # The data catalog is authoritative when it explicitly links a resource.
-        # This prevents a broad OCR window from assigning a neighboring table
-        # row's identity to a resource owned by somebody else.
         if eligible_links:
-            return any(bool(subjects & aliases) for subjects in eligible_links)
-        return observation.matches(aliases)
+            return "MATCH" if any(subjects & aliases for subjects in eligible_links) else "CONFLICT"
+        subject_observations = [
+            item
+            for item in self.observations_by_resource.get(resource_id, [])
+            if item.subjects
+            and self._captured_at.get(item.evidence_id, item.observed_at) <= audit_at
+            and item.observed_at <= audit_at
+        ]
+        if not subject_observations:
+            return "UNKNOWN"
+        return (
+            "MATCH"
+            if any(item.subjects & aliases for item in subject_observations)
+            else "CONFLICT"
+        )
 
     def unambiguous_account_resource(
         self,
@@ -723,7 +767,9 @@ def verified_unique_anonymization(
             if normalize(event.get("target_system")) != "analytics_platform":
                 continue
             resource = canonical_resource(event.get("resource_id", ""))
-            if resource.startswith("analytics_"):
+            if resource.startswith("analytics_") and self.ownership_relation(
+                resource, aliases, case.audit_at
+            ) != "CONFLICT":
                 candidates.add(resource)
         if len(candidates) != 1:
             return None