refactor(agent): gate sustained goals behind explicit /goal

Replace the legacy long-goal skill contract with command-scoped goal tools and runtime guidance. Keep goal state durable across continuations while restricting create and replace mutations to explicit user /goal turns.

tests/agent/test_context_builder.py

@@ -5,6 +5,7 @@ from pathlib import Path
 import pytest
 
 from nanobot.agent.context import ContextBuilder
+from nanobot.bus.events import InboundMessage
 from nanobot.session.goal_state import GOAL_STATE_KEY
 
 # ---------------------------------------------------------------------------
@@ -334,9 +335,59 @@ class TestBuildMessages:
             session_metadata=meta,
         )
         user_msg = str(messages[-1]["content"])
+        assert ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG in user_msg
+        assert "Execute sustained work" in user_msg
+        assert "Start or replace the sustained goal" not in user_msg
         assert "Goal (active):" in user_msg
         assert "Finish docs migration." in user_msg
 
+    def test_goal_start_turn_injects_objective_guidance_after_user_text(self, tmp_path):
+        builder = _builder(tmp_path)
+        normal_messages = builder.build_messages([], "hi", channel="cli", chat_id="direct")
+        messages = builder.build_messages(
+            [],
+            "/goal audit the repo",
+            channel="cli",
+            chat_id="direct",
+            goal_start_requested=True,
+        )
+        stale_messages = builder.build_messages(
+            [],
+            "/goal stale request",
+            channel="cli",
+            chat_id="direct",
+            inbound_message=InboundMessage(
+                channel="cli",
+                sender_id="system",
+                chat_id="direct",
+                content="/goal stale request",
+                metadata={"original_command": "/goal", "goal_requested": True},
+            ),
+        )
+
+        user_msg = str(messages[-1]["content"])
+        assert "Write a durable objective" in user_msg
+        assert "complete `/goal <task>` command" in user_msg
+        guidance = user_msg[
+            user_msg.index(ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG) :
+            user_msg.index(ContextBuilder._GOAL_RUNTIME_GUIDANCE_END)
+        ].lower()
+        assert "authorization" not in guidance
+        assert "host-issued" not in guidance
+        assert user_msg.index("/goal audit the repo") < user_msg.index(
+            ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG
+        )
+        assert user_msg.index(ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG) < user_msg.index(
+            ContextBuilder._RUNTIME_CONTEXT_TAG
+        )
+        assert normal_messages[0]["content"] == messages[0]["content"]
+        assert ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG not in str(
+            normal_messages[-1]["content"]
+        )
+        assert ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG not in str(
+            stale_messages[-1]["content"]
+        )
+
     def test_goal_state_does_not_leak_without_session_metadata(self, tmp_path):
         builder = _builder(tmp_path)
         other_session_meta = {

tests/agent/test_loop_runner_integration.py

@@ -7,9 +7,12 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
+from nanobot.agent.context import ContextBuilder
+from nanobot.agent.goal_permission import goal_mutation_allowed, goal_mutation_permission
 from nanobot.bus.outbound_events import StreamedResponseEvent
 from nanobot.config.schema import AgentDefaults
 from nanobot.providers.base import GenerationSettings, LLMResponse, ToolCallRequest
+from nanobot.session.goal_state import GOAL_STATE_KEY
 from nanobot.utils.llm_runtime import LLMRuntime
 
 _MAX_TOOL_RESULT_CHARS = AgentDefaults().max_tool_result_chars
@@ -31,6 +34,146 @@ def _make_loop(tmp_path):
         loop = AgentLoop(bus=bus, provider=provider, workspace=tmp_path)
     return loop
 
+
+@pytest.mark.asyncio
+async def test_ephemeral_runner_enters_and_restores_turn_scopes(tmp_path):
+    loop = _make_loop(tmp_path)
+
+    async def chat_with_retry(**_kwargs):
+        assert goal_mutation_allowed() is True
+        return LLMResponse(content="done", tool_calls=[], usage={})
+
+    loop.provider.chat_with_retry = AsyncMock(side_effect=chat_with_retry)
+    loop.tools.get_definitions = MagicMock(return_value=[])
+
+    await loop._run_agent_loop(
+        [],
+        runtime=loop.llm_runtime(),
+        ephemeral=True,
+        turn_scopes=[goal_mutation_permission(True)],
+    )
+
+    assert goal_mutation_allowed() is False
+
+
+@pytest.mark.asyncio
+async def test_goal_command_can_implement_plan_from_prior_discussion(tmp_path):
+    from nanobot.agent.loop import AgentLoop
+    from nanobot.bus.events import InboundMessage
+    from nanobot.bus.queue import MessageBus
+
+    provider = MagicMock()
+    provider.get_default_model.return_value = "test-model"
+    provider.chat_with_retry = AsyncMock(side_effect=[
+        LLMResponse(
+            content="recording the agreed plan",
+            tool_calls=[
+                ToolCallRequest(
+                    id="call_create",
+                    name="create_goal",
+                    arguments={
+                        "objective": "Implement the agreed migration plan and run its tests.",
+                    },
+                )
+            ],
+            usage={},
+        ),
+        LLMResponse(
+            content="closing goal",
+            tool_calls=[
+                ToolCallRequest(
+                    id="call_update",
+                    name="update_goal",
+                    arguments={"action": "complete", "recap": "Implemented and tested."},
+                )
+            ],
+            usage={},
+        ),
+        LLMResponse(
+            content="trying to start another goal",
+            tool_calls=[
+                ToolCallRequest(
+                    id="call_create_again",
+                    name="create_goal",
+                    arguments={"objective": "Start an unrelated follow-up."},
+                )
+            ],
+            usage={},
+        ),
+        LLMResponse(content="done", tool_calls=[], usage={}),
+    ])
+    loop = AgentLoop(bus=MessageBus(), provider=provider, workspace=tmp_path, model="test-model")
+    loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=None)
+    session = loop.sessions.get_or_create("cli:direct")
+    session.add_message("user", "Let's agree on the migration implementation.")
+    session.add_message("assistant", "Use the staged migration plan and run integration tests.")
+
+    result = await loop._process_message(
+        InboundMessage(
+            channel="cli",
+            sender_id="user",
+            chat_id="direct",
+            content="/goal implement the plan above",
+        )
+    )
+
+    assert result is not None
+    assert result.content == "done"
+    assert goal_mutation_allowed() is False
+    assert session.metadata[GOAL_STATE_KEY]["status"] == "completed"
+    first_request = provider.chat_with_retry.await_args_list[0].kwargs["messages"]
+    assert "staged migration plan" in str(first_request)
+    assert "/goal implement the plan above" in str(first_request)
+    assert ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG in str(first_request)
+    final_request = provider.chat_with_retry.await_args_list[-1].kwargs["messages"]
+    assert "create_goal is unavailable for this turn" in str(final_request)
+    assert all(
+        ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG not in str(message.get("content") or "")
+        for message in session.messages
+    )
+
+
+@pytest.mark.asyncio
+async def test_non_goal_direct_turn_cannot_reuse_prior_goal_command(tmp_path):
+    from nanobot.agent.loop import AgentLoop
+    from nanobot.bus.queue import MessageBus
+
+    provider = MagicMock()
+    provider.get_default_model.return_value = "test-model"
+    provider.chat_with_retry = AsyncMock(side_effect=[
+        LLMResponse(
+            content="trying to create a goal",
+            tool_calls=[
+                ToolCallRequest(
+                    id="call_create",
+                    name="create_goal",
+                    arguments={"objective": "Unauthorized persistent objective."},
+                )
+            ],
+            usage={},
+        ),
+        LLMResponse(content="handled as a one-time task", tool_calls=[], usage={}),
+    ])
+    loop = AgentLoop(bus=MessageBus(), provider=provider, workspace=tmp_path, model="test-model")
+    loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=None)
+    session = loop.sessions.get_or_create("api:default")
+    session.add_message("user", "/goal old completed request")
+    session.add_message("assistant", "The old request is complete.")
+
+    result = await loop.process_direct(
+        "Handle this as an ordinary one-time task.",
+        session_key=session.key,
+        channel="api",
+        chat_id="default",
+        persist_user_message=False,
+    )
+
+    assert result is not None
+    assert result.content == "handled as a one-time task"
+    assert GOAL_STATE_KEY not in session.metadata
+    second_request = provider.chat_with_retry.await_args_list[1].kwargs["messages"]
+    assert "create_goal is unavailable for this turn" in str(second_request)
+
 @pytest.mark.asyncio
 async def test_loop_max_iterations_message_stays_stable(tmp_path):
     loop = _make_loop(tmp_path)

