MAI/docs/superpowers/plans/2026-06-24-fast-handoff-stale-lock-implementation.md

Fast Handoff And Stale Lock Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Make the tracker abandon stale lock anchors quickly during sharp turns, switch to valid new YOLO/ByteTrack candidates in 1-2 frames when safe, and reset guidance smoothing so the blue point and yellow line immediately follow the new target region.

Architecture: Extract the stale-lock and fast-handoff decision logic into a small pure-Python helper module so it can be unit-tested without running the full video pipeline. Then integrate that logic into main.py, add event-driven guidance reset hooks in guidance.py, and extend decision_logger.py so before/after behavior can be measured from logs. The plan assumes the current workspace is not a git repository; commit commands are included for use once the code is executed inside a repo-backed workspace.

Tech Stack: Python 3.12, pytest, NumPy, OpenCV-based tracking pipeline, existing local CSV/JSON diagnostics.

---

File Structure

• Create: target_handoff.py
  • Responsibility: pure decision helpers for stale-lock detection, fast-handoff threshold selection, and lightweight switch event representation.
• Create: tests/test_target_handoff.py
  • Responsibility: unit tests for stale-lock break and fast-handoff policy.
• Modify: guidance.py
  • Responsibility: add explicit reset hook for smoothing state on stale-break and target switch.
• Create: tests/test_guidance_reset.py
  • Responsibility: unit tests that prove aim/command smoothing is reset correctly.
• Modify: main.py
  • Responsibility: compute stale-lock inputs, invoke handoff helpers, suppress stale prediction hold, propagate reset events to guidance, and emit diagnostics.
• Modify: decision_logger.py
  • Responsibility: add CSV fields and summary accounting for stale-lock/handoff/guidance-reset events.
• Create: tests/test_decision_logger_events.py
  • Responsibility: verify new CSV fields and event serialization.
• Modify: config.py
  • Responsibility: add narrowly scoped tuning knobs for stale-lock break and fast handoff.

Task 1: Add Pure Handoff Decision Module

Files:

• Create: target_handoff.py

• Test: tests/test_target_handoff.py

• Step 1: Write the failing tests

from pathlib import Path
import sys


PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from target_handoff import (
    HandoffDecision,
    compute_fast_handoff_hits,
    evaluate_stale_lock,
)


def test_evaluate_stale_lock_requires_fresh_candidate_and_weak_anchor():
    decision = evaluate_stale_lock(
        confirmed=True,
        have_fresh_candidate=True,
        target_has_fresh_update=False,
        candidate_dist_px=420.0,
        pred_diag_px=70.0,
        miss_streak=8,
        klt_valid=False,
        klt_quality=0.18,
        klt_iou=0.12,
        candidate_motion_ok=True,
        target_motion_ok=False,
        stale_break_dist_diag=3.5,
        stale_break_min_miss=2,
        stale_break_klt_quality=0.35,
        stale_break_klt_iou=0.25,
    )

    assert decision.stale_lock_active is True
    assert "target_not_fresh" in decision.reasons
    assert "weak_klt_quality" in decision.reasons


def test_evaluate_stale_lock_rejects_close_candidate_when_anchor_is_healthy():
    decision = evaluate_stale_lock(
        confirmed=True,
        have_fresh_candidate=True,
        target_has_fresh_update=True,
        candidate_dist_px=90.0,
        pred_diag_px=70.0,
        miss_streak=0,
        klt_valid=True,
        klt_quality=0.92,
        klt_iou=0.71,
        candidate_motion_ok=True,
        target_motion_ok=True,
        stale_break_dist_diag=3.5,
        stale_break_min_miss=2,
        stale_break_klt_quality=0.35,
        stale_break_klt_iou=0.25,
    )

    assert decision.stale_lock_active is False
    assert decision.reasons == []


def test_compute_fast_handoff_hits_reduces_confirmation_under_stale_lock():
    hits = compute_fast_handoff_hits(
        stale_lock_active=True,
        motion_switch_mode=False,
        approach_extra_hits=2,
        fast_maneuver_extra_hits=2,
        default_switch_hits=4,
        fast_handoff_hits=2,
        motion_switch_hits=6,
    )

    assert hits == 2


def test_handoff_decision_repr_is_stable_for_logging():
    decision = HandoffDecision(
        stale_lock_active=True,
        allow_fast_handoff=True,
        required_switch_hits=2,
        disable_prediction_hold=True,
        reasons=["target_not_fresh", "weak_klt_quality"],
    )

    assert decision.reason_text() == "target_not_fresh|weak_klt_quality"

