Sessions Audio Presigned PUT Migration — Implementation Plan

MAX_AUDIO_BYTES = 10 * 1024 * 1024  # 10 MB
URL_EXPIRES_SECONDS = 300            # 5 min
S3_KEY_REGEX = r"^sessions/(?P<session_id>[^/]+)/window_(?P<window_index>\d{3})/audio\.wav$"

# main/server/tests/unit/test_audio_url_minter.py
"""Unit tests for the rewritten SessionAudioFunction (URL minter).

Each test cites a row in docs/plans/2026-05-19-sessions-audio-presigned.md
either from Happy Path step 2 or from the Failure Modes table.
"""

from __future__ import annotations

import json
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

from ..worker_utils import load_worker_module

_APP_PATH = (
    Path(__file__).resolve().parents[2] / "api" / "sessions" / "audio" / "app.py"
)


def _load_module():
    return load_worker_module("sessions_audio_app", _APP_PATH)


def _event(session_id: str = "sess-1", window_index: int | str | None = 7, body: dict | None = None) -> dict:
    event: dict = {
        "pathParameters": {"sessionId": session_id} if session_id else {},
    }
    if window_index is not None:
        event["queryStringParameters"] = {"windowIndex": str(window_index)}
    else:
        event["queryStringParameters"] = None
    event["body"] = json.dumps(body) if body is not None else json.dumps({"sizeBytes": 1_923_456})
    return event


def _ddb_session_exists(session_id: str = "sess-1") -> MagicMock:
    """Mock DynamoDB table whose get_item returns a session row."""
    mock_table = MagicMock()
    mock_table.get_item.return_value = {"Item": {"sessionId": session_id}}
    return mock_table


class TestURLMinterHappyPath:
    @staticmethod
    def test_returns_presigned_url_with_correct_s3_key_for_valid_request():
        module = _load_module()
        mock_s3 = MagicMock()
        mock_s3.generate_presigned_url.return_value = "https://s3.example/presigned"
        with (
            patch.object(module, "s3", mock_s3),
            patch.object(module, "table", _ddb_session_exists("sess-1")),
        ):
            result = module.lambda_handler(_event("sess-1", 7, {"sizeBytes": 1_923_456}), None)

        assert result["statusCode"] == 200
        body = json.loads(result["body"])
        assert body["s3Key"] == "sessions/sess-1/window_007/audio.wav"
        assert body["windowIndex"] == 7
        assert body["url"] == "https://s3.example/presigned"
        assert body["expiresIn"] == 300

    @staticmethod
    def test_url_binds_content_type_audio_wav():
        module = _load_module()
        mock_s3 = MagicMock()
        mock_s3.generate_presigned_url.return_value = "https://s3.example/presigned"
        with (
            patch.object(module, "s3", mock_s3),
            patch.object(module, "table", _ddb_session_exists("sess-1")),
        ):
            module.lambda_handler(_event("sess-1", 7, {"sizeBytes": 1000}), None)

        kwargs = mock_s3.generate_presigned_url.call_args.kwargs
        assert kwargs["Params"]["ContentType"] == "audio/wav"

    @staticmethod
    def test_url_binds_exact_content_length_from_size_bytes():
        module = _load_module()
        mock_s3 = MagicMock()
        mock_s3.generate_presigned_url.return_value = "u"
        with (
            patch.object(module, "s3", mock_s3),
            patch.object(module, "table", _ddb_session_exists("sess-1")),
        ):
            module.lambda_handler(_event("sess-1", 7, {"sizeBytes": 1_923_456}), None)

        kwargs = mock_s3.generate_presigned_url.call_args.kwargs
        assert kwargs["Params"]["ContentLength"] == 1_923_456

    @staticmethod
    def test_url_expires_in_300_seconds():
        module = _load_module()
        mock_s3 = MagicMock()
        mock_s3.generate_presigned_url.return_value = "u"
        with (
            patch.object(module, "s3", mock_s3),
            patch.object(module, "table", _ddb_session_exists("sess-1")),
        ):
            module.lambda_handler(_event(), None)

        kwargs = mock_s3.generate_presigned_url.call_args.kwargs
        assert kwargs["ExpiresIn"] == 300

# main/server/api/sessions/audio/app.py
"""Audio Upload — issues a presigned PUT URL for per-window WAV chunks.

This handler does not accept the audio body. The client uses the returned
URL to PUT bytes directly to S3, bypassing the 29s API Gateway cap.
Completion is handled by AudioUploadCompleteFunction via S3 ObjectCreated.
"""

from __future__ import annotations

import json
import os

import boto3
from shared.logger import create_logger

BUCKET = os.environ.get("BUCKET_NAME", "encache-raw-memory")
SESSIONS_TABLE = os.environ.get("SESSIONS_TABLE_NAME", "encache-sessions")
URL_EXPIRES_SECONDS = 300
MAX_AUDIO_BYTES = 10 * 1024 * 1024  # 10 MB

s3 = boto3.client("s3")
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(SESSIONS_TABLE)

logger = create_logger({"flow": "audio_url_grant"})


def _bad_request(message: str) -> dict:
    return {
        "statusCode": 400,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps({"error": message}),
    }


def _not_found(message: str) -> dict:
    return {
        "statusCode": 404,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps({"error": message}),
    }