tests/agent/test_loop_save_turn.py

@@ -18,7 +18,7 @@ from nanobot.bus.outbound_events import (
 )
 from nanobot.bus.queue import MessageBus
 from nanobot.cron.session_turns import CRON_HISTORY_META, CRON_TRIGGER_META
-from nanobot.providers.base import LLMResponse
+from nanobot.providers.base import LLMProvider, LLMResponse
 from nanobot.providers.factory import ProviderSnapshot
 from nanobot.session.automation_turns import AUTOMATION_HISTORY_META
 from nanobot.session.goal_state import GOAL_STATE_KEY
@@ -48,6 +48,22 @@ def _mk_loop() -> AgentLoop:
     return loop
 
 
+def _host_text_message(content: str, suffix: str) -> dict:
+    return {
+        "role": "user",
+        "content": content,
+        "_meta": {ContextBuilder._HOST_TEXT_SUFFIX_META_KEY: suffix},
+    }
+
+
+def _host_text_block(text: str) -> dict:
+    return {
+        "type": "text",
+        "text": text,
+        "_meta": {ContextBuilder._HOST_BLOCK_META_KEY: True},
+    }
+
+
 def _make_full_loop(tmp_path: Path) -> AgentLoop:
     provider = MagicMock()
     provider.get_default_model.return_value = "test-model"
@@ -348,7 +364,7 @@ def test_save_turn_skips_multimodal_user_when_only_runtime_context() -> None:
 
     loop._save_turn(
         session,
-        [{"role": "user", "content": [{"type": "text", "text": runtime}]}],
+        [{"role": "user", "content": [_host_text_block(runtime)]}],
         skip=0,
     )
     assert session.messages == []
@@ -365,7 +381,7 @@ def test_save_turn_keeps_image_placeholder_with_path_after_runtime_strip() -> No
             "role": "user",
             "content": [
                 {"type": "image_url", "image_url": {"url": "data:image/png;base64,abc"}, "_meta": {"path": "/media/feishu/photo.jpg"}},
-                {"type": "text", "text": runtime},
+                _host_text_block(runtime),
             ],
         }],
         skip=0,
@@ -384,7 +400,7 @@ def test_save_turn_keeps_image_placeholder_without_meta() -> None:
             "role": "user",
             "content": [
                 {"type": "image_url", "image_url": {"url": "data:image/png;base64,abc"}},
-                {"type": "text", "text": runtime},
+                _host_text_block(runtime),
             ],
         }],
         skip=0,
@@ -392,23 +408,73 @@ def test_save_turn_keeps_image_placeholder_without_meta() -> None:
     assert session.messages[0]["content"] == [{"type": "text", "text": "[image]"}]
 
 
-def test_save_turn_strips_runtime_context_suffix_from_string() -> None:
+def test_save_turn_strips_host_guidance_suffix_from_string() -> None:
     loop = _mk_loop()
     session = Session(key="test:suffix-strip")