• Step 2: Run test to verify it fails

Run: pytest tests/test_target_handoff.py -v

Expected: FAIL with ModuleNotFoundError: No module named 'target_handoff'

• Step 3: Write minimal implementation

from dataclasses import dataclass


@dataclass(frozen=True)
class HandoffDecision:
    stale_lock_active: bool
    allow_fast_handoff: bool
    required_switch_hits: int
    disable_prediction_hold: bool
    reasons: list[str]

    def reason_text(self) -> str:
        return "|".join(self.reasons)


def evaluate_stale_lock(
    *,
    confirmed,
    have_fresh_candidate,
    target_has_fresh_update,
    candidate_dist_px,
    pred_diag_px,
    miss_streak,
    klt_valid,
    klt_quality,
    klt_iou,
    candidate_motion_ok,
    target_motion_ok,
    stale_break_dist_diag,
    stale_break_min_miss,
    stale_break_klt_quality,
    stale_break_klt_iou,
):
    reasons = []
    if not confirmed or not have_fresh_candidate:
        return HandoffDecision(False, False, 0, False, [])

    far_enough = candidate_dist_px >= max(40.0, stale_break_dist_diag * max(pred_diag_px, 1.0))
    weak_anchor = (
        (not klt_valid)
        or (klt_quality < stale_break_klt_quality)
        or (klt_iou < stale_break_klt_iou)
        or (miss_streak >= stale_break_min_miss)
        or (candidate_motion_ok and not target_motion_ok)
        or (not target_has_fresh_update)
    )

    if not target_has_fresh_update:
        reasons.append("target_not_fresh")
    if not klt_valid:
        reasons.append("klt_invalid")
    if klt_quality < stale_break_klt_quality:
        reasons.append("weak_klt_quality")
    if klt_iou < stale_break_klt_iou:
        reasons.append("weak_klt_iou")
    if miss_streak >= stale_break_min_miss:
        reasons.append("miss_streak")
    if candidate_motion_ok and not target_motion_ok:
        reasons.append("motion_conflict")

    stale = bool(far_enough and weak_anchor)
    return HandoffDecision(
        stale_lock_active=stale,
        allow_fast_handoff=stale,
        required_switch_hits=0,
        disable_prediction_hold=stale,
        reasons=reasons if stale else [],
    )


def compute_fast_handoff_hits(
    *,
    stale_lock_active,
    motion_switch_mode,
    approach_extra_hits,
    fast_maneuver_extra_hits,
    default_switch_hits,
    fast_handoff_hits,
    motion_switch_hits,
):
    if stale_lock_active:
        return int(fast_handoff_hits)

    hits = int(default_switch_hits) + int(approach_extra_hits) + int(fast_maneuver_extra_hits)
    if motion_switch_mode:
        hits = max(hits, int(motion_switch_hits))
    return int(hits)

• Step 4: Run test to verify it passes

Run: pytest tests/test_target_handoff.py -v

Expected: 4 passed

• Step 5: Commit

git add target_handoff.py tests/test_target_handoff.py
git commit -m "feat: add stale-lock and fast-handoff decision helpers"

Task 2: Add Guidance Reset Hook

Files:

• Modify: guidance.py

• Create: tests/test_guidance_reset.py

• Step 1: Write the failing tests

from pathlib import Path
import sys

import numpy as np


PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from guidance import ScreenGuidanceController


def test_guidance_reset_reinitializes_smoothed_point():
    ctrl = ScreenGuidanceController()
    ctrl.smoothed_point = np.array([100.0, 200.0], dtype=np.float32)
    ctrl.smoothed_cmd = np.array([0.7, -0.4], dtype=np.float32)

    ctrl.reset_for_target_switch(np.array([500.0, 300.0], dtype=np.float32))

    assert np.allclose(ctrl.smoothed_point, np.array([500.0, 300.0], dtype=np.float32))
    assert np.allclose(ctrl.smoothed_cmd, np.zeros(2, dtype=np.float32))


def test_guidance_reset_can_damp_instead_of_zero_when_requested():
    ctrl = ScreenGuidanceController()
    ctrl.smoothed_cmd = np.array([0.8, -0.6], dtype=np.float32)

    ctrl.reset_for_target_switch(np.array([10.0, 20.0], dtype=np.float32), cmd_damp=0.25)

    assert np.allclose(ctrl.smoothed_cmd, np.array([0.2, -0.15], dtype=np.float32))