def lambda_handler(event, _context):
    session_id = (event.get("pathParameters") or {}).get("sessionId", "")
    if not session_id:
        return _bad_request("sessionId required")

    query_params = event.get("queryStringParameters") or {}
    window_index_str = query_params.get("windowIndex")
    if window_index_str is None:
        return _bad_request("windowIndex required")
    try:
        window_index = int(window_index_str)
    except (ValueError, TypeError):
        return _bad_request("windowIndex must be an integer")
    if window_index < 0:
        return _bad_request("windowIndex must be non-negative")

    raw_body = event.get("body") or ""
    try:
        body = json.loads(raw_body) if raw_body else {}
    except json.JSONDecodeError:
        return _bad_request("body must be valid JSON")
    size_bytes = body.get("sizeBytes")
    if not isinstance(size_bytes, int) or size_bytes <= 0:
        return _bad_request("sizeBytes must be a positive integer")
    if size_bytes > MAX_AUDIO_BYTES:
        return _bad_request(f"sizeBytes exceeds cap of {MAX_AUDIO_BYTES}")

    resp = table.get_item(Key={"sessionId": session_id}, ProjectionExpression="sessionId")
    if "Item" not in resp:
        return _not_found("session not found")

    s3_key = f"sessions/{session_id}/window_{window_index:03d}/audio.wav"
    url = s3.generate_presigned_url(
        "put_object",
        Params={
            "Bucket": BUCKET,
            "Key": s3_key,
            "ContentType": "audio/wav",
            "ContentLength": size_bytes,
        },
        ExpiresIn=URL_EXPIRES_SECONDS,
    )

    logger(
        {
            "step": "audio_url_granted",
            "additional": {
                "session_id": session_id,
                "window_index": window_index,
                "size_bytes": size_bytes,
                "s3_key": s3_key,
            },
        }
    )

    return {
        "statusCode": 200,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps(
            {
                "url": url,
                "s3Key": s3_key,
                "windowIndex": window_index,
                "expiresIn": URL_EXPIRES_SECONDS,
            }
        ),
    }

git add main/server/api/sessions/audio/app.py main/server/tests/unit/test_audio_url_minter.py
git commit -m "feat(audio): rewrite SessionAudioFunction as presigned URL minter

Returns {url, s3Key, windowIndex, expiresIn} with strict
ContentType=audio/wav and exact ContentLength binding. Body is no
longer accepted through API Gateway. Completion handled separately by
AudioUploadCompleteFunction via S3 ObjectCreated event.

Refs: docs/plans/2026-05-19-sessions-audio-presigned.md"

class TestURLMinterInputValidation:
    @staticmethod
    def test_returns_400_when_session_id_path_param_missing():
        module = _load_module()
        with patch.object(module, "table", MagicMock()):
            event = {"pathParameters": {}, "queryStringParameters": {"windowIndex": "0"}, "body": json.dumps({"sizeBytes": 1})}
            result = module.lambda_handler(event, None)
        assert result["statusCode"] == 400

    @staticmethod
    def test_returns_400_when_window_index_missing():
        module = _load_module()
        with patch.object(module, "table", _ddb_session_exists()):
            event = _event(window_index=None, body={"sizeBytes": 1})
            result = module.lambda_handler(event, None)
        assert result["statusCode"] == 400

    @staticmethod
    def test_returns_400_when_window_index_negative():
        module = _load_module()
        with patch.object(module, "table", _ddb_session_exists()):
            result = module.lambda_handler(_event(window_index=-1, body={"sizeBytes": 1}), None)
        assert result["statusCode"] == 400

    @staticmethod
    def test_returns_400_when_window_index_non_integer():
        module = _load_module()
        with patch.object(module, "table", _ddb_session_exists()):
            result = module.lambda_handler(_event(window_index="abc", body={"sizeBytes": 1}), None)
        assert result["statusCode"] == 400

    @staticmethod
    def test_returns_400_when_size_bytes_missing():
        module = _load_module()
        with patch.object(module, "table", _ddb_session_exists()):
            result = module.lambda_handler(_event(body={}), None)
        assert result["statusCode"] == 400

    @staticmethod
    def test_returns_400_when_size_bytes_zero_or_negative():
        module = _load_module()
        with patch.object(module, "table", _ddb_session_exists()):
            assert module.lambda_handler(_event(body={"sizeBytes": 0}), None)["statusCode"] == 400
            assert module.lambda_handler(_event(body={"sizeBytes": -1}), None)["statusCode"] == 400

    @staticmethod
    def test_returns_400_when_size_bytes_exceeds_10mb_cap():
        module = _load_module()
        with patch.object(module, "table", _ddb_session_exists()):
            result = module.lambda_handler(_event(body={"sizeBytes": 10 * 1024 * 1024 + 1}), None)
        assert result["statusCode"] == 400

git add main/server/tests/unit/test_audio_url_minter.py
git commit -m "test(audio): cover URL minter input validation 400 cases"

class TestURLMinterSessionLookup:
    @staticmethod
    def test_returns_404_when_session_id_not_in_dynamodb():
        module = _load_module()
        mock_table = MagicMock()
        mock_table.get_item.return_value = {}  # no Item key = not found
        with patch.object(module, "table", mock_table):
            result = module.lambda_handler(_event("unknown-sess", 0, {"sizeBytes": 1}), None)
        assert result["statusCode"] == 404

git add main/server/tests/unit/test_audio_url_minter.py
git commit -m "test(audio): cover URL minter 404 on unknown session"

mkdir -p main/server/events/audio_upload_complete
touch main/server/events/__init__.py
touch main/server/events/audio_upload_complete/__init__.py

# main/server/tests/unit/test_audio_upload_complete.py
"""Unit tests for AudioUploadCompleteFunction (S3 event handler).

Each test cites a Failure Modes table row or Happy Path step in
docs/plans/2026-05-19-sessions-audio-presigned.md.
"""

