src/scribe/mcp/server.py

@@ -1,6 +1,8 @@
 """FastMCP instance + Quart mount-point. Tools are registered in mcp/tools/."""
 from __future__ import annotations
 
+import difflib
+
 from mcp.server.fastmcp import FastMCP
 from mcp.server.transport_security import TransportSecuritySettings
 from quart import Quart
@@ -165,6 +167,46 @@ def _body_calls_write_tool(body: bytes) -> bool:
     return False
 
 
+class StrictArgsFastMCP(FastMCP):
+    """A FastMCP that REJECTS tool calls carrying undeclared arguments.
+
+    FastMCP validates arguments with a pydantic model built from the tool
+    signature, and pydantic's default extra-field policy is "ignore" — so a
+    misnamed argument simply vanishes and the tool runs with that field's
+    default. On the create/update tools the default is "", which turns a
+    plausible near-miss (`content=` for `body=`, primed by add_task_log's
+    `content`) into SILENT DATA LOSS: the call reports success and stores an
+    empty body, leaving a record search cannot see (#2709). Two notes were
+    persisted body-less that way before anyone noticed.
+
+    An error the caller sees once is strictly better than data half-written
+    forever, so the policy is applied to every tool, not just the two that
+    bit: nothing here knows tool semantics, only that an argument nobody
+    declared cannot have been meant to be dropped.
+    """
+
+    async def call_tool(self, name, arguments):
+        try:
+            tool = self._tool_manager.get_tool(name)
+        except Exception:
+            tool = None  # unknown tool → let upstream produce its own error
+        if tool is not None:
+            declared = set((tool.parameters or {}).get("properties", {}))
+            unknown = sorted(set(arguments or {}) - declared)
+            if unknown:
+                hints = []
+                for arg in unknown:
+                    close = difflib.get_close_matches(arg, sorted(declared), n=1)
+                    suggestion = f" (did you mean '{close[0]}'?)" if close else ""
+                    hints.append(f"'{arg}'{suggestion}")
+                raise ValueError(
+                    f"{name} does not accept argument(s) {', '.join(hints)}. "
+                    f"It accepts: {', '.join(sorted(declared))}. Nothing was "
+                    "created or changed — retry with the declared names."
+                )
+        return await super().call_tool(name, arguments)
+
+
 def build_mcp_server() -> FastMCP:
     """Build the FastMCP instance with all tools registered.
 
@@ -185,7 +227,7 @@ def build_mcp_server() -> FastMCP:
     # every request self-contained (bearer-auth only), so a post-deploy
     # reconnect just works. Trade-off: no server-pushed list_changed stream,
     # which we don't use — tools are re-fetched on reconnect anyway.
-    mcp = FastMCP(
+    mcp = StrictArgsFastMCP(
         "scribe",
         instructions=_INSTRUCTIONS.strip(),
         stateless_http=True,

src/scribe/mcp/tools/projects.py

@@ -17,6 +17,7 @@ keeps working.
 from __future__ import annotations
 
 from scribe.mcp._context import current_user_id
+from scribe.mcp.tools import systems as systems_tools
 from scribe.services import coverage as coverage_svc
 from scribe.services import design_systems as design_systems_svc
 from scribe.services import milestones as milestones_svc
@@ -57,7 +58,8 @@ async def enter_project(project_id: int) -> dict:
 
     Returns a dict with keys: project, milestone_summary, applicable_rules,
     project_rules, subscribed_rulebooks, applicable_rules_truncated,
-    open_tasks, recent_notes, design_system, systems, pattern_coverage.
+    open_tasks, recent_notes, design_system, systems, pattern_coverage —
+    plus systems_bootstrap, present only when it applies (see below).
 
     `pattern_coverage` (usually null) is a one-line estimate of how much of
     the bound repo's code has recorded snippets — e.g. "pattern-library
@@ -72,6 +74,12 @@ async def enter_project(project_id: int) -> dict:
     create it with create_system rather than leaving the area unmodelled. Read
     a subsystem's accumulated records with list_system_records.
 
+    `systems_bootstrap` appears ONLY when the project has many records and no
+    Systems at all — act on it before starting other work: propose a starter
+    vocabulary from the areas the project's records name, confirm it with the
+    operator, and create_system the confirmed set. It stops appearing the
+    moment the first System exists.
+
     `design_system` is null unless the project points at one. When present it
     carries the chain-merged guidance (the house style AND this project's
     departures from it) plus a summary of the token set — treat it as binding
@@ -116,6 +124,17 @@ async def enter_project(project_id: int) -> dict:
     # three days of the feature landing (#2546's audit).
     systems = await systems_svc.list_systems(uid, project_id)
 
+    # The arrival-moment half of the bootstrap ask (#2683): session start is
+    # when the agent has just read the project map and is not yet deep in a
+    # task — the one moment "propose a starter vocabulary" is cheap. The
+    # write-moment half rides untagged-record responses (attach_systems);
+    # both retire the instant the first System exists.
+    systems_bootstrap = None
+    if not systems:
+        systems_bootstrap = await systems_tools.bootstrap_systems_ask(
+            uid, project_id
+        )
+
     # Probably the largest surfacing by volume, and it emitted nothing — so
     # the pulls it caused floated unattributed and the surfaced:pulled ratio
     # ran against a denominator missing its biggest contributor (#2477). An
@@ -143,7 +162,7 @@ async def enter_project(project_id: int) -> dict:
         project.user_id or uid, project_id
     )
 
-    return {
+    out = {
         "project": project.to_dict(),
         "pattern_coverage": coverage_svc.coverage_line(coverage) if coverage else None,
         # Trimmed to what tagging needs. The full charter is get_system's job —
@@ -179,6 +198,11 @@ async def enter_project(project_id: int) -> dict:
             for n in recent_notes
         ],
     }
+    # Attached only when it applies — a key that usually says null trains
+    # readers to skip it (#2483), and this one exists to be acted on.
+    if systems_bootstrap:
+        out["systems_bootstrap"] = systems_bootstrap
+    return out
 
 
 async def get_project(project_id: int) -> dict:

src/scribe/mcp/tools/systems.py

@@ -13,8 +13,57 @@ Sentinels (match the milestone/task tool conventions):
 from __future__ import annotations
 
 from scribe.mcp._context import current_user_id
+from scribe.services import notes as notes_svc
 from scribe.services import systems as systems_svc
 
+# Below this, a project is young enough that the mild "which area is this
+# about?" question stays proportionate; at or above it, a zero-Systems project
+# has demonstrated that the question never converts (#2683 — Minstrel reached
+# 282 records without a single System) and the ask escalates.
+_BOOTSTRAP_MIN_RECORDS = 20
+_BOOTSTRAP_TITLES = 6
+
+
+async def bootstrap_systems_ask(user_id: int, project_id: int) -> str | None:
+    """The escalated vocabulary-bootstrap ask for a mature zero-Systems project.
+
+    The generic zero-state question habituates: identical on every record,
+    maximal in scope ("invent the taxonomy"), asked at wrap-up time — so
+    organic sessions skip it forever and only audit-shaped sessions ever mint
+    (#2683). What separates the nudges that convert from the prose that
+    doesn't (the duplicate gate, the prior-art "already defined in 2 files")
+    is the project's OWN evidence in the ask — so this one carries the record
+    count and the recent titles, and asks for a concrete deliverable: propose
+    a starter set, confirm, create.
+
+    Self-retiring by construction: callers only reach for it while the
+    project has zero Systems, so the first create_system ends it everywhere.
+    Returns None below the record threshold or on any failure (fail-open —
+    a hint must never break the call it rides on).
+    """
+    try:
+        recent, total = await notes_svc.list_notes(
+            user_id, project_id=project_id, sort="updated_at",
+            limit=_BOOTSTRAP_TITLES,
+        )
+    except Exception:
+        return None
+    if total < _BOOTSTRAP_MIN_RECORDS:
+        return None
+    titles = "; ".join(
+        '"' + " ".join((n.title or "").split())[:70] + '"' for n in recent
+    )
+    return (
+        f"This project has {total} records and NO Systems modelled — none of "
+        "them can be tagged to an area, so recurring problem-spots stay "
+        "invisible. Bootstrap the vocabulary now, in this session: from the "
+        f"areas the records themselves name (recent: {titles}), propose 3-6 "
+        "Systems to the operator, create_system each confirmed one with a "
+        "one-paragraph charter, then tag this record (system_ids=[...]). "
+        "This ask repeats until the first System exists; answering it once "
+        "retires it for every future record."
+    )
+
 
 async def untagged_systems_hint(user_id: int, project_id: int) -> str | None:
     """The Systems question, for an untagged project record.
@@ -26,6 +75,10 @@ async def untagged_systems_hint(user_id: int, project_id: int) -> str | None:
     vocabulary is not an exemption — it is the question at its most urgent
     (#2562, #2569). Instruction prose alone demonstrably doesn't fire at write
     time; in-band behavior (the duplicate gate) does.
+
+    In a MATURE zero-Systems project the question escalates to the bootstrap
+    ask instead (#2683): the mild form demonstrably never converts there, and
+    an evidence-carrying, deliverable-shaped ask is the form that does.
     """
     # Fail-open like the dedup gate: a hint must never break the call.
     try:
@@ -37,6 +90,9 @@ async def untagged_systems_hint(user_id: int, project_id: int) -> str | None:
             f"#{s.id} {s.name}" for s in systems
         ) + "."
     else:
+        ask = await bootstrap_systems_ask(user_id, project_id)
+        if ask:
+            return ask
         vocab = "This project has no Systems yet."
     return (
         "This record is untagged — which area(s) of the project is it about? "

tests/test_mcp_strict_args.py

@@ -0,0 +1,75 @@
+"""Unknown tool arguments are rejected, never silently dropped (#2709).
+
+The failure this pins: FastMCP validates tool arguments with a pydantic model
+whose extra-field policy is "ignore", so `create_note(content=...)` — a
+plausible near-miss for `body=`, primed by add_task_log's `content` — ran
+successfully, stored `body: ""`, and left a record embedding/search cannot
+see. The call REPORTED SUCCESS. Two real notes were persisted body-less
+before the pattern was noticed.
+
+The fix is at the dispatch seam, not per-tool: StrictArgsFastMCP rejects any
+call carrying arguments the tool does not declare, with a did-you-mean when
+one is close. An error the caller sees once beats data half-written forever.
+"""
+import pytest
+
+from scribe.mcp.server import StrictArgsFastMCP
+
+
+def _echo_server() -> StrictArgsFastMCP:
+    mcp = StrictArgsFastMCP("strict-test")
+
+    @mcp.tool()
+    def echo(text: str = "") -> str:
+        return text
+
+    return mcp
+
+
+async def test_declared_arguments_still_dispatch():
+    mcp = _echo_server()
+    result = await mcp.call_tool("echo", {"text": "hi"})
+    assert "hi" in str(result)
+
+
+async def test_unknown_argument_is_an_error_not_a_silent_drop():
+    """The load-bearing property: the call must FAIL, because succeeding is
+    what turned a typo into data loss."""
+    mcp = _echo_server()
+    with pytest.raises(ValueError) as exc:
+        await mcp.call_tool("echo", {"txt": "hi"})
+    msg = str(exc.value)
+    assert "'txt'" in msg
+    assert "did you mean 'text'?" in msg          # difflib near-miss hint
+    assert "Nothing was created or changed" in msg
+
+
+async def test_empty_arguments_pass():
+    mcp = _echo_server()
+    assert await mcp.call_tool("echo", {}) is not None
+
+
+async def test_unknown_tool_keeps_the_upstream_error():
+    """The gate must not swallow or reshape 'no such tool' — that error path
+    belongs to FastMCP and clients already understand it."""
+    mcp = _echo_server()
+    with pytest.raises(Exception) as exc:
+        await mcp.call_tool("no_such_tool", {"text": "hi"})
+    assert "no_such_tool" in str(exc.value)
+
+
+async def test_the_original_regression_create_note_with_content():
+    """Pin #2709 itself against the REAL server: `content=` on create_note
+    must raise before dispatch — naming the bad argument and listing `body`
+    among the accepted ones — instead of creating a body-less note."""
+    from scribe.mcp.server import build_mcp_server
+
+    mcp = build_mcp_server()
+    assert isinstance(mcp, StrictArgsFastMCP)   # the guard is actually mounted
+    with pytest.raises(ValueError) as exc:
+        await mcp.call_tool(
+            "create_note", {"title": "t", "content": "the body text"}
+        )
+    msg = str(exc.value)
+    assert "'content'" in msg
+    assert "body" in msg

tests/test_mcp_tool_projects.py

@@ -40,6 +40,18 @@ def _no_coverage():
         yield
 
 
+@pytest.fixture(autouse=True)
+def _no_bootstrap():
+    """With _no_systems stubbing an empty vocabulary, every test here reaches
+    the zero-Systems branch, whose bootstrap ask (#2683) counts the project's
+    records — a database read. Stub the common case (young project, no ask);
+    the firing shape has its own test below.
+    """
+    with patch("scribe.mcp.tools.projects.systems_tools.bootstrap_systems_ask",
+               AsyncMock(return_value=None)) as mock:
+        yield mock
+
+
 def _fake_project(design_system_id=None, **overrides) -> MagicMock:
     p = MagicMock()
     base = {"id": 1, "title": "P", "description": "", "goal": "",
@@ -211,6 +223,9 @@ async def test_enter_project_composes_full_context():
     # vocabulary, and "this project has no named areas yet" is information the
     # create-the-System instruction acts on.
     assert out["systems"] == []
+    # Young project, no bootstrap ask -> the key is ABSENT, not null: it exists
+    # to be acted on, and a key that usually says null gets skipped (#2483).
+    assert "systems_bootstrap" not in out
 
 
 @pytest.mark.asyncio
@@ -251,6 +266,62 @@ async def test_enter_project_surfaces_the_systems_vocabulary():
     ]
 
 
+def _enter_project_stubs(p):
+    """The four patches every enter_project test repeats, as one context list."""
+    return [
+        patch("scribe.mcp.tools.projects.projects_svc.get_project",
+              AsyncMock(return_value=p)),
+        patch("scribe.mcp.tools.projects.rulebooks_svc.get_applicable_rules",
+              AsyncMock(return_value={"rules": [], "truncated": False,
+                                      "subscribed_rulebooks": []})),
+        patch("scribe.mcp.tools.projects.milestones_svc.get_project_milestone_summary",
+              AsyncMock(return_value=[])),
+        patch("scribe.mcp.tools.projects.notes_svc.list_notes",
+              AsyncMock(side_effect=[([], 0), ([], 0)])),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_enter_project_carries_the_bootstrap_ask_when_it_fires():
+    """The arrival-moment half of #2683: a mature zero-Systems project greets
+    the session with the concrete bootstrap ask, before it is deep in a task —
+    the moment "propose a starter vocabulary" is cheapest."""
+    import contextlib
+
+    ask = "This project has 282 records and NO Systems modelled — ..."
+    with contextlib.ExitStack() as stack:
+        for cm in _enter_project_stubs(_fake_project(id=5)):
+            stack.enter_context(cm)
+        stack.enter_context(patch(
+            "scribe.mcp.tools.projects.systems_tools.bootstrap_systems_ask",
+            AsyncMock(return_value=ask),
+        ))
+        out = await enter_project(project_id=5)
+    assert out["systems_bootstrap"] == ask
+
+
+@pytest.mark.asyncio
+async def test_enter_project_never_asks_bootstrap_once_a_vocabulary_exists(
+    _no_bootstrap,
+):
+    """The ask is self-retiring: the first System ends it — enter_project must
+    not even evaluate it once the vocabulary is non-empty."""
+    import contextlib
+
+    sys1 = MagicMock()
+    sys1.id = 3; sys1.name = "retrieval"; sys1.description = ""
+    with contextlib.ExitStack() as stack:
+        for cm in _enter_project_stubs(_fake_project(id=5)):
+            stack.enter_context(cm)
+        stack.enter_context(patch(
+            "scribe.mcp.tools.projects.systems_svc.list_systems",
+            AsyncMock(return_value=[sys1]),
+        ))
+        out = await enter_project(project_id=5)
+    assert "systems_bootstrap" not in out
+    _no_bootstrap.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_enter_project_hands_back_the_design_system_when_the_project_has_one():
     """The handshake is where an agent learns what binds it, and a design

tests/test_mcp_tool_systems.py

@@ -88,13 +88,22 @@ async def test_untagged_hint_names_the_projects_systems():
     assert "create_system" in hint
 
 
+def _fake_note(title):
+    n = MagicMock()
+    n.title = title
+    return n
+
+
 @pytest.mark.asyncio
 async def test_untagged_hint_zero_systems_prompts_first_create_and_fails_open():
     from scribe.mcp.tools.systems import untagged_systems_hint
-    with patch("scribe.mcp.tools.systems.systems_svc") as svc:
-        # Zero Systems is the state nothing else nudges out of — the hint must
-        # prompt the FIRST create_system, not go silent.
+    with patch("scribe.mcp.tools.systems.systems_svc") as svc, \
+         patch("scribe.mcp.tools.systems.notes_svc") as notes:
+        # Zero Systems in a YOUNG project (below the bootstrap threshold):
+        # the mild question stays proportionate — it must prompt the FIRST
+        # create_system, not go silent.
         svc.list_systems = AsyncMock(return_value=[])
+        notes.list_notes = AsyncMock(return_value=([], 3))
         hint = await untagged_systems_hint(1, 5)
         assert "no Systems yet" in hint and "create_system" in hint
     with patch("scribe.mcp.tools.systems.systems_svc") as svc:
@@ -103,6 +112,57 @@ async def test_untagged_hint_zero_systems_prompts_first_create_and_fails_open():
         assert await untagged_systems_hint(1, 5) is None
 
 
+@pytest.mark.asyncio
+async def test_untagged_hint_escalates_in_a_mature_zero_systems_project():
+    """#2683: at 282 records and zero Systems (Minstrel), the generic question
+    had demonstrably never converted. The escalated ask must carry the
+    project's OWN evidence — record count and recent titles — and demand a
+    concrete deliverable, because that is the property separating the nudges
+    that convert from the prose that doesn't."""
+    from scribe.mcp.tools.systems import untagged_systems_hint
+    recent = [_fake_note("Fix scrape retry backoff"), _fake_note("Worker pool sizing")]
+    with patch("scribe.mcp.tools.systems.systems_svc") as svc, \
+         patch("scribe.mcp.tools.systems.notes_svc") as notes:
+        svc.list_systems = AsyncMock(return_value=[])
+        notes.list_notes = AsyncMock(return_value=(recent, 282))
+        hint = await untagged_systems_hint(1, 5)
+    assert "282 records" in hint
+    assert "Fix scrape retry backoff" in hint      # the project's own evidence
+    assert "3-6" in hint and "create_system" in hint
+    assert "propose" in hint
+    # The generic wording is REPLACED, not appended — two questions is noise.
+    assert "no Systems yet" not in hint
+
+
+@pytest.mark.asyncio
+async def test_bootstrap_ask_stays_quiet_below_threshold_and_fails_open():
+    from scribe.mcp.tools import systems as tools
+    with patch("scribe.mcp.tools.systems.notes_svc") as notes:
+        notes.list_notes = AsyncMock(
+            return_value=([], tools._BOOTSTRAP_MIN_RECORDS - 1))
+        assert await tools.bootstrap_systems_ask(1, 5) is None
+    with patch("scribe.mcp.tools.systems.notes_svc") as notes:
+        # The record count is decoration on a hint on a create — three layers
+        # deep, nothing there may break the call.
+        notes.list_notes = AsyncMock(side_effect=RuntimeError("db down"))
+        assert await tools.bootstrap_systems_ask(1, 5) is None
+
+
+@pytest.mark.asyncio
+async def test_populated_vocabulary_never_counts_records():
+    """The bootstrap query is zero-state-only: once Systems exist the hint
+    must not spend a count query per untagged record."""
+    from scribe.mcp.tools.systems import untagged_systems_hint
+    sys_a = MagicMock(); sys_a.id = 1; sys_a.name = "workers"
+    with patch("scribe.mcp.tools.systems.systems_svc") as svc, \
+         patch("scribe.mcp.tools.systems.notes_svc") as notes:
+        svc.list_systems = AsyncMock(return_value=[sys_a])
+        notes.list_notes = AsyncMock()
+        hint = await untagged_systems_hint(1, 5)
+    assert "workers" in hint
+    notes.list_notes.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_create_system_same_normalized_name_is_duplicate_gated():
     existing = _fake_system(sid=7, name="Scrape Pipeline")
@@ -204,11 +264,15 @@ async def test_add_task_log_on_untagged_project_task_asks_the_question():
     with patch("scribe.mcp.tools.tasks.current_user_id", return_value=1), \
          patch("scribe.mcp.tools.tasks.task_logs_svc") as logs_svc, \
          patch("scribe.mcp.tools.tasks.notes_svc") as notes_svc, \
-         patch("scribe.mcp.tools.systems.systems_svc") as seam_svc:
+         patch("scribe.mcp.tools.systems.systems_svc") as seam_svc, \
+         patch("scribe.mcp.tools.systems.notes_svc") as seam_notes:
         logs_svc.create_log = AsyncMock(return_value=log)
         notes_svc.get_note_for_user = AsyncMock(return_value=(task, "owner"))
         seam_svc.list_record_systems = AsyncMock(return_value=[])
         seam_svc.list_systems = AsyncMock(return_value=[])
+        # Young project: below the #2683 bootstrap threshold, so the mild
+        # question is the expected form here.
+        seam_notes.list_notes = AsyncMock(return_value=([], 2))
         from scribe.mcp.tools.tasks import add_task_log
         result = await add_task_log(task_id=70, content="progress")
     assert "no Systems yet" in result["systems_hint"]