emrg: a delegated child inherits the tier its spawning turn runs under

emrg/server/daemon.py

@@ -4370,6 +4370,16 @@ def _inject_tool_arguments(
             args["_parent_tools"] = (
                 list(tool_names) if tool_names is not None else None
             )
+            # The child inherits **this turn's** tier, which is the tier the
+            # parent's own shell calls are running under — `req.sandbox`, the same
+            # value injected three lines above. Reading it from the session object
+            # instead was a hole: a scheduled task's tier is resolved from
+            # `tasks.yml` into the request and never written onto the session, so
+            # every child a task spawned inherited nothing and ran at
+            # `policy.DEFAULT_MODE` — unconfined — while the task body itself was
+            # confined. Injected unconditionally for the reason `_parent_tools` is:
+            # a tier the model could name for its own child is not a boundary.
+            args["_parent_sandbox"] = req.sandbox
 
     async def _apply_escalation(
         self, tc_name: str, args: dict, session, req: TaskRequest,
@@ -7615,6 +7625,7 @@ async def _await_memory_reflection(self, session_id: str) -> None:
     async def _run_subagent_tool(
         self, *, parent_session_id: str, parent_cwd: str, prompt: str,
         timeout_s: int, intent: str = "", parent_tools: list[str] | None = None,
+        parent_sandbox: str | None = None,
     ) -> dict:
         """Run one child session for one `subagent` call. Answers text or an error.
 
@@ -7639,6 +7650,7 @@ async def _run_subagent_tool(
             intent=intent,
             state=state,
             parent_tools=parent_tools,
+            parent_sandbox=parent_sandbox,
         ))
         try:
             done, _ = await asyncio.wait({child_task}, timeout=timeout_s)
@@ -7670,6 +7682,7 @@ async def _run_subagent_tool(
     async def _run_subagent_child(
         self, *, parent_session_id: str, parent_cwd: str, prompt: str,
         intent: str, state: dict, parent_tools: list[str] | None = None,
+        parent_sandbox: str | None = None,
     ) -> dict:
         """The body of one delegation: pick a slot, run the child, report its answer."""
         started = time.monotonic()
@@ -7723,8 +7736,13 @@ async def _run_subagent_child(
             child.set_depth(depth + 1)
             # The child's tier is the parent's, never the child's own: authorization
             # a caller can choose for the thing it is authorizing is not a boundary.
-            if parent.sandbox:
-                child.set_sandbox(parent.sandbox)
+            # The session's own tier first — the source of truth for a client's
+            # session — then the tier this turn is running under, which is the only
+            # one a scheduled task has. Silence in both means unconfined, and that
+            # stays the meaning: `policy.DEFAULT_MODE`.
+            child_tier = getattr(parent, "sandbox", None) or parent_sandbox
+            if child_tier:
+                child.set_sandbox(child_tier)
             if parent.sandbox_roots:
                 try:
                     child.set_sandbox_roots(parent.sandbox_roots)
@@ -7739,7 +7757,7 @@ async def _run_subagent_child(
                 cwd=str(slot_cwd),
                 prompt=prompt,
                 timestamp="",
-                sandbox=parent.sandbox,
+                sandbox=child_tier,
                 # Issue #2465: the child's tool set is the parent's — inherited
                 # the same way its tier is, because authorisation a caller can
                 # choose for the thing it is authorising is not a boundary.
@@ -7824,6 +7842,7 @@ def _subagent_report(
     async def _run_boss_tool(
         self, *, parent_session_id: str, parent_cwd: str, focus: str = "",
         timeout_s: int, intent: str = "", parent_tools: list[str] | None = None,
+        parent_sandbox: str | None = None,
     ) -> dict:
         """Run the owner session for one `boss` call. Answers text or an error.
 
@@ -7844,6 +7863,7 @@ async def _run_boss_tool(
             intent=intent,
             state=state,
             parent_tools=parent_tools,
+            parent_sandbox=parent_sandbox,
         ))
         try:
             done, _ = await asyncio.wait({child_task}, timeout=timeout_s)
@@ -7871,6 +7891,7 @@ async def _run_boss_tool(
     async def _run_boss_child(
         self, *, parent_session_id: str, parent_cwd: str, focus: str,
         intent: str, state: dict, parent_tools: list[str] | None = None,
+        parent_sandbox: str | None = None,
     ) -> dict:
         """The body of one owner call: open the retained session, run it, report."""
         started = time.monotonic()
@@ -7917,8 +7938,11 @@ async def _run_boss_child(
         child.set_depth(depth + 1)
         # The owner's tier is the parent's, never its own: authorization a caller
         # can choose for the thing it is authorizing is not a boundary.
-        if parent.sandbox:
-            child.set_sandbox(parent.sandbox)
+        # Same rule as `subagent`: the session's own tier first, then the tier this
+        # turn runs under — the only one a scheduled task has.
+        child_tier = getattr(parent, "sandbox", None) or parent_sandbox
+        if child_tier:
+            child.set_sandbox(child_tier)
         if parent.sandbox_roots:
             try:
                 child.set_sandbox_roots(parent.sandbox_roots)
@@ -7931,7 +7955,7 @@ async def _run_boss_child(
             cwd=str(boss_cwd),
             prompt=prompt,
             timestamp="",
-            sandbox=parent.sandbox,
+            sandbox=child_tier,
             # The owner inherits the executor's tool set the way any child does
             # (#2465). It includes `boss` itself; the depth limit above is what
             # bounds an owner that asks for an owner.

emrg/tools/boss_tool.py

@@ -177,6 +177,15 @@ async def execute(self, arguments: dict) -> ToolResult:
         # The parent turn's tool set, injected by the daemon (#2465). A value that
         # is not a list is "no statement", the same meaning the message field
         # carries, so silence cannot be read two ways.
+        # The tier this turn is running under, injected by the daemon. A value that
+        # is not a non-empty string is "no statement": silence means the turn itself
+        # declared no tier, which the daemon reads as unconfined.
+        raw_parent_sandbox = arguments.get("_parent_sandbox")
+        parent_sandbox = (
+            raw_parent_sandbox
+            if isinstance(raw_parent_sandbox, str) and raw_parent_sandbox else None
+        )
+
         raw_parent_tools = arguments.get("_parent_tools")
         parent_tools = (
             [str(name) for name in raw_parent_tools]
@@ -190,6 +199,7 @@ async def execute(self, arguments: dict) -> ToolResult:
             timeout_s=int(timeout_s),
             intent=intent,
             parent_tools=parent_tools,
+            parent_sandbox=parent_sandbox,
         )
         if outcome.get("error"):
             return ToolResult(name="boss", content=outcome["error"], error=True)

emrg/tools/subagent_tool.py

@@ -170,6 +170,15 @@ async def execute(self, arguments: dict) -> ToolResult:
         # value that is not a list is treated as "no statement", which the
         # daemon reads as its default set — the same meaning the message field
         # has, so the two cannot disagree about silence.
+        # The tier this turn is running under, injected by the daemon. A value that
+        # is not a non-empty string is "no statement": silence means the turn itself
+        # declared no tier, which the daemon reads as unconfined.
+        raw_parent_sandbox = arguments.get("_parent_sandbox")
+        parent_sandbox = (
+            raw_parent_sandbox
+            if isinstance(raw_parent_sandbox, str) and raw_parent_sandbox else None
+        )
+
         raw_parent_tools = arguments.get("_parent_tools")
         parent_tools = (
             [str(name) for name in raw_parent_tools]
@@ -183,6 +192,7 @@ async def execute(self, arguments: dict) -> ToolResult:
             timeout_s=int(timeout_s),
             intent=str(arguments.get("intent") or ""),
             parent_tools=parent_tools,
+            parent_sandbox=parent_sandbox,
         )
         if outcome.get("error"):
             return ToolResult(name="subagent", content=outcome["error"], error=True)

tests/test_boss_tool.py

@@ -49,6 +49,7 @@
     default_tool_names,
     scheduled_task_tool_names,
 )
+from emrg.protocol import TaskRequest
 from emrg.session import Session
 from emrg.tools.boss_tool import DEFAULT_TIMEOUT_S, BossTool
 
@@ -419,3 +420,41 @@ def test_the_depth_limit_bounds_an_owner_that_asks_for_an_owner(tmp_path, monkey
     assert out.get("error"), "an owner was started past the delegation depth limit"
     assert "delegations deep" in out["error"]
     assert seen == [], "the refused call still ran a turn"
+
+
+def test_an_owner_of_a_task_inherits_the_turns_tier(tmp_path, monkeypatch):
+    """The owner of a scheduled task runs where the task runs, not unconfined.
+
+    A task's tier is resolved from `tasks.yml` into the request and never written
+    onto the session object, so an owner that read only the session inherited
+    nothing — and an owner with a wider tier than its executor is the one shape of
+    this bug that can act on the answer it then gives. Driven through
+    `_inject_tool_arguments` and then the tool, because the injection is the half
+    that was missing.
+    """
+    server = _server(tmp_path, monkeypatch)
+    _stub_the_loop(server)
+    parent = Session.create_with_id("task-tier-owner", tmp_path)
+    assert parent.sandbox is None, (
+        "this test models a scheduled task, whose tier lives on the request"
+    )
+    req = TaskRequest(
+        id="owner-tier-turn", session_id=parent.session_id, cwd=str(tmp_path),
+        prompt="a task's tick", sandbox="read-only",
+    )
+
+    args = {"timeout_s": 10, "intent": "ask the owner"}
+    EmrgServer._inject_tool_arguments("boss", args, parent, req, None)
+    assert args.get("_parent_sandbox") == "read-only", (
+        f"the turn's tier never reached the call: {args}"
+    )
+    result = asyncio.run(BossTool(server._run_boss_tool).execute(args))
+    assert not result.error, result.content
+
+    owner = Session.load(
+        f"boss-{parent.session_id}", _boss_cwd(tmp_path, parent.session_id),
+    )
+    assert owner.sandbox == "read-only", (
+        "the owner of a confined task ran at a wider tier: an owner that may go "
+        "where its executor may not is the escape this inheritance refuses"
+    )

tests/test_subagent_tool.py

@@ -43,6 +43,7 @@
 from emrg.config import LlmConfig
 from emrg.server import daemon as daemon_mod
 from emrg.server.daemon import EmrgServer
+from emrg.protocol import TaskRequest
 from emrg.session import Session
 from emrg.tools.subagent_tool import DEFAULT_TIMEOUT_S, SubagentTool
 
@@ -524,3 +525,45 @@ async def runner(**kwargs):
     assert seen["timeout_s"] == DEFAULT_TIMEOUT_S
     assert seen["parent_session_id"] == "p"
     assert seen["intent"] == "because"
+
+
+def test_a_task_delegation_inherits_the_turns_tier(tmp_path, monkeypatch):
+    """A scheduled task's tier reaches its child, and it is the *turn's* tier.
+
+    A task's tier is resolved from `tasks.yml` into the request and never written
+    onto the session object, so a child that read the session inherited nothing and
+    ran unconfined while the task body itself was confined — the escalation this
+    test exists to refuse. Driven through `_inject_tool_arguments` and then the
+    tool, never by handing the tier straight to the runner: the injection is the
+    half that was missing, and a test below it would have stayed green.
+    """
+    server = _server(tmp_path, monkeypatch)
+    _stub_reply(server, "child answer")
+    parent = Session.create_with_id("parent-task-turn", tmp_path)
+    assert parent.sandbox is None, (
+        "this test models a scheduled task, whose tier lives on the request; a "
+        "parent that carries one is the client shape the other test covers"
+    )
+    req = TaskRequest(
+        id="tier-turn", session_id=parent.session_id, cwd=str(tmp_path),
+        prompt="a task's tick", sandbox="read-only",
+    )
+
+    args = {"prompt": "do the job", "timeout_s": 10}
+    EmrgServer._inject_tool_arguments("subagent", args, parent, req, None)
+
+    assert args.get("_parent_sandbox") == "read-only", (
+        "the turn's tier never reached the call, so the child cannot inherit it: "
+        f"{args}"
+    )
+    result = _run(SubagentTool(server._run_subagent_tool).execute(args))
+    assert not result.error, result.content
+
+    child = Session.load(
+        _report_field(result.content, "session"),
+        Path(_report_field(result.content, "cwd")),
+    )
+    assert child.sandbox == "read-only", (
+        "a child of a confined task ran at a wider tier; a delegation is the same "
+        "authorisation one level down, never a way around it"
+    )