from __future__ import annotations

from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

from ..worker_utils import load_worker_module

_APP_PATH = (
    Path(__file__).resolve().parents[2]
    / "events"
    / "audio_upload_complete"
    / "app.py"
)


def _load_module():
    return load_worker_module("audio_upload_complete_app", _APP_PATH)


def _s3_event(key: str = "sessions/sess-1/window_007/audio.wav") -> dict:
    return {
        "Records": [
            {
                "s3": {
                    "bucket": {"name": "encache-raw-memory"},
                    "object": {"key": key},
                }
            }
        ]
    }


class TestS3KeyParser:
    @staticmethod
    def test_parses_valid_key():
        module = _load_module()
        result = module._parse_key("sessions/sess-1/window_007/audio.wav")
        assert result == ("sess-1", 7)

    @staticmethod
    def test_returns_none_for_malformed_key():
        module = _load_module()
        for bad in [
            "wrong/prefix/window_000/audio.wav",
            "sessions/sess-1/audio.wav",
            "sessions/sess-1/window_xyz/audio.wav",
            "sessions/sess-1/window_007/audio.mp3",
            "",
        ]:
            assert module._parse_key(bad) is None

# main/server/events/audio_upload_complete/app.py
"""Audio Upload Complete — S3 ObjectCreated handler.

Triggered by S3 events on sessions/*/window_*/audio.wav. Updates
DynamoDB completedAudioWindows and conditionally triggers ingest.
"""

from __future__ import annotations

import json
import os
import re

import boto3
from shared.logger import create_logger

SESSIONS_TABLE = os.environ.get("SESSIONS_TABLE_NAME", "encache-sessions")
INGEST_FUNCTION = os.environ.get("INGEST_FUNCTION_NAME", "")
DEFAULT_FRAME_COUNT_PER_WINDOW = 60

_KEY_RE = re.compile(
    r"^sessions/(?P<session_id>[^/]+)/window_(?P<window_index>\d{3})/audio\.wav$"
)

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(SESSIONS_TABLE)
lambda_client = boto3.client("lambda")

logger = create_logger({"flow": "audio_upload_complete"})


def _parse_key(key: str) -> tuple[str, int] | None:
    match = _KEY_RE.match(key)
    if not match:
        return None
    return match.group("session_id"), int(match.group("window_index"))


def lambda_handler(event, _context):
    # Implementation grows across subsequent tasks.
    return {"ok": True}

git add main/server/events/ main/server/tests/unit/test_audio_upload_complete.py
git commit -m "feat(audio-upload-complete): scaffold S3 event handler + key parser

New event-driven Lambda under main/server/events/. Top-level events/
directory introduced for non-API Lambdas (no existing precedent;
keeps separation from api/ for HTTP and worldmm/ for the ingest
pipeline).

Refs: docs/plans/2026-05-19-sessions-audio-presigned.md"

class TestAudioOnlyMode:
    @staticmethod
    def test_happy_path_audio_only_mode_triggers_ingest_immediately():
        module = _load_module()
        mock_table = MagicMock()
        mock_table.update_item.return_value = {
            "Attributes": {
                "sessionId": "sess-1",
                "captureMode": "audio_only",
                "completedAudioWindows": {7},
                "completedFrameWindows": set(),
                "userId": "user-1",
            }
        }
        mock_lambda = MagicMock()
        with (
            patch.object(module, "table", mock_table),
            patch.object(module, "lambda_client", mock_lambda),
            patch.dict(os.environ, {"INGEST_FUNCTION_NAME": "IngestFn"}),
        ):
            module.lambda_handler(_s3_event("sessions/sess-1/window_007/audio.wav"), None)

        # DDB ADD called with correct window index
        update_kwargs = mock_table.update_item.call_args.kwargs
        assert update_kwargs["Key"] == {"sessionId": "sess-1"}
        assert ":win" in update_kwargs["ExpressionAttributeValues"]
        assert 7 in update_kwargs["ExpressionAttributeValues"][":win"]

        # Ingest invoked
        invoke_kwargs = mock_lambda.invoke.call_args.kwargs
        assert invoke_kwargs["FunctionName"] == "IngestFn"
        assert invoke_kwargs["InvocationType"] == "Event"
        payload = json.loads(invoke_kwargs["Payload"])
        assert payload["sessionId"] == "sess-1"
        assert payload["windowIndex"] == 7

    @staticmethod
    def test_adds_window_index_to_completed_audio_windows_in_ddb():
        module = _load_module()
        mock_table = MagicMock()
        mock_table.update_item.return_value = {
            "Attributes": {
                "captureMode": "audio_only",
                "completedAudioWindows": {3},
                "completedFrameWindows": set(),
                "userId": "u",
            }
        }
        with (
            patch.object(module, "table", mock_table),
            patch.object(module, "lambda_client", MagicMock()),
        ):
            module.lambda_handler(_s3_event("sessions/sess-1/window_003/audio.wav"), None)

        kwargs = mock_table.update_item.call_args.kwargs
        assert kwargs["UpdateExpression"] == "ADD completedAudioWindows :win"

