FEAT-103 Code-Review Fixes 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: Fix all issues found in the FEAT-103 code review — covering a dead-import cleanup, a critical API contract regression in ChatStorage, DRY handler extraction, ArtifactStore correctness bugs, query pagination, and test reliability.

Architecture: All changes are confined to the feat-103-agent-artifact-persistency worktree at .claude/worktrees/feat-103-agent-artifact-persistency. The asyncdb DynamoDB driver does NOT exist in the installed version (2.15.0), so the aioboto3 implementation is correct and kept as-is; only a clarifying comment is added. Changes proceed layer-by-layer: storage backend → ChatStorage → handlers → tests → completion notes.

Tech Stack: Python 3.11, aioboto3, botocore, aiohttp, pydantic v2, pytest-asyncio

Working directory for all commands: .claude/worktrees/feat-103-agent-artifact-persistency

File Map¶

File	Action	What changes
`packages/ai-parrot/src/parrot/storage/dynamodb.py`	Modify	Remove dead imports; move `DKey`/`Attr` to top; add `delete_turn()`; fix `query_threads()` pagination; add aioboto3 justification comment
`packages/ai-parrot/src/parrot/storage/artifacts.py`	Modify	Fix `datetime.utcnow()` → `datetime.now(timezone.utc)`; recalculate TTL in `update_artifact()`; guard enum in `_deserialize()`
`packages/ai-parrot/src/parrot/storage/chat.py`	Modify	Add `update_thread_metadata()`; fix `delete_turn()` to use `self._dynamo.delete_turn()`; add warning log in `update_conversation_title()` + `delete_turn()` when ids missing; add `document_db` deprecation warning; increment `turn_count` atomically in `_save_to_dynamodb()`
`packages/ai-parrot/src/parrot/handlers/_mixins.py`	Create	`UserSessionMixin` with `_get_user_id()`
`packages/ai-parrot/src/parrot/handlers/threads.py`	Modify	Use `UserSessionMixin`; replace `_dynamo` leak with `storage.update_thread_metadata()`; require `agent_id` in DELETE
`packages/ai-parrot/src/parrot/handlers/artifacts.py`	Modify	Use `UserSessionMixin`; require `agent_id` in list and detail endpoints
`packages/ai-parrot/src/parrot/handlers/agent.py`	Modify	Move inline `Artifact`/`datetime` imports to module top; fix f-string log → `%s`
`packages/ai-parrot/src/parrot/handlers/infographic.py`	Modify	Move inline `Artifact`/`ArtifactType`/`ArtifactCreator` imports to module top
`tests/storage/test_dynamodb_backend.py`	Modify	Add tests for `delete_turn()`; add test for `query_threads()` pagination behaviour
`tests/storage/test_chat_storage_dynamodb.py`	Modify	Add tests for `update_thread_metadata()`; add test for `delete_turn()` via backend
`tests/storage/test_integration_artifact_persistence.py`	Modify	Replace `asyncio.sleep(0.05)` with direct `_save_to_dynamodb()` call
`sdd/tasks/completed/TASK-71[7-9].md` + `TASK-72[0-6].md`	Modify	Fill in completion notes for all 10 tasks

Task 1 — `dynamodb.py` cleanup and `delete_turn()` addition¶

Files: - Modify: packages/ai-parrot/src/parrot/storage/dynamodb.py - Modify: tests/storage/test_dynamodb_backend.py

1.1 — Write failing tests for `delete_turn()` and `query_threads()` pagination¶

Open tests/storage/test_dynamodb_backend.py and append:

class TestDeleteTurn:
    """Tests for the new delete_turn method."""

    @pytest.mark.asyncio
    async def test_delete_turn_calls_delete_item(self, dynamo_backend):
        mock_table = _make_mock_table()
        dynamo_backend._conv_table = mock_table
        dynamo_backend._art_table = _make_mock_table()

        await dynamo_backend.delete_turn("u1", "agent1", "sess1", "t001")

        mock_table.delete_item.assert_called_once()
        key = mock_table.delete_item.call_args.kwargs["Key"]
        assert key["PK"] == "USER#u1#AGENT#agent1"
        assert key["SK"] == "THREAD#sess1#TURN#t001"

    @pytest.mark.asyncio
    async def test_delete_turn_not_initialized(self, dynamo_backend):
        # Should not raise when not connected
        await dynamo_backend.delete_turn("u1", "agent1", "sess1", "t001")

    @pytest.mark.asyncio
    async def test_delete_turn_handles_client_error(self, dynamo_backend):
        from botocore.exceptions import ClientError
        mock_table = _make_mock_table()
        mock_table.delete_item.side_effect = ClientError(
            {"Error": {"Code": "500", "Message": "Internal"}}, "DeleteItem"
        )
        dynamo_backend._conv_table = mock_table
        dynamo_backend._art_table = _make_mock_table()
        # Should not raise
        await dynamo_backend.delete_turn("u1", "agent1", "sess1", "t001")


class TestQueryThreadsPagination:
    """Tests that query_threads paginates to collect enough thread-type items."""

    @pytest.mark.asyncio
    async def test_returns_empty_when_not_connected(self, dynamo_backend):
        result = await dynamo_backend.query_threads("u1", "agent1", limit=10)
        assert result == []

    @pytest.mark.asyncio
    async def test_paginates_when_last_evaluated_key_present(self, dynamo_backend):
        """Simulates first page returning only turns (filtered out),
        second page returning a thread item."""
        mock_table = _make_mock_table()
        dynamo_backend._conv_table = mock_table
        dynamo_backend._art_table = _make_mock_table()

        page1 = {
            "Items": [],  # All turns filtered — no thread items
            "LastEvaluatedKey": {"PK": "X", "SK": "Y"},
        }
        page2 = {
            "Items": [
                {"PK": "USER#u1#AGENT#agent1", "SK": "THREAD#sess1",
                 "type": "thread", "session_id": "sess1", "title": "T1"}
            ],
        }
        mock_table.query = AsyncMock(side_effect=[page1, page2])

        results = await dynamo_backend.query_threads("u1", "agent1", limit=5)
        assert len(results) == 1
        assert results[0]["session_id"] == "sess1"
        # Two calls: first page + second page
        assert mock_table.query.call_count == 2

