src/scribe/services/embeddings.py

@@ -23,6 +23,7 @@ from sqlalchemy import delete, or_, select
 from scribe.models import async_session
 from scribe.models.embedding import NoteEmbedding, RuleEmbedding
 from scribe.models.note import Note
+from scribe.models.task_log import TaskLog
 from scribe.services.access import can_read_project, notes_visibility_clause
 
 if TYPE_CHECKING:  # resolves forward refs without importing at runtime
@@ -271,11 +272,20 @@ def untrigger_title(title: str | None, trigger: str | None) -> str:
 # search. The fix is the document shape: one vector per meaningful chunk, and a
 # record is as findable as its best-matching section.
 
-# Bumped whenever chunk_document's output can change for the same input. Stored
-# on every note_embeddings row so the startup backfill can re-embed exactly the
-# notes whose stored shape is stale — a version comparison instead of the table
-# wipe migrations 0067/0077 had to do.
-CHUNKER_VERSION = 1
+# Bumped whenever THE DOCUMENT A RECORD IS EMBEDDED AS can change for a record
+# that itself has not changed. Stored on every note_embeddings row so the
+# startup backfill re-embeds exactly the stale notes — a version comparison
+# instead of the table wipe migrations 0067/0077 had to do.
+#
+# Stated that way rather than as "chunk_document's output for the same input",
+# which is what it used to say: `chunk_document` is only the last step, and
+# version 2 moves without touching it. A task's document now carries its work
+# logs (#4251), so every task that has one embeds differently than it did while
+# its title, body and the chunker are all untouched — exactly the case the
+# narrower wording would have read as "nothing to re-embed".
+#
+#   1 → 2: work logs joined the task document.
+CHUNKER_VERSION = 2
 
 
 # The public name of the space every score lives in, and the two facts that
@@ -483,6 +493,47 @@ async def _claim_parent_row(session, id_column, row_id: int, label: str) -> bool
     return True
 
 
+async def _work_log_sections(note_id: int) -> list[tuple[object, str | None]]:
+    """A task's work logs, oldest first, for its embedded document (#4251).
+
+    Reads the table directly rather than through `task_logs.logs_for_task`,
+    because that function asks a PERMISSION question — may this user read this
+    task — and there is no user here. An index build acts for the record, and
+    the record's vectors carry the owner's `user_id`, so the access decision is
+    made once at search time by the clause that already scopes every hit.
+
+    That also settles what happens on a shared task: a collaborator's log is
+    part of the task's document, so it becomes findable by everyone who can
+    read the task and by nobody else. The same answer `logs_for_task` gives a
+    reader (#4241), which is the point — a log that can be read and not found
+    is the half-surface that issue was about.
+
+    Asked for every note, not only tasks. `upsert_note_embedding` is handed a
+    note_id and no kind — and the three writers that call it would each have to
+    learn to pass one — so a non-task simply has no rows and gets []. One
+    indexed lookup beside an ONNX forward pass over every chunk is not the
+    expense worth adding a parameter to three call sites for.
+
+    Returns [] on any failure. A task whose logs could not be read should embed
+    as its own prose rather than not embed at all: less findable is recoverable
+    at the next write, unindexed is not.
+    """
+    try:
+        async with async_session() as session:
+            result = await session.execute(
+                select(TaskLog.created_at, TaskLog.content)
+                .where(TaskLog.task_id == note_id)
+                .order_by(TaskLog.created_at.asc(), TaskLog.id.asc())
+            )
+            return list(result.all())
+    except Exception:
+        logger.warning(
+            "Could not read work logs for note %d; embedding its own prose only",
+            note_id, exc_info=True,
+        )
+        return []
+
+
 async def upsert_note_embedding(
     note_id: int, user_id: int, title: str | None, body: str | None
 ) -> None:
