coulomb/state-hub
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Preview source-owned dependency edges in the projection dry run (STATE-WP-0094-T05)

Browse source 
diff_against_hub accepts hub_edges/edge_targets and reports edges to create,
remove, retain, unresolved and external, using the same pure planner that
receiving now calls (plan_source_edges + classify_source_edges). The classifier
works on a row list, fixing a collapse where an external duplicate of an owned
edge key could hide the owned row from removal. 7 new pure tests.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 692292@bnt-lap001
Assistant-Session: 89c7d8e8-e1b3-47eb-9411-a0ab227fca2f
This commit is contained in:
tegwick 2026-10-04 13:19:23 +02:00
parent 216876c0cb
commit c287ed8259
3 changed files with 407 additions and 90 deletions
Download patch file Download diff file Expand all files Collapse all files

321
api/services/forge_projection.py
View file

@ -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()

19
docs/forge-dependency-edge-receiving.md
View file

@ -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.

157
tests/test_forge_dependency_edge_preview.py Normal file
View file

@ -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")]