def _claim_and_trigger(
    session_id: str, window_index: int, user_id: str, frame_count: int
) -> bool:
    """Conditional claim + async invoke of IngestWindowFunction.

    If the invoke fails, the claim is rolled back so AWS async retry can
    re-claim and re-invoke. Without the rollback, a transient invoke
    failure would leave the claim in place and silently drop the window
    on retry (the conditional would reject the second attempt).

    Returns True if we won the claim race AND the invoke succeeded.
    Returns False if another invocation already claimed this window.
    Raises if the invoke fails — caller (lambda_handler) propagates so
    AWS async-invoke retry kicks in. By the time the retry runs, the
    claim has been rolled back so the new attempt can succeed.
    """
    from botocore.exceptions import ClientError

    try:
        table.update_item(
            Key={"sessionId": session_id},
            UpdateExpression="ADD ingestTriggeredWindows :win",
            ConditionExpression=(
                "attribute_not_exists(ingestTriggeredWindows) "
                "OR NOT contains(ingestTriggeredWindows, :winVal)"
            ),
            ExpressionAttributeValues={
                ":win": {window_index},
                ":winVal": window_index,
            },
        )
    except ClientError as e:
        if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
            logger(
                {
                    "step": "ingest_claim_lost",
                    "additional": {
                        "session_id": session_id,
                        "window_index": window_index,
                    },
                }
            )
            return False
        raise

    try:
        lambda_client.invoke(
            FunctionName=INGEST_FUNCTION,
            InvocationType="Event",
            Payload=json.dumps(
                {
                    "sessionId": session_id,
                    "userId": user_id,
                    "windowIndex": window_index,
                    "frameCount": frame_count,
                }
            ),
        )
    except Exception:
        # Roll back the claim so AWS async retry can re-claim + re-invoke.
        # Without this, the conditional rejects the retry and the window
        # is silently dropped.
        try:
            table.update_item(
                Key={"sessionId": session_id},
                UpdateExpression="DELETE ingestTriggeredWindows :win",
                ExpressionAttributeValues={":win": {window_index}},
            )
            logger(
                {
                    "step": "ingest_claim_rolled_back",
                    "additional": {
                        "session_id": session_id,
                        "window_index": window_index,
                    },
                }
            )
        except Exception as rollback_err:
            # If rollback fails, log loudly but still re-raise the original
            # invoke failure. The rollback failure is recoverable manually
            # (operator can delete the window from ingestTriggeredWindows).
            logger(
                {
                    "step": "ingest_claim_rollback_failed",
                    "additional": {
                        "session_id": session_id,
                        "window_index": window_index,
                        "error": str(rollback_err),
                    },
                }
            )
        raise

    return True


def lambda_handler(event, _context):
    for record in event.get("Records", []):
        key = record.get("s3", {}).get("object", {}).get("key", "")
        parsed = _parse_key(key)
        if parsed is None:
            logger(
                {"step": "audio_upload_malformed_key", "additional": {"key": key}}
            )
            continue
        session_id, window_index = parsed

        resp = table.update_item(
            Key={"sessionId": session_id},
            UpdateExpression="ADD completedAudioWindows :win",
            ExpressionAttributeValues={":win": {window_index}},
            ReturnValues="ALL_NEW",
        )
        attrs = resp.get("Attributes") or {}
        capture_mode = attrs.get("captureMode", "audio_video")
        user_id = attrs.get("userId", "")
        completed_frames = attrs.get("completedFrameWindows") or set()

        should_trigger = capture_mode == "audio_only" or window_index in completed_frames
        if should_trigger and INGEST_FUNCTION:
            frame_count = 0 if capture_mode == "audio_only" else DEFAULT_FRAME_COUNT_PER_WINDOW
            _claim_and_trigger(session_id, window_index, user_id, frame_count)

        logger(
            {
                "step": "audio_upload_registered",
                "additional": {
                    "session_id": session_id,
                    "window_index": window_index,
                    "capture_mode": capture_mode,
                },
            }
        )

    return {"ok": True}

git add main/server/events/audio_upload_complete/app.py main/server/tests/unit/test_audio_upload_complete.py
git commit -m "feat(audio-upload-complete): DDB ADD and ingest trigger in audio_only mode"

class TestAudioVideoMode:
    @staticmethod
    def test_audio_video_mode_triggers_ingest_only_when_frames_ready():
        module = _load_module()
        mock_table = MagicMock()
        mock_table.update_item.return_value = {
            "Attributes": {
                "captureMode": "audio_video",
                "completedAudioWindows": {7},
                "completedFrameWindows": {7},  # frames already done
                "userId": "u",
            }
        }
        mock_lambda = MagicMock()
        with (
            patch.object(module, "table", mock_table),
            patch.object(module, "lambda_client", mock_lambda),
            patch.dict(os.environ, {"INGEST_FUNCTION_NAME": "IngestFn"}),
        ):
            module.lambda_handler(_s3_event("sessions/sess-1/window_007/audio.wav"), None)

        # Invoke should fire — frames are ready
        assert mock_lambda.invoke.called

    @staticmethod
    def test_audio_video_mode_does_not_trigger_when_frames_pending():
        module = _load_module()
        mock_table = MagicMock()
        mock_table.update_item.return_value = {
            "Attributes": {
                "captureMode": "audio_video",
                "completedAudioWindows": {7},
                "completedFrameWindows": set(),  # frames NOT done
                "userId": "u",
            }
        }
        mock_lambda = MagicMock()
        with (
            patch.object(module, "table", mock_table),
            patch.object(module, "lambda_client", mock_lambda),
        ):
            module.lambda_handler(_s3_event("sessions/sess-1/window_007/audio.wav"), None)

        # Invoke should NOT fire — frames pending
        assert not mock_lambda.invoke.called

git add main/server/tests/unit/test_audio_upload_complete.py
git commit -m "test(audio-upload-complete): cover audio_video mode trigger race"