Run to confirm failures:

cd .claude/worktrees/feat-103-agent-artifact-persistency
source .venv/bin/activate 2>/dev/null || true
pytest tests/storage/test_dynamodb_backend.py::TestDeleteTurn tests/storage/test_dynamodb_backend.py::TestQueryThreadsPagination -v 2>&1 | tail -20

Expected: ERROR or FAILED — delete_turn and new pagination behaviour don't exist yet.

1.2 — Fix `dynamodb.py`¶

Open packages/ai-parrot/src/parrot/storage/dynamodb.py and apply these changes in order:

a) Replace the top import block (remove dead imports, hoist late imports):

Old block (approximately lines 1–18):

from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional

import aioboto3
from botocore.exceptions import ClientError, BotoCoreError
from navconfig.logging import logging

from boto3.dynamodb.types import TypeSerializer, TypeDeserializer

New block:

"""DynamoDB backend for conversation and artifact storage.

Domain wrapper around aioboto3's DynamoDB resource API. Provides PK/SK
construction, TTL setting, and domain-specific query patterns for the
two-table design (conversations + artifacts).

NOTE: The FEAT-103 spec originally specified asyncdb's AsyncDB("dynamodb")
driver, but asyncdb 2.15.0 does not ship a DynamoDB driver.  We use
aioboto3 directly here — it is aioboto3/aiobotocore that asyncdb itself
would wrap.  If asyncdb ships a DynamoDB driver in a future release this
class can be refactored to delegate to it.

FEAT-103: agent-artifact-persistency — Module 2.
"""

from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional

import aioboto3
from boto3.dynamodb.conditions import Key as DKey, Attr
from botocore.exceptions import ClientError, BotoCoreError
from navconfig.logging import logging

b) Remove all from boto3.dynamodb.conditions import ... lines inside methods — there are five occurrences (in query_threads, query_turns, delete_thread_cascade, query_artifacts, delete_session_artifacts). Delete those inline import lines in each method.

c) Add delete_turn() method directly after query_turns() (before delete_thread_cascade()):

async def delete_turn(
    self,
    user_id: str,
    agent_id: str,
    session_id: str,
    turn_id: str,
) -> None:
    """Delete a single conversation turn.

    Args:
        user_id: User identifier.
        agent_id: Agent/bot identifier.
        session_id: Conversation session identifier.
        turn_id: Turn identifier.
    """
    if not self.is_connected:
        return
    pk = self._build_pk(user_id, agent_id)
    sk = f"THREAD#{session_id}#TURN#{turn_id}"
    try:
        await self._conv_table.delete_item(Key={"PK": pk, "SK": sk})
    except (ClientError, BotoCoreError, Exception) as exc:
        self.logger.warning(
            "DynamoDB delete_turn failed for session %s turn %s: %s",
            session_id, turn_id, exc,
        )

d) Replace query_threads() body with a version that paginates to collect enough type=thread items despite Limit+FilterExpression interaction:

async def query_threads(
    self,
    user_id: str,
    agent_id: str,
    limit: int = 50,
) -> List[dict]:
    """List thread metadata items for a user+agent pair.

    Uses over-read pagination (Limit = limit * 5) to compensate for
    DynamoDB applying Limit before FilterExpression.  For users with many
    turns per thread, a small Limit would return zero thread-type items
    even though threads exist — the pagination loop ensures we collect
    enough thread items.

    Args:
        user_id: User identifier.
        agent_id: Agent/bot identifier.
        limit: Maximum number of thread items to return.

    Returns:
        List of thread metadata dicts, newest first.
    """
    if not self.is_connected:
        return []
    pk = self._build_pk(user_id, agent_id)
    threads: List[dict] = []
    last_key = None
    try:
        while len(threads) < limit:
            kwargs: Dict[str, Any] = {
                "KeyConditionExpression": DKey("PK").eq(pk) & DKey("SK").begins_with("THREAD#"),
                "FilterExpression": Attr("type").eq("thread"),
                "ScanIndexForward": False,
                # Over-read: fetch limit*5 raw items to get at least `limit` threads
                "Limit": min(limit * 5, 1000),
            }
            if last_key:
                kwargs["ExclusiveStartKey"] = last_key
            response = await self._conv_table.query(**kwargs)
            threads.extend(response.get("Items", []))
            last_key = response.get("LastEvaluatedKey")
            if not last_key:
                break
    except (ClientError, BotoCoreError, Exception) as exc:
        self.logger.warning(
            "DynamoDB query_threads failed for user %s: %s", user_id, exc
        )
        return []
    return threads[:limit]

Run tests to confirm they pass:

pytest tests/storage/test_dynamodb_backend.py -v 2>&1 | tail -30

Expected: All tests pass.

1.3 — Commit¶

Commit:

git add packages/ai-parrot/src/parrot/storage/dynamodb.py \
        tests/storage/test_dynamodb_backend.py
git commit -m "fix(storage): dynamodb.py cleanup — remove dead imports, hoist DKey/Attr, add delete_turn(), fix query_threads pagination"

Task 2 — `artifacts.py` correctness fixes¶

Files: - Modify: packages/ai-parrot/src/parrot/storage/artifacts.py - Modify: tests/storage/test_artifact_store.py