+    guidance = ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG + "\ninternal guidance"
     runtime = (
         ContextBuilder._RUNTIME_CONTEXT_TAG
         + "\nCurrent Time: now\n"
         + ContextBuilder._RUNTIME_CONTEXT_END
     )
+    suffix = f"{guidance}\n\n{runtime}"
 
     loop._save_turn(
         session,
-        [{"role": "user", "content": f"hello world\n\n{runtime}"}],
+        [_host_text_message(f"hello world\n\n{suffix}", suffix)],
         skip=0,
     )
     assert session.messages[0]["content"] == "hello world"
 
 
+def test_build_and_save_preserves_user_text_containing_goal_guidance_tag(tmp_path: Path) -> None:
+    loop = _mk_loop()
+    session = Session(key="test:user-guidance-literal")
+    user_text = (
+        "Keep this prefix\n"
+        f"{ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG}\n"
+        "This label and everything after it are user-authored."
+    )
+    messages = ContextBuilder(tmp_path).build_messages(
+        [],
+        user_text,
+        channel="cli",
+        chat_id="direct",
+        goal_start_requested=True,
+    )
+    assert "_meta" in messages[-1]
+    assert "_meta" not in LLMProvider._sanitize_empty_content(messages)[-1]
+
+    loop._save_turn(session, messages, skip=1)
+
+    assert session.messages[0]["content"] == user_text
+
+
+def test_build_and_save_preserves_multimodal_user_block_starting_with_runtime_tag(
+    tmp_path: Path,
+) -> None:
+    loop = _mk_loop()
+    session = Session(key="test:user-runtime-literal-block")
+    image = tmp_path / "user-tag.png"
+    image.write_bytes(_PNG_1X1)
+    user_text = (
+        f"{ContextBuilder._RUNTIME_CONTEXT_TAG}\n"
+        "This entire block is user-authored and must remain in history."
+    )
+    messages = ContextBuilder(tmp_path).build_messages(
+        [],
+        user_text,
+        media=[str(image)],
+        channel="cli",
+        chat_id="direct",
+        goal_start_requested=True,
+    )
+
+    loop._save_turn(session, messages, skip=1)
+
+    assert {"type": "text", "text": user_text} in session.messages[0]["content"]
+
+
 def test_save_turn_skips_string_user_when_only_runtime_context_suffix() -> None:
     loop = _mk_loop()
     session = Session(key="test:suffix-only")
@@ -420,7 +486,7 @@ def test_save_turn_skips_string_user_when_only_runtime_context_suffix() -> None:
 
     loop._save_turn(
         session,
-        [{"role": "user", "content": runtime}],
+        [_host_text_message(runtime, runtime)],
         skip=0,
     )
     assert session.messages == []

tests/agent/test_subagent_lifecycle.py

@@ -254,10 +254,10 @@ class TestSpawn:
             return AgentRunResult(final_content="done", messages=[], stop_reason="completed")
         sm.runner.run = _slow_run
 
-        long_task = "A" * 50
-        await sm.spawn(long_task, runtime=_runtime(), session_key="s1")
+        long_label_source = "A" * 50
+        await sm.spawn(long_label_source, runtime=_runtime(), session_key="s1")
         status = next(iter(sm._task_statuses.values()))
-        assert status.label == long_task[:30] + "..."
+        assert status.label == long_label_source[:30] + "..."
 
         block.set()
         await _drain_subagent_tasks(sm)

tests/agent/tools/test_long_task.py

@@ -1,4 +1,4 @@
-"""Tests for sustained goal tools (`long_task`, `complete_goal`)."""
+"""Tests for sustained goal tools (``create_goal``, ``update_goal``)."""
 
 from __future__ import annotations
 
@@ -7,42 +7,79 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
+from nanobot.agent.goal_permission import goal_mutation_allowed, goal_mutation_permission
 from nanobot.agent.loop import AgentLoop
