src/scribe/mcp/tools/tasks.py

@@ -28,6 +28,7 @@ from scribe.services import notes as notes_svc
 # avoid the database would otherwise stub the validation too — turning a
 # guard into a MagicMock that approves anything.
 from scribe.services.notes import minted_kind
+from scribe.services import placement as placement_svc
 from scribe.services import planning as planning_svc
 from scribe.services import record_batch as batch_svc
 from scribe.services import rulebooks as rulebooks_svc
@@ -185,7 +186,7 @@ async def create_task(
 
     Returns the created task, OR — when a near-duplicate is found and force is
     false — {"duplicate": true, "existing_id": ..., "message": ...} (nothing
-    created). A tagged record shows its `systems`; created untagged in a
+    created). A task in a project carries `placement` (see update_task). A tagged record shows its `systems`; created untagged in a
     project, the response carries the `systems_hint` question instead —
     answer it: tag the record, create the missing System, or deliberately
     leave it untagged.
@@ -223,7 +224,7 @@ async def create_task(
         await systems_svc.set_record_systems(uid, note.id, system_ids)
     data = note.to_dict()
     await systems_tools.attach_systems(uid, uid, data, note.id, project_id or None)
-    return data
+    return await placement_svc.attach_placement(uid, data, note)
 
 
 async def update_task(
@@ -262,6 +263,14 @@ async def update_task(
             work that becomes an investigation should say so. 'plan' is
             refused: plans are milestones (start_planning), and the value
             survives only so historical plan-tasks stay writable.
+
+    The response carries `placement` for a task in a project: its `project`,
+    and for a task in a milestone its `milestone`, `position` ({step, of}),
+    `progress` ({completed, total, pct}) and `next` (the next open step, or
+    null). These are the facts to use when telling the operator where the
+    work sits and what comes next — read them from here rather than
+    reconstructing them, because a remembered milestone title or "next step"
+    reads exactly like a real one when it is wrong.
     """
     uid = current_user_id()
     fields: dict = {}
@@ -299,7 +308,7 @@ async def update_task(
     await systems_tools.attach_systems(
         uid, getattr(note, "user_id", uid) or uid, data, task_id, note.project_id
     )
-    return data
+    return await placement_svc.attach_placement(uid, data, note)
 
 
 async def add_task_log(task_id: int, content: str) -> dict:

src/scribe/routes/tasks.py

@@ -6,6 +6,7 @@ from scribe.auth import login_required, get_current_user_id
 from scribe.models.note import TaskPriority, TaskStatus
 from scribe.routes.utils import not_found, parse_iso_date, parse_pagination
 from scribe.services.access import can_write_note
+from scribe.services import placement as placement_svc
 from scribe.services import systems as systems_svc
 from scribe.services.notes import (
     create_note,
@@ -151,6 +152,7 @@ async def create_task_route():
         await systems_svc.set_record_systems(uid, task.id, data["system_ids"])
     out = task.to_dict()
     out["systems"] = [s.to_dict() for s in await systems_svc.list_record_systems(uid, task.id)]
+    await placement_svc.attach_placement(uid, out, task)
     return jsonify(out), 201
 
 
@@ -267,6 +269,7 @@ async def update_task_route(task_id: int):
         await systems_svc.set_record_systems(uid, task_id, data["system_ids"])
     out = task.to_dict()
     out["systems"] = [s.to_dict() for s in await systems_svc.list_record_systems(uid, task_id)]
+    await placement_svc.attach_placement(uid, out, task)
     return jsonify(out)
 
 

src/scribe/services/placement.py

@@ -0,0 +1,145 @@
+"""Where a task sits — its project, its milestone, its step position, what is next.
+
+WHY THIS EXISTS (milestone 409 step 1)
+
+An agent reporting finished work to the operator is asked to place it: which
+milestone, which step of how many, what comes next. Without those facts in
+hand it reconstructs them from memory, and a reconstruction reads exactly like
+the real thing while being wrong — the drafting of the feature note that
+started this milestone invented a milestone title and named a "next" step that
+was already done. So the facts come back on the write that changes a task,
+where the report is about to be written, instead of being left to recall.
+
+THE SHAPE
+
+    {"project":   {"id", "title"},
+     "milestone": {"id", "title", "status"},
+     "position":  {"step": 3, "of": 6},
+     "progress":  {"completed", "total", "pct"},
+     "next":      {"id", "title", "status"} | None}
+
+`milestone`, `position`, `progress` and `next` appear only for a task in a
+milestone; a task with a project and no milestone gets `project` alone; a task
+with neither gets no placement at all (None), which the doors omit rather than
+send empty.
+
+WHAT "STEP" AND "NEXT" MEAN
+
+Notes carry no order column, so a milestone's steps are in CREATION order
+(created_at, then id) — the order a plan's steps are written in, and the order
+a batch create inserts them. Deliberately NOT get_milestone's listing order
+(status, then last update): that is a display choice, and it reshuffles every
+time anything is touched.
+
+`next` is the first open step (todo or in_progress) AFTER this one; when every
+later step is closed it falls back to the earliest open step before it, since
+that is still the milestone's next piece of work. None when nothing else is open.
+
+ACCESS
+
+Siblings are read through access.readable_notes_clause, so a collaborator on a
+shared task is never shown the title of a step they cannot open. Position and
+progress are computed over that same readable set — one list, so the counts can
+never disagree with the titles they sit beside. For an owner the readable set
+is every step, and the numbers equal get_milestone's.
+"""
+from __future__ import annotations
+
+import logging
+
+from sqlalchemy import select
+
+from scribe.models import async_session
+from scribe.models.milestone import Milestone
+from scribe.models.note import Note
+from scribe.models.project import Project
+from scribe.services import access as access_svc
+from scribe.services.milestones import _progress_from_counts
+
+_OPEN = ("todo", "in_progress")
+
+logger = logging.getLogger(__name__)
+
+
+def _next_open(steps: list, current_id: int) -> dict | None:
+    """The next open step after `current_id`, else the earliest open one before it."""
+    index = next((i for i, s in enumerate(steps) if s.id == current_id), -1)
+    later = [s for s in steps[index + 1:] if s.status in _OPEN]
+    earlier = [s for s in steps[:max(index, 0)] if s.status in _OPEN]
+    pick = (later or earlier or [None])[0]
+    if pick is None:
+        return None
+    return {"id": pick.id, "title": pick.title, "status": pick.status}
+
+
+async def task_placement(user_id: int, task) -> dict | None:
+    """Placement for `task` as `user_id` may see it, or None when it has none.
+
+    `task` is the Note the caller already holds — it was just written or read —
+    so its own row is not fetched again.
+    """
+    project_id = getattr(task, "project_id", None)
+    milestone_id = getattr(task, "milestone_id", None)
+    if not project_id and not milestone_id:
+        return None
+
+    async with async_session() as session:
+        milestone = None
+        if milestone_id:
+            milestone = (await session.execute(
+                select(Milestone).where(
+                    Milestone.id == milestone_id, Milestone.deleted_at.is_(None),
+                )
+            )).scalars().first()
+            project_id = project_id or (milestone.project_id if milestone else None)
+
+        project = None
+        if project_id and await access_svc.can_read_project(user_id, project_id):
+            project = await session.get(Project, project_id)
+            if project is not None and project.deleted_at is not None:
+                project = None
+
+        steps: list = []
+        if milestone is not None:
+            steps = list((await session.execute(
+                select(Note).where(
+                    Note.milestone_id == milestone.id,
+                    Note.status.isnot(None),
+                    Note.deleted_at.is_(None),
+                    access_svc.readable_notes_clause(user_id),
+                ).order_by(Note.created_at.asc(), Note.id.asc())
+            )).scalars().all())
+
+    out: dict = {}
+    if project is not None:
+        out["project"] = {"id": project.id, "title": project.title}
+    # A milestone is shown only to someone who can read its project (or who
+    # owns it): the task being readable does not make its plan readable.
+    if milestone is not None and (project is not None or milestone.user_id == user_id):
+        counts: dict[str, int] = {}
+        for step in steps:
+            counts[step.status] = counts.get(step.status, 0) + 1
+        progress = _progress_from_counts(counts)
+        position = next((i for i, s in enumerate(steps, start=1) if s.id == task.id), None)
+        out["milestone"] = {"id": milestone.id, "title": milestone.title, "status": milestone.status}
+        out["position"] = {"step": position, "of": len(steps)}
+        out["progress"] = {k: progress[k] for k in ("completed", "total", "pct")}
+        out["next"] = _next_open(steps, task.id)
+    return out or None
+
+
+async def attach_placement(user_id: int, data: dict, task) -> dict:
+    """Add `placement` to a task payload the door is about to return.
+
+    Fail-open, like every in-band decoration: the write it rides on has
+    already happened, and a placement lookup that errors must not turn a
+    successful update into a reported failure. Omitted, never sent empty.
+    """
+    try:
+        placement = await task_placement(user_id, task)
+    except Exception:  # noqa: BLE001 - a decoration never breaks its payload
+        logger.warning("placement lookup failed for task %s", getattr(task, "id", None), exc_info=True)
+        return data
+    if placement:
+        data["placement"] = placement
+    return data

tests/conftest.py

@@ -69,6 +69,21 @@ async def _dispose_engine():
     await engine.dispose()
 
 
+@pytest.fixture
+def _no_embedding():
+    """Stub the fire-and-forget embedding refresh a note write detaches.
+
+    For integration tests about ids, transactions and access rather than
+    recall: `embed_note` spawns a task that loads the embedding model, which
+    outlives the test's event loop and makes the lane slower for nothing. Opt
+    in alongside `_dispose_engine`.
+    """
+    from unittest.mock import MagicMock
+
+    with patch("scribe.services.notes.embed_note", MagicMock()):
+        yield
+
+
 @pytest.fixture
 def _no_supersession():
     """Stub the auto-inject menu's "which lines are superseded?" lookup (#278).

tests/test_integration_placement.py

@@ -0,0 +1,96 @@
+"""Real-Postgres tests for task placement (milestone 409 step 1).
+
+What a mock cannot show: that step order is creation order rather than update
+order, that position and progress are computed over the steps the CALLER may
+read, and that a collaborator holding one shared task is not shown the titles
+of steps they cannot open.
+"""
+import pytest
+import pytest_asyncio
+
+from scribe.models import async_session
+from scribe.models.note import Note
+from scribe.models.project import Project
+from scribe.models.share import NoteShare, ProjectShare
+from scribe.services import notes as notes_svc
+from scribe.services.placement import task_placement
+from scribe.services.record_batch import BatchItem, BatchMilestone, create_batch
+from tests.helpers import ensure_user
+
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine", "_no_embedding")]
+
+
+@pytest_asyncio.fixture
+async def plan():
+    """Owner, a collaborator, a project and a four-step milestone."""
+    async with async_session() as s:
+        owner = await ensure_user(s, "placement_owner")
+        other = await ensure_user(s, "placement_collaborator")
+        project = Project(user_id=owner.id, title="Placement target")
+        s.add(project)
+        await s.flush()
+        ids = {"owner": owner.id, "other": other.id, "pid": project.id}
+        await s.commit()
+    ms, steps = await create_batch(ids["owner"], [
+        BatchItem(title="Step 1", status="done"),
+        BatchItem(title="Step 2", status="in_progress"),
+        BatchItem(title="Step 3"),
+        BatchItem(title="Step 4"),
+    ], project_id=ids["pid"], milestone=BatchMilestone(title="The plan", body="b"))
+    ids.update({"mid": ms.id, "steps": steps})
+    return ids
+
+
+async def test_the_owner_sees_the_whole_placement(plan):
+    step2 = plan["steps"][1]
+    out = await task_placement(plan["owner"], step2)
+    assert out["project"] == {"id": plan["pid"], "title": "Placement target"}
+    assert out["milestone"] == {"id": plan["mid"], "title": "The plan", "status": "active"}
+    assert out["position"] == {"step": 2, "of": 4}
+    assert out["progress"] == {"completed": 1, "total": 4, "pct": 25.0}
+    assert out["next"] == {"id": plan["steps"][2].id, "title": "Step 3", "status": "todo"}
+
+
+async def test_step_order_is_creation_order_not_last_update(plan):
+    """Touching step 1 must not move it to the end of the list."""
+    first = plan["steps"][0]
+    updated = await notes_svc.update_note(plan["owner"], first.id, body="edited later")
+    out = await task_placement(plan["owner"], updated)
+    assert out["position"] == {"step": 1, "of": 4}
+
+
+async def test_the_last_step_points_back_at_the_earliest_open_one(plan):
+    last = await notes_svc.update_note(plan["owner"], plan["steps"][3].id, status="done")
+    out = await task_placement(plan["owner"], last)
+    assert out["next"]["title"] == "Step 2"
+    assert out["progress"]["completed"] == 2
+
+
+async def test_a_task_outside_any_milestone_gets_its_project_alone(plan):
+    loose = await notes_svc.create_note(plan["owner"], title="Loose task", status="todo",
+                                        project_id=plan["pid"])
+    out = await task_placement(plan["owner"], loose)
+    assert out == {"project": {"id": plan["pid"], "title": "Placement target"}}
+
+
+async def test_a_collaborator_holding_one_shared_task_sees_no_plan(plan):
+    """A note share opens one task. It does not open the project, so neither the
+    milestone nor its other steps' titles may leak through the placement."""
+    step3 = plan["steps"][2]
+    async with async_session() as s:
+        s.add(NoteShare(note_id=step3.id, shared_with_user_id=plan["other"],
+                        permission="editor", invited_by=plan["owner"]))
+        await s.commit()
+    assert await task_placement(plan["other"], step3) is None
+
+
+async def test_a_project_collaborator_sees_the_plan(plan):
+    async with async_session() as s:
+        s.add(ProjectShare(project_id=plan["pid"], shared_with_user_id=plan["other"],
+                           permission="viewer", invited_by=plan["owner"]))
+        await s.commit()
+        step3 = await s.get(Note, plan["steps"][2].id)
+    out = await task_placement(plan["other"], step3)
+    assert out["milestone"]["title"] == "The plan"
+    assert out["position"] == {"step": 3, "of": 4}
+    assert out["next"]["title"] == "Step 4"

tests/test_integration_record_batch.py

@@ -7,7 +7,6 @@ properties of the sequence and the transaction, so they are measured against
 the real ones.
 """
 import asyncio
-from unittest.mock import MagicMock, patch
 
 import pytest
 import pytest_asyncio
@@ -22,16 +21,7 @@ from scribe.services.planning import start_planning
 from scribe.services.record_batch import BatchItem, create_batch
 from tests.helpers import ensure_user
 
-pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
-
-
-@pytest.fixture(autouse=True)
-def _no_embedding():
-    # embed_note detaches a task that loads the embedding model; this lane is
-    # about ids and transactions, and a background model load would outlive
-    # the test's event loop.
-    with patch("scribe.services.notes.embed_note", MagicMock()):
-        yield
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine", "_no_embedding")]
 
 
 @pytest_asyncio.fixture

tests/test_services_placement.py

@@ -0,0 +1,48 @@
+"""Placement — the pure half: what "next" means, and that a lookup never breaks a write.
+
+The query half (position over readable steps, what a collaborator may see)
+needs Postgres and lives in test_integration_placement.py.
+"""
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+from scribe.services import placement
+from scribe.services.placement import _next_open
+
+
+def _steps(*statuses):
+    return [SimpleNamespace(id=i, title=f"step {i}", status=s) for i, s in enumerate(statuses, start=1)]
+
+
+def test_next_is_the_first_open_step_after_this_one():
+    steps = _steps("done", "in_progress", "done", "todo", "todo")
+    assert _next_open(steps, current_id=2)["id"] == 4
+
+
+def test_next_falls_back_to_an_earlier_open_step():
+    """The last step finishing does not mean the milestone is finished."""
+    steps = _steps("todo", "done", "done")
+    assert _next_open(steps, current_id=3) == {"id": 1, "title": "step 1", "status": "todo"}
+
+
+def test_nothing_open_means_no_next():
+    assert _next_open(_steps("done", "cancelled", "done"), current_id=3) is None
+
+
+async def test_a_task_with_no_project_or_milestone_has_no_placement_and_no_query():
+    boom = AsyncMock(side_effect=AssertionError("must not open a session"))
+    with patch.object(placement, "async_session", boom):
+        task = SimpleNamespace(id=1, project_id=None, milestone_id=None)
+        assert await placement.task_placement(7, task) is None
+
+
+async def test_attach_omits_placement_rather_than_sending_it_empty():
+    with patch.object(placement, "task_placement", AsyncMock(return_value=None)):
+        data = await placement.attach_placement(7, {"id": 1}, SimpleNamespace(id=1))
+    assert "placement" not in data
+
+
+async def test_a_failing_lookup_returns_the_payload_untouched():
+    with patch.object(placement, "task_placement", AsyncMock(side_effect=RuntimeError("db down"))):
+        data = await placement.attach_placement(7, {"id": 1}, SimpleNamespace(id=1))
+    assert data == {"id": 1}