Function claim_handoff_delivery

plugins/codex-security/scripts/workbench_db.py:1902–1936

(
    connection: sqlite3.Connection, args: argparse.Namespace
)

def claim_handoff_delivery(
    connection: sqlite3.Connection, args: argparse.Namespace
) -> dict[str, Any]:
    scan_id = require_uuid(args.scan_id, "scan-id")
    claim_token = require_handoff_claim_token(args.claim_token)
    timestamp = now()
    with connection:
        scan = require_scan(connection, scan_id)
        if scan["handoff_status"] != "pending" or scan["handoff_claim_token"] == claim_token:
            return workspace_state(connection, scan["workspace_id"])
        updated = connection.execute(
            """
            UPDATE scans
            SET handoff_claimed_at = ?, handoff_claim_token = ?, updated_at = ?
            WHERE id = ? AND handoff_status = 'pending'
                AND (
                    handoff_claim_token IS NULL
                    OR (
                        ? = 1
                        AND (handoff_claimed_at IS NULL OR handoff_claimed_at <= ?)
                    )
                )
            """,
            (
                timestamp,
                claim_token,
                timestamp,
                scan["id"],
                int(args.take_over_stale),
                stale_claim_before(),
            ),
        )
        if updated.rowcount != 1:
            return workspace_state(connection, scan["workspace_id"])
    return workspace_state(connection, scan["workspace_id"])


def release_handoff_delivery(