-from nanobot.agent.tools.context import RequestContext, request_context
-from nanobot.agent.tools.long_task import (
-    CompleteGoalTool,
-    LongTaskTool,
+from nanobot.agent.tools.context import (
+    RequestContext,
+    current_request_context,
+    request_context,
 )
+from nanobot.agent.tools.long_task import (
+    CreateGoalTool,
+    UpdateGoalTool,
+)
+from nanobot.agent.tools.registry import ToolRegistry
 from nanobot.bus.outbound_events import GoalStateSyncEvent
 from nanobot.bus.queue import MessageBus
 from nanobot.bus.runtime_events import RuntimeEventBus
-from nanobot.session.goal_state import GOAL_STATE_KEY
+from nanobot.session.goal_state import GOAL_STATE_KEY, MAX_GOAL_OBJECTIVE_CHARS
 from nanobot.session.manager import SessionManager
+from nanobot.session.turn_continuation import should_finalize_on_max_iterations
 from nanobot.session.webui_turns import WebuiTurnCoordinator
 
 
-def _tools(sm: SessionManager) -> tuple[LongTaskTool, CompleteGoalTool]:
-    lt = LongTaskTool(sessions=sm)
-    cg = CompleteGoalTool(sessions=sm)
-    return lt, cg
+def _goal_metadata() -> dict[str, object]:
+    return {
+        "original_command": "/goal",
+        "original_content": "/goal implement the agreed plan",
+        "goal_requested": True,
+    }
 
 
-def _request_context(chat_id: str = "c1") -> RequestContext:
+def _request_context(
+    *,
+    chat_id: str = "c1",
+    metadata: dict[str, object] | None = None,
+    original_user_text: str | None = "/goal implement the agreed plan",
+    channel: str = "websocket",
+) -> RequestContext:
     return RequestContext(
-        channel="websocket",
+        channel=channel,
         chat_id=chat_id,
-        session_key=f"websocket:{chat_id}",
-        metadata={},
+        session_key=f"{channel}:{chat_id}",
+        original_user_text=original_user_text,
+        metadata=metadata if metadata is not None else _goal_metadata(),
     )
 
 
-@pytest.mark.asyncio
-async def test_long_task_records_goal_metadata(tmp_path):
-    sm = SessionManager(tmp_path)
-    lt, _cg = _tools(sm)
+def _tools(
+    sm: SessionManager,
+    *,
+    metadata: dict[str, object] | None = None,
+) -> tuple[CreateGoalTool, UpdateGoalTool, RequestContext]:
+    create = CreateGoalTool(sessions=sm)
+    update = UpdateGoalTool(sessions=sm)
+    rc = _request_context(metadata=metadata)
+    return create, update, rc
 
-    with request_context(_request_context()):
-        out = await lt.execute(goal="Do the thing", ui_summary="thing")
+
+async def _execute(tool, ctx: RequestContext, *, allowed: bool = True, **kwargs):
+    with request_context(ctx), goal_mutation_permission(allowed):
+        return await tool.execute(**kwargs)
+
+
+@pytest.mark.asyncio
+async def test_create_goal_records_goal_metadata(tmp_path):
+    sm = SessionManager(tmp_path)
+    create, _update, ctx = _tools(sm)
+    sm.get_or_create("websocket:c1").metadata["_sustained_goal_continuation_rounds"] = 12
+
+    out = await _execute(
+        create,
+        ctx,
+        objective="Do the thing",
+        ui_summary="thing",
+    )
     assert "Goal recorded" in out
 
     sess = sm.get_or_create("websocket:c1")
@@ -51,28 +88,50 @@ async def test_long_task_records_goal_metadata(tmp_path):
     assert blob["status"] == "active"
     assert blob["objective"] == "Do the thing"
     assert blob["ui_summary"] == "thing"
+    assert "_sustained_goal_continuation_rounds" not in sess.metadata
+    assert "_sustained_goal_continuation_rounds" not in (
+        SessionManager(tmp_path).get_or_create("websocket:c1").metadata
+    )
+    assert not should_finalize_on_max_iterations(
+        pending_queue_available=True,
+        session_metadata=sess.metadata,
+    )
 
 
 @pytest.mark.asyncio
-async def test_long_task_rejects_second_active_goal(tmp_path):
+async def test_create_goal_rejects_without_explicit_goal_permission(tmp_path):
     sm = SessionManager(tmp_path)
-    lt, _cg = _tools(sm)
+    create, _update, ctx = _tools(sm)
+    sess = sm.get_or_create("websocket:c1")
+    sess.add_message("user", "/goal implement the old plan")
+    sess.add_message("assistant", "The old goal is complete.")
+    sess.add_message("user", "Handle this as an ordinary one-time task.")
 
-    with request_context(_request_context()):
-        await lt.execute(goal="First")
-        out = await lt.execute(goal="Second")
-    assert "already active" in out
+    out = await _execute(
+        create,
+        ctx,
+        allowed=False,
+        objective="Implement another plan.",
+    )
+
+    assert "create_goal is unavailable for this turn" in str(out)
+    assert "/goal <task>" in str(out)
+    assert GOAL_STATE_KEY not in sess.metadata
 
 
 @pytest.mark.asyncio
-async def test_complete_goal_closes_active_goal(tmp_path):
+async def test_update_goal_complete_closes_active_goal(tmp_path):
     sm = SessionManager(tmp_path)
-    lt, cg = _tools(sm)
+    create, update, ctx = _tools(sm)
+
+    with request_context(ctx), goal_mutation_permission(True):
+        await create.execute(objective="X")
+        out = await update.execute(action="complete", recap="Done.")
+        denied = await create.execute(objective="Another")
+        assert goal_mutation_allowed() is False
 
-    with request_context(_request_context()):
-        await lt.execute(goal="X")
-        out = await cg.execute(recap="Done.")
     assert "marked complete" in out
+    assert "create_goal is unavailable for this turn" in str(denied)
 
     sess = sm.get_or_create("websocket:c1")
     blob = sess.metadata.get(GOAL_STATE_KEY)
@@ -80,55 +139,245 @@ async def test_complete_goal_closes_active_goal(tmp_path):
     assert blob["recap"] == "Done."
 
 
+@pytest.mark.asyncio
+async def test_update_goal_replace_keeps_goal_active_with_new_objective(tmp_path):
+    sm = SessionManager(tmp_path)
+    create, update, ctx = _tools(sm)
+
+    await _execute(create, ctx, objective="Old")
+    sess = sm.get_or_create("websocket:c1")
+    sess.metadata["_sustained_goal_continuation_rounds"] = 12
+    sm.save(sess)
+    out = await _execute(
+        update,
+        _request_context(),
+        action="replace",
+        objective="New",
+        ui_summary="new",
+    )
+
+    assert "Goal replaced" in out
+    blob = sm.get_or_create("websocket:c1").metadata[GOAL_STATE_KEY]
+    assert blob["status"] == "active"
+    assert blob["objective"] == "New"
+    assert blob["previous_objective"] == "Old"
+    assert blob["ui_summary"] == "new"
+    assert "_sustained_goal_continuation_rounds" not in sess.metadata
+    assert "_sustained_goal_continuation_rounds" not in (
+        SessionManager(tmp_path).get_or_create("websocket:c1").metadata
+    )
+    assert not should_finalize_on_max_iterations(
+        pending_queue_available=True,
+        session_metadata=sess.metadata,
+    )
+
+
+@pytest.mark.asyncio
+async def test_goal_state_mutations_roll_back_on_save_failure(tmp_path, monkeypatch):
+    sm = SessionManager(tmp_path)
+    create, update, _context = _tools(sm)
+    sess = sm.get_or_create("websocket:c1")
+    sess.metadata["marker"] = {"keep": True}
+    sess.metadata["_sustained_goal_continuation_rounds"] = 12
+    original_save = sm.save
+    create_context = _request_context()
+
+    def fail_save(_session, **_kwargs):
+        raise OSError("disk unavailable")
+
+    monkeypatch.setattr(sm, "save", fail_save)
+    with pytest.raises(OSError, match="disk unavailable"):
+        await _execute(create, create_context, objective="Old")
+
+    assert sess.metadata == {
+        "marker": {"keep": True},
+        "_sustained_goal_continuation_rounds": 12,
+    }
+    assert GOAL_STATE_KEY not in SessionManager(tmp_path).get_or_create("websocket:c1").metadata
+
+    monkeypatch.setattr(sm, "save", original_save)
+    assert "Goal recorded" in await _execute(create, create_context, objective="Old")
+    sess.metadata["_sustained_goal_continuation_rounds"] = 12
+    sm.save(sess)
+    replace_context = _request_context()
+
+    monkeypatch.setattr(sm, "save", fail_save)
+    with pytest.raises(OSError, match="disk unavailable"):
+        await _execute(update, replace_context, action="replace", objective="New")
+
+    assert sess.metadata[GOAL_STATE_KEY]["objective"] == "Old"
+    assert sess.metadata["_sustained_goal_continuation_rounds"] == 12
+    persisted = SessionManager(tmp_path).get_or_create("websocket:c1").metadata
+    assert persisted[GOAL_STATE_KEY]["objective"] == "Old"
+    assert persisted["_sustained_goal_continuation_rounds"] == 12
+
+    monkeypatch.setattr(sm, "save", original_save)
+    assert "Goal replaced" in await _execute(
+        update,
+        replace_context,
+        action="replace",
+        objective="New",
+    )
+    assert "_sustained_goal_continuation_rounds" not in sess.metadata
+    assert (
+        SessionManager(tmp_path).get_or_create("websocket:c1").metadata[GOAL_STATE_KEY]["objective"]
+        == "New"
+    )
+
+
+@pytest.mark.asyncio
+async def test_goal_tools_reject_oversized_objectives(tmp_path):
+    sm = SessionManager(tmp_path)
+    create = CreateGoalTool(sessions=sm)
+    create_context = _request_context()
+    oversized = "x" * (MAX_GOAL_OBJECTIVE_CHARS + 1)
+
+    create_out = await _execute(create, create_context, objective=oversized)
+
+    assert f"must not exceed {MAX_GOAL_OBJECTIVE_CHARS}" in str(create_out)
+    assert GOAL_STATE_KEY not in sm.get_or_create("websocket:c1").metadata
+    assert "Goal recorded" in await _execute(
+        create,
+        create_context,
+        objective="x" * MAX_GOAL_OBJECTIVE_CHARS,
+    )
+
+    update = UpdateGoalTool(sessions=sm)
+    replace_context = _request_context()
+    replace_out = await _execute(update, replace_context, action="replace", objective=oversized)
+
+    assert f"must not exceed {MAX_GOAL_OBJECTIVE_CHARS}" in str(replace_out)
+    assert len(sm.get_or_create("websocket:c1").metadata[GOAL_STATE_KEY]["objective"]) == (
+        MAX_GOAL_OBJECTIVE_CHARS
+    )
+
+
+@pytest.mark.asyncio
+async def test_active_goal_create_failure_preserves_permission_for_replace(tmp_path):
+    sm = SessionManager(tmp_path)
+    create, update, initial_context = _tools(sm)
+    assert "Goal recorded" in await _execute(create, initial_context, objective="Old")
+
+    replacement_context = _request_context()
+    with request_context(replacement_context), goal_mutation_permission(True):
+        create_out = await create.execute(objective="New")
+        assert goal_mutation_allowed() is True
+        replace_out = await update.execute(action="replace", objective="New")
+        assert goal_mutation_allowed() is True
+
+    assert "already active" in str(create_out)
+    assert "Goal replaced" in replace_out
+
+
+@pytest.mark.asyncio
+async def test_update_goal_replace_requires_explicit_goal_permission(tmp_path):
+    sm = SessionManager(tmp_path)
+    create, update, initial_context = _tools(sm)
+    assert "Goal recorded" in await _execute(create, initial_context, objective="Old")
+    ordinary_context = _request_context(original_user_text="Continue the existing objective.")
+
+    unauthorized = await _execute(
+        update,
+        ordinary_context,
+        allowed=False,
+        action="replace",
+        objective="Unrequested",
+    )
+
+    assert "replacing the goal is unavailable for this turn" in str(unauthorized)
+    assert "/goal <task>" in str(unauthorized)
+    assert sm.get_or_create("websocket:c1").metadata[GOAL_STATE_KEY]["objective"] == "Old"
+    replace_context = _request_context()
+
+    with request_context(replace_context), goal_mutation_permission(True):
+        assert "Goal replaced" in await update.execute(action="replace", objective="New")
+        reused = await update.execute(action="replace", objective="Another")
+        assert goal_mutation_allowed() is True
+
+    assert "Goal replaced" in reused
+    assert sm.get_or_create("websocket:c1").metadata[GOAL_STATE_KEY]["objective"] == "Another"
+
+
 @pytest.mark.asyncio
 async def test_goal_tools_keep_request_context_per_task(tmp_path):
     sm = SessionManager(tmp_path)
-    lt = LongTaskTool(sessions=sm)
-    cg = CompleteGoalTool(sessions=sm)
-    ctx_a = RequestContext(channel="websocket", chat_id="a", session_key="websocket:a")
-    ctx_b = RequestContext(channel="websocket", chat_id="b", session_key="websocket:b")
+    create = CreateGoalTool(sessions=sm)
+    update = UpdateGoalTool(sessions=sm)
+    ctx_a = RequestContext(
+        channel="websocket",
+        chat_id="a",
+        session_key="websocket:a",
+        metadata=_goal_metadata(),
+    )
+    ctx_b = RequestContext(
+        channel="websocket",
+        chat_id="b",
+        session_key="websocket:b",
+        metadata=_goal_metadata(),
+    )
 
-    async def start_goal(ctx: RequestContext, goal: str) -> str:
-        with request_context(ctx):
-            return await lt.execute(goal=goal)
-
-    task_a = asyncio.create_task(start_goal(ctx_a, "Goal A"))
-    task_b = asyncio.create_task(start_goal(ctx_b, "Goal B"))
+    task_a = asyncio.create_task(_execute(create, ctx_a, objective="Goal A"))
+    task_b = asyncio.create_task(_execute(create, ctx_b, objective="Goal B"))
     await asyncio.gather(task_a, task_b)
 
     assert sm.get_or_create("websocket:a").metadata[GOAL_STATE_KEY]["objective"] == "Goal A"
     assert sm.get_or_create("websocket:b").metadata[GOAL_STATE_KEY]["objective"] == "Goal B"
 
-    async def complete_goal(ctx: RequestContext, recap: str) -> str:
-        with request_context(ctx):
-            return await cg.execute(recap=recap)
+    a_revoked = asyncio.Event()
 
-    done_a = asyncio.create_task(complete_goal(ctx_a, "Done A"))
-    done_b = asyncio.create_task(complete_goal(ctx_b, "Done B"))
-    await asyncio.gather(done_a, done_b)
+    async def complete_a() -> None:
+        with request_context(ctx_a), goal_mutation_permission(True):
+            await update.execute(action="complete", recap="Done A")
+            assert goal_mutation_allowed() is False
+            a_revoked.set()
+
+    async def replace_b() -> None:
+        with request_context(ctx_b), goal_mutation_permission(True):
+            await a_revoked.wait()
+            assert goal_mutation_allowed() is True
+            await update.execute(action="replace", objective="Goal B2")
+
+    await asyncio.gather(complete_a(), replace_b())
 
     assert sm.get_or_create("websocket:a").metadata[GOAL_STATE_KEY]["recap"] == "Done A"
-    assert sm.get_or_create("websocket:b").metadata[GOAL_STATE_KEY]["recap"] == "Done B"
+    assert sm.get_or_create("websocket:b").metadata[GOAL_STATE_KEY]["objective"] == "Goal B2"
 
 
 @pytest.mark.asyncio
-async def test_goal_tools_share_authoritative_request_context(tmp_path):
-    """Both goal tools resolve routing from the same request snapshot."""
+async def test_registry_does_not_reuse_goal_context_after_request_scope(tmp_path):
     sm = SessionManager(tmp_path)
-    lt = LongTaskTool(sessions=sm)
-    cg = CompleteGoalTool(sessions=sm)
-    ctx = RequestContext(channel="websocket", chat_id="a", session_key="websocket:a")
+    create = CreateGoalTool(sessions=sm)
+    update = UpdateGoalTool(sessions=sm)
+    registry = ToolRegistry()
+    registry.register(create)
+    registry.register(update)
+    sess = sm.get_or_create("websocket:c1")
+    sess.metadata[GOAL_STATE_KEY] = {"status": "active", "objective": "Old"}
+    sm.save(sess)
+    ctx = _request_context()
 
-    with request_context(ctx):
-        assert lt._session() is sm.get_or_create("websocket:a")
-        assert cg._session() is sm.get_or_create("websocket:a")
+    with request_context(ctx), goal_mutation_permission(True):
+        create_out = await registry.execute("create_goal", {"objective": "New"})
+        complete_out = await registry.execute(
+            "update_goal",
+            {"action": "complete", "recap": "Old goal done."},
+        )
+        denied_out = await registry.execute("create_goal", {"objective": "Denied"})
+        assert goal_mutation_allowed() is False
 
-    assert lt._session() is None
-    assert cg._session() is None
+    assert "already active" in str(create_out)
+    assert "marked complete" in str(complete_out)
+    assert "create_goal is unavailable for this turn" in str(denied_out)
+    assert current_request_context() is None
+
+    leaked_out = await registry.execute("create_goal", {"objective": "Leaked"})
+
+    assert "missing routing context" in str(leaked_out)
+    assert sess.metadata[GOAL_STATE_KEY]["status"] == "completed"
 
 
 @pytest.mark.asyncio
-async def test_long_task_publishes_goal_state_ws_after_save(tmp_path):
+async def test_goal_state_events_publish_active_then_inactive(tmp_path):
     bus = MagicMock()
     bus.publish_outbound = AsyncMock()
     runtime_events = RuntimeEventBus()
@@ -138,15 +387,15 @@ async def test_long_task_publishes_goal_state_ws_after_save(tmp_path):
         sessions=sm,
         schedule_background=lambda _coro: None,
     ).subscribe(runtime_events)