class TestErrorHandling:
    @staticmethod
    def test_malformed_s3_key_logs_and_returns_success():
        module = _load_module()
        mock_table = MagicMock()
        with patch.object(module, "table", mock_table):
            result = module.lambda_handler(_s3_event("garbage/path/no-match"), None)
        assert result["ok"] is True
        assert not mock_table.update_item.called

    @staticmethod
    def test_unknown_session_returns_success_without_invoking_ingest():
        """DDB returns no Attributes when session row doesn't exist (or row exists but no fields).
        Handler should still succeed without invoking ingest."""
        module = _load_module()
        mock_table = MagicMock()
        # update_item on a non-existent row creates an empty row with the
        # ADD attribute — Attributes will contain only completedAudioWindows.
        mock_table.update_item.return_value = {
            "Attributes": {"completedAudioWindows": {7}}
        }
        mock_lambda = MagicMock()
        with (
            patch.object(module, "table", mock_table),
            patch.object(module, "lambda_client", mock_lambda),
        ):
            result = module.lambda_handler(_s3_event("sessions/ghost/window_007/audio.wav"), None)
        assert result["ok"] is True
        # captureMode defaults to "audio_video"; completedFrameWindows is empty → no trigger
        assert not mock_lambda.invoke.called

    @staticmethod
    def test_ddb_throttle_raises_for_aws_retry():
        from botocore.exceptions import ClientError
        module = _load_module()
        mock_table = MagicMock()
        mock_table.update_item.side_effect = ClientError(
            {"Error": {"Code": "ProvisionedThroughputExceededException", "Message": "throttled"}},
            "UpdateItem",
        )
        with patch.object(module, "table", mock_table):
            with pytest.raises(ClientError):
                module.lambda_handler(_s3_event(), None)

    @staticmethod
    def test_ingest_invoke_failure_raises_for_aws_retry():
        module = _load_module()
        mock_table = MagicMock()
        mock_table.update_item.return_value = {
            "Attributes": {
                "captureMode": "audio_only",
                "userId": "u",
                "completedFrameWindows": set(),
            }
        }
        mock_lambda = MagicMock()
        mock_lambda.invoke.side_effect = RuntimeError("boom")
        with (
            patch.object(module, "table", mock_table),
            patch.object(module, "lambda_client", mock_lambda),
            patch.dict(os.environ, {"INGEST_FUNCTION_NAME": "IngestFn"}),
        ):
            with pytest.raises(RuntimeError):
                module.lambda_handler(_s3_event(), None)

git add main/server/tests/unit/test_audio_upload_complete.py
git commit -m "test(audio-upload-complete): cover error/recovery handling"

class TestIdempotency:
    @staticmethod
    def test_duplicate_s3_event_does_not_double_trigger_ingest():
        """Second event for same key fails the conditional claim → no second invoke."""
        from botocore.exceptions import ClientError
        module = _load_module()

        mock_table = MagicMock()
        # First call to update_item (audio ADD): succeeds with frames ready
        # Second call (claim): succeeds — first invocation
        # Third call (audio ADD again, second event): succeeds idempotently
        # Fourth call (claim, second event): conditional fails — already claimed
        mock_table.update_item.side_effect = [
            # First event: audio ADD
            {
                "Attributes": {
                    "captureMode": "audio_video",
                    "completedAudioWindows": {7},
                    "completedFrameWindows": {7},
                    "userId": "u",
                }
            },
            # First event: claim
            {"Attributes": {"ingestTriggeredWindows": {7}}},
            # Second event: audio ADD (idempotent — set semantics)
            {
                "Attributes": {
                    "captureMode": "audio_video",
                    "completedAudioWindows": {7},
                    "completedFrameWindows": {7},
                    "userId": "u",
                }
            },
            # Second event: claim fails conditional
            ClientError(
                {"Error": {"Code": "ConditionalCheckFailedException", "Message": "x"}},
                "UpdateItem",
            ),
        ]
        mock_lambda = MagicMock()
        with (
            patch.object(module, "table", mock_table),
            patch.object(module, "lambda_client", mock_lambda),
            patch.dict(os.environ, {"INGEST_FUNCTION_NAME": "IngestFn"}),
        ):
            evt = _s3_event("sessions/sess-1/window_007/audio.wav")
            module.lambda_handler(evt, None)
            module.lambda_handler(evt, None)

        # invoke called exactly once
        assert mock_lambda.invoke.call_count == 1

    @staticmethod
    def test_concurrent_audio_and_frame_completion_invoke_ingest_exactly_once():
        """Both an audio S3 event and a frame completion (from FramePostFunction)
        can race to call _claim_and_trigger for the same window. The conditional
        ensures exactly one invoke fires.

        Simulated by calling _claim_and_trigger twice in sequence — second call
        returns False because the conditional rejects the duplicate.
        """
        from botocore.exceptions import ClientError
        module = _load_module()
        mock_table = MagicMock()
        mock_table.update_item.side_effect = [
            {"Attributes": {"ingestTriggeredWindows": {7}}},  # first claim wins
            ClientError(  # second claim loses
                {"Error": {"Code": "ConditionalCheckFailedException", "Message": "x"}},
                "UpdateItem",
            ),
        ]
        mock_lambda = MagicMock()
        with (
            patch.object(module, "table", mock_table),
            patch.object(module, "lambda_client", mock_lambda),
            patch.dict(os.environ, {"INGEST_FUNCTION_NAME": "IngestFn"}),
        ):
            first = module._claim_and_trigger("sess-1", 7, "u", 60)
            second = module._claim_and_trigger("sess-1", 7, "u", 60)

        assert first is True
        assert second is False
        assert mock_lambda.invoke.call_count == 1