2.1 — Write failing tests¶

Append to tests/storage/test_artifact_store.py:

class TestUpdateArtifactTTL:
    """update_artifact() must recalculate TTL and use timezone-aware datetime."""

    @pytest.mark.asyncio
    async def test_update_artifact_recalculates_ttl(self, store, mock_dynamo, mock_overflow):
        mock_dynamo.get_artifact.return_value = {
            "artifact_id": "chart-x1",
            "artifact_type": "chart",
            "title": "Revenue",
            "created_at": "2025-04-16T12:00:00",
            "updated_at": "2025-04-16T12:00:00",
            "definition": {"old": "def"},
            "definition_ref": None,
        }
        await store.update_artifact("u1", "bot1", "sess1", "chart-x1",
                                    definition={"new": "def"})
        call_data = mock_dynamo.put_artifact.call_args.kwargs["data"]
        # updated_at must be a timezone-aware ISO string (contains '+' or 'Z')
        assert "+" in call_data["updated_at"] or "Z" in call_data["updated_at"], (
            f"updated_at is not timezone-aware: {call_data['updated_at']}"
        )

    @pytest.mark.asyncio
    async def test_deserialize_unknown_artifact_type_falls_back(
        self, store
    ):
        """_deserialize must not raise ValueError on unknown artifact_type."""
        from parrot.storage.artifacts import ArtifactStore
        raw = {
            "artifact_id": "x1",
            "artifact_type": "video",   # not a valid ArtifactType
            "title": "X",
            "created_at": "2025-04-16T12:00:00",
            "updated_at": "2025-04-16T12:00:00",
        }
        # Should not raise — falls back to EXPORT
        artifact = ArtifactStore._deserialize(raw, None)
        from parrot.storage.models import ArtifactType
        assert artifact.artifact_type == ArtifactType.EXPORT

Run to confirm failures:

pytest tests/storage/test_artifact_store.py::TestUpdateArtifactTTL -v 2>&1 | tail -20

2.2 — Fix `artifacts.py`¶

In packages/ai-parrot/src/parrot/storage/artifacts.py, make these three changes:

a) Fix datetime.utcnow() in update_artifact(). Find:

        now = datetime.utcnow().isoformat()
        update_data["updated_at"] = now

Replace with:

        now = datetime.now(timezone.utc)
        update_data["updated_at"] = now.isoformat()
        update_data["ttl"] = ConversationDynamoDB._ttl_epoch(now, ConversationDynamoDB.DEFAULT_TTL_DAYS)

Add timezone to the imports at the top of the file. Find:

from datetime import datetime

Replace with:

from datetime import datetime, timezone