@@ -496,6 +547,7 @@ async def upsert_note_embedding(
     inserted in one transaction, so a concurrent read sees the old shape or the
     new one, never a mixture.
     """
+    title, body = task_document(title, body, await _work_log_sections(note_id))
     chunks = chunk_document(title, body)
     try:
         if not chunks:
@@ -908,6 +960,71 @@ async def backfill_note_embeddings() -> None:
 
 # ── Rules (milestone 307, note 3026) ────────────────────────────────────
 
+# The heading a work log gets inside its task's embedded document. A CONSTANT
+# because it is load-bearing twice over: `_split_sections` splits on it, so it
+# is what keeps a log from being merged into the task's own prose, and it is
+# what a reader sees at the top of a matched passage — "this is a log entry,
+# not the task's description". Changing it changes the chunk boundaries of
+# every task that has one, which is a CHUNKER_VERSION move.
+WORK_LOG_HEADING = "## Work log"
+
+
+def task_document(
+    title: str | None,
+    body: str | None,
+    logs: "Sequence[tuple[object, str | None]]" = (),
+) -> tuple[str | None, str | None]:
+    """The (title, body) a TASK is EMBEDDED as — its prose plus its work logs.
+
+    A synthesised embed-time shape, like `rule_document` and unlike a lesson:
+    the stored record is the task row, and the logs live in their own table, so
+    the document that should be searchable exists nowhere until it is built
+    here (#4251).
+
+    WHY THE LOGS BELONG IN THE TASK'S DOCUMENT rather than in rows of their
+    own. "Has anyone tried this before?" is answered by a log and asked of a
+    task — a hit on a bare log would have to be resolved back to its task to be
+    worth anything, so the useful result is the task either way. The objection
+    to folding them in is that a long log drowns a short title, and that was
+    true before #280: one vector per record meant a 2,000-word log averaged the
+    task's own subject away, and everything past ~400 words was truncated
+    unread. Chunking removed both. Each log becomes its own section, each
+    section its own title-anchored vector, each scored separately — so a task
+    is as findable as its best-matching log, and the task's own prose keeps the
+    chunk it always had.
+
+    That the result is LEGIBLE is the other half, and it is this session's
+    other build: a search now hands back the chunk that won (#4243), so a hit
+    earned by a log shows that log's passage under the task's title. Without it
+    the reader would get `body[:240]` of the task — the opening of a record
+    whose relevance lives three hundred lines further down.
+
+    Ordering is oldest-first, matching how the web renders the narrative. Only
+    the heading date distinguishes the sections, so it is part of the shape:
+    "when was this tried" is half of what a log answers.
+
+    An entry with no content is skipped rather than emitted as a bare heading —
+    an empty section is a vector with nothing in it but the task's title, which
+    competes with the task's real chunk and says nothing.
+    """
+    sections = []
+    for created_at, content in logs:
+        text = (content or "").strip()
+        if not text:
+            continue
+        stamp = getattr(created_at, "date", None)
+        heading = (
+            f"{WORK_LOG_HEADING} — {stamp()}" if callable(stamp)
+            else WORK_LOG_HEADING
+        )
+        sections.append(f"{heading}\n\n{text}")
+    if not sections:
+        return title, body
+    prose = (body or "").strip()
+    joined = "\n\n".join(sections)
+    return title, f"{prose}\n\n{joined}" if prose else joined
+
+
 def rule_document(
     title: str | None, statement: str | None, when_to_apply: str | None,
 ) -> tuple[str | None, str | None]:

src/scribe/services/task_logs.py

@@ -14,6 +14,36 @@ logger = logging.getLogger(__name__)
 _UNSET = object()
 
 
+async def _refresh_task_document(session, task_id: int) -> None:
+    """Re-embed the task whose work log just changed (#4251).
+
+    A task's embedded document carries its logs, so a log written and not
+    indexed is #4241's half-surface wearing different clothes: the entry is
+    readable and unfindable, and the next session rebuilds what this one ruled
+    out. Create, edit and delete all go through here — an edited log that keeps
+    matching its old wording is the stale-vector problem `upsert_note_embedding`
+    already refuses to leave behind for a body.
+
+    Loads the NOTE rather than synthesising one. `embed_note` reads
+    `title`, `body` and the OWNER's `user_id` off what it is handed, so a
+    stand-in row would index the logs under an empty title — throwing away the
+    per-chunk topical anchor that makes any of this discriminative — and file
+    the vectors under the wrong user.
+
+    Failure is swallowed the way `embed_note`'s own is: a log that saved must
+    not fail on its index refresh. The startup backfill is the backstop.
+    """
+    from scribe.services.notes import embed_note
+
+    try:
+        result = await session.execute(select(Note).where(Note.id == task_id))
+        note = result.scalars().first()
+        if note is not None:
+            embed_note(note)
+    except Exception:  # noqa: BLE001 - indexing never breaks a write
+        logger.exception("embedding refresh failed for task %s", task_id)
+
+
 async def create_log(
     user_id: int,
     task_id: int,
@@ -36,6 +66,7 @@ async def create_log(
         session.add(log)
         await session.commit()
         await session.refresh(log)
+        await _refresh_task_document(session, task_id)
         return log
 
 
@@ -144,6 +175,7 @@ async def update_log(
         log.updated_at = datetime.now(timezone.utc)
         await session.commit()
         await session.refresh(log)
+        await _refresh_task_document(session, log.task_id)
         return log
 
 
@@ -155,6 +187,8 @@ async def delete_log(user_id: int, log_id: int) -> bool:
         log = result.scalars().first()
         if log is None:
             return False
+        task_id = log.task_id
         await session.delete(log)
         await session.commit()
+        await _refresh_task_document(session, task_id)
         return True

tests/helpers.py

@@ -89,6 +89,24 @@ def make_mock_session() -> AsyncMock:
     return s
 
 
+def session_returning(note):
+    """``make_mock_session`` whose every ``execute()`` yields `note` (or None).
+
+    The commonest shape above the bare session: a service that loads one
+    record by id and acts on it. Two files had spelled this out identically
+    before a third was about to (#3207 — derive before the third copy), and
+    it belongs beside `make_mock_session` for the same reason that one does.
+
+    For a service making SEVERAL different reads, set
+    ``session.execute.side_effect`` to a list of results instead.
+    """
+    session = make_mock_session()
+    result = MagicMock()
+    result.scalars.return_value.first.return_value = note
+    session.execute = AsyncMock(return_value=result)
+    return session
+
+
 async def ensure_user(session, username: str, role: str = "user"):
     """Get-or-create a User by username inside an open session (flushed, not
     committed).

tests/test_chunking.py

@@ -185,11 +185,17 @@ async def test_search_collapses_chunk_rows_to_best_chunk_per_note():
 # --- the write path: one row per chunk (#280 step 3) -------------------------
 
 
-def _session_ctx():
+def _session_ctx(log_rows=()):
+    """A session stand-in. `log_rows` answers the work-log read a task's
+    document now needs (#4251) — answered explicitly rather than left to
+    autovivify, because `list(result.all())` on a bare MagicMock raises and is
+    swallowed, which would quietly make every one of these a no-log test."""
     from unittest.mock import AsyncMock, MagicMock
 
     session = MagicMock()
-    session.execute = AsyncMock()
+    result = MagicMock()
+    result.all.return_value = list(log_rows)
+    session.execute = AsyncMock(return_value=result)
     session.commit = AsyncMock()
     ctx = MagicMock()
     ctx.__aenter__ = AsyncMock(return_value=session)
@@ -226,6 +232,32 @@ async def test_upsert_stores_one_versioned_row_per_chunk():
     session.execute.assert_awaited()  # the delete that makes replacement atomic
 
 
+async def test_a_tasks_work_logs_reach_the_rows_it_is_stored_as():
+    """The wiring half of #4251: the shaper is pure and tested next door, so
+    what is pinned here is that the WRITER actually asks for the logs and
+    embeds what comes back. A log that is written, readable (#4241) and absent
+    from the index is the same half-surface one layer down."""
+    import datetime
+    from unittest.mock import AsyncMock, patch
+
+    from scribe.services import embeddings as emb
+
+    session, ctx = _session_ctx(
+        log_rows=[(datetime.datetime(2026, 9, 20), "ruled out the cache theory")]
+    )
+    with (
+        patch.object(emb, "async_session", return_value=ctx),
+        patch.object(emb, "get_embeddings", AsyncMock(return_value=[[0.0] * 384])),
+    ):
+        await emb.upsert_note_embedding(7, 42, "A task", "Its own prose.")
+
+    rows = [call.args[0] for call in session.add.call_args_list]
+    stored = "\n".join(r.chunk_text for r in rows)
+    assert "ruled out the cache theory" in stored
+    assert "Its own prose." in stored
+    assert emb.WORK_LOG_HEADING in stored
+
+
 async def test_upsert_of_an_emptied_record_clears_rows_instead_of_embedding():
     """An empty embedding is worse than none, and a STALE one is worse than
     that — a record emptied of content must stop being findable by what it no

tests/test_embedding_yields_to_a_delete.py

@@ -32,6 +32,11 @@ def _mock_session(lock_result: object = 7, execute_side_effect=None):
     session = MagicMock()
     claimed = MagicMock()
     claimed.scalar_one_or_none.return_value = lock_result
+    # A task's document carries its work logs (#4251), so the refresh reads
+    # task_logs before it builds chunks. Answered explicitly — left to
+    # autovivify, `list(result.all())` would raise and be swallowed, and these
+    # tests would be exercising the failure path without saying so.
+    claimed.all.return_value = []
     if execute_side_effect is not None:
         session.execute = AsyncMock(side_effect=execute_side_effect)
     else:
@@ -65,10 +70,19 @@ async def test_a_note_refresh_claims_the_row_before_rewriting_its_vectors():
     ):
         await emb.upsert_note_embedding(7, 42, "T", "a short body")
 
-    claim, replace = [c.args[0] for c in session.execute.call_args_list][:2]
-    assert compiled_sql(claim, dialect=PG).startswith("SELECT notes.id")
-    assert "FOR KEY SHARE NOWAIT" in compiled_sql(claim, dialect=PG)
-    assert compiled_sql(replace, dialect=PG).startswith("DELETE FROM note_embeddings")
+    sqls = [compiled_sql(c.args[0], dialect=PG) for c in session.execute.call_args_list]
+    # Located by what they ARE rather than by position: the work-log read a
+    # task's document needs (#4251) runs before any of this, and an index is
+    # not what the claim is about.
+    claim = next(i for i, sql in enumerate(sqls) if "FOR KEY SHARE NOWAIT" in sql)
+    replace = next(
+        i for i, sql in enumerate(sqls) if sql.startswith("DELETE FROM note_embeddings")
+    )
+    assert sqls[claim].startswith("SELECT notes.id")
+    assert claim < replace, "the claim goes first — that is the whole fix"
+    assert not any("note_embeddings" in sql for sql in sqls[:claim]), (
+        "nothing may touch the chunk rows before the parent is claimed"
+    )
     session.add.assert_called()
 
 
@@ -113,6 +127,8 @@ async def test_a_record_already_gone_is_not_re_embedded():
     ):
         await emb.upsert_note_embedding(7, 42, "T", "a short body")
 
-    assert session.execute.await_count == 1
+    sqls = [compiled_sql(c.args[0], dialect=PG) for c in session.execute.call_args_list]
+    assert any("FOR KEY SHARE NOWAIT" in sql for sql in sqls), "it did reach the claim"
+    assert not any("note_embeddings" in sql for sql in sqls), "and stopped there"
     session.add.assert_not_called()
     session.commit.assert_not_awaited()

tests/test_services_dedup.py

@@ -14,23 +14,14 @@ from scribe.services.dedup import (
     plan_candidate_text,
     plan_match_response,
 )
-from tests.helpers import fake_note, make_mock_session
-
-
-def _session_returning(note):
-    """A mocked async_session() whose single execute() yields `note` (or None)."""
-    s = make_mock_session()
-    result = MagicMock()
-    result.scalars.return_value.first.return_value = note
-    s.execute = AsyncMock(return_value=result)
-    return s
+from tests.helpers import fake_note, make_mock_session, session_returning
 
 
 @pytest.mark.asyncio
 async def test_title_exact_match_returns_title_duplicate():
     note = fake_note(id=10, title="Setup CI")
     with patch("scribe.services.dedup.async_session",
-               return_value=_session_returning(note)):
+               return_value=session_returning(note)):
         # whitespace/case differences are normalized away
         dup = await find_duplicate_note(7, "  setup   ci ", project_id=2, is_task=True)
     assert dup is not None
@@ -43,7 +34,7 @@ async def test_title_exact_match_returns_title_duplicate():
 async def test_short_body_skips_semantic_check():
     sem = AsyncMock()
     with patch("scribe.services.dedup.async_session",
-               return_value=_session_returning(None)), \
+               return_value=session_returning(None)), \
          patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
         dup = await find_duplicate_note(7, "Unique", body="too short", project_id=2)
     assert dup is None
@@ -55,7 +46,7 @@ async def test_semantic_match_when_body_substantial():
     hit = fake_note(id=20, title="Existing", note_type="note")
     sem = AsyncMock(return_value=[(0.93, hit)])
     with patch("scribe.services.dedup.async_session",
-               return_value=_session_returning(None)), \
+               return_value=session_returning(None)), \
          patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
         dup = await find_duplicate_note(
             7, "Title", body="x" * 250, project_id=2, is_task=False, note_type="note",
@@ -84,7 +75,7 @@ async def test_gate_catches_a_duplicate_hiding_in_a_later_chunk():
     # Every chunk misses except the LAST one the gate will ask about.
     sem = AsyncMock(side_effect=[[] for _ in range(n_chunks - 1)] + [[(0.94, hit)]])
     with patch("scribe.services.dedup.async_session",
-               return_value=_session_returning(None)), \
+               return_value=session_returning(None)), \
          patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
         dup = await find_duplicate_note(
             7, "Title", body=body, project_id=2, is_task=False, note_type="note",
@@ -98,7 +89,7 @@ async def test_semantic_match_of_other_note_type_is_ignored():
     other = fake_note(id=21, title="X", note_type="process")
     sem = AsyncMock(return_value=[(0.97, other)])
     with patch("scribe.services.dedup.async_session",
-               return_value=_session_returning(None)), \
+               return_value=session_returning(None)), \
          patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
         dup = await find_duplicate_note(7, "Title", body="x" * 250, note_type="note")
     assert dup is None  # type mismatch must not block
@@ -108,7 +99,7 @@ async def test_semantic_match_of_other_note_type_is_ignored():
 async def test_rule_title_match_in_topic():
     rule = fake_note(id=47, title="Honor the multi-user sharing ACL")
     with patch("scribe.services.dedup.async_session",
-               return_value=_session_returning(rule)):
+               return_value=session_returning(rule)):
         dup = await find_duplicate_rule(
             "honor the   multi-user sharing acl", topic_id=7,
         )
@@ -361,7 +352,7 @@ async def test_plan_title_match_short_circuits_the_semantic_arm():
     ms = MagicMock(id=415, title="Plan gate")
     sem = AsyncMock()
     with patch("scribe.services.dedup.can_read_project", AsyncMock(return_value=True)), \
-         patch("scribe.services.dedup.async_session", return_value=_session_returning(ms)), \
+         patch("scribe.services.dedup.async_session", return_value=session_returning(ms)), \
          patch("scribe.services.dedup.embeddings_svc.semantic_search_milestones", sem):
         dup = await find_matching_plan(7, 2, " plan  GATE", "x" * 300)
     assert (dup.id, dup.reason, dup.similarity) == (415, "title", 1.0)
@@ -373,7 +364,7 @@ async def test_plan_semantic_arm_asks_for_active_plans_in_the_project_at_the_set
     ms = MagicMock(id=9, title="Metadata")
     sem = AsyncMock(return_value=[(0.912345, ms)])
     with patch("scribe.services.dedup.can_read_project", AsyncMock(return_value=True)), \
-         patch("scribe.services.dedup.async_session", return_value=_session_returning(None)), \
+         patch("scribe.services.dedup.async_session", return_value=session_returning(None)), \
          patch("scribe.services.dedup.embeddings_svc.semantic_search_milestones", sem), \
          patch("scribe.services.settings.get_setting", AsyncMock(return_value="0.8")):
         dup = await find_matching_plan(7, 2, "Book metadata", "x" * 300)
@@ -396,7 +387,7 @@ async def test_plan_gate_fails_open():
 @pytest.mark.asyncio
 async def test_plan_gate_says_nothing_about_a_project_the_caller_cannot_read():
     ms = MagicMock(id=415, title="Their plan")
-    session = MagicMock(return_value=_session_returning(ms))
+    session = MagicMock(return_value=session_returning(ms))
     with patch("scribe.services.dedup.can_read_project", AsyncMock(return_value=False)), \
          patch("scribe.services.dedup.async_session", session):
         assert await find_matching_plan(8, 2, "Their plan") is None

tests/test_task_document_shape.py

@@ -0,0 +1,200 @@
+"""task_document — the document a TASK is embedded as, work logs included (#4251).
+
+A work log is the richest record Scribe holds of *why* something is the way it
+is: prose written during the work, saying what was tried and ruled out. #4241
+made it readable. It was still not findable, so "has anyone tried this
+approach?" — the question a log answers — could not reach one, and #4208 was
+rebuilt in this very session because its logs were unreachable.
+
+These tests pin the shape and the two properties the decision rested on: a task
+without logs embeds EXACTLY as it always did, and a task with them stays as
+discriminative as it was, because each log is its own title-anchored chunk
+rather than prose averaged into one vector.
+"""
+import datetime
+
+from scribe.services.embeddings import (
+    WORK_LOG_HEADING,
+    chunk_document,
+    embedding_text,
+    task_document,
+)
+
+D1 = datetime.datetime(2026, 9, 20, 10, 0)
+D2 = datetime.datetime(2026, 9, 21, 11, 0)
+
+
+# --- the identity half: a task with no logs is untouched ---------------------
+
+
+def test_a_task_with_no_logs_embeds_exactly_as_it_did_before():
+    """Most notes are not tasks and most tasks carry no log. Their vectors are
+    the corpus this instance's thresholds were measured against, and changing
+    them for nothing would move every tuned number underneath itself (#4225)."""
+    assert task_document("A title", "A body") == ("A title", "A body")
+    assert task_document("A title", "A body", []) == ("A title", "A body")
+    assert task_document("T", None) == ("T", None)
+
+
+def test_an_entry_with_no_content_is_skipped_rather_than_emitted_empty():
+    """A bare heading is a vector containing nothing but the task's title — it
+    would compete with the task's real chunk and say nothing."""
+    _t, body = task_document("T", "prose", [(D1, "   "), (D2, None), (D1, "real")])
+    assert body.count(WORK_LOG_HEADING) == 1
+    assert "real" in body
+
+
+def test_a_task_that_is_only_logs_still_yields_a_document():
+    """A task opened with a title and filled in entirely through its log — the
+    shape every step of a milestone starts as."""
+    title, body = task_document("T", "", [(D1, "the only content")])
+    assert title == "T"
+    assert body.startswith(WORK_LOG_HEADING)
+    assert "the only content" in body
+
+
+# --- the shape: the logs are sections, oldest first --------------------------
+
+
+def test_logs_are_appended_as_dated_sections_after_the_tasks_own_prose():
+    title, body = task_document(
+        "Fix the thing", "The task prose.",
+        [(D1, "Tried X, ruled out."), (D2, "Y worked.")],
+    )
+    assert title == "Fix the thing"
+    assert body.index("The task prose.") < body.index("Tried X") < body.index("Y worked.")
+    assert f"{WORK_LOG_HEADING} — 2026-09-20" in body
+    assert f"{WORK_LOG_HEADING} — 2026-09-21" in body
+
+
+def test_an_entry_with_no_timestamp_still_gets_its_own_section():
+    """Degrades to an undated heading rather than dropping the entry or
+    crashing — a restored row or a hand-built one is not a reason to lose the
+    richest prose in the record."""
+    _t, body = task_document("T", "p", [(None, "content from nowhere")])
+    assert WORK_LOG_HEADING in body
+    assert "content from nowhere" in body
+
+
+# --- why folding them in is safe: chunking, not averaging --------------------
+
+
+def test_each_log_becomes_its_own_title_anchored_chunk():
+    """THE decision this build turns on. The objection to putting logs in the
+    task's document is that a long log drowns a short title — true before #280,
+    when one vector per record meant a 2,000-word log averaged the task's
+    subject away and everything past ~400 words was truncated unread. Chunking
+    answers both: separate vectors, each carrying the title as its anchor."""
+    def _entry(subject: str) -> str:
+        para = " ".join([f"This log entry discusses the {subject} in detail."] * 8)
+        return "\n\n".join(f"{para} (p{i})" for i in range(6))
+
+    title, body = task_document(
+        "Short task title", "Short body.",
+        [(D1, _entry("approach")), (D2, _entry("alternative"))],
+    )
+    chunks = chunk_document(title, body)
+    assert len(chunks) > 1, "a long log must split rather than truncate"
+    for chunk in chunks:
+        assert chunk.startswith("Short task title\n")
+
+    # The task's own prose still has a chunk of its own — it is not merged into
+    # a log section and scored as part of it.
+    assert any("Short body." in c and "alternative" not in c for c in chunks)
+
+    # And nothing is lost: this is what #280 exists for. Paragraph-shaped,
+    # because that is what a log is and because a single unbroken 3,000-char
+    # line is hard-split mid-line by design — `test_chunking` owns that case.
+    joined = "\n".join(chunks)
+    for line in body.splitlines():
+        if line.strip():
+            assert line.strip() in joined, f"content dropped: {line[:60]!r}"
+
+
+def test_a_short_task_with_a_short_log_is_still_one_sharp_chunk():
+    """Over-splitting is not the goal either. A task and one short log stay a
+    single document, the shape note #2485 measured as the sharpest."""
+    title, body = task_document("T", "body", [(D1, "a short note about it")])
+    assert chunk_document(title, body) == [embedding_text(title, body)]
+
+
+# --- keeping it current: a log write refreshes the task's vectors ------------
+#
+# The shaper above is pure. These pin that the writers actually call it — a
+# task whose logs are in the document but never re-indexed after one is written
+# is #4241's half-surface one layer down: the entry is readable, and the search
+# still answers as though it were never written.
+
+from unittest.mock import MagicMock, patch          # noqa: E402
+
+import pytest                                       # noqa: E402
+
+from scribe.services import task_logs as svc        # noqa: E402
+from tests.helpers import (                      # noqa: E402
+    fake_note, make_mock_session, session_returning,
+)
+
+
+@pytest.mark.asyncio
+async def test_writing_a_work_log_refreshes_the_tasks_embedding():
+    note = fake_note(id=7, title="A task", body="prose", is_task=True)
+    session = session_returning(note)
+    with (
+        patch.object(svc, "async_session", return_value=session),
+        patch("scribe.services.notes.embed_note") as embed,
+    ):
+        await svc.create_log(42, 7, "what I tried")
+    embed.assert_called_once()
+    assert embed.call_args.args[0] is note
+
+
+@pytest.mark.asyncio
+async def test_editing_a_work_log_refreshes_it_too():
+    """An edited log whose vectors still carry the old wording keeps matching
+    what it no longer says — the stale-vector case `upsert_note_embedding`
+    already refuses to leave behind for a body."""
+    note = fake_note(id=7, title="A task", body="prose", is_task=True)
+    log = MagicMock(id=3, task_id=7)
+    session = make_mock_session()
+    found_log, found_note = MagicMock(), MagicMock()
+    found_log.scalars.return_value.first.return_value = log
+    found_note.scalars.return_value.first.return_value = note
+    session.execute.side_effect = [found_log, found_note]
+    with (
+        patch.object(svc, "async_session", return_value=session),
+        patch("scribe.services.notes.embed_note") as embed,
+    ):
+        await svc.update_log(42, 3, content="reworded")
+    embed.assert_called_once_with(note)
+
+
+@pytest.mark.asyncio
+async def test_deleting_a_work_log_refreshes_it_too():
+    """And reads the task id BEFORE the row goes — after the delete there is
+    nothing left to ask which task it belonged to."""
+    note = fake_note(id=7, title="A task", body="prose", is_task=True)
+    log = MagicMock(id=3, task_id=7)
+    session = make_mock_session()
+    found_log, found_note = MagicMock(), MagicMock()
+    found_log.scalars.return_value.first.return_value = log
+    found_note.scalars.return_value.first.return_value = note
+    session.execute.side_effect = [found_log, found_note]
+    with (
+        patch.object(svc, "async_session", return_value=session),
+        patch("scribe.services.notes.embed_note") as embed,
+    ):
+        assert await svc.delete_log(42, 3) is True
+    embed.assert_called_once_with(note)
+
+
+@pytest.mark.asyncio
+async def test_a_failed_refresh_does_not_fail_the_log_that_saved():
+    """Indexing never breaks a write. The startup backfill is the backstop."""
+    session = session_returning(fake_note(id=7, title="T", body="b", is_task=True))
+    with (
+        patch.object(svc, "async_session", return_value=session),
+        patch("scribe.services.notes.embed_note", side_effect=RuntimeError("boom")),
+    ):
+        log = await svc.create_log(42, 7, "what I tried")
+    assert log is not None
+    session.commit.assert_awaited()