git add main/server/tests/unit/test_audio_upload_complete.py
git commit -m "test(audio-upload-complete): cover duplicate S3 event idempotency"

  AudioPostFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: api/sessions/audio/
      Handler: app.lambda_handler
      Runtime: python3.12
      Layers:
        - !Ref SharedLayer
      Environment:
        Variables:
          BUCKET_NAME: !Ref RawMemoryBucket
          SESSIONS_TABLE_NAME: !Ref SessionsTable
      Policies:
        - !Ref S3AccessPolicy
        - Statement:
            - Effect: Allow
              Action: dynamodb:GetItem
              Resource: !GetAtt SessionsTable.Arn
      Events:
        AudioPost:
          Type: Api
          Properties:
            Path: /sessions/{sessionId}/audio
            Method: post

  AudioUploadCompleteDLQ:
    Type: AWS::SQS::Queue
    Properties:
      VisibilityTimeout: 720
      MessageRetentionPeriod: 1209600

  AudioUploadCompleteFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: events/audio_upload_complete/
      Handler: app.lambda_handler
      Runtime: python3.12
      Timeout: 30
      MemorySize: 256
      Layers:
        - !Ref SharedLayer
      Environment:
        Variables:
          SESSIONS_TABLE_NAME: !Ref SessionsTable
          INGEST_FUNCTION_NAME: !Ref IngestWindowFunction
      Policies:
        - !Ref DynamoDBPolicy
        - !Ref LambdaInvokePolicy
      EventInvokeConfig:
        MaximumRetryAttempts: 2
        DestinationConfig:
          OnFailure:
            Type: SQS
            Destination: !GetAtt AudioUploadCompleteDLQ.Arn

  AudioUploadCompletePermission:
    Type: AWS::Lambda::Permission
    Properties:
      Action: lambda:InvokeFunction
      FunctionName: !GetAtt AudioUploadCompleteFunction.Arn
      Principal: s3.amazonaws.com
      SourceArn: !GetAtt RawMemoryBucket.Arn
      SourceAccount: !Ref AWS::AccountId

git add main/server/template.yaml
git commit -m "chore(infra): add AudioUploadCompleteFunction + DLQ + S3 notification