• Step 2: Run test to verify it fails

Run: pytest tests/test_guidance_reset.py -v

Expected: FAIL with AttributeError: 'ScreenGuidanceController' object has no attribute 'reset_for_target_switch'

• Step 3: Write minimal implementation

    def reset_for_target_switch(self, aim_point, cmd_damp=0.0):
        aim = np.array(aim_point, dtype=np.float32)
        self.smoothed_point = aim.copy()
        damp = float(clamp(cmd_damp, 0.0, 1.0))
        self.smoothed_cmd = (self.smoothed_cmd * damp).astype(np.float32)

Add it inside ScreenGuidanceController near status_line() / idle_state() in guidance.py.

• Step 4: Run test to verify it passes

Run: pytest tests/test_guidance_reset.py -v

Expected: 2 passed

• Step 5: Commit

git add guidance.py tests/test_guidance_reset.py
git commit -m "feat: add guidance reset hook for target handoff"

Task 3: Integrate Stale Lock And Fast Handoff Into Main Loop

Files:

• Modify: config.py

• Modify: main.py

• Modify: decision_logger.py

• Create: tests/test_decision_logger_events.py

• Step 1: Write the failing logger test

from pathlib import Path
import sys


PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from decision_logger import FIELDNAMES


def test_decision_logger_contains_handoff_fields():
    assert "stale_lock_active" in FIELDNAMES
    assert "stale_lock_reason" in FIELDNAMES
    assert "fast_handoff_active" in FIELDNAMES
    assert "guidance_reset_event" in FIELDNAMES

• Step 2: Run test to verify it fails

Run: pytest tests/test_decision_logger_events.py -v

Expected: FAIL because the new field names are missing from FIELDNAMES

• Step 3: Add config defaults

Add these exact settings near the existing single-target lock policy section in config.py:

STALE_LOCK_BREAK_ENABLE = True
STALE_LOCK_BREAK_DIST_DIAG = 3.5
STALE_LOCK_BREAK_MIN_MISS = 2
STALE_LOCK_BREAK_KLT_QUALITY = 0.35
STALE_LOCK_BREAK_KLT_IOU = 0.25
FAST_HANDOFF_ENABLE = True
FAST_HANDOFF_CONFIRM_HITS = 2
FAST_HANDOFF_GUIDANCE_CMD_DAMP = 0.0

• Step 4: Wire stale-lock and fast-handoff in main.py

Add imports near the top of main.py:

from target_handoff import compute_fast_handoff_hits, evaluate_stale_lock

Add per-frame state defaults near the other per-frame flags:

        stale_lock_active = False
        stale_lock_reason = ""
        fast_handoff_active = False
        guidance_reset_event = ""

After best candidate selection is known and before switch confirmation is finalized, add stale-lock evaluation using current values:

                handoff_decision = evaluate_stale_lock(
                    confirmed=confirmed,
                    have_fresh_candidate=(best is not None and best.time_since_update == 0),
                    target_has_fresh_update=(target_track is not None and target_track.time_since_update == 0),
                    candidate_dist_px=float(dist) if pred_ref is not None else 0.0,
                    pred_diag_px=float(pred_diag),
                    miss_streak=int(miss_streak),
                    klt_valid=bool(klt_valid),
                    klt_quality=float(klt.quality),
                    klt_iou=float(iou(klt.box, locked_box_eff)) if (klt.box is not None and locked_box_eff is not None) else 0.0,
                    candidate_motion_ok=bool(residual_ok),
                    target_motion_ok=bool(target_track is not None and target_track.time_since_update == 0),
                    stale_break_dist_diag=float(STALE_LOCK_BREAK_DIST_DIAG),
                    stale_break_min_miss=int(STALE_LOCK_BREAK_MIN_MISS),
                    stale_break_klt_quality=float(STALE_LOCK_BREAK_KLT_QUALITY),
                    stale_break_klt_iou=float(STALE_LOCK_BREAK_KLT_IOU),
                )
                stale_lock_active = bool(STALE_LOCK_BREAK_ENABLE and handoff_decision.stale_lock_active)
                stale_lock_reason = handoff_decision.reason_text()

