bvandeusen
bda6e6c80f
consolidate_task reads the task title, description (read-only context), and chronological work logs; builds a prompt via _build_consolidation_prompt; calls generate_completion with the user's background_model setting; on a non-empty result, writes back to Note.body, stamps consolidated_at, and re-runs the embedding pipeline. Errors are caught and logged. LLM failures leave body untouched so the next trigger retries cleanly. Per-task asyncio lock prevents simultaneous passes for the same task.
318 lines
11 KiB
Python
318 lines
11 KiB
Python
"""Tests for the task-body consolidation pipeline (gate + full pass).
|
|
|
|
Design: docs/superpowers/specs/2026-05-13-task-as-durable-record-design.md
|
|
"""
|
|
from datetime import datetime, timezone
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
|
|
# ── Gate logic ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
async def test_maybe_consolidate_below_threshold_does_not_fire():
|
|
"""log_added with only 2 logs since last pass → no consolidation."""
|
|
from fabledassistant.services import consolidation
|
|
|
|
with patch.object(
|
|
consolidation, "_logs_since_last_consolidation",
|
|
new=AsyncMock(return_value=2),
|
|
), patch.object(
|
|
consolidation, "consolidate_task", new=AsyncMock(),
|
|
) as mock_consolidate, patch.object(
|
|
consolidation, "_auto_consolidate_enabled", new=AsyncMock(return_value=True),
|
|
):
|
|
await consolidation.maybe_consolidate(
|
|
user_id=1, task_id=42, reason="log_added",
|
|
)
|
|
|
|
mock_consolidate.assert_not_awaited()
|
|
|
|
|
|
async def test_maybe_consolidate_at_threshold_fires():
|
|
"""log_added with N logs since last pass → consolidation scheduled."""
|
|
from fabledassistant.services import consolidation
|
|
|
|
# asyncio.create_task is the actual scheduler; replace it so the test
|
|
# doesn't need an event-loop background task to settle.
|
|
with patch.object(
|
|
consolidation, "_logs_since_last_consolidation",
|
|
new=AsyncMock(return_value=3),
|
|
), patch.object(
|
|
consolidation, "_auto_consolidate_enabled", new=AsyncMock(return_value=True),
|
|
), patch(
|
|
"fabledassistant.services.consolidation.asyncio.create_task",
|
|
) as mock_create_task:
|
|
await consolidation.maybe_consolidate(
|
|
user_id=1, task_id=42, reason="log_added",
|
|
)
|
|
|
|
mock_create_task.assert_called_once()
|
|
|
|
|
|
async def test_maybe_consolidate_task_closed_always_fires():
|
|
"""task_closed bypasses the log-count gate."""
|
|
from fabledassistant.services import consolidation
|
|
|
|
with patch.object(
|
|
consolidation, "_logs_since_last_consolidation",
|
|
new=AsyncMock(return_value=0),
|
|
), patch.object(
|
|
consolidation, "_auto_consolidate_enabled", new=AsyncMock(return_value=True),
|
|
), patch(
|
|
"fabledassistant.services.consolidation.asyncio.create_task",
|
|
) as mock_create_task:
|
|
await consolidation.maybe_consolidate(
|
|
user_id=1, task_id=42, reason="task_closed",
|
|
)
|
|
|
|
mock_create_task.assert_called_once()
|
|
|
|
|
|
async def test_maybe_consolidate_setting_off_blocks_both_reasons():
|
|
"""auto_consolidate_tasks=false → neither trigger fires."""
|
|
from fabledassistant.services import consolidation
|
|
|
|
with patch.object(
|
|
consolidation, "_logs_since_last_consolidation",
|
|
new=AsyncMock(return_value=5),
|
|
), patch.object(
|
|
consolidation, "_auto_consolidate_enabled", new=AsyncMock(return_value=False),
|
|
), patch(
|
|
"fabledassistant.services.consolidation.asyncio.create_task",
|
|
) as mock_create_task:
|
|
await consolidation.maybe_consolidate(
|
|
user_id=1, task_id=42, reason="log_added",
|
|
)
|
|
await consolidation.maybe_consolidate(
|
|
user_id=1, task_id=42, reason="task_closed",
|
|
)
|
|
|
|
mock_create_task.assert_not_called()
|
|
|
|
|
|
async def test_maybe_consolidate_unknown_reason_is_noop():
|
|
"""Unknown reasons get logged and skipped — not raised."""
|
|
from fabledassistant.services import consolidation
|
|
|
|
with patch.object(
|
|
consolidation, "_logs_since_last_consolidation",
|
|
new=AsyncMock(return_value=99),
|
|
), patch.object(
|
|
consolidation, "_auto_consolidate_enabled", new=AsyncMock(return_value=True),
|
|
), patch(
|
|
"fabledassistant.services.consolidation.asyncio.create_task",
|
|
) as mock_create_task:
|
|
await consolidation.maybe_consolidate(
|
|
user_id=1, task_id=42, reason="some_other_reason",
|
|
)
|
|
|
|
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"
|