- New SAM function under events/audio_upload_complete/.
- Dedicated SQS DLQ with 2 retries before drop.
- S3 ObjectCreated event on sessions/*/audio.wav prefix.
- Downgrades AudioPostFunction IAM to S3 sign + DDB GetItem only.

Refs: docs/plans/2026-05-19-sessions-audio-presigned.md"

# main/server/tests/integration/test_audio_presigned_flow.py
"""End-to-end test of the presigned audio flow against moto.

Covers Happy Path steps 2-5 and the URL-signature rejection cases
from the Failure Modes table in
docs/plans/2026-05-19-sessions-audio-presigned.md.
"""

from __future__ import annotations

import json
import time
from pathlib import Path

import boto3
import pytest
import requests
from moto import mock_aws

from ..worker_utils import load_worker_module

_URL_MINTER_PATH = (
    Path(__file__).resolve().parents[2] / "api" / "sessions" / "audio" / "app.py"
)
_COMPLETE_PATH = (
    Path(__file__).resolve().parents[2] / "events" / "audio_upload_complete" / "app.py"
)


@pytest.fixture
def aws_env(monkeypatch):
    monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
    monkeypatch.setenv("BUCKET_NAME", "test-bucket")
    monkeypatch.setenv("SESSIONS_TABLE_NAME", "test-sessions")
    monkeypatch.setenv("INGEST_FUNCTION_NAME", "IngestFn")


def _setup_aws():
    """Create the bucket + sessions table inside an active mock_aws context."""
    s3 = boto3.client("s3", region_name="us-east-1")
    s3.create_bucket(Bucket="test-bucket")

    ddb = boto3.resource("dynamodb", region_name="us-east-1")
    ddb.create_table(
        TableName="test-sessions",
        KeySchema=[{"AttributeName": "sessionId", "KeyType": "HASH"}],
        AttributeDefinitions=[{"AttributeName": "sessionId", "AttributeType": "S"}],
        BillingMode="PAY_PER_REQUEST",
    )
    table = ddb.Table("test-sessions")
    table.put_item(Item={"sessionId": "sess-1", "captureMode": "audio_only", "userId": "u"})
    return table


@mock_aws
def test_full_upload_flow_writes_audio_and_triggers_ingest(aws_env, monkeypatch):
    table = _setup_aws()

    # Reload modules under the mock context so they pick up moto's clients
    minter = load_worker_module("audio_minter_int", _URL_MINTER_PATH)
    complete = load_worker_module("audio_complete_int", _COMPLETE_PATH)

    # Step 1: client POSTs for URL
    grant_event = {
        "pathParameters": {"sessionId": "sess-1"},
        "queryStringParameters": {"windowIndex": "7"},
        "body": json.dumps({"sizeBytes": 4}),
    }
    grant_resp = minter.lambda_handler(grant_event, None)
    assert grant_resp["statusCode"] == 200
    grant_body = json.loads(grant_resp["body"])
    url = grant_body["url"]

    # Step 2: client PUTs bytes
    put_resp = requests.put(
        url,
        data=b"DATA",
        headers={"Content-Type": "audio/wav", "Content-Length": "4"},
        timeout=10,
    )
    assert put_resp.status_code == 200

    # Step 3: simulate S3 event firing → complete handler
    invoke_calls: list = []
    monkeypatch.setattr(
        complete.lambda_client,
        "invoke",
        lambda **kw: invoke_calls.append(kw),
    )
    s3_event = {
        "Records": [
            {
                "s3": {
                    "bucket": {"name": "test-bucket"},
                    "object": {"key": "sessions/sess-1/window_007/audio.wav"},
                }
            }
        ]
    }
    complete.lambda_handler(s3_event, None)

    # DDB row reflects window completion
    row = table.get_item(Key={"sessionId": "sess-1"})["Item"]
    assert 7 in row.get("completedAudioWindows", set())

    # Ingest invoke captured
    assert len(invoke_calls) == 1
    payload = json.loads(invoke_calls[0]["Payload"])
    assert payload["sessionId"] == "sess-1"
    assert payload["windowIndex"] == 7


@mock_aws
def test_url_signature_rejects_wrong_content_length(aws_env):
    _setup_aws()
    minter = load_worker_module("audio_minter_int2", _URL_MINTER_PATH)
    grant_resp = minter.lambda_handler(
        {
            "pathParameters": {"sessionId": "sess-1"},
            "queryStringParameters": {"windowIndex": "0"},
            "body": json.dumps({"sizeBytes": 100}),
        },
        None,
    )
    url = json.loads(grant_resp["body"])["url"]

    # PUT with mismatched Content-Length should fail
    resp = requests.put(
        url,
        data=b"X" * 50,  # 50 bytes != signed 100
        headers={"Content-Type": "audio/wav", "Content-Length": "50"},
        timeout=10,
    )
    assert resp.status_code >= 400


@mock_aws
def test_url_signature_rejects_wrong_content_type(aws_env):
    _setup_aws()
    minter = load_worker_module("audio_minter_int3", _URL_MINTER_PATH)
    grant_resp = minter.lambda_handler(
        {
            "pathParameters": {"sessionId": "sess-1"},
            "queryStringParameters": {"windowIndex": "0"},
            "body": json.dumps({"sizeBytes": 4}),
        },
        None,
    )
    url = json.loads(grant_resp["body"])["url"]

    resp = requests.put(
        url,
        data=b"DATA",
        headers={"Content-Type": "video/mp4", "Content-Length": "4"},
        timeout=10,
    )
    assert resp.status_code >= 400


@mock_aws
def test_expired_url_returns_403_on_put(aws_env, monkeypatch):
    """Generate a URL with 1s expiry, sleep past it, attempt PUT, expect rejection.

    Adjusts URL_EXPIRES_SECONDS via monkeypatch instead of waiting 5 minutes.
    """
    _setup_aws()
    minter = load_worker_module("audio_minter_int4", _URL_MINTER_PATH)
    monkeypatch.setattr(minter, "URL_EXPIRES_SECONDS", 1)
    grant_resp = minter.lambda_handler(
        {
            "pathParameters": {"sessionId": "sess-1"},
            "queryStringParameters": {"windowIndex": "0"},
            "body": json.dumps({"sizeBytes": 4}),
        },
        None,
    )
    url = json.loads(grant_resp["body"])["url"]

    time.sleep(2)

    resp = requests.put(
        url,
        data=b"DATA",
        headers={"Content-Type": "audio/wav", "Content-Length": "4"},
        timeout=10,
    )
    # Real S3 returns 403 with "Request has expired"; moto may return 403 or 400.
    # Either way the PUT must NOT succeed.
    assert resp.status_code >= 400, (
        f"expected expired URL to reject PUT, got {resp.status_code} "
        f"(if moto skips expiry validation, skip this test and rely on issue #497)"
    )

git add main/server/tests/integration/test_audio_presigned_flow.py
git commit -m "test(audio): integration tests for presigned upload flow with moto"

rm main/server/tests/unit/test_audio_chunk_api.py

git add -A main/server/tests/unit/
git commit -m "chore(test): drop test_audio_chunk_api.py — superseded by URL minter + complete tests"

// main/app/__tests__/capture-session-audio-upload.test.ts
/**
 * Tests for the rewritten audio branch of the capture-session uploadFn
 * (see main/app/lib/capture-session.ts, replacing the lines that used to
 * POST raw audio bytes through API Gateway).
 *
 * Each test cites a Happy Path step or Failure Modes table row in
 * docs/plans/2026-05-19-sessions-audio-presigned.md.
 */

import { uploadAudioItem } from "../lib/capture-session";

const sessionId = "sess-1";
const baseItem = {
  id: "audio-1",
  type: "audio" as const,
  uri: "file:///tmp/audio.wav",
  sessionId,
  windowIndex: 7,
  retryCount: 0,
  sizeBytes: 1234,
};

const flushMicrotasks = () => new Promise((r) => setImmediate(r));

