feat(consolidation): full consolidate_task with background model
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.
This commit is contained in:
@@ -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,
|
||||
)
|
||||
|
||||
+210
-1
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user