b) Guard enum in _deserialize(). Find:

        return Artifact(
            artifact_id=raw.get("artifact_id", ""),
            artifact_type=raw.get("artifact_type", ArtifactType.CHART),

Replace with:

        _raw_type = raw.get("artifact_type", ArtifactType.CHART)
        try:
            _a_type = ArtifactType(_raw_type)
        except ValueError:
            # Unknown artifact type stored in DynamoDB (e.g. from a future version)
            # Fall back to EXPORT rather than crashing the caller.
            import logging as _log
            _log.getLogger("parrot.storage.ArtifactStore").warning(
                "Unknown artifact_type '%s' in DynamoDB item %s — defaulting to EXPORT",
                _raw_type, raw.get("artifact_id"),
            )
            _a_type = ArtifactType.EXPORT
        return Artifact(
            artifact_id=raw.get("artifact_id", ""),
            artifact_type=_a_type,

Run tests:

pytest tests/storage/test_artifact_store.py -v 2>&1 | tail -20

Expected: All tests pass.

2.3 — Commit¶

Commit:

git add packages/ai-parrot/src/parrot/storage/artifacts.py \
        tests/storage/test_artifact_store.py
git commit -m "fix(storage): artifacts.py — timezone-aware updated_at, TTL recalculation in update_artifact, enum guard in _deserialize"

Task 3 — `chat.py` API contract and encapsulation fixes¶

Files: - Modify: packages/ai-parrot/src/parrot/storage/chat.py - Modify: tests/storage/test_chat_storage_dynamodb.py

3.1 — Write failing tests¶

Append to tests/storage/test_chat_storage_dynamodb.py:

class TestUpdateThreadMetadata:
    """Tests for the new update_thread_metadata() method."""

    @pytest.mark.asyncio
    async def test_delegates_to_dynamo(self, chat_storage, mock_dynamo):
        result = await chat_storage.update_thread_metadata(
            user_id="u1", session_id="sess1", agent_id="bot1",
            pinned=True, tags=["sales"],
        )
        assert result is True
        mock_dynamo.update_thread.assert_called_once_with(
            user_id="u1", agent_id="bot1", session_id="sess1",
            pinned=True, tags=["sales"],
        )

    @pytest.mark.asyncio
    async def test_returns_false_when_no_dynamo(self, mock_redis):
        storage = ChatStorage(redis_conversation=mock_redis, dynamodb=None)
        storage._initialized = True
        result = await storage.update_thread_metadata(
            user_id="u1", session_id="sess1", agent_id="bot1", pinned=True,
        )
        assert result is False


class TestDeleteTurnViaBackend:
    """delete_turn() must use ConversationDynamoDB.delete_turn(), not _conv_table."""

    @pytest.mark.asyncio
    async def test_delete_turn_calls_dynamo_delete_turn(self, chat_storage, mock_dynamo):
        mock_dynamo.delete_turn = AsyncMock()
        result = await chat_storage.delete_turn(
            session_id="sess1", turn_id="t001",
            user_id="u1", agent_id="bot1",
        )
        assert result is True
        mock_dynamo.delete_turn.assert_called_once_with(
            user_id="u1", agent_id="bot1",
            session_id="sess1", turn_id="t001",
        )

    @pytest.mark.asyncio
    async def test_update_conversation_title_logs_warning_when_ids_missing(
        self, chat_storage, caplog
    ):
        import logging
        with caplog.at_level(logging.WARNING, logger="parrot.storage.ChatStorage"):
            result = await chat_storage.update_conversation_title(
                session_id="sess1", title="New Title"
                # no user_id, no agent_id
            )
        assert result is False
        assert "requires user_id and agent_id" in caplog.text


class TestTurnCountIncrement:
    """save_turn() must increment turn_count in DynamoDB."""

    @pytest.mark.asyncio
    async def test_save_turn_increments_turn_count(self, chat_storage, mock_dynamo):
        await chat_storage.save_turn(
            user_id="u1", session_id="sess1", agent_id="bot1",
            user_message="hi", assistant_response="hello",
        )
        # Allow background task to complete
        import asyncio
        await asyncio.sleep(0.1)
        # update_thread must be called with turn_count_increment=1
        calls = mock_dynamo.update_thread.call_args_list
        assert any(
            c.kwargs.get("turn_count_increment") == 1
            for c in calls
        ), f"update_thread calls: {calls}"

Run to confirm failures:

pytest tests/storage/test_chat_storage_dynamodb.py::TestUpdateThreadMetadata \
       tests/storage/test_chat_storage_dynamodb.py::TestDeleteTurnViaBackend \
       tests/storage/test_chat_storage_dynamodb.py::TestTurnCountIncrement -v 2>&1 | tail -25

3.2 — Update `ConversationDynamoDB.update_thread()` to support atomic turn_count increment¶

In dynamodb.py, extend update_thread() to accept a special turn_count_increment kwarg that uses DynamoDB ADD instead of SET:

Find the start of update_thread() and replace its body:

async def update_thread(
    self,
    user_id: str,
    agent_id: str,
    session_id: str,
    turn_count_increment: int = 0,
    **updates,
) -> None:
    """Update specific attributes on a thread metadata item.

    Supports atomic turn_count increment via the ``turn_count_increment``
    parameter, which uses DynamoDB ``ADD`` rather than ``SET`` to avoid
    read-modify-write races.

    Args:
        user_id: User identifier.
        agent_id: Agent/bot identifier.
        session_id: Conversation session identifier.
        turn_count_increment: If > 0, atomically adds this value to turn_count.
        **updates: Key-value pairs to SET.
    """
    if not self.is_connected:
        return
    if not updates and not turn_count_increment:
        return
    pk = self._build_pk(user_id, agent_id)
    sk = f"THREAD#{session_id}"

    set_parts = []
    add_parts = []
    expr_names: Dict[str, str] = {}
    expr_values: Dict[str, Any] = {}

    for i, (key, value) in enumerate(updates.items()):
        alias_name = f"#k{i}"
        alias_value = f":v{i}"
        set_parts.append(f"{alias_name} = {alias_value}")
        expr_names[alias_name] = key
        if isinstance(value, datetime):
            value = value.isoformat()
        expr_values[alias_value] = value

    if turn_count_increment:
        expr_names["#tc"] = "turn_count"
        expr_values[":tc_inc"] = turn_count_increment
        add_parts.append("#tc :tc_inc")

    expression_parts = []
    if set_parts:
        expression_parts.append("SET " + ", ".join(set_parts))
    if add_parts:
        expression_parts.append("ADD " + ", ".join(add_parts))

    update_expression = " ".join(expression_parts)

    try:
        await self._conv_table.update_item(
            Key={"PK": pk, "SK": sk},
            UpdateExpression=update_expression,
            ExpressionAttributeNames=expr_names,
            ExpressionAttributeValues=expr_values,
        )
    except (ClientError, BotoCoreError, Exception) as exc:
        self.logger.warning(
            "DynamoDB update_thread failed for session %s: %s", session_id, exc
        )

3.3 — Fix `chat.py`¶

Add update_thread_metadata() method to ChatStorage (after update_conversation_title()):

async def update_thread_metadata(
    self,
    user_id: str,
    session_id: str,
    agent_id: str,
    **fields: Any,
) -> bool:
    """Update arbitrary thread metadata fields in DynamoDB.

    Use this instead of accessing ``_dynamo`` directly from handlers.

    Args:
        user_id: User identifier.
        session_id: Conversation session identifier.
        agent_id: Agent/bot identifier.
        **fields: Metadata fields to update (e.g. pinned=True, tags=[...]).

    Returns:
        True if update succeeded, False otherwise.
    """
    if not self._dynamo:
        return False
    try:
        await self._dynamo.update_thread(
            user_id=user_id,
            agent_id=agent_id,
            session_id=session_id,
            **fields,
        )
        return True
    except Exception as exc:
        self.logger.warning(
            "update_thread_metadata failed for session %s: %s", session_id, exc
        )
        return False

Fix update_conversation_title() — add warning when ids are missing. Find:

    async def update_conversation_title(
        self,
        session_id: str,
        title: str,
        user_id: Optional[str] = None,
        agent_id: Optional[str] = None,
    ) -> bool:
        if not self._dynamo or not user_id or not agent_id:
            return False

Replace with:

    async def update_conversation_title(
        self,
        session_id: str,
        title: str,
        user_id: Optional[str] = None,
        agent_id: Optional[str] = None,
    ) -> bool:
        if not self._dynamo:
            return False
        if not user_id or not agent_id:
            self.logger.warning(
                "update_conversation_title requires user_id and agent_id for "
                "DynamoDB PK construction (session: %s) — skipping cold storage update",
                session_id,
            )
            return False

Fix delete_turn() — replace _conv_table direct access with self._dynamo.delete_turn(). Find and replace the body of delete_turn():

async def delete_turn(
    self,
    session_id: str,
    turn_id: str,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
) -> bool:
    """Delete a single turn from DynamoDB.

    Note: DynamoDB requires user_id and agent_id to build the PK.
    Callers that previously omitted these will now receive a False
    return and a warning log.

    Returns:
        True if deletion succeeded.
    """
    if not self._dynamo:
        return False
    if not user_id or not agent_id:
        self.logger.warning(
            "delete_turn requires user_id and agent_id for DynamoDB PK "
            "construction (session: %s turn: %s) — skipping",
            session_id, turn_id,
        )
        return False
    try:
        await self._dynamo.delete_turn(
            user_id=user_id,
            agent_id=agent_id,
            session_id=session_id,
            turn_id=turn_id,
        )
        # Update thread metadata to reflect the deletion
        await self._dynamo.update_thread(
            user_id=user_id,
            agent_id=agent_id,
            session_id=session_id,
            updated_at=datetime.now(timezone.utc),
        )
        self.logger.debug("Deleted turn %s from session %s", turn_id, session_id)
        return True
    except Exception as exc:
        self.logger.warning(
            "delete_turn failed for %s in %s: %s", turn_id, session_id, exc
        )
        return False

Add deprecation warning for document_db param in __init__. Find:

        self._docdb = document_db

Replace with:

        if document_db is not None:
            import warnings
            warnings.warn(
                "ChatStorage document_db parameter is deprecated and will be "
                "removed in a future version. DocumentDB has been replaced by "
                "DynamoDB via the dynamodb parameter.",
                DeprecationWarning,
                stacklevel=2,
            )
        self._docdb = document_db

Increment turn_count in _save_to_dynamodb(). Find the update_thread call in _save_to_dynamodb():

            # Upsert thread metadata
            await self._dynamo.update_thread(
                user_id=user_id,
                agent_id=agent_id,
                session_id=session_id,
                updated_at=now,
                last_user_message=user_msg.content[:200],
                last_assistant_message=assistant_msg.content[:200],
                model=assistant_msg.model,
                provider=assistant_msg.provider,
            )

Replace with:

            # Upsert thread metadata and atomically increment turn_count
            await self._dynamo.update_thread(
                user_id=user_id,
                agent_id=agent_id,
                session_id=session_id,
                updated_at=now,
                last_user_message=user_msg.content[:200],
                last_assistant_message=assistant_msg.content[:200],
                model=assistant_msg.model,
                provider=assistant_msg.provider,
                turn_count_increment=1,
            )

Run tests:

pytest tests/storage/test_chat_storage_dynamodb.py -v 2>&1 | tail -30

Expected: All tests pass.

3.4 — Commit¶

Commit:

git add packages/ai-parrot/src/parrot/storage/dynamodb.py \
        packages/ai-parrot/src/parrot/storage/chat.py \
        tests/storage/test_chat_storage_dynamodb.py
git commit -m "fix(storage): chat.py — add update_thread_metadata(), fix delete_turn() encapsulation, add id-missing warnings, document_db deprecation, increment turn_count atomically"

Task 4 — Extract `UserSessionMixin` for handlers¶

Files: - Create: packages/ai-parrot/src/parrot/handlers/_mixins.py - Modify: packages/ai-parrot/src/parrot/handlers/threads.py - Modify: packages/ai-parrot/src/parrot/handlers/artifacts.py

4.1 — Create `_mixins.py`¶

Create packages/ai-parrot/src/parrot/handlers/_mixins.py:

"""Shared handler mixins for AI-Parrot aiohttp views.

FEAT-103: extracted from threads.py and artifacts.py to eliminate
4× duplication of _get_user_id().
"""

from typing import Optional

from navigator_auth.conf import AUTH_SESSION_OBJECT
from navigator_session import get_session


class UserSessionMixin:
    """Mixin that provides ``_get_user_id()`` for aiohttp BaseView subclasses.

    Extracts the authenticated user's ID from ``request.user`` (set by
    ``@is_authenticated`` + ``@user_session`` decorators) with fallback
    to the navigator session object.
    """

    async def _get_user_id(self) -> Optional[str]:
        """Extract user_id from the authenticated session.

        Returns:
            The user ID string, or None if not resolvable.
        """
        # Primary: decorator-populated request.user attribute
        user = getattr(self.request, "user", None)
        if user:
            uid = getattr(user, "user_id", None) or getattr(user, "id", None)
            if uid:
                return str(uid)

        # Fallback: navigator session object
        try:
            session = await get_session(self.request)
        except Exception:
            return None
        if not session:
            return None

        # Try AUTH_SESSION_OBJECT dict first (navigator-auth pattern)
        userinfo = session.get(AUTH_SESSION_OBJECT, {})
        if isinstance(userinfo, dict):
            uid = userinfo.get("user_id")
            if uid:
                return str(uid)

        # Try top-level session key
        uid = session.get("user_id")
        return str(uid) if uid else None

4.2 — Update `threads.py` to use `UserSessionMixin`¶

In threads.py, add the import at the top:
```
from ._mixins import UserSessionMixin
```

Replace class ThreadListView(BaseView): with:

class ThreadListView(UserSessionMixin, BaseView):

Replace class ThreadDetailView(BaseView): with:

class ThreadDetailView(UserSessionMixin, BaseView):

Delete the _get_user_id() method bodies from both ThreadListView and ThreadDetailView (they are now inherited).

Fix the PATCH handler — replace the _dynamo leak with update_thread_metadata(). In ThreadDetailView.patch(), find:

        # For pinned/tags, update directly through DynamoDB if available
        dynamo = getattr(storage, "_dynamo", None)
        if dynamo:
            update_fields = {}
            if "pinned" in body:
                update_fields["pinned"] = bool(body["pinned"])
            if "tags" in body:
                update_fields["tags"] = body["tags"]
            if "archived" in body:
                update_fields["archived"] = bool(body["archived"])
            if update_fields:
                update_fields["updated_at"] = datetime.utcnow()
                await dynamo.update_thread(
                    user_id=user_id,
                    agent_id=agent_id,
                    session_id=session_id,
                    **update_fields,
                )

Replace with:

        # Update pinned/tags/archived via ChatStorage interface (no direct _dynamo access)
        metadata_fields: dict = {}
        if "pinned" in body:
            metadata_fields["pinned"] = bool(body["pinned"])
        if "tags" in body:
            metadata_fields["tags"] = body["tags"]
        if "archived" in body:
            metadata_fields["archived"] = bool(body["archived"])
        if metadata_fields:
            from datetime import datetime, timezone
            metadata_fields["updated_at"] = datetime.now(timezone.utc)
            await storage.update_thread_metadata(
                user_id=user_id,
                session_id=session_id,
                agent_id=agent_id,
                **metadata_fields,
            )

Fix agent_id default in DELETE — in ThreadDetailView.delete(), find:

        qs = self.get_arguments(self.request)
        agent_id = qs.get("agent_id", "")

Replace with:

        qs = self.get_arguments(self.request)
        agent_id = qs.get("agent_id")
        if not agent_id:
            return self.error(
                response={"message": "agent_id query parameter is required"},
                status=400,
            )

4.3 — Update `artifacts.py` to use `UserSessionMixin`¶

In artifacts.py, add the import at the top:
```
from ._mixins import UserSessionMixin
```

Replace both view class declarations:

class ArtifactListView(UserSessionMixin, BaseView):
class ArtifactDetailView(UserSessionMixin, BaseView):

Delete the _get_user_id() method bodies from both classes.

Fix agent_id default in ArtifactListView.get() — find:

        qs = self.get_arguments(self.request)
        agent_id = qs.get("agent_id", "")

Replace with:

        qs = self.get_arguments(self.request)
        agent_id = qs.get("agent_id")
        if not agent_id:
            return self.error(
                response={"message": "agent_id query parameter is required"},
                status=400,
            )

Fix agent_id default in ArtifactDetailView.get(), put(), and delete()** — apply the same pattern (3 occurrences, each with agent_id = qs.get("agent_id", "")):

        qs = self.get_arguments(self.request)
        agent_id = qs.get("agent_id")
        if not agent_id:
            return self.error(
                response={"message": "agent_id query parameter is required"},
                status=400,
            )

4.4 — Run handler tests¶

Run:
```
pytest tests/handlers/test_threads.py tests/handlers/test_artifacts.py -v 2>&1 | tail -30
```
Expected: All tests pass (tests already mock agent_id; the 400 case is new and not tested yet — that's fine for now).

4.5 — Commit¶

Commit:

git add packages/ai-parrot/src/parrot/handlers/_mixins.py \
        packages/ai-parrot/src/parrot/handlers/threads.py \
        packages/ai-parrot/src/parrot/handlers/artifacts.py
git commit -m "fix(handlers): extract UserSessionMixin, fix PATCH _dynamo leak via update_thread_metadata(), require agent_id in DELETE/list/detail endpoints"

Task 5 — `agent.py` handler: move inline imports and fix f-string logging¶

Files: - Modify: packages/ai-parrot/src/parrot/handlers/agent.py - Modify: packages/ai-parrot/src/parrot/handlers/infographic.py

5.1 — Fix `agent.py`¶

In agent.py, add these imports to the existing top-level import block (after the existing import uuid and import asyncio lines):

from datetime import datetime, timezone
from parrot.storage.models import Artifact, ArtifactType, ArtifactCreator

Note: uuid and asyncio are already imported at the top (lines 14–15). Do NOT add them again.

In the FEAT-103 auto-save block (around line 1820–1860), remove the inline import lines:

                    from datetime import datetime as _dt, timezone as _tz
                    from parrot.storage.models import (  # noqa: E501 pylint: disable=import-outside-toplevel
                        Artifact,
                        ArtifactType,
                        ArtifactCreator,
                    )
                    import uuid as _uuid

Update the code that used those aliased names to use the top-level ones. Find occurrences of _dt, _tz, _uuid in the block and replace:
_dt.now(_tz.utc) → datetime.now(timezone.utc)
_uuid.uuid4() → uuid.uuid4()

Fix the f-string log at the end of the block. Find:

                self.logger.warning(f"Error scheduling artifact auto-save: {ex}")

Replace with:

                self.logger.warning("Error scheduling artifact auto-save: %s", ex)

5.2 — Fix `infographic.py`¶

In infographic.py, add these imports after the existing from ..helpers.infographics import ( block:

from ..storage.models import Artifact, ArtifactType, ArtifactCreator

In _auto_save_infographic_artifact(), remove the inline import block:

        try:
            from ..storage.models import (  # noqa: E501 pylint: disable=import-outside-toplevel
                Artifact,
                ArtifactType,
                ArtifactCreator,
            )

The try block itself stays but now starts at the now = datetime.now(timezone.utc) line.

Run a quick import check to verify no circular imports:

cd .claude/worktrees/feat-103-agent-artifact-persistency
source .venv/bin/activate 2>/dev/null || true
python -c "from parrot.handlers.agent import AgentTalk; print('agent ok')" 2>&1
python -c "from parrot.handlers.infographic import InfographicTalk; print('infographic ok')" 2>&1

Expected: both print ok.

5.3 — Run auto-save tests¶

Run:

pytest tests/handlers/test_auto_save.py -v 2>&1 | tail -20

Expected: All tests pass.

5.4 — Commit¶

Commit:

git add packages/ai-parrot/src/parrot/handlers/agent.py \
        packages/ai-parrot/src/parrot/handlers/infographic.py
git commit -m "fix(handlers): move inline Artifact/datetime imports to module top in agent.py and infographic.py; fix f-string log"

Task 6 — Fix integration test fragile `asyncio.sleep`¶

Files: - Modify: tests/storage/test_integration_artifact_persistence.py

6.1 — Fix the sleep¶

In test_integration_artifact_persistence.py, find TestConversationLifecycle.test_save_turn_and_fire_dynamodb:

    @pytest.mark.asyncio
    async def test_save_turn_and_fire_dynamodb(self, chat_storage, mock_dynamo):
        turn_id = await chat_storage.save_turn(
            user_id="u1", session_id="sess1", agent_id="bot1",
            user_message="What are Q4 sales?",
            assistant_response="Q4 sales were $2.5M",
        )
        assert turn_id is not None
        # Allow background task to run
        await asyncio.sleep(0.05)
        mock_dynamo.put_turn.assert_called_once()

Replace with:

    @pytest.mark.asyncio
    async def test_save_turn_writes_to_dynamodb(self, chat_storage, mock_dynamo):
        """Verify _save_to_dynamodb writes the correct turn data.

        Calls _save_to_dynamodb directly instead of racing asyncio.create_task.
        """
        from datetime import datetime
        from parrot.storage.models import ChatMessage, MessageRole

        now = datetime.now()
        turn_id = "test-turn-001"

        user_msg = ChatMessage(
            message_id=f"{turn_id}_user",
            session_id="sess1",
            user_id="u1",
            agent_id="bot1",
            role=MessageRole.USER.value,
            content="What are Q4 sales?",
            timestamp=now,
        )
        assistant_msg = ChatMessage(
            message_id=f"{turn_id}_assistant",
            session_id="sess1",
            user_id="u1",
            agent_id="bot1",
            role=MessageRole.ASSISTANT.value,
            content="Q4 sales were $2.5M",
            timestamp=now,
            tool_calls=[],
            sources=[],
            metadata={},
        )

        await chat_storage._save_to_dynamodb(user_msg, assistant_msg, "bot1", now)

        mock_dynamo.put_turn.assert_called_once()
        call_kwargs = mock_dynamo.put_turn.call_args.kwargs
        assert call_kwargs["user_id"] == "u1"
        assert call_kwargs["session_id"] == "sess1"
        assert call_kwargs["data"]["user_message"] == "What are Q4 sales?"
        assert call_kwargs["data"]["assistant_response"] == "Q4 sales were $2.5M"

Run:

pytest tests/storage/test_integration_artifact_persistence.py -v 2>&1 | tail -30

Expected: All tests pass.

6.2 — Commit¶

Commit:

git add tests/storage/test_integration_artifact_persistence.py
git commit -m "fix(tests): replace fragile asyncio.sleep(0.05) with direct _save_to_dynamodb() call in integration test"

Task 7 — Fill in SDD task completion notes¶

Files: All 10 task files in sdd/tasks/completed/

For each task file below, replace *(Agent fills this in when done)* with a summary of what was done (use the template below). Then commit all 10 at once.

Template:

### Completion Note
Implemented as specified. [Any deviations or noteworthy decisions].
All acceptance criteria met. Tests pass: `pytest tests/storage/ -v`.

TASK-717 (sdd/tasks/completed/TASK-717-artifact-thread-models.md):

### Completion Note
Implemented all 8 Pydantic models/enums in `parrot/storage/models.py` as specified.
Existing `ChatMessage` and `Conversation` dataclasses left unchanged.
All acceptance criteria met. Tests pass: `pytest tests/storage/test_artifact_models.py -v`.

TASK-718 (sdd/tasks/completed/TASK-718-dynamodb-backend.md):

### Completion Note
Implemented `ConversationDynamoDB` in `parrot/storage/dynamodb.py`.
DEVIATION: Used `aioboto3` directly instead of `asyncdb("dynamodb")` because
asyncdb 2.15.0 does not ship a DynamoDB driver. Justification comment added to
the module docstring. All domain methods implemented with graceful degradation,
TTL setting, and pagination. Tests pass: `pytest tests/storage/test_dynamodb_backend.py -v`.

TASK-719 (sdd/tasks/completed/TASK-719-s3-overflow-manager.md):

### Completion Note
Implemented `S3OverflowManager` in `parrot/storage/s3_overflow.py`.
`resolve()` uses `io.BytesIO` with `S3FileManager.download_file()`'s `BinaryIO`
overload (confirmed: the method signature is `destination: Path | BinaryIO`).
All acceptance criteria met. Tests pass: `pytest tests/storage/test_s3_overflow.py -v`.

TASK-720 (sdd/tasks/completed/TASK-720-artifact-store.md):

### Completion Note
Implemented `ArtifactStore` in `parrot/storage/artifacts.py`.
Post-review fixes applied: `datetime.utcnow()` replaced with `datetime.now(timezone.utc)`,
TTL recalculated on `update_artifact()`, enum guard added in `_deserialize()`.
All acceptance criteria met. Tests pass: `pytest tests/storage/test_artifact_store.py -v`.

TASK-721 (sdd/tasks/completed/TASK-721-configuration.md):

### Completion Note
Added DYNAMODB_CONVERSATIONS_TABLE, DYNAMODB_ARTIFACTS_TABLE, DYNAMODB_REGION,
DYNAMODB_ENDPOINT_URL, S3_ARTIFACT_BUCKET to `parrot/conf.py` following the
existing navconfig pattern. No existing config variables modified.
All acceptance criteria met.

TASK-722 (sdd/tasks/completed/TASK-722-chatstorage-migration.md):

### Completion Note
Migrated `ChatStorage` from DocumentDB to DynamoDB. Redis hot-cache path unchanged.
Post-review fixes applied: `update_thread_metadata()` added, `delete_turn()`
refactored to use `ConversationDynamoDB.delete_turn()`, deprecation warning added
for `document_db` param, `turn_count` now atomically incremented on each turn save.
Signature changes in `update_conversation_title()` and `delete_turn()` now log
explicit warnings when `user_id`/`agent_id` are missing.
All acceptance criteria met. Tests pass: `pytest tests/storage/test_chat_storage_dynamodb.py -v`.

TASK-723 (sdd/tasks/completed/TASK-723-api-thread-views.md):

### Completion Note
Created `parrot/handlers/threads.py` with ThreadListView and ThreadDetailView.
Post-review fixes applied: `UserSessionMixin` extracted for DRY user-id extraction,
PATCH handler no longer accesses `storage._dynamo` directly (uses `update_thread_metadata()`),
DELETE endpoint now requires `agent_id` query parameter.
All acceptance criteria met. Tests pass: `pytest tests/handlers/test_threads.py -v`.

TASK-724 (sdd/tasks/completed/TASK-724-api-artifact-views.md):

### Completion Note
Created `parrot/handlers/artifacts.py` with ArtifactListView and ArtifactDetailView.
Post-review fixes applied: `UserSessionMixin` used, `agent_id` now required (not
defaulting to empty string) in all 5 endpoints to prevent wrong-partition queries.
All acceptance criteria met. Tests pass: `pytest tests/handlers/test_artifacts.py -v`.

TASK-725 (sdd/tasks/completed/TASK-725-handler-auto-save.md):

### Completion Note
Wired auto-save into `agent.py` and `infographic.py`.
Post-review fixes applied: inline `Artifact`/`datetime` imports moved to module top
in both files, f-string log in `agent.py` replaced with `%s` format.
All acceptance criteria met. Tests pass: `pytest tests/handlers/test_auto_save.py -v`.

TASK-726 (sdd/tasks/completed/TASK-726-integration-tests.md):

### Completion Note
Created comprehensive integration test suite in
`tests/storage/test_integration_artifact_persistence.py`.
Post-review fix: `asyncio.sleep(0.05)` replaced with a direct call to
`_save_to_dynamodb()` for deterministic background-task testing.
All acceptance criteria met. Tests pass:
`pytest tests/storage/test_integration_artifact_persistence.py -v`.

After editing all 10 files, commit:

git add sdd/tasks/completed/
git commit -m "docs(sdd): fill in completion notes for all FEAT-103 tasks (TASK-717 through TASK-726)"

Task 8 — Full test suite verification¶

Run the complete test suite for affected modules:

cd .claude/worktrees/feat-103-agent-artifact-persistency
source .venv/bin/activate 2>/dev/null || true
pytest tests/storage/ tests/handlers/test_threads.py tests/handlers/test_artifacts.py \
       tests/handlers/test_auto_save.py -v 2>&1 | tail -40

Expected: All tests pass, 0 failures.

If any test fails, fix the root cause before proceeding (do not skip or mock away real failures).

FEAT-103 Code-Review Fixes Implementation Plan¶

File Map¶

Task 1 — dynamodb.py cleanup and delete_turn() addition¶

1.1 — Write failing tests for delete_turn() and query_threads() pagination¶

1.2 — Fix dynamodb.py¶

1.3 — Commit¶

Task 2 — artifacts.py correctness fixes¶

2.1 — Write failing tests¶

2.2 — Fix artifacts.py¶

2.3 — Commit¶

Task 3 — chat.py API contract and encapsulation fixes¶

3.1 — Write failing tests¶

3.2 — Update ConversationDynamoDB.update_thread() to support atomic turn_count increment¶

3.3 — Fix chat.py¶

3.4 — Commit¶

Task 4 — Extract UserSessionMixin for handlers¶

4.1 — Create _mixins.py¶

4.2 — Update threads.py to use UserSessionMixin¶

4.3 — Update artifacts.py to use UserSessionMixin¶

4.4 — Run handler tests¶

4.5 — Commit¶

Task 5 — agent.py handler: move inline imports and fix f-string logging¶

5.1 — Fix agent.py¶

5.2 — Fix infographic.py¶

5.3 — Run auto-save tests¶

5.4 — Commit¶

Task 6 — Fix integration test fragile asyncio.sleep¶

6.1 — Fix the sleep¶

6.2 — Commit¶

Task 7 — Fill in SDD task completion notes¶

Task 8 — Full test suite verification¶

Self-Review Checklist¶

Task 1 — `dynamodb.py` cleanup and `delete_turn()` addition¶

1.1 — Write failing tests for `delete_turn()` and `query_threads()` pagination¶

1.2 — Fix `dynamodb.py`¶

Task 2 — `artifacts.py` correctness fixes¶

2.2 — Fix `artifacts.py`¶

Task 3 — `chat.py` API contract and encapsulation fixes¶

3.2 — Update `ConversationDynamoDB.update_thread()` to support atomic turn_count increment¶

3.3 — Fix `chat.py`¶

Task 4 — Extract `UserSessionMixin` for handlers¶

4.1 — Create `_mixins.py`¶

4.2 — Update `threads.py` to use `UserSessionMixin`¶

4.3 — Update `artifacts.py` to use `UserSessionMixin`¶

Task 5 — `agent.py` handler: move inline imports and fix f-string logging¶

5.1 — Fix `agent.py`¶

5.2 — Fix `infographic.py`¶

Task 6 — Fix integration test fragile `asyncio.sleep`¶