-    lt = LongTaskTool(sessions=sm, runtime_events=runtime_events)
-    rc = RequestContext(
-        channel="websocket",
-        chat_id="chat-99",
-        session_key="websocket:chat-99",
-        metadata={},
+    create = CreateGoalTool(sessions=sm, runtime_events=runtime_events)
+    update = UpdateGoalTool(sessions=sm, runtime_events=runtime_events)
+    rc = _request_context(chat_id="chat-99")
+    await _execute(
+        create,
+        rc,
+        objective="Objective alpha",
+        ui_summary="alpha",
     )
-    with request_context(rc):
-        await lt.execute(goal="Objective alpha", ui_summary="alpha")
 
     bus.publish_outbound.assert_awaited_once()
     call = bus.publish_outbound.await_args.args[0]
@@ -159,31 +408,17 @@ async def test_long_task_publishes_goal_state_ws_after_save(tmp_path):
         "objective": "Objective alpha",
     }
 
-
-@pytest.mark.asyncio
-async def test_complete_goal_publishes_inactive_goal_state_ws(tmp_path):
-    bus = MagicMock()
-    bus.publish_outbound = AsyncMock()
-    runtime_events = RuntimeEventBus()
-    sm = SessionManager(tmp_path)
-    WebuiTurnCoordinator(
-        bus=bus,
-        sessions=sm,
-        schedule_background=lambda _coro: None,
-    ).subscribe(runtime_events)
-    lt = LongTaskTool(sessions=sm, runtime_events=runtime_events)
-    cg = CompleteGoalTool(sessions=sm, runtime_events=runtime_events)
-    rc = RequestContext(
-        channel="websocket",
-        chat_id="chat-z",
-        session_key="websocket:chat-z",
-        metadata={},
+    bus.publish_outbound.reset_mock()
+    await _execute(
+        update,
+        RequestContext(
+            channel="websocket",
+            chat_id="chat-99",
+            session_key="websocket:chat-99",
+        ),
+        action="complete",
+        recap="Done.",
     )
-    with request_context(rc):
-        await lt.execute(goal="X")
-
-        bus.publish_outbound.reset_mock()
-        await cg.execute(recap="Done.")
 
     bus.publish_outbound.assert_awaited_once()
     call = bus.publish_outbound.await_args.args[0]
@@ -192,32 +427,42 @@ async def test_complete_goal_publishes_inactive_goal_state_ws(tmp_path):
 
 
 @pytest.mark.asyncio
-async def test_complete_goal_without_active_is_noop_message(tmp_path):
+async def test_update_goal_without_active_is_noop_message(tmp_path):
     sm = SessionManager(tmp_path)
-    _lt, cg = _tools(sm)
+    _create, update, ctx = _tools(sm)
 
-    with request_context(_request_context()):
-        out = await cg.execute(recap="n/a")
+    out = await _execute(update, ctx, action="complete", recap="n/a")
     assert "No active" in out
 
 
 @pytest.mark.asyncio
-async def test_long_task_skips_ws_publish_without_bus(tmp_path):
-    sm = SessionManager(tmp_path)
-    lt, _cg = _tools(sm)
-    with request_context(_request_context()):
-        out = await lt.execute(goal="Solo", ui_summary="s")
-    assert "Goal recorded" in out
-
-
-@pytest.mark.asyncio
-async def test_long_task_and_complete_goal_registered(tmp_path):
+async def test_goal_tools_registered_in_base_registry(tmp_path):
     bus = MessageBus()
     provider = MagicMock()
     provider.get_default_model.return_value = "test-model"
     loop = AgentLoop(bus=bus, provider=provider, workspace=tmp_path, model="test-model")
 
-    lt = loop.tools.get("long_task")
-    cg = loop.tools.get("complete_goal")
-    assert lt is not None and lt.name == "long_task"
-    assert cg is not None and cg.name == "complete_goal"
+    create = loop.tools.get("create_goal")
+    update = loop.tools.get("update_goal")
+    assert create is not None and create.name == "create_goal"
+    assert update is not None and update.name == "update_goal"
+    assert set(create.parameters["properties"]) == {"objective", "ui_summary"}
+    assert create.parameters["required"] == ["objective"]
+    assert (
+        create.parameters["properties"]["objective"]["maxLength"]
+        == MAX_GOAL_OBJECTIVE_CHARS
+    )
+    assert (
+        update.parameters["properties"]["objective"]["maxLength"]
+        == MAX_GOAL_OBJECTIVE_CHARS
+    )
+    model_visible_contract = " ".join(
+        (
+            create.description,
+            str(create.parameters),
+            update.description,
+            str(update.parameters),
+        )
+    ).lower()
+    assert "authoriz" not in model_visible_contract
+    assert "/goal" not in model_visible_contract

tests/command/test_model_command.py

@@ -3,6 +3,7 @@ from unittest.mock import MagicMock
 
 import pytest
 
+from nanobot.agent.goal_permission import goal_mutation_allowed
 from nanobot.agent.loop import AgentLoop
 from nanobot.bus.events import InboundMessage
 from nanobot.bus.queue import MessageBus
@@ -59,6 +60,7 @@ def _ctx_session(loop: AgentLoop, raw: str, args: str = "") -> CommandContext:
     msg = InboundMessage(channel="cli", sender_id="user", chat_id="direct", content=raw)
     return CommandContext(
         msg=msg, session=MagicMock(), key=msg.session_key, raw=raw, args=args, loop=loop,
+        is_user_turn=True,
     )
 
 
@@ -183,16 +185,20 @@ async def test_goal_command_rejects_mid_turn_without_session(tmp_path) -> None:
 
 
 @pytest.mark.asyncio
-async def test_goal_command_rewrites_to_agent_prompt(tmp_path) -> None:
+async def test_goal_command_marks_turn_and_preserves_explicit_request(tmp_path) -> None:
     loop = _make_loop(tmp_path)
     ctx = _ctx_session(loop, "/goal audit the repo", args="audit the repo")
     out = await cmd_goal(ctx)
     assert out is None
-    assert "audit the repo" in ctx.msg.content
-    assert "long_task" in ctx.msg.content
+    assert ctx.msg.content == "/goal audit the repo"
     assert ctx.msg.metadata.get("original_command") == "/goal"
     assert ctx.msg.metadata.get("original_content") == "/goal audit the repo"
+    assert ctx.msg.metadata.get("goal_requested") is True
     assert isinstance(ctx.msg.metadata.get("goal_started_at"), int | float)
+    assert len(ctx.turn_scopes) == 1
+    with ctx.turn_scopes[0]:
+        assert goal_mutation_allowed() is True
+    assert goal_mutation_allowed() is False
 
 
 @pytest.mark.asyncio
@@ -204,6 +210,35 @@ async def test_goal_command_registered_on_router(tmp_path) -> None:
     out = await router.dispatch(ctx)
     assert out is None
     assert "ship it" in ctx.msg.content
+    assert len(ctx.turn_scopes) == 1
+    with ctx.turn_scopes[0]:
+        assert goal_mutation_allowed() is True
+    assert goal_mutation_allowed() is False
+
+
+@pytest.mark.asyncio
+async def test_goal_command_does_not_allow_internal_turn(tmp_path) -> None:
+    loop = _make_loop(tmp_path)
+    ctx = CommandContext(
+        msg=InboundMessage(
+            channel="cli",
+            sender_id="system",
+            chat_id="direct",
+            content="/goal internal work",
+        ),
+        session=MagicMock(),
+        key="cli:direct",
+        raw="/goal internal work",
+        args="internal work",
+        loop=loop,
+        is_user_turn=False,
+    )
+
+    out = await cmd_goal(ctx)
+
+    assert out is not None
+    assert "only be started by a user" in out.content
+    assert ctx.turn_scopes == []
 
 
 def test_goal_command_in_help_and_palette() -> None:

tests/command/test_skill_command.py

@@ -126,6 +126,14 @@ async def test_skill_command_no_render_as_text(tmp_path: Path) -> None:
     assert out.metadata.get("render_as") != "text"
 
 
+@pytest.mark.asyncio
+async def test_skill_command_does_not_list_goal_runtime_protocol(tmp_path: Path) -> None:
+    loop = _make_loop(tmp_path)
+    out = await cmd_skill(_ctx(loop))
+
+    assert "long-goal" not in out.content
+
+
 @pytest.mark.asyncio
 async def test_skill_command_registered_on_router(tmp_path: Path) -> None:
     router = CommandRouter()

tests/session/test_goal_state.py

@@ -4,7 +4,9 @@ from __future__ import annotations
 
 from nanobot.session.goal_state import (
     GOAL_STATE_KEY,
+    MAX_GOAL_OBJECTIVE_CHARS,
     discard_legacy_goal_state_key,
+    explicit_goal_requested,
     goal_state_runtime_lines,
     goal_state_ws_blob,
     parse_goal_state,
@@ -40,6 +42,16 @@ def test_runtime_lines_include_objective_when_active():
     assert any("Summary: fix" in ln for ln in lines)
 
 
+def test_runtime_lines_preserve_maximum_accepted_objective():
+    objective = "x" * MAX_GOAL_OBJECTIVE_CHARS
+
+    lines = goal_state_runtime_lines(
+        {GOAL_STATE_KEY: {"status": "active", "objective": objective}}
+    )
+
+    assert lines == ["Goal (active):", objective]
+
+
 def test_runtime_lines_read_legacy_thread_goal_key():
     meta = {"thread_goal": {"status": "active", "objective": "Legacy key.", "ui_summary": "L"}}
     lines = goal_state_runtime_lines(meta)
@@ -109,6 +121,12 @@ def test_sustained_goal_active_respects_legacy_thread_goal_key():
     assert sustained_goal_active(meta) is True
 
 
+def test_explicit_goal_requested_only_reads_command_metadata():
+    assert explicit_goal_requested({}) is False
+    message_meta = {"original_command": "/goal", "goal_requested": True}
+    assert explicit_goal_requested(message_meta) is True
+
+
 def test_runner_wall_llm_timeout_uses_metadata_override(tmp_path):
     sm = SessionManager(tmp_path)
     assert (

tests/session/test_turn_continuation.py

@@ -8,7 +8,11 @@ from types import SimpleNamespace
 import pytest
 
 from nanobot.bus.events import InboundMessage
-from nanobot.session.goal_state import GOAL_STATE_KEY
+from nanobot.session.goal_state import (
+    GOAL_STATE_KEY,
+    explicit_goal_requested,
+    sustained_goal_turn,
+)
 from nanobot.session.turn_continuation import (
     INTERNAL_CONTINUATION_KIND_META,
     INTERNAL_CONTINUATION_META,
@@ -50,6 +54,8 @@ async def test_maybe_continue_turn_queues_internal_message():
                 "origin_message_id": "msg-0",
                 "_wants_stream": True,
                 "webui": True,
+                "original_command": "/goal",
+                "goal_requested": True,
             },
         ),
         session_key="feishu:c1",
@@ -74,6 +80,8 @@ async def test_maybe_continue_turn_queues_internal_message():
     assert queued.metadata["message_id"] == "msg-1"
     assert queued.metadata["origin_message_id"] == "msg-0"
     assert queued.metadata["_wants_stream"] is True
+    assert not explicit_goal_requested(queued.metadata)
+    assert sustained_goal_turn(meta, message_metadata=queued.metadata)
     assert "Finish the migration." in queued.content
     assert ctx.all_messages == messages[:-1]
     assert ctx.final_content == ""

tests/tools/test_filesystem_tools.py

@@ -79,10 +79,10 @@ class TestReadFileTool:
 
     @pytest.mark.asyncio
     async def test_workspace_relative_builtin_skill_read_falls_back_to_packaged_skill(self, tool):
-        result = await tool.execute(path="skills/long-goal/SKILL.md", limit=5)
+        result = await tool.execute(path="skills/cron/SKILL.md", limit=5)
 
         assert "Error" not in result
-        assert "long-goal" in result.lower()
+        assert "cron" in result.lower()
 
     @pytest.mark.asyncio
     async def test_missing_path_returns_clear_error(self, tool):