describe("uploadAudioItem", () => {
  let originalFetch: typeof global.fetch;
  let mockApiPost: jest.Mock;
  let mockFetch: jest.Mock;
  let mockReadBytes: jest.Mock;

  beforeEach(() => {
    originalFetch = global.fetch;
    mockApiPost = jest.fn().mockResolvedValue({
      url: "https://s3.example/presigned",
      s3Key: `sessions/${sessionId}/window_007/audio.wav`,
      windowIndex: 7,
      expiresIn: 300,
    });
    mockFetch = jest.fn().mockResolvedValue({ ok: true, status: 200 });
    mockReadBytes = jest.fn().mockResolvedValue(new Uint8Array(1234));
    global.fetch = mockFetch as unknown as typeof global.fetch;
  });

  afterEach(() => {
    global.fetch = originalFetch;
    jest.restoreAllMocks();
  });

  test("uploads_audio_when_both_url_grant_and_put_succeed", async () => {
    await uploadAudioItem({
      item: baseItem,
      sessionId,
      apiPost: mockApiPost,
      readBytes: mockReadBytes,
    });

    expect(mockApiPost).toHaveBeenCalledWith(
      `/sessions/${sessionId}/audio?windowIndex=7`,
      JSON.stringify({ sizeBytes: 1234 }),
      expect.objectContaining({
        headers: { "Content-Type": "application/json" },
      }),
    );

    expect(mockFetch).toHaveBeenCalledWith(
      "https://s3.example/presigned",
      expect.objectContaining({
        method: "PUT",
        headers: {
          "Content-Type": "audio/wav",
          "Content-Length": "1234",
        },
      }),
    );
  });

  test("does_not_remove_item_from_queue_if_put_throws", async () => {
    mockFetch.mockRejectedValueOnce(new Error("network down"));
    await expect(
      uploadAudioItem({
        item: baseItem,
        sessionId,
        apiPost: mockApiPost,
        readBytes: mockReadBytes,
      }),
    ).rejects.toThrow(/network down/);
  });

  test("parks_item_when_url_grant_returns_5xx", async () => {
    mockApiPost.mockRejectedValueOnce(Object.assign(new Error("500"), { status: 500 }));
    await expect(
      uploadAudioItem({
        item: baseItem,
        sessionId,
        apiPost: mockApiPost,
        readBytes: mockReadBytes,
      }),
    ).rejects.toMatchObject({ status: 500 });
    expect(mockFetch).not.toHaveBeenCalled();
  });

  test("parks_item_when_put_to_s3_returns_5xx", async () => {
    mockFetch.mockResolvedValueOnce({ ok: false, status: 503 });
    await expect(
      uploadAudioItem({
        item: baseItem,
        sessionId,
        apiPost: mockApiPost,
        readBytes: mockReadBytes,
      }),
    ).rejects.toThrow(/S3 PUT failed/);
  });

  test("parks_item_when_put_returns_403_signature_mismatch", async () => {
    mockFetch.mockResolvedValueOnce({ ok: false, status: 403 });
    await expect(
      uploadAudioItem({
        item: baseItem,
        sessionId,
        apiPost: mockApiPost,
        readBytes: mockReadBytes,
      }),
    ).rejects.toThrow(/S3 PUT failed/);
  });

  test("refetches_url_after_5xx_retry_does_not_reuse_stale_url", async () => {
    // Simulate two consecutive uploads of the same item — each must
    // request its own URL, not reuse a cached one.
    await uploadAudioItem({
      item: baseItem,
      sessionId,
      apiPost: mockApiPost,
      readBytes: mockReadBytes,
    });
    await uploadAudioItem({
      item: baseItem,
      sessionId,
      apiPost: mockApiPost,
      readBytes: mockReadBytes,
    });

    expect(mockApiPost).toHaveBeenCalledTimes(2);
    expect(mockFetch).toHaveBeenCalledTimes(2);
  });
});

export interface UploadAudioDeps {
  item: {
    uri: string;
    windowIndex?: number;
    sizeBytes: number;
  };
  sessionId: string;
  apiPost: (
    path: string,
    body: string,
    options: { headers: Record<string, string>; timeout?: number },
  ) => Promise<{ url: string; s3Key: string; windowIndex: number; expiresIn: number }>;
  readBytes: (uri: string) => Promise<Uint8Array>;
}

export async function uploadAudioItem(deps: UploadAudioDeps): Promise<void> {
  const { item, sessionId, apiPost, readBytes } = deps;
  const windowIndex = item.windowIndex ?? 0;
  const bytes = await readBytes(item.uri);
  const sizeBytes = bytes.byteLength;

  const grant = await apiPost(
    `/sessions/${sessionId}/audio?windowIndex=${windowIndex}`,
    JSON.stringify({ sizeBytes }),
    {
      headers: { "Content-Type": "application/json" },
      timeout: 5000,
    },
  );

  const PUT_TIMEOUT_MS = 60000;
  const controller = new AbortController();
  const timeoutId = setTimeout(() => controller.abort(), PUT_TIMEOUT_MS);
  let putResp: Response;
  try {
    putResp = await fetch(grant.url, {
      method: "PUT",
      headers: {
        "Content-Type": "audio/wav",
        "Content-Length": String(sizeBytes),
      },
      body: bytes as unknown as BodyInit,
      signal: controller.signal,
    });
  } finally {
    clearTimeout(timeoutId);
  }
  if (!putResp.ok) {
    throw new Error(`S3 PUT failed: ${putResp.status}`);
  }
}

    if (item.type === "frame") {
      await api.post(`/sessions/${sessionId}/frames`, bytes, {
        headers: { "Content-Type": "image/jpeg" },
        timeout: 10000,
      });
      const filename = item.uri.split("/").pop() || "";
      await markUploaded(sessionId, filename);
    } else {
      const file = new File(toFileUri(item.uri));
      await uploadAudioItem({
        item,
        sessionId,
        apiPost: async (path, body, options) => {
          const res = await api.post<{
            url: string;
            s3Key: string;
            windowIndex: number;
            expiresIn: number;
          }>(path, body, options);
          return res;
        },
        readBytes: async (_uri) => await file.bytes(),
      });
    }

git add main/app/lib/capture-session.ts main/app/__tests__/capture-session-audio-upload.test.ts
git commit -m "feat(client): switch audio upload to presigned PUT flow

uploadAudioItem requests presigned URL from server, then PUTs raw
WAV bytes directly to S3 — bypasses API Gateway 29s cap. Frame
upload unchanged.

Refs: docs/plans/2026-05-19-sessions-audio-presigned.md"

git push -u origin plan/sessions-audio-presigned