R2 hold-model correctness fixes (PR #56 review: #2,3,4,5,8,11)

src/opal/api/routes/execution.py

@@ -30,8 +30,8 @@
     clear_focus,
     complete_step_flow,
     focus_step,
-    held_scope_blockers,
     mark_instance_in_work,
+    skip_blockers,
     touch_user_presence,
 )
 from opal.core.genealogy import record_assembly_genealogy
@@ -820,14 +820,12 @@ async def skip_step(
     if step_status in (StepStatus.COMPLETED.value, StepStatus.SIGNED_OFF.value):
         raise HTTPException(status_code=400, detail="Cannot skip completed step")
 
-    # Containment gate: SKIP is a terminal commitment, same scope check as
-    # COMPLETE — skipping held work would sweep the hold.
-    skip_blockers = get_hold_state(db, instance_id).blockers_for_complete(step_exec)
-    if skip_blockers:
-        raise HTTPException(status_code=400, detail=_hold_blocker_detail("skip", skip_blockers))
-
-    # Skip is a commitment moment too — held work cannot be skipped around.
-    holds = held_scope_blockers(db, instance, step_exec)
+    # SKIP is a terminal commitment: it inherits the held scope (a hold cannot
+    # be skipped around) and any open redline-rework op on the gate — skipping
+    # the host step must not strand authorized rework (F5). It does not inherit
+    # strict_sequence/dependency ordering: skipping legitimately need not wait
+    # on predecessors.
+    holds = skip_blockers(db, instance, step_exec)
     if holds:
         raise HTTPException(
             status_code=400,

src/opal/api/routes/issues.py

@@ -16,6 +16,7 @@
 from opal.core.designators import generate_issue_number
 from opal.core.events import emit_issue_dispositioned
 from opal.core.holds import holding_readout
+from opal.db.models.execution import StepExecution
 from opal.db.models.issue import (
     CONTAINMENT_RANK,
     Containment,
@@ -36,6 +37,39 @@ def _get_enum_val(obj: object, attr: str) -> str:
     return val.value if hasattr(val, "value") else val
 
 
+def _resolve_containment_instance(
+    db,
+    containment: Containment,
+    containment_step_id: int | None,
+    raised_step_id: int | None,
+    procedure_instance_id: int | None,
+) -> int | None:
+    """Reconcile a non-advisory issue's work order against its step anchors.
+
+    Derives procedure_instance_id from a step anchor when unset, and rejects a
+    procedure_instance_id (or a second anchor) that names a different work order
+    (F3). Advisory issues pass through untouched — they hold nothing.
+    """
+    if containment == Containment.ADVISORY:
+        return procedure_instance_id
+    resolved = procedure_instance_id
+    for step_id in (containment_step_id, raised_step_id):
+        if step_id is None:
+            continue
+        step = db.query(StepExecution).filter(StepExecution.id == step_id).first()
+        if step is None:
+            raise HTTPException(status_code=400, detail=f"Step {step_id} not found")
+        if resolved is None:
+            resolved = step.instance_id
+        elif step.instance_id != resolved:
+            raise HTTPException(
+                status_code=400,
+                detail=f"Step {step_id} belongs to a different work order "
+                "than procedure_instance_id",
+            )
+    return resolved
+
+
 # ============ Schemas ============
 
 
@@ -342,6 +376,15 @@ def create_issue(
             status_code=400, detail=f"Invalid containment: {data.containment}"
         ) from err
 
+    # A non-advisory containment holds a specific work order via the anchor's
+    # procedure_instance_id (core/holds filters on it). Derive it from any step
+    # anchor and reject anchors that name a different work order — otherwise the
+    # hold reports success but blocks nothing, or blocks the wrong WO (F3). A
+    # bare draft (no anchor, no WO) is still allowed; the WO is linked later.
+    procedure_instance_id = _resolve_containment_instance(
+        db, containment, data.containment_step_id, data.raised_step_id, data.procedure_instance_id
+    )
+
     issue = Issue(
         issue_number=generate_issue_number(db),
         title=data.title,
@@ -359,7 +402,7 @@ def create_issue(
         expected_benefit=data.expected_benefit,
         part_id=data.part_id,
         procedure_id=data.procedure_id,
-        procedure_instance_id=data.procedure_instance_id,
+        procedure_instance_id=procedure_instance_id,
         raised_step_id=data.raised_step_id,
         raised_by_id=user_id,
         assigned_to_id=data.assigned_to_id,
@@ -614,8 +657,25 @@ def set_containment(
 
     old_values = get_model_dict(issue)
     issue.containment = new_containment
-    if data.containment_step_id is not None:
-        issue.containment_step_id = data.containment_step_id
+    # Boundary re-binding uses explicit presence, not None-skip: an omitted
+    # containment_step_id leaves the boundary untouched, while an explicit null
+    # clears it (a mis-bound boundary was previously immutable — F8). A non-null
+    # boundary must belong to this issue's work order; it seeds the WO on a
+    # still-unlinked issue and is rejected when it names a different one.
+    if "containment_step_id" in data.model_fields_set:
+        new_step_id = data.containment_step_id
+        if new_step_id is not None:
+            step = db.query(StepExecution).filter(StepExecution.id == new_step_id).first()
+            if step is None:
+                raise HTTPException(status_code=404, detail=f"Step {new_step_id} not found")
+            if issue.procedure_instance_id is None:
+                issue.procedure_instance_id = step.instance_id
+            elif step.instance_id != issue.procedure_instance_id:
+                raise HTTPException(
+                    status_code=400,
+                    detail=f"Step {new_step_id} belongs to a different work order than this issue",
+                )
+        issue.containment_step_id = new_step_id
     log_update(db, issue, old_values, user_id)
 
     if narrowing and (data.note or "").strip():

src/opal/core/execution_flow.py

@@ -26,7 +26,8 @@
 from sqlalchemy.dialects.sqlite import insert as sqlite_insert
 from sqlalchemy.orm import Session
 
-from opal.core.holds import get_hold_state
+from opal.core.holds import _val as _status_value  # enum-unwrap, one home in core/holds
+from opal.core.holds import blocking_issues_for_instance, get_hold_state
 from opal.db.models.attachment import Attachment
 from opal.db.models.execution import (
     InstanceStatus,
@@ -61,10 +62,6 @@ def __init__(self, message: str):
         self.message = message
 
 
-def _status_value(obj: Any) -> str:
-    return obj.value if hasattr(obj, "value") else obj
-
-
 def user_initials(name: str | None) -> str:
     if not name:
         return "?"
@@ -244,11 +241,26 @@ def complete_blockers(
     step_exec: StepExecution,
 ) -> list[Blocker]:
     """Everything holding this step's (or OP's) COMPLETE: the held scope
-    plus the structural sequence gates. SKIP checks only the held scope —
-    sequencing gates COMPLETE alone."""
-    return held_scope_blockers(db, instance, step_exec) + sequence_blockers(
-        db, instance, step_exec
-    )
+    plus the structural sequence gates."""
+    return held_scope_blockers(db, instance, step_exec) + sequence_blockers(db, instance, step_exec)
+
+
+def skip_blockers(
+    db: Session,
+    instance: ProcedureInstance,
+    step_exec: StepExecution,
+) -> list[Blocker]:
+    """Everything holding this step's SKIP: the held scope plus any open
+    redline-rework op on the gate.
+
+    SKIP is a terminal commitment, so it inherits the held scope (a hold
+    cannot be skipped around) and the redline gate — skipping the host step
+    must not strand an authorized rework op (F5). It deliberately omits
+    strict_sequence and OP-dependency ordering: skipping legitimately does
+    not require predecessors to be done."""
+    scope = held_scope_blockers(db, instance, step_exec)
+    redlines = [b for b in sequence_blockers(db, instance, step_exec) if b.kind == "redline"]
+    return scope + redlines
 
 
 # ============ Presence (focus) ============
@@ -464,7 +476,6 @@ def complete_step_flow(
     old_instance_status = _status_value(instance.status)
     check_instance_completion(db, instance)
 
-
     instance_completed = (
         old_instance_status != InstanceStatus.COMPLETED.value
         and _status_value(instance.status) == InstanceStatus.COMPLETED.value
@@ -490,6 +501,12 @@ def check_instance_completion(db: Session, instance: ProcedureInstance) -> None:
     all_steps = db.query(StepExecution).filter(StepExecution.instance_id == instance.id).all()
     hold_state = get_hold_state(db, instance.id)
 
+    def _row_held(se: StepExecution) -> bool:
+        # A bound "resolve by" hold (start_blocked) gates a row's COMPLETE just
+        # as a raised/op hold does — fold both so an OP never auto-completes,
+        # and the WO never closes, over an open hold anchored on a terminal row.
+        return bool(hold_state.blockers_for_complete(se) or hold_state.blockers_for_start(se))
+
     for step_exec in all_steps:
         if step_exec.level == 0:
             children = [s for s in all_steps if s.parent_step_order == step_exec.step_number]
@@ -501,9 +518,7 @@ def check_instance_completion(db: Session, instance: ProcedureInstance) -> None:
                         in (StepStatus.COMPLETED.value, StepStatus.SKIPPED.value)
                         for c in children
                     )
-                    scope_held = bool(hold_state.blockers_for_complete(step_exec)) or any(
-                        hold_state.blockers_for_complete(c) for c in children
-                    )
+                    scope_held = _row_held(step_exec) or any(_row_held(c) for c in children)
                     if all_children_done and not scope_held:
                         step_exec.status = StepStatus.COMPLETED
                         step_exec.completed_at = datetime.now(UTC)
@@ -534,9 +549,14 @@ def _to_aware(dt: datetime) -> datetime:
         ):
             return
 
-    # WO containment holds close-out: every step may be terminal while an
-    # undispositioned wo-scoped issue keeps the work order open.
-    if hold_state.wo_blocked:
+    # Close-out is authoritative on the issue set, not on per-row hold buckets:
+    # ANY non-advisory, containment-bearing, undispositioned issue open on this
+    # work order keeps it open — a wo-scoped hold, an op/step hold whose anchor
+    # already went terminal, or a "resolve by" boundary bound to a terminal or
+    # OP row. Inferring close-out from wo_blocked alone let those slip through
+    # (a WO closing COMPLETED over an open NC). Dispositioned/advisory issues
+    # are excluded here, so legitimate completion is never deadlocked.
+    if blocking_issues_for_instance(db, instance.id):
         return
 
     instance.status = InstanceStatus.COMPLETED

src/opal/core/holds.py

@@ -170,8 +170,8 @@ def holding_readout(db: Session, issue: Issue) -> list[dict]:
     wo_label = instance.work_order_number or f"WO #{instance.id}"
 
     def exec_href(op_order: int | None = None) -> str:
-        base = f"/executions/{instance.id}?tab=operations"
-        return f"{base}&op={op_order}" if op_order is not None else base
+        base = f"/executions/{instance.id}"
+        return f"{base}?op={op_order}" if op_order is not None else base
 
     anchor = None
     anchor_id = issue.containment_step_id or issue.raised_step_id

src/opal/mcp/server.py

@@ -2908,6 +2908,29 @@ async def _raise_issue(db, args: dict) -> list[TextContent]:
     priority = args.get("priority", "medium")
     containment = args.get("containment", "advisory")
 
+    # A non-advisory containment holds a specific work order (core/holds filters
+    # on procedure_instance_id). Derive it from any step anchor, reject anchors
+    # naming a different WO — otherwise the hold blocks nothing or the wrong WO
+    # (F3). A bare draft (no anchor, no WO) is still allowed.
+    procedure_instance_id = args.get("procedure_instance_id")
+    if containment != Containment.ADVISORY.value:
+        for step_id in (args.get("containment_step_id"), args.get("raised_step_id")):
+            if step_id is None:
+                continue
+            step = db.query(StepExecution).filter(StepExecution.id == step_id).first()
+            if step is None:
+                return json_response({"success": False, "error": f"Step {step_id} not found"})
+            if procedure_instance_id is None:
+                procedure_instance_id = step.instance_id
+            elif step.instance_id != procedure_instance_id:
+                return json_response(
+                    {
+                        "success": False,
+                        "error": f"Step {step_id} belongs to a different work order "
+                        "than procedure_instance_id",
+                    }
+                )
+
     issue = Issue(
         issue_number=generate_issue_number(db),
         title=args["title"],
@@ -2919,7 +2942,7 @@ async def _raise_issue(db, args: dict) -> list[TextContent]:
         containment_step_id=args.get("containment_step_id"),
         should_be=args.get("should_be"),
         actual=args.get("actual"),
-        procedure_instance_id=args.get("procedure_instance_id"),
+        procedure_instance_id=procedure_instance_id,
         raised_step_id=args.get("raised_step_id"),
         raised_by_id=args.get("user_id"),
         part_id=args.get("part_id"),
@@ -3023,8 +3046,26 @@ async def _set_containment(db, args: dict) -> list[TextContent]:
     user_id = args.get("user_id")
     old_values = get_model_dict(issue)
     issue.containment = new_containment
-    if args.get("containment_step_id") is not None:
-        issue.containment_step_id = args["containment_step_id"]
+    # Explicit presence, not None-skip: an omitted containment_step_id leaves
+    # the boundary untouched; an explicit null clears it (F8). A non-null
+    # boundary must belong to this issue's work order.
+    if "containment_step_id" in args:
+        new_step_id = args["containment_step_id"]
+        if new_step_id is not None:
+            step = db.query(StepExecution).filter(StepExecution.id == new_step_id).first()
+            if step is None:
+                return json_response({"success": False, "error": f"Step {new_step_id} not found"})
+            if issue.procedure_instance_id is None:
+                issue.procedure_instance_id = step.instance_id
+            elif step.instance_id != issue.procedure_instance_id:
+                return json_response(
+                    {
+                        "success": False,
+                        "error": f"Step {new_step_id} belongs to a different work order "
+                        "than this issue",
+                    }
+                )
+        issue.containment_step_id = new_step_id
     log_update(db, issue, old_values, user_id)
 
     if narrowing and note:
@@ -6280,6 +6321,21 @@ async def _bind_issue_hold(db, args: dict) -> list[TextContent]:
     if not issue:
         return json_response({"error": f"Issue {args['issue_id']} not found"})
 
+    # A hold that names one work order cannot be bound into another's step — the
+    # containment filter is per-instance, so a cross-WO bind would report
+    # success and block nothing (F3). Adopt the target WO when the issue names
+    # none yet; reject when it names a different one.
+    if issue.procedure_instance_id is None:
+        issue.procedure_instance_id = instance.id
+        db.flush()
+    elif issue.procedure_instance_id != instance.id:
+        return json_response(
+            {
+                "error": f"{issue.issue_number} belongs to work order "
+                f"{issue.procedure_instance_id}, not this execution ({instance.id})"
+            }
+        )
+
     containment = issue.containment.value if hasattr(issue.containment, "value") else issue.containment
     if containment == Containment.STEP.value and issue.containment_step_id == step_exec.id:
         label = step_exec.step_number_str or str(step_exec.step_number)