fix(#4016): records that cite each other are created together, and a guessed id is refused

Sessions predicted the ids their next creates would get and wrote them into
plan bodies and reference notes before the records existed. The database
never collides; the sequence is shared by every session and user, so any
concurrent create took the guessed numbers and the references pointed at
someone else's records.

- create_records (new MCP tool) and start_planning(body=, steps=) create
  their records in ONE transaction: insert, flush for the real ids, rewrite
  {{ref:N}} / {{ref:milestone}} placeholders as #id "title", commit. No
  prediction, no waiting, no stub records left behind when a batch fails.
  Ids need not be consecutive and nothing depends on it.
- Every MCP create/update of a note, task or milestone refuses a #N sitting
  just above the highest assigned id (within 50): that can only be a guess.
  Refusal, not warning. Numbers far above the max (PRs, forge issues) pass.
- notes.build_note splits validation out of create_note so the batch
  validates records exactly as a single create does.
- writing-plans and using-scribe say to pass steps up front and never write
  an unassigned id; plugin version minted.

Integration test runs six concurrent batches and checks each resolves its
placeholders to its own records, and that a failing batch writes nothing.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/scribe/mcp/tools/tasks.py

@@ -29,11 +29,13 @@ from scribe.services import notes as notes_svc
 # guard into a MagicMock that approves anything.
 from scribe.services.notes import minted_kind
 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
 from scribe.services import systems as systems_svc
 from scribe.services import task_logs as task_logs_svc
 from scribe.services import trash as trash_svc
 from scribe.services.note_usage import record_pulled
+from scribe.services.record_refs import refuse_guessed_ids
 
 
 async def list_tasks(
@@ -174,6 +176,13 @@ async def create_task(
             BLOCKED and the existing task's id is returned so you update it
             instead. Set true only for a genuinely distinct task.
 
+    AN ID EXISTS ONLY ONCE A CREATE RETURNS IT. Never write the id you expect
+    a record to get: every session and user draws from one sequence, so the
+    number is taken by whoever creates next. A body citing a `#N` that has not
+    been assigned yet is refused. Creating several records that cite each
+    other? Use create_records (or start_planning(steps=...)) and write
+    {{ref:N}} — the real ids are filled in as they are created.
+
     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
@@ -189,6 +198,7 @@ async def create_task(
             "milestone + seeds the design), then create each step as its own "
             "task with create_task(milestone_id=<that milestone>)."
         )
+    await refuse_guessed_ids(title, body)
     if not force:
         dup = await dedup_svc.find_duplicate_note(
             uid, title, body, project_id=project_id or None,
@@ -279,6 +289,7 @@ async def update_task(
         fields["arose_from_id"] = arose_from_id
     if kind:
         fields["task_kind"] = minted_kind(kind)
+    await refuse_guessed_ids(title, body)
     note = await notes_svc.update_note(uid, task_id, **fields)
     if note is None:
         raise ValueError(f"task {task_id} not found")
@@ -324,7 +335,117 @@ async def add_task_log(task_id: int, content: str) -> dict:
     return data
 
 
-async def start_planning(project_id: int, title: str) -> dict:
+_ITEM_KEYS = {"title", "body", "type", "status", "priority", "kind", "tags", "system_ids"}
+
+
+def _batch_items(records: list[dict], *, what: str = "record") -> list[batch_svc.BatchItem]:
+    """Parse the door's plain dicts into BatchItems, refusing unknown keys.
+
+    Strict for the same reason StrictArgsFastMCP is (#2709): a misspelt key
+    silently dropped — `milestone` for a per-record milestone, `desc` for body —
+    creates a record that looks right and is missing what the caller sent.
+    """
+    items: list[batch_svc.BatchItem] = []
+    for i, raw in enumerate(records or [], start=1):
+        if not isinstance(raw, dict):
+            raise ValueError(f"{what} {i} must be an object with at least a title")
+        unknown = set(raw) - _ITEM_KEYS
+        if unknown:
+            raise ValueError(
+                f"{what} {i} has unknown field(s) {sorted(unknown)}; "
+                f"allowed: {sorted(_ITEM_KEYS)}"
+            )
+        rtype = raw.get("type") or "task"
+        if rtype not in ("task", "note"):
+            raise ValueError(f"{what} {i}: type must be 'task' or 'note', got {rtype!r}")
+        items.append(batch_svc.BatchItem(
+            title=raw.get("title") or "",
+            body=raw.get("body") or "",
+            is_task=rtype == "task",
+            status=raw.get("status") or "todo",
+            priority=raw.get("priority") or None,
+            task_kind=raw.get("kind") or "work",
+            tags=list(raw.get("tags") or []),
+            system_ids=list(raw.get("system_ids") or []),
+        ))
+    return items
+
+
+async def _first_duplicate(uid: int, items: list, project_id: int | None) -> dict | None:
+    """The duplicate gate over a whole batch — the first hit blocks all of it."""
+    for i, item in enumerate(items, start=1):
+        dup = await dedup_svc.find_duplicate_note(
+            uid, item.title, item.body, project_id=project_id,
+            is_task=item.is_task, note_type="note",
+        )
+        if dup is not None:
+            payload = dedup_svc.duplicate_response(dup, "task" if item.is_task else "note")
+            payload["record"] = i
+            payload["message"] = f"Record {i} of the batch: {payload['message']} Nothing in the batch was created."
+            return payload
+    return None
+
+
+async def create_records(
+    records: list[dict],
+    project_id: int = 0,
+    milestone_id: int = 0,
+    force: bool = False,
+) -> dict:
+    """Create several tasks and/or notes in ONE call, so they can cite each other.
+
+    Reach for this whenever records you are about to create need to reference
+    one another — a set of tasks that name their siblings, a reference note
+    listing the tasks it indexes. Writing the ids you EXPECT them to get is
+    always wrong: every session and user draws from one sequence, so another
+    create takes those numbers, and a body citing an unassigned `#N` is refused.
+    A plan (a milestone plus its steps) is start_planning(steps=...), which
+    uses the same mechanism. One record with nothing to cite is create_task or
+    create_note.
+
+    In any record's body write {{ref:N}} where the Nth record's id belongs
+    (1-based, in the order listed). It is replaced with `#<id> "<title>"`
+    once the ids exist. Everything is created in one transaction: a bad
+    placeholder, an invalid field or a duplicate creates NOTHING.
+
+    Args:
+        records: The records, in order. Each is an object with `title`
+            (required) and optionally `body`, `type` ('task', the default, or
+            'note'), and for tasks `status`, `priority` and `kind`
+            ('work' | 'issue' | 'spike'); plus `tags` and `system_ids`.
+        project_id: The project every record belongs to (0 = none, or taken
+            from milestone_id).
+        milestone_id: File every record under this existing milestone (0 = none).
+        force: Bypass the near-duplicate gate for the whole batch. By default
+            the first record that near-duplicates an existing one BLOCKS the
+            batch, and its existing id comes back so you can update it instead.
+
+    Returns {"ids": [...], "records": [...]} in the order given — the ids
+    cite-able from here on — OR a duplicate payload naming which `record`
+    matched, with nothing created. The ids are the real ones and are not
+    necessarily consecutive: other sessions keep creating meanwhile, and
+    nothing here depends on the numbers being adjacent.
+    """
+    uid = current_user_id()
+    items = _batch_items(records)
+    await refuse_guessed_ids(*[t for item in items for t in (item.title, item.body)])
+    if not force:
+        dup = await _first_duplicate(uid, items, project_id or None)
+        if dup is not None:
+            return dup
+    _ms, notes = await batch_svc.create_batch(
+        uid, items, project_id=project_id or None, milestone_id=milestone_id or None,
+    )
+    return {"ids": [n.id for n in notes], "records": [n.to_dict() for n in notes]}
+
+
+async def start_planning(
+    project_id: int,
+    title: str,
+    body: str = "",
+    steps: list[dict] | None = None,
+    force: bool = False,
+) -> dict:
     """Begin a plan in Scribe (the preferred home for plans — not a local .md file).
 
     Reach for this when the work has an ARC — several steps toward one goal,
@@ -341,11 +462,17 @@ async def start_planning(project_id: int, title: str) -> dict:
     context. The milestone is the plan container — the individual steps live as
     first-class child tasks under it, not as checkboxes in the body.
 
-    Afterwards:
-    - Edit the plan/design with update_milestone(milestone_id, body=...).
-    - Create each step as its own task with create_task(milestone_id=<this id>);
-      track it with status + add_task_log. Do NOT put steps as checkboxes in the
-      milestone body.
+    PASS THE STEPS HERE when you already know them. The milestone and every
+    step are then created in one transaction, and the plan body and the steps
+    can cite each other with placeholders that become real ids: {{ref:N}} is
+    the Nth step (1-based) and {{ref:milestone}} is this milestone. Never write
+    the ids you expect records to get — other sessions and users draw from the
+    same sequence, and a body citing an unassigned `#N` is refused.
+
+    Without steps: edit the design afterwards with update_milestone(
+    milestone_id, body=...), and add steps with create_records(milestone_id=
+    <this id>, ...) — or create_task for a single one. Track each with status
+    + add_task_log. Do NOT put steps as checkboxes in the milestone body.
 
     (kind=plan tasks are retired — use this instead. Existing historical
     plan-tasks remain readable but new planning goes through milestones.)
@@ -353,10 +480,29 @@ async def start_planning(project_id: int, title: str) -> dict:
     Args:
         project_id: The project this plan is for.
         title: A short title for the plan/milestone.
+        body: The plan's design (markdown). Omit to seed the Goal/Approach/
+            Verification template.
+        steps: The plan's step-tasks, in order — each an object with `title`
+            (required) and optionally `body`, `status`, `priority`, `kind`
+            ('work' | 'issue' | 'spike'), `tags`, `system_ids`.
+        force: Bypass the near-duplicate gate on the steps. By default a step
+            that near-duplicates an existing task blocks the whole plan, and
+            nothing — milestone included — is created.
+
+    Returns the milestone, the project's applicable rules and brief context,
+    plus `steps` (the created tasks, in order) when steps were given — OR a
+    duplicate payload naming the `record` that matched, with nothing created.
     """
     uid = current_user_id()
+    items = _batch_items(steps or [], what="step")
+    await refuse_guessed_ids(body, *[t for item in items for t in (item.title, item.body)])
+    if items and not force:
+        dup = await _first_duplicate(uid, items, project_id or None)
+        if dup is not None:
+            return dup
     return await planning_svc.start_planning(
         user_id=uid, project_id=project_id, title=title,
+        body=body or None, steps=items or None,
     )
 
 
@@ -388,6 +534,7 @@ def register(mcp) -> None:
         list_tasks,
         get_task,
         create_task,
+        create_records,
         update_task,
         add_task_log,
         start_planning,