diff --git a/api/services/forge_projection.py b/api/services/forge_projection.py index 9af8884..2645f2f 100644 --- a/api/services/forge_projection.py +++ b/api/services/forge_projection.py @@ -646,10 +646,15 @@ class ProjectionDiff: # Set when the source could not support a claim of absence, so `stale` was # deliberately left empty rather than computed (`STATE-WP-0084-T01`). stale_withheld: str | None = None + # Source-owned dependency-edge preview (STATE-WP-0094-T05); None when the + # caller supplied no hub edge state. Keys: create, remove, retain, + # unresolved, external. + edges: dict[str, list[dict[str, Any]]] | None = None @property def clean(self) -> bool: - return not (self.missing or self.stale or self.differing or self.ambiguous) + edge_changes = bool(self.edges and (self.edges["create"] or self.edges["remove"])) + return not (self.missing or self.stale or self.differing or self.ambiguous or edge_changes) @property def would_remove(self) -> int: @@ -672,6 +677,10 @@ class ProjectionDiff: "stale": self.stale, "differing": self.differing, "ambiguous": self.ambiguous, + **({} if self.edges is None else { + "edge_counts": {k: len(v) for k, v in self.edges.items()}, + "edges": self.edges, + }), } @@ -712,11 +721,20 @@ def diff_against_hub( derived: DerivedProjection, hub_workplans: list[dict[str, Any]], hub_tasks_by_workplan: dict[str, list[dict[str, Any]]], + *, + hub_edges: list[dict[str, Any]] | None = None, + edge_targets: dict[str, list[str]] | None = None, ) -> ProjectionDiff: """Compare a derived projection with the hub's current records. Pure: takes the hub's state as data rather than reading it, so the comparison is testable without a database and cannot accidentally mutate anything. + + Dependency edges are previewed only when ``hub_edges`` is given: the + ``workplan_dependencies`` rows whose from-workplan belongs to this + repository (with ``source_owner``). ``edge_targets`` maps a target record id + to its fleet-wide, non-retired candidate ids; a target absent from it but + declared in this projection resolves to the id the receipt would create. """ d = ProjectionDiff(repo_slug=derived.repo_slug, commit=derived.commit) d.ambiguous.extend(_derived_identity_ambiguities(derived)) @@ -896,9 +914,72 @@ def diff_against_hub( {"kind": "task", "record_id": t.record_id, "uuid": t.uuid, "workplan": w.record_id, "changed": changed} ) + if hub_edges is not None: + d.edges = _preview_source_edges( + derived, want_wp, have_wp, hub_tasks_by_workplan, hub_edges, edge_targets or {} + ) return d +def _preview_source_edges( + derived: DerivedProjection, + want_wp: dict[str, DerivedWorkplan], + have_wp: dict[str, dict[str, Any]], + hub_tasks_by_workplan: dict[str, list[dict[str, Any]]], + hub_edges: list[dict[str, Any]], + edge_targets: dict[str, list[str]], +) -> dict[str, list[dict[str, Any]]]: + """What receiving would do to source-owned edges, using the same planner.""" + from_ids: dict[str, str] = {} + task_ids: dict[str, dict[str, str]] = {} + pending: dict[str, str] = {} # record id -> id the receipt would create + for key, w in want_wp.items(): + cur = have_wp.get(key) + wp_id = str(cur["id"]) if cur else w.uuid + from_ids[key] = wp_id + if not cur: + pending[w.record_id] = wp_id + hub_rows = hub_tasks_by_workplan.get(wp_id) or hub_tasks_by_workplan.get(w.uuid) or [] + by_record = {(r.get("record_id") or "").strip().lower(): str(r["id"]) + for r in hub_rows if r.get("record_id")} + for t in w.tasks: + k = t.record_id.strip().lower() + if k not in by_record: + by_record[k] = t.uuid + pending[t.record_id] = t.uuid + task_ids[wp_id] = by_record + targets: dict[str, tuple[str, Any]] = {} + for target in source_edge_targets(want_wp): + cands = sorted({str(c) for c in edge_targets.get(target, [])}) + if not cands and target in pending: + cands = [pending[target]] + if not cands: + targets[target] = ("missing", None) + elif len(cands) > 1: + targets[target] = ("ambiguous", cands) + else: + targets[target] = ("ok", cands[0]) + desired, unresolved, held_sides = plan_source_edges(want_wp, from_ids, task_ids, targets) + + def _s(v: Any) -> str | None: + return str(v) if v else None + + existing = [ + ((str(e["from_workplan_id"]), _s(e.get("from_task_id")), _s(e.get("to_workplan_id")), + _s(e.get("to_task_id")), e.get("relationship_type") or "blocks"), + str(e.get("id")), e.get("source_owner")) + for e in hub_edges + ] + plan = classify_source_edges(desired, held_sides, existing, edge_source_owner(derived.repo_slug)) + return { + "create": [_edge_dict(k, i) for k, i in plan["create"]], + "remove": [_edge_dict(k, {"id": rid}) for k, rid in plan["remove"]], + "retain": [_edge_dict(k, {"id": rid}) for k, rid in plan["retain"]], + "unresolved": unresolved, + "external": [_edge_dict(k, i) for k, i in plan["external"]], + } + + # --------------------------------------------------------------------------- # Applying the reset (STATE-WP-0083-T03) # --------------------------------------------------------------------------- @@ -1717,6 +1798,123 @@ def edge_source_owner(repo_slug: str) -> str: return f"{EDGE_SOURCE_PREFIX}{repo_slug}" +def _iter_edge_declarations(want: dict[str, DerivedWorkplan]): + """Yield ``(key, workplan, task_or_None, target)`` for every desired declaration. + + Shared by receiving and the dry-run preview so both apply the same rules: + historical workplans declare nothing; task-block edges of closed workplans + and closed tasks are history (C-20 parity). + """ + for key, w in want.items(): + if w.historical_preservation: + continue + for target in w.depends_on_workplans + w.depends_on_tasks: + yield key, w, None, target + if _edge_workplan_closed(w.status): + continue + for t in w.tasks: + if not t.depends_on or _edge_task_closed(t.status): + continue + for target in t.depends_on: + yield key, w, t, target + + +def source_edge_targets(want: dict[str, DerivedWorkplan]) -> set[str]: + """Every target record id a receipt of ``want`` would resolve.""" + return {target for _, _, _, target in _iter_edge_declarations(want)} + + +def plan_source_edges( + want: dict[str, DerivedWorkplan], + from_workplan_ids: dict[str, Any], + task_ids: dict[Any, dict[str, Any]], + targets: dict[str, tuple[str, Any]], +) -> tuple[dict[tuple, dict[str, Any]], list[dict[str, Any]], set[tuple]]: + """Pure: desired edge keys, unresolved declarations and held from-sides. + + ``from_workplan_ids`` maps a ``want`` key to the from workplan's id (keys + absent are skipped); ``task_ids`` maps that id to ``{record_id.lower(): task + id}``; ``targets`` maps a target record id to ``("ok", id)`` or + ``("missing"|"ambiguous", detail)`` (absent means missing). + Edge key: ``(from_workplan_id, from_task_id, to_workplan_id, to_task_id, rel)``. + """ + desired: dict[tuple, dict[str, Any]] = {} + unresolved: list[dict[str, Any]] = [] + # From-sides with an unresolved declaration keep every owned edge: the + # unresolved target may be one of them (e.g. newly ambiguous). + held_sides: set[tuple] = set() + for key, w, t, target in _iter_edge_declarations(want): + wp_id = from_workplan_ids.get(key) + if wp_id is None: + continue + from_task_id = None + if t is not None: + from_task_id = task_ids.get(wp_id, {}).get(t.record_id.strip().lower()) + if from_task_id is None: + held_sides.add((wp_id, "*")) + unresolved.append({"workplan": w.record_id, "from": t.record_id, + "target": target, "reason": "from-task-missing"}) + continue + side = (wp_id, from_task_id) + is_task = bool(_TASK_TARGET_RE.search(target)) + state, value = targets.get(target, ("missing", None)) + base = {"workplan": w.record_id, "from": t.record_id if t is not None else w.record_id, + "target": target} + if state == "ok" and ((not is_task and value == wp_id) or (is_task and value == from_task_id)): + state, value = "self", None + if state != "ok": + held_sides.add(side) + unresolved.append({**base, "reason": state, **({"candidates": value} if value else {})}) + continue + rel = "starts_after" if is_task else "blocks" + edge_key = (wp_id, from_task_id, None if is_task else value, value if is_task else None, rel) + desired.setdefault(edge_key, {**base, "relationship_type": rel}) + return desired, unresolved, held_sides + + +def _edge_dict(edge_key: tuple, info: dict[str, Any] | None = None) -> dict[str, Any]: + from_wp, from_task, to_wp, to_task, rel = edge_key + return { + **(info or {}), + "from_workplan_id": str(from_wp), + "from_task_id": str(from_task) if from_task else None, + "to_workplan_id": str(to_wp) if to_wp else None, + "to_task_id": str(to_task) if to_task else None, + "relationship_type": rel, + } + + +def classify_source_edges( + desired: dict[tuple, dict[str, Any]], + held_sides: set[tuple], + existing: list[tuple[tuple, Any, str | None]], + owner: str, +) -> dict[str, list]: + """Pure: split desired/existing edges into create/external/remove/retain. + + ``existing`` lists ``(edge key, row id, source_owner)`` for every row whose + from-workplan belongs to the repository. Several rows may share a key (an + owned row and an API-authored duplicate); they are never collapsed, so an + external duplicate can neither hide nor protect an owned row. + """ + owned_keys = {k for k, _, o in existing if o == owner} + any_keys = {k for k, _, _ in existing} + plan: dict[str, list] = {"create": [], "external": [], "remove": [], "retain": []} + for edge_key in sorted(desired, key=str): + if edge_key in owned_keys: + continue + bucket = "external" if edge_key in any_keys else "create" + plan[bucket].append((edge_key, desired[edge_key])) + for edge_key, row_id, src_owner in sorted(existing, key=lambda r: (str(r[0]), str(r[1]))): + if src_owner != owner or edge_key in desired: + continue + if (edge_key[0], edge_key[1]) in held_sides or (edge_key[0], "*") in held_sides: + plan["retain"].append((edge_key, row_id)) + else: + plan["remove"].append((edge_key, row_id)) + return plan + + async def _resolve_edge_target(session: Any, record_id: str, cache: dict) -> tuple[str, Any]: """Resolve a declared target fleet-wide to ``("ok", uuid)`` or a reason. @@ -1779,63 +1977,26 @@ async def _reconcile_source_edges( owner = edge_source_owner(repo_slug) await session.flush() - # (from_workplan_id, from_task_id, to_workplan_id, to_task_id, relationship) - desired: dict[tuple, dict[str, Any]] = {} - # From-sides with an unresolved declaration keep every owned edge: the - # unresolved target may be one of them (e.g. newly ambiguous). - held_sides: set[tuple] = set() - cache: dict[str, tuple[str, Any]] = {} - - async def _declare(row: Any, w: DerivedWorkplan, from_task: Any, from_label: str, target: str) -> None: - side = (row.id, from_task.id if from_task is not None else None) - is_task = bool(_TASK_TARGET_RE.search(target)) - state, value = await _resolve_edge_target(session, target, cache) - base = {"workplan": w.record_id, "from": from_label, "target": target} - if state == "ok" and ((not is_task and value == row.id) or (is_task and from_task is not None and value == from_task.id)): - state, value = "self", None - if state != "ok": - held_sides.add(side) - outcome.unresolved_edges.append({**base, "reason": state, **({"candidates": value} if value else {})}) - return - rel = "starts_after" if is_task else "blocks" - key = (row.id, side[1], None if is_task else value, value if is_task else None, rel) - desired.setdefault(key, {**base, "relationship_type": rel}) - - for key, w in want.items(): - row = received_rows.get(key) - if row is None or w.historical_preservation: + from_ids = {key: row.id for key, row in received_rows.items() if key in want} + targets: dict[str, tuple[str, Any]] = {} + for target in sorted(source_edge_targets(want)): + targets[target] = await _resolve_edge_target(session, target, targets) + task_ids: dict[Any, dict[str, Any]] = {} + for key, w, t, _ in _iter_edge_declarations(want): + wp_id = from_ids.get(key) + if t is None or wp_id is None or wp_id in task_ids: continue - for target in w.depends_on_workplans + w.depends_on_tasks: - await _declare(row, w, None, w.record_id, target) - # C-20 parity: closed workplans and closed tasks are history, not live - # ordering, so their task-block edges are not desired. - if _edge_workplan_closed(w.status): - continue - task_rows = None - for t in w.tasks: - if not t.depends_on or _edge_task_closed(t.status): - continue - if task_rows is None: - task_rows = { - (tr.record_id or "").strip().lower(): tr - for tr in (await session.execute(select(Task).where(Task.workplan_id == row.id))).scalars() - } - from_task = task_rows.get(t.record_id.strip().lower()) - if from_task is None: - held_sides.add((row.id, "*")) - for target in t.depends_on: - outcome.unresolved_edges.append( - {"workplan": w.record_id, "from": t.record_id, "target": target, - "reason": "from-task-missing"} - ) - continue - for target in t.depends_on: - await _declare(row, w, from_task, t.record_id, target) + task_ids[wp_id] = { + (tr.record_id or "").strip().lower(): tr.id + for tr in (await session.execute(select(Task).where(Task.workplan_id == wp_id))).scalars() + } + desired, unresolved, held_sides = plan_source_edges(want, from_ids, task_ids, targets) + outcome.unresolved_edges.extend(unresolved) repo_workplan_ids = list( (await session.execute(select(Workplan.id).where(Workplan.repo_id == repo.id))).scalars() ) - existing = list( + existing_rows = list( ( await session.execute( select(WorkplanDependency).where( @@ -1844,29 +2005,17 @@ async def _reconcile_source_edges( ) ).scalars() ) if repo_workplan_ids else [] - by_key: dict[tuple, Any] = { - (e.from_workplan_id, e.from_task_id, e.to_workplan_id, e.to_task_id, e.relationship_type): e - for e in existing - } - - def _edge_dict(edge_key: tuple, info: dict[str, Any] | None = None) -> dict[str, Any]: - from_wp, from_task, to_wp, to_task, rel = edge_key - return { - **(info or {}), - "from_workplan_id": str(from_wp), - "from_task_id": str(from_task) if from_task else None, - "to_workplan_id": str(to_wp) if to_wp else None, - "to_task_id": str(to_task) if to_task else None, - "relationship_type": rel, - } - - for edge_key in sorted(desired, key=str): - info = desired[edge_key] - current = by_key.get(edge_key) - if current is not None: - if current.source_owner != owner: - outcome.external_edges.append(_edge_dict(edge_key, info)) - continue + rows_by_id = {e.id: e for e in existing_rows} + plan = classify_source_edges( + desired, + held_sides, + [((e.from_workplan_id, e.from_task_id, e.to_workplan_id, e.to_task_id, e.relationship_type), + e.id, e.source_owner) for e in existing_rows], + owner, + ) + for edge_key, info in plan["external"]: + outcome.external_edges.append(_edge_dict(edge_key, info)) + for edge_key, info in plan["create"]: from_wp, from_task, to_wp, to_task, rel = edge_key session.add( WorkplanDependency( @@ -1881,19 +2030,13 @@ async def _reconcile_source_edges( ) ) outcome.created_edges.append(_edge_dict(edge_key, info)) - - for edge_key, edge in sorted(by_key.items(), key=lambda kv: str(kv[0])): - if edge.source_owner != owner or edge_key in desired: - continue - if (edge.from_workplan_id, edge.from_task_id) in held_sides or ( - edge.from_workplan_id, "*" - ) in held_sides: - outcome.notes.append( - f"Retained source-owned edge {edge.id}: its from-side has an unresolved declaration." - ) - continue - await session.delete(edge) - outcome.removed_edges.append(_edge_dict(edge_key, {"id": str(edge.id)})) + for edge_key, row_id in plan["retain"]: + outcome.notes.append( + f"Retained source-owned edge {row_id}: its from-side has an unresolved declaration." + ) + for edge_key, row_id in plan["remove"]: + await session.delete(rows_by_id[row_id]) + outcome.removed_edges.append(_edge_dict(edge_key, {"id": str(row_id)})) # Integrity failures surface here, inside the caller's transaction. await session.flush() diff --git a/docs/forge-dependency-edge-receiving.md b/docs/forge-dependency-edge-receiving.md index b3ec5f2..b85ef3e 100644 --- a/docs/forge-dependency-edge-receiving.md +++ b/docs/forge-dependency-edge-receiving.md @@ -67,6 +67,24 @@ ownership contract first. Outcome schema additions (`state-hub.projection-reset.v1`): `created_edges`, `removed_edges`, `unresolved_edges`, `external_edges` lists and their counts. +## Dry-run preview + +`diff_against_hub(..., hub_edges=[...], edge_targets={...})` previews edges +with the same pure planner receiving uses (`plan_source_edges` + +`classify_source_edges`). `hub_edges` are the rows whose from-workplan belongs +to the repository (with `source_owner`); `edge_targets` maps a target record id +to its fleet-wide non-retired candidate ids. A target the hub lacks but this +projection declares resolves to the id the receipt would create. The diff gains +`edges` (`create`, `remove`, `retain`, `unresolved`, `external`) and +`edge_counts`; only `create`/`remove` make it non-clean. Without `hub_edges` +the output is unchanged (no `edges` key). + +Duplicate keys: several rows may share one edge key (an owned row plus an +API-authored duplicate). The classifier works on the row list, so an external +duplicate can neither hide an owned row from removal nor turn a satisfied +declaration into an external report (fixed 2026-10-04; the first +implementation keyed rows by edge key and let the last row win). + ## Coexistence with C-20 C-20 keys existence on (from task, target, relationship) regardless of owner, @@ -78,7 +96,6 @@ explicit adoption decision. - Pre-existing C-20/API edges are never adopted, so a removed declaration does not remove an edge C-20 created earlier; adoption needs its own decision. -- `diff_against_hub` (dry run) does not yet preview edge changes. - Ambiguity by slug for workplans is mostly prevented by the global slug index; task `record_id` ambiguity across workplans is the realistic case. - Not deployed; no evidence of scheduled freshness. T05 stays in progress. diff --git a/tests/test_forge_dependency_edge_preview.py b/tests/test_forge_dependency_edge_preview.py new file mode 100644 index 0000000..1f75d12 --- /dev/null +++ b/tests/test_forge_dependency_edge_preview.py @@ -0,0 +1,157 @@ +"""Dry-run preview of source-owned dependency edges (STATE-WP-0094-T05 follow-up). + +``diff_against_hub`` is pure, so these tests need no database: hub state is +passed as data. They pin that the preview uses the same rules as receiving. +""" +from pathlib import Path + +from api.services import forge_projection as fp +from tests.test_forge_dependency_edges import _source, _write + +TGT_WP = "11111111-1111-1111-1111-111111111111" +TGT_T01 = "22222222-2222-2222-2222-222222222222" +OWNER = "forge:edge-source" + + +def _derive(tmp_path: Path, frontmatter_deps: str, **kw): + _write(tmp_path, _source(frontmatter_deps, **kw)) + return fp.derive_from_checkout(tmp_path, "edge-source", "b" * 40) + + +def _hub(derived): + """Hub already holds every derived workplan and task under derived UUIDs.""" + workplans = [{"id": w.uuid, "slug": w.record_id.lower(), "status": w.status, + "flavor": w.flavor, "backing_relative_path": w.relative_path} + for w in derived.workplans] + tasks = {w.uuid: [{"id": t.uuid, "record_id": t.record_id, "title": t.title, + "status": t.status} for t in w.tasks] for w in derived.workplans} + return workplans, tasks + + +def _ids(derived): + wp = {w.record_id: w.uuid for w in derived.workplans} + tasks = {t.record_id: t.uuid for w in derived.workplans for t in w.tasks} + return wp, tasks + + +def test_preview_absent_without_hub_edges(tmp_path): + derived = _derive(tmp_path, "depends_on: [TGT-WP-0001]\n") + workplans, tasks = _hub(derived) + diff = fp.diff_against_hub(derived, workplans, tasks) + assert diff.edges is None + assert "edges" not in diff.to_dict() + + +def test_preview_lists_create_unresolved_and_external(tmp_path): + derived = _derive(tmp_path, "depends_on: [TGT-WP-0001, TGT-WP-0001-T01]\n") + workplans, tasks = _hub(derived) + wp, tk = _ids(derived) + # The sibling edge already exists as an externally authored (C-20/API) row. + external = {"id": "e1", "from_workplan_id": wp["SRC-WP-0001"], + "from_task_id": tk["SRC-WP-0001-T01"], "to_workplan_id": wp["SRC-WP-0002"], + "to_task_id": None, "relationship_type": "blocks", "source_owner": None} + diff = fp.diff_against_hub( + derived, workplans, tasks, hub_edges=[external], + edge_targets={"TGT-WP-0001": [TGT_WP], "TGT-WP-0001-T01": [TGT_T01], + "SRC-WP-0002": [wp["SRC-WP-0002"]]}, + ) + e = diff.edges + created = {(c["from"], c["target"], c["relationship_type"]) for c in e["create"]} + assert created == {("SRC-WP-0001", "TGT-WP-0001", "blocks"), + ("SRC-WP-0001", "TGT-WP-0001-T01", "starts_after")} + assert [(x["from"], x["target"]) for x in e["external"]] == [("SRC-WP-0001-T01", "SRC-WP-0002")] + # T02 declares a missing target; the closed T03 declares nothing. + assert [(u["from"], u["reason"]) for u in e["unresolved"]] == [("SRC-WP-0001-T02", "missing")] + assert e["remove"] == [] and e["retain"] == [] + assert not diff.clean + out = diff.to_dict() + assert out["edge_counts"] == {"create": 2, "remove": 0, "retain": 0, "unresolved": 1, "external": 1} + + +def test_preview_removes_only_undeclared_owned_edges(tmp_path): + derived = _derive(tmp_path, "", t02_target="SRC-WP-0002") + workplans, tasks = _hub(derived) + wp, tk = _ids(derived) + targets = {"SRC-WP-0002": [wp["SRC-WP-0002"]]} + owned_live = {"id": "o1", "from_workplan_id": wp["SRC-WP-0001"], + "from_task_id": tk["SRC-WP-0001-T01"], "to_workplan_id": wp["SRC-WP-0002"], + "to_task_id": None, "relationship_type": "blocks", "source_owner": OWNER} + owned_live_t02 = {**owned_live, "id": "o2", "from_task_id": tk["SRC-WP-0001-T02"]} + owned_gone = {"id": "o3", "from_workplan_id": wp["SRC-WP-0001"], "from_task_id": None, + "to_workplan_id": TGT_WP, "to_task_id": None, "relationship_type": "blocks", + "source_owner": OWNER} + foreign_gone = {**owned_gone, "id": "x1", "source_owner": "forge:other"} + null_gone = {**owned_gone, "id": "x2", "source_owner": None} + diff = fp.diff_against_hub( + derived, workplans, tasks, + hub_edges=[owned_live, owned_live_t02, owned_gone, foreign_gone, null_gone], + edge_targets=targets, + ) + e = diff.edges + assert [r["id"] for r in e["remove"]] == ["o3"] + assert e["create"] == [] and e["external"] == [] and e["unresolved"] == [] + assert not diff.clean + + +def test_preview_holds_owned_edges_of_unresolved_side_and_is_clean_when_settled(tmp_path): + derived = _derive(tmp_path, "depends_on: [TGT-WP-0001]\n") + workplans, tasks = _hub(derived) + wp, tk = _ids(derived) + owned = [ + {"id": "o1", "from_workplan_id": wp["SRC-WP-0001"], "from_task_id": None, + "to_workplan_id": TGT_WP, "to_task_id": None, "relationship_type": "blocks", + "source_owner": OWNER}, + {"id": "o2", "from_workplan_id": wp["SRC-WP-0001"], "from_task_id": tk["SRC-WP-0001-T01"], + "to_workplan_id": wp["SRC-WP-0002"], "to_task_id": None, "relationship_type": "blocks", + "source_owner": OWNER}, + # Owned edge of T02, whose declared target is now ambiguous: retained. + {"id": "o3", "from_workplan_id": wp["SRC-WP-0001"], "from_task_id": tk["SRC-WP-0001-T02"], + "to_workplan_id": TGT_WP, "to_task_id": None, "relationship_type": "blocks", + "source_owner": OWNER}, + ] + diff = fp.diff_against_hub( + derived, workplans, tasks, hub_edges=owned, + edge_targets={"TGT-WP-0001": [TGT_WP], "SRC-WP-0002": [wp["SRC-WP-0002"]], + "MISSING-WP-0009": ["a", "b"]}, + ) + e = diff.edges + assert e["create"] == [] and e["remove"] == [] + assert [r["id"] for r in e["retain"]] == ["o3"] + assert e["unresolved"][0]["reason"] == "ambiguous" + assert e["unresolved"][0]["candidates"] == ["a", "b"] + assert diff.clean # unresolved/retained/external alone do not change the hub + + +def test_preview_resolves_targets_created_in_the_same_receipt(tmp_path): + derived = _derive(tmp_path, "") + workplans, tasks = _hub(derived) + wp, tk = _ids(derived) + # The hub lacks SRC-WP-0002 entirely; the receipt would create it. + workplans = [w for w in workplans if w["slug"] != "src-wp-0002"] + diff = fp.diff_against_hub(derived, workplans, tasks, hub_edges=[], edge_targets={}) + created = [(c["from"], c["to_workplan_id"]) for c in diff.edges["create"]] + assert created == [("SRC-WP-0001-T01", wp["SRC-WP-0002"])] + assert any(m["record_id"] == "SRC-WP-0002" for m in diff.missing) + + +def test_preview_flags_self_edge(tmp_path): + derived = _derive(tmp_path, "depends_on: [SRC-WP-0001]\n") + workplans, tasks = _hub(derived) + wp, _ = _ids(derived) + diff = fp.diff_against_hub( + derived, workplans, tasks, hub_edges=[], + edge_targets={"SRC-WP-0001": [wp["SRC-WP-0001"]], "SRC-WP-0002": [wp["SRC-WP-0002"]]}, + ) + assert ("SRC-WP-0001", "self") in {(u["from"], u["reason"]) for u in diff.edges["unresolved"]} + + +def test_classifier_never_collapses_duplicate_keys(): + key = ("wp", None, "tgt", None, "blocks") + rows = [(key, "owned", OWNER), (key, "api", None)] + # Declared: the owned row satisfies it; the API duplicate is not reported. + plan = fp.classify_source_edges({key: {}}, set(), rows, OWNER) + assert plan == {"create": [], "external": [], "remove": [], "retain": []} + # Undeclared: only the owned row is removed, whatever the row order. + for ordered in (rows, rows[::-1]): + plan = fp.classify_source_edges({}, set(), ordered, OWNER) + assert plan["remove"] == [(key, "owned")]