When computing switch confirmation hits, replace the current manual calculation with:

                    req_switch_hits = compute_fast_handoff_hits(
                        stale_lock_active=bool(FAST_HANDOFF_ENABLE and stale_lock_active),
                        motion_switch_mode=bool(motion_switch_mode),
                        approach_extra_hits=int(approach_switch_extra_hits),
                        fast_maneuver_extra_hits=int(max(0, SWITCH_FAST_MANEUVER_EXTRA_HITS)) if fast_maneuver_guard else 0,
                        default_switch_hits=int(TARGET_SWITCH_CONFIRM_HITS),
                        fast_handoff_hits=int(FAST_HANDOFF_CONFIRM_HITS),
                        motion_switch_hits=int(MOTION_CONF_SWITCH_CONFIRM_HITS),
                    )
                    fast_handoff_active = bool(FAST_HANDOFF_ENABLE and stale_lock_active and req_switch_hits == int(FAST_HANDOFF_CONFIRM_HITS))

Disable prediction hold while stale-lock is active by modifying the existing hold branch:

            if confirmed and pred_box_eff is not None and (not stale_lock_active) and miss_streak < KALMAN_HOLD_MAX:

When a fast handoff or target switch is accepted, reset guidance:

                old_target_id = target_id

before replacing target_id, then after locked_box_eff = chosen:

            if confirmed and target_track is not None and old_target_id is not None and target_track.track_id != old_target_id:
                guidance_ctrl.reset_for_target_switch(
                    box_center(locked_box_eff),
                    cmd_damp=float(FAST_HANDOFF_GUIDANCE_CMD_DAMP),
                )
                guidance_reset_event = "target_switch"
            elif stale_lock_active and locked_box_eff is not None:
                guidance_ctrl.reset_for_target_switch(
                    box_center(locked_box_eff),
                    cmd_damp=float(FAST_HANDOFF_GUIDANCE_CMD_DAMP),
                )
                guidance_reset_event = "stale_lock_break"

• Step 5: Add logger fields

Extend decision_logger.py field names and row payload with:

        "stale_lock_active",
        "stale_lock_reason",
        "fast_handoff_active",
        "guidance_reset_event",

and map them in log_frame():

            "stale_lock_active": int(bool(kwargs.get("stale_lock_active", False))),
            "stale_lock_reason": kwargs.get("stale_lock_reason", ""),
            "fast_handoff_active": int(bool(kwargs.get("fast_handoff_active", False))),
            "guidance_reset_event": kwargs.get("guidance_reset_event", ""),

Also append event names if present:

        if kwargs.get("stale_lock_active", False):
            events.append("stale_lock_break")
        if kwargs.get("fast_handoff_active", False):
            events.append("fast_handoff_switch")
        if kwargs.get("guidance_reset_event", ""):
            events.append("guidance_reset")

• Step 6: Run focused tests

Run:

pytest tests/test_target_handoff.py tests/test_guidance_reset.py tests/test_decision_logger_events.py tests/test_track_score_gates.py -v

Expected: all tests PASS

• Step 7: Run a smoke video pass

Run:

py -3 main.py

Expected:

• no import/runtime errors

• new CSV fields appear in track_log_*.csv

• problem segment near 231.5s - 232.3s shows earlier stale-lock break and faster visual realignment

• Step 8: Commit

git add config.py main.py guidance.py decision_logger.py tests/test_decision_logger_events.py
git commit -m "feat: add stale-lock break and fast handoff integration"

Self-Review

Spec coverage

• Stale-lock detection: Task 1 + Task 3
• Fast handoff switch policy: Task 1 + Task 3
• Prediction hold restrictions: Task 3
• Guidance reset on switch: Task 2 + Task 3
• Logging and diagnostics: Task 3
• Verification on known failure segment: Task 3 smoke pass

No spec gaps remain for the first-pass implementation.

Placeholder scan

• No TBD, TODO, or deferred placeholders included.
• Every code-changing step contains explicit code snippets.
• Every verification step contains an exact command and expected result.

Type consistency

• HandoffDecision property names are consistent across tests and integration steps.
• reset_for_target_switch() is referenced consistently in tests and integration.
• New logger field names match the values referenced in main.py.

Execution Handoff

Plan complete and saved to docs/superpowers/plans/2026-06-24-fast-handoff-stale-lock-implementation.md. Two execution options:

1. Subagent-Driven (recommended) - I dispatch a fresh subagent per task, review between tasks, fast iteration

2. Inline Execution - Execute tasks in this session using executing-plans, batch execution with checkpoints

Which approach?