diff --git a/src/fabledassistant/services/consolidation.py b/src/fabledassistant/services/consolidation.py index 183f066..ee77b5e 100644 --- a/src/fabledassistant/services/consolidation.py +++ b/src/fabledassistant/services/consolidation.py @@ -81,6 +81,126 @@ async def maybe_consolidate(user_id: int, task_id: int, *, reason: str) -> None: asyncio.create_task(consolidate_task(user_id, task_id)) +def _build_consolidation_prompt( + *, title: str, description: str | None, logs: list, +) -> str: + """Build the LLM prompt for one consolidation pass. + + Caps total log content at MAX_PROMPT_INPUT_CHARS; logs that don't fit + are dropped. Caller is expected to slice to the most-recent window + before calling. + """ + log_lines: list[str] = [] + chars = 0 + for log in logs: + ts = log.created_at.isoformat() if getattr(log, "created_at", None) else "?" + line = f"- [{ts}] {log.content}" + if chars + len(line) > MAX_PROMPT_INPUT_CHARS: + break + log_lines.append(line) + chars += len(line) + + return ( + "You are summarizing the work done on a task. The user wrote the goal " + "below; do not restate it. Read the chronological work-log entries and " + "produce a 1-3 paragraph summary of: what was attempted, what worked, " + "what failed, what's current state. Use the user's voice; cite specific " + "commands/decisions; favor brevity over completeness. Output plain " + "markdown body content only — no preamble.\n\n" + f"TITLE: {title}\n" + f"GOAL (read-only context): {description or '(no goal recorded)'}\n" + f"WORK LOG (chronological):\n" + "\n".join(log_lines) + ) + + async def consolidate_task(user_id: int, task_id: int) -> None: - """Stub — full implementation in the next task in the plan.""" - raise NotImplementedError + """Run a consolidation pass: read description + logs, write summary to body. + + Errors are logged and swallowed so the fire-and-forget caller is never + interrupted; on LLM failure the body and consolidated_at are left + untouched and the next trigger retries. + """ + lock = _locks[task_id] + if lock.locked(): + logger.debug( + "consolidate_task: skipping — already running for task %d", task_id + ) + return + async with lock: + try: + async with async_session() as session: + task = ( + await session.execute( + select(Note).where( + Note.id == task_id, Note.user_id == user_id, + ) + ) + ).scalars().first() + if task is None or not task.status: + return # not a task, or missing + logs = ( + await session.execute( + select(TaskLog) + .where( + TaskLog.task_id == task_id, + TaskLog.user_id == user_id, + ) + .order_by(TaskLog.created_at.asc()) + ) + ).scalars().all() + + if not logs: + return # nothing to summarize yet + + title = task.title or "" + description = task.description + window = logs[-MAX_LOGS_FOR_PROMPT:] + + prompt = _build_consolidation_prompt( + title=title, description=description, logs=window, + ) + + from fabledassistant.services.llm import generate_completion + from fabledassistant.services.settings import get_setting + from fabledassistant.config import Config + + bg_model = await get_setting( + user_id, "background_model", Config.OLLAMA_BACKGROUND_MODEL, + ) + summary = await generate_completion( + [{"role": "user", "content": prompt}], + model=bg_model, + max_tokens=800, + num_ctx=4096, + ) + if not summary or not summary.strip(): + return + + now = datetime.now(timezone.utc) + async with async_session() as session: + task = ( + await session.execute( + select(Note).where( + Note.id == task_id, Note.user_id == user_id, + ) + ) + ).scalars().first() + if task is None: + return + task.body = summary.strip() + task.consolidated_at = now + task.updated_at = now + await session.commit() + + from fabledassistant.services.embeddings import upsert_note_embedding + await upsert_note_embedding( + task_id, user_id, f"{title}\n{summary.strip()}".strip(), + ) + logger.info( + "consolidate_task: refreshed task %d body (%d chars)", + task_id, len(summary), + ) + except Exception: + logger.exception( + "consolidate_task failed for task %d", task_id, + ) diff --git a/tests/test_consolidation.py b/tests/test_consolidation.py index fe4766f..fb86782 100644 --- a/tests/test_consolidation.py +++ b/tests/test_consolidation.py @@ -2,7 +2,9 @@ Design: docs/superpowers/specs/2026-05-13-task-as-durable-record-design.md """ -from unittest.mock import AsyncMock, patch +from datetime import datetime, timezone +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch # ── Gate logic ─────────────────────────────────────────────────────────────── @@ -106,3 +108,210 @@ async def test_maybe_consolidate_unknown_reason_is_noop(): ) mock_create_task.assert_not_called() + + +# ── Prompt builder ─────────────────────────────────────────────────────────── + + +def test_build_consolidation_prompt_includes_title_goal_and_logs(): + from fabledassistant.services.consolidation import _build_consolidation_prompt + + logs = [ + SimpleNamespace( + content="tried REJECT, logs noisy", + created_at=datetime(2026, 5, 13, 10, 0, tzinfo=timezone.utc), + ), + SimpleNamespace( + content="switched to DROP", + created_at=datetime(2026, 5, 13, 11, 0, tzinfo=timezone.utc), + ), + ] + prompt = _build_consolidation_prompt( + title="firewall tuning", description="quiet the logs", logs=logs, + ) + assert "firewall tuning" in prompt + assert "quiet the logs" in prompt + assert "tried REJECT" in prompt + assert "switched to DROP" in prompt + + +def test_build_consolidation_prompt_handles_missing_description(): + from fabledassistant.services.consolidation import _build_consolidation_prompt + + logs = [ + SimpleNamespace( + content="a", created_at=datetime(2026, 5, 13, tzinfo=timezone.utc), + ), + ] + prompt = _build_consolidation_prompt(title="t", description=None, logs=logs) + assert "no goal recorded" in prompt + + +def test_build_consolidation_prompt_truncates_to_char_budget(): + from fabledassistant.services.consolidation import ( + _build_consolidation_prompt, MAX_PROMPT_INPUT_CHARS, + ) + + big = "x" * (MAX_PROMPT_INPUT_CHARS + 1000) + logs = [ + SimpleNamespace( + content=big, created_at=datetime(2026, 5, 13, tzinfo=timezone.utc), + ), + SimpleNamespace( + content="should be dropped", + created_at=datetime(2026, 5, 13, 1, tzinfo=timezone.utc), + ), + ] + prompt = _build_consolidation_prompt(title="t", description=None, logs=logs) + # First log consumes the budget; the second is dropped. + assert "should be dropped" not in prompt + + +# ── consolidate_task orchestration (mocked DB + LLM + embedding) ───────────── + + +def _make_mock_session(task, logs): + """Return a mock async_session that hands out task and logs to successive + .execute() calls. consolidate_task calls execute three times: select task, + select logs (same context), then re-select task in the write-back block.""" + mock_session = AsyncMock() + + task_result = MagicMock() + task_result.scalars.return_value.first.return_value = task + logs_result = MagicMock() + logs_result.scalars.return_value.all.return_value = logs + + mock_session.execute = AsyncMock( + side_effect=[task_result, logs_result, task_result], + ) + mock_session.commit = AsyncMock() + mock_session.refresh = AsyncMock() + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=False) + return mock_session + + +async def test_consolidate_task_writes_body_and_timestamp_and_reembeds(): + task = MagicMock() + task.id = 42 + task.title = "Cert renewal" + task.description = "renew LE before Nov 30" + task.status = "in_progress" + task.body = "old body" + task.consolidated_at = None + logs = [ + SimpleNamespace( + content="tried certbot renew", + created_at=datetime(2026, 5, 13, tzinfo=timezone.utc), + ), + SimpleNamespace( + content="switched to manual", + created_at=datetime(2026, 5, 13, 1, tzinfo=timezone.utc), + ), + ] + fake_summary = ( + "Started with certbot renew; DNS split-horizon failed. " + "Switched to manual flow." + ) + + # Each consolidate_task call needs its own per-task lock state. + # _locks is module-level defaultdict — patch it on the module to avoid + # cross-test leakage of the held-lock state. + from collections import defaultdict + import asyncio as _asyncio + + with patch( + "fabledassistant.services.consolidation._locks", + new=defaultdict(_asyncio.Lock), + ), patch( + "fabledassistant.services.consolidation.async_session", + return_value=_make_mock_session(task, logs), + ), patch( + "fabledassistant.services.llm.generate_completion", + new=AsyncMock(return_value=fake_summary), + ) as mock_llm, patch( + "fabledassistant.services.embeddings.upsert_note_embedding", + new=AsyncMock(), + ) as mock_embed, patch( + "fabledassistant.services.settings.get_setting", + new=AsyncMock(return_value="gemma3:4b"), + ): + from fabledassistant.services.consolidation import consolidate_task + await consolidate_task(1, 42) + + mock_llm.assert_awaited_once() + mock_embed.assert_awaited_once() + assert task.body == fake_summary + assert task.consolidated_at is not None + + +async def test_consolidate_task_llm_failure_leaves_body_untouched(): + task = MagicMock() + task.id = 42 + task.title = "X" + task.description = "y" + task.status = "in_progress" + task.body = "old body content" + task.consolidated_at = None + logs = [ + SimpleNamespace( + content="a", created_at=datetime(2026, 5, 13, tzinfo=timezone.utc), + ), + ] + + from collections import defaultdict + import asyncio as _asyncio + + with patch( + "fabledassistant.services.consolidation._locks", + new=defaultdict(_asyncio.Lock), + ), patch( + "fabledassistant.services.consolidation.async_session", + return_value=_make_mock_session(task, logs), + ), patch( + "fabledassistant.services.llm.generate_completion", + new=AsyncMock(side_effect=RuntimeError("ollama down")), + ), patch( + "fabledassistant.services.embeddings.upsert_note_embedding", + new=AsyncMock(), + ) as mock_embed, patch( + "fabledassistant.services.settings.get_setting", + new=AsyncMock(return_value="gemma3:4b"), + ): + from fabledassistant.services.consolidation import consolidate_task + await consolidate_task(1, 42) # must not raise + + assert task.body == "old body content" + assert task.consolidated_at is None + mock_embed.assert_not_awaited() + + +async def test_consolidate_task_skips_when_no_logs(): + task = MagicMock() + task.id = 42 + task.title = "X" + task.description = "y" + task.status = "in_progress" + task.body = "old" + task.consolidated_at = None + + from collections import defaultdict + import asyncio as _asyncio + + with patch( + "fabledassistant.services.consolidation._locks", + new=defaultdict(_asyncio.Lock), + ), patch( + "fabledassistant.services.consolidation.async_session", + return_value=_make_mock_session(task, []), + ), patch( + "fabledassistant.services.llm.generate_completion", new=AsyncMock(), + ) as mock_llm, patch( + "fabledassistant.services.settings.get_setting", + new=AsyncMock(return_value="gemma3:4b"), + ): + from fabledassistant.services.consolidation import consolidate_task + await consolidate_task(1, 42) + + mock_llm.assert_not_awaited() + assert task.body == "old"