# 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**

```python
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**

```python
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**

```bash
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**

```python
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**

```python
    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**

```bash
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**

```python
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`:

```python
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`:

```python
from target_handoff import compute_fast_handoff_hits, evaluate_stale_lock
```

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

```python
        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:

```python
                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:

```python
                    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:

```python
            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:

```python
                old_target_id = target_id
```

before replacing `target_id`, then after `locked_box_eff = chosen`:

```python
            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:

```python
        "stale_lock_active",
        "stale_lock_reason",
        "fast_handoff_active",
        "guidance_reset_event",
```

and map them in `log_frame()`:

```python
            "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:

```python
        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:

```bash
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:

```bash
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**

```bash
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?**