From ee646fc8edcd97d4584a7970debf28368089a182 Mon Sep 17 00:00:00 2001 From: EMRG Evolution Date: Thu, 22 Oct 2026 21:32:18 +0800 Subject: [PATCH] emrg: the owner chooses its own question, and is not asked twice inside four hours --- emrg/server/daemon.py | 78 +++++++++++++---- emrg/server/prompts/boss.j2 | 4 - emrg/session.py | 45 ++++++++++ emrg/tools/boss_tool.py | 57 ++++++------- tests/test_boss_tool.py | 166 +++++++++++++++++++++++++++++++++--- 5 files changed, 289 insertions(+), 61 deletions(-) diff --git a/emrg/server/daemon.py b/emrg/server/daemon.py index a772d63d..5ca7b3e2 100644 --- a/emrg/server/daemon.py +++ b/emrg/server/daemon.py @@ -25,7 +25,7 @@ import sys import time import traceback -from datetime import datetime +from datetime import datetime, timezone from pathlib import Path from types import SimpleNamespace from typing import Any, Optional @@ -169,7 +169,7 @@ def resolve_client_tier(session, req_sandbox: str | None = None) -> str: from emrg.tools.grep_tool import GrepTool from emrg.tools.submit_rant_tool import SubmitRantTool from emrg.tools.subagent_tool import SubagentTool -from emrg.tools.boss_tool import BossTool +from emrg.tools.boss_tool import BUSY_MARKER, REASK_WINDOW_S, BossTool from emrg.skills.loader import load_skills from emrg.skills.registry import ensure_catalog_file, load_catalog_skills, skill_is_managed from emrg.server.rants import append_rant, list_rants @@ -8027,7 +8027,7 @@ def _subagent_report( BOSS_DIR = "sub-boss" async def _run_boss_tool( - self, *, parent_session_id: str, parent_cwd: str, focus: str = "", + self, *, parent_session_id: str, parent_cwd: str, timeout_s: int, intent: str = "", parent_tools: list[str] | None = None, parent_sandbox: str | None = None, ) -> dict: @@ -8046,7 +8046,6 @@ async def _run_boss_tool( child_task = asyncio.create_task(self._run_boss_child( parent_session_id=parent_session_id, parent_cwd=parent_cwd, - focus=focus, intent=intent, state=state, parent_tools=parent_tools, @@ -8071,12 +8070,12 @@ async def _run_boss_tool( return {"error": ( f"boss: the owner did not answer within {timeout_s}s, so its turn was " "cancelled. Nothing from it was delivered. Owner session: " - f"{state['child']}. Raise `timeout_s`, or narrow `focus` to the one " - "thing you need judged." + f"{state['child']}. Raise `timeout_s` if the owner needs longer to read " + "this session's history." )} async def _run_boss_child( - self, *, parent_session_id: str, parent_cwd: str, focus: str, + self, *, parent_session_id: str, parent_cwd: str, intent: str, state: dict, parent_tools: list[str] | None = None, parent_sandbox: str | None = None, ) -> dict: @@ -8109,19 +8108,39 @@ async def _run_boss_child( boss_cwd.mkdir(parents=True, exist_ok=True) parent_history = parent_session_dir / "history.jsonl" - prompt = _get_jinja_env().get_template("boss.j2").render( - parent_history=str(parent_history), - parent_session=parent_session_id, - # Read by the template's own `{% if focus %}`, so an empty string and a - # missing key agree: no particular question was asked. - focus=focus or "", - ) - # Load-or-create, never delete: the retained session keeps its history, its # memory and its own compaction between rounds. That is what makes the # several thousand characters of mission a cost paid once rather than # every round (issue #2467, requirement 3). child = self._get_or_create_session(child_id, boss_cwd) + + # The quiet window (issue #2495). Direction does not change inside four + # hours, so a second call in that time would spend a whole owner turn to + # re-buy a judgement that cannot have moved. The clock is the owner + # session's own `meta.json`, so a restart neither resets nor forgets it, + # and it is read before anything is started: no turn, no prompt, no depth. + asked_at = time.time() + last_asked = child.boss_asked_at + if last_asked is not None and asked_at - last_asked < REASK_WINDOW_S: + free_at = last_asked + REASK_WINDOW_S + logger.info( + "[boss] busy: parent=%s child=%s remaining=%.0fs", + parent_session_id, child_id, free_at - asked_at, + ) + return {"text": self._boss_busy_report( + child_id, last_asked, free_at, free_at - asked_at, + )} + + # Stamped before the turn runs, never after it: a turn that dies mid-way + # must still leave the clock written, or the next call pays for the same + # judgement again with nothing to show for the first. + child.set_boss_asked_at(asked_at) + + prompt = _get_jinja_env().get_template("boss.j2").render( + parent_history=str(parent_history), + parent_session=parent_session_id, + ) + 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. @@ -8218,6 +8237,35 @@ def _boss_report( f"{answer}" ) + @staticmethod + def _boss_busy_report( + child_id: str, last_asked: float, free_at: float, remaining: float, + ) -> str: + """The answer a call inside the quiet window gets: a notice, and no stance. + + An ordinary result rather than an error, because a busy owner is not a + failure of the round that asked — and its first line carries `BUSY_MARKER` + so that "the owner was left alone" is readable from the answer's shape. + The executor is told in as many words that no direction was decided, since + the tool otherwise promises every answer is a stance. + """ + def _stamp(when: float) -> str: + return datetime.fromtimestamp(when, tz=timezone.utc).strftime( + "%Y-%m-%dT%H:%M:%SZ" + ) + + return ( + f"{BUSY_MARKER} — the owner was left alone; this is not a stance.\n" + f"- owner session: {child_id}\n" + f"- last asked: {_stamp(last_asked)}\n" + f"- free at: {_stamp(free_at)}\n" + f"- remaining: {int(remaining)}s\n\n" + "No question was put to the owner this round, so no direction was " + "decided: carry on with the plan you already have, or finish and " + "report. Ask again once the window has passed if the direction " + "genuinely has to change." + ) + def _maybe_reflect_memory( self, session: Session, user_prompt: str, assistant_content: str, tool_names: list[str] | None = None, diff --git a/emrg/server/prompts/boss.j2 b/emrg/server/prompts/boss.j2 index 91ffce06..80109e79 100644 --- a/emrg/server/prompts/boss.j2 +++ b/emrg/server/prompts/boss.j2 @@ -38,8 +38,4 @@ Write one short answer, in the imperative, addressed to the executor. Do not restate the mission, do not narrate what you read, and do not hedge: an answer that lists considerations without choosing between them leaves the executor where it started. -{% if focus %} -The executor especially wants your judgement on this: -{{ focus }} -{% endif %} diff --git a/emrg/session.py b/emrg/session.py index cfd75c39..afa66bc3 100644 --- a/emrg/session.py +++ b/emrg/session.py @@ -411,6 +411,14 @@ def __init__(self, session_id: str, cwd: Path) -> None: # `subagent` call reads this, and a daemon restart must not reset the # count of a session that already exists on disk. self._depth: int = 0 + # When this session — acting as an owner — was last asked for a stance, in + # seconds since the epoch (issue #2495). An owner is left alone for + # `REASK_WINDOW_S` after a call, because direction does not change inside + # four hours and a second call buys the same judgement again. It lives in + # `meta.json` for the same reason `depth` does: the session is retained on + # disk, so a daemon restart that reset this clock would let the very next + # round re-buy a decision it already paid for. + self._boss_asked_at: float | None = None # Lazy-initialized memory store self._memory_store = None @@ -524,6 +532,15 @@ def load(cls, session_id: str, cwd: Path) -> Session: # a JSON `true` in this field would then read as depth 1. if isinstance(stored_depth, int) and not isinstance(stored_depth, bool): session._depth = max(0, stored_depth) + stored_boss_asked_at = meta.get("boss_asked_at") + # The same care as `depth`: `True` is an int, and a JSON `true` here + # would read as the epoch's first second. Absent, malformed and + # non-numeric all mean "never asked", which is the state that permits + # a call — the safe direction for a value that only ever blocks one. + if isinstance(stored_boss_asked_at, (int, float)) and not isinstance( + stored_boss_asked_at, bool + ): + session._boss_asked_at = float(stored_boss_asked_at) logger.info("session loaded: %s (%d messages)", session_id, session._message_count) return session @@ -592,6 +609,30 @@ def set_depth(self, depth: int) -> None: self._save_meta() logger.info("session delegation depth set: %s -> %d", self.session_id, self._depth) + # ── The owner's clock ───────────────────────────────────── + + @property + def boss_asked_at(self) -> float | None: + """When this session was last asked for a stance, or ``None`` if never. + + Seconds since the epoch, so the quiet window's comparison is arithmetic + rather than a parse: a malformed instant would otherwise have to be told + apart from an absent one at the point where it decides something. + """ + return self._boss_asked_at + + def set_boss_asked_at(self, when: float) -> None: + """Persist the instant this session was last asked, into ``meta.json``. + + The only writer of the key, and it is called *before* the owner's turn + runs rather than after it: a stamp written on completion leaves a turn that + died looking like one that never happened, and the next call would then pay + for the same judgement a second time. + """ + self._boss_asked_at = float(when) + self._save_meta() + logger.info("session owner clock set: %s -> %.0f", self.session_id, self._boss_asked_at) + # ── Sandbox tier ────────────────────────────────────────── @property @@ -1098,6 +1139,10 @@ def _save_meta_with_title(self, title: str | None) -> None: # key saying it is zero deep. if self._depth: meta["depth"] = self._depth + # Same rule again: a session that was never asked for a stance must not + # gain a key claiming one was. + if self._boss_asked_at is not None: + meta["boss_asked_at"] = self._boss_asked_at if title is not None: meta["title"] = title else: diff --git a/emrg/tools/boss_tool.py b/emrg/tools/boss_tool.py index c6f6254f..aeb9079f 100644 --- a/emrg/tools/boss_tool.py +++ b/emrg/tools/boss_tool.py @@ -16,9 +16,15 @@ ``/sub-agent/sub-boss``, so the parent's own session directory is two levels up and its ``history.jsonl`` is right there — the last ``user`` record is the entire task statement, ``extra_prompt`` included. The parent supplies -no text: ``focus`` is an optional *question*, never the brief. Measured on the -evolution task, that statement runs to tens of thousands of characters, and every -one of them is the position this role needs. +no steering text at all, not even a question: naming what the owner should judge is +choosing the direction, and a session that chose the question would be deciding how +its own work is judged. What the caller supplies is ``intent`` — the one-line reason +the log keeps — and nothing else. + +**It is left alone between calls.** Direction does not change inside four hours, so +a call in that window starts no turn and buys no judgement; it answers with +``BUSY_MARKER`` and the time the owner frees up, which is an ordinary result rather +than an error because a busy owner is not a failure of the round that asked. **It is not cleared between rounds.** The nine ``subagent`` slots are least-recently-closed and emptied on use, which is right for a job that ends when @@ -47,15 +53,27 @@ from emrg.tools.base import ToolExecutor #: The runner the daemon supplies: ``await runner(parent_session_id=…, -#: parent_cwd=…, focus=…, timeout_s=…, intent=…, parent_tools=…)`` answering -#: either ``{"text": …}`` or ``{"error": …}``. ``parent_tools`` is the parent -#: turn's effective tool set (#2465), which the child inherits. +#: parent_cwd=…, timeout_s=…, intent=…, parent_tools=…, parent_sandbox=…)`` +#: answering either ``{"text": …}`` or ``{"error": …}``. ``parent_tools`` is the +#: parent turn's effective tool set (#2465), which the child inherits. Runner = Callable[..., Awaitable[dict]] #: Generous enough for a real read of the parent's history, short enough that a #: wedged owner session cannot hold the parent's round for ever. DEFAULT_TIMEOUT_S = 600 +#: How long an owner is left alone after a call (issue #2495). Direction does not +#: change inside four hours, so a second call in that window spends a full owner +#: turn to re-buy a judgement that cannot have changed. The constant lives here, +#: beside the description that tells the caller about it, so the advertised window +#: and the enforced one are one number: the daemon reads this same constant. +REASK_WINDOW_S = 4 * 60 * 60 + +#: What a busy answer opens with, so that "the owner was left alone" is readable +#: from the answer's shape rather than by running the prose through a reader. A +#: stance never begins with it. +BUSY_MARKER = "boss-busy-window" + class BossTool(ToolExecutor): """Ask this task's owner session what the coming rounds' stance is.""" @@ -75,20 +93,14 @@ def definition(self) -> ToolDefinition: "for this round, not a list of problems to consider. Anything the " "answer does not mention is still yours to decide. It is offered " "to scheduled tasks only, keeps its memory across rounds, and its " - "round stands still until it answers." + "round stands still until it answers. It is also left alone for " + f"{REASK_WINDOW_S // 3600}h after a call: an answer that opens with " + f"`{BUSY_MARKER}` is that notice, carries no stance, and means carry " + "on with your own round rather than waiting." ), parameters={ "type": "object", "properties": { - "focus": { - "type": "string", - "description": ( - "Optional. The one thing you want the owner's " - "judgement on — a question, not a summary of the work: " - "it already reads the history. Omit it and the owner " - "states its own stance." - ), - }, "timeout_s": { "type": "integer", "minimum": 1, @@ -123,18 +135,6 @@ async def execute(self, arguments: dict) -> ToolResult: error=True, ) - focus = arguments.get("focus") - if focus is not None and not isinstance(focus, str): - return ToolResult( - name="boss", - content=( - f"boss: `focus` must be a string, so got {focus!r}. It is an " - "optional question, so leaving it out is how a caller states " - "'no particular question'." - ), - error=True, - ) - timeout_s = arguments.get("timeout_s") if timeout_s is None: timeout_s = DEFAULT_TIMEOUT_S @@ -195,7 +195,6 @@ async def execute(self, arguments: dict) -> ToolResult: outcome = await self._runner( parent_session_id=str(parent_session_id), parent_cwd=str(parent_cwd), - focus=(focus or ""), timeout_s=int(timeout_s), intent=intent, parent_tools=parent_tools, diff --git a/tests/test_boss_tool.py b/tests/test_boss_tool.py index 4d7d51c3..dceea8f6 100644 --- a/tests/test_boss_tool.py +++ b/tests/test_boss_tool.py @@ -16,8 +16,11 @@ authorization story: it lives in one whitelist and nowhere else, so a second mechanism inside the tool asking "who called me?" would be a copy of a fact the producer already holds; -* **the arguments** — ``intent`` is required, ``focus`` is optional and is not a - brief, and every refusal names the parameter it refused; +* **the arguments** — ``intent`` is required, the schema carries no ``focus`` at + all, and every refusal names the parameter it refused; +* **the quiet window** — a second call inside four hours starts no owner turn and + says so in a shape a caller can detect; the clock is the session's own, so a + rebuilt session is still busy and a turn that died still leaves its stamp; * **where the mission comes from** — driven for real: the owner's first request does **not** contain the mission and its second one does, which can only happen if the owner read the parent's history itself. A test that asserted the prompt @@ -36,6 +39,7 @@ import asyncio import json +import time from pathlib import Path import pytest @@ -51,7 +55,12 @@ ) from emrg.protocol import TaskRequest from emrg.session import Session -from emrg.tools.boss_tool import DEFAULT_TIMEOUT_S, BossTool +from emrg.tools.boss_tool import ( + BUSY_MARKER, + DEFAULT_TIMEOUT_S, + REASK_WINDOW_S, + BossTool, +) REPO_ROOT = Path(__file__).resolve().parent.parent PROMPTS = REPO_ROOT / "emrg" / "server" / "prompts" @@ -154,16 +163,29 @@ def test_a_call_without_a_parent_session_is_refused(): assert runner.calls == [] -def test_focus_is_optional_and_must_be_a_string(): +def test_the_schema_carries_no_focus_parameter(): + """Issue #2495: choosing the question is choosing the direction. + + Asserted on the schema rather than on behaviour, and the difference is the + guard: the tool ignores keys it does not know, so a test that passed `focus` in + and watched for no effect would stay green with the parameter restored. + """ + properties = BossTool(_Runner()).definition().parameters["properties"] + assert "focus" not in properties, ( + "the executor can still name what the owner should judge, which hands the " + "direction back to the session being directed" + ) + assert set(properties) == {"intent", "timeout_s"} + + +def test_the_runner_is_handed_no_focus(): + """The other half: a removed parameter leaves the call as well as the schema.""" runner = _Runner() asyncio.run(BossTool(runner).execute(_injected())) - assert runner.calls[0]["focus"] == "", ( - "an absent focus must read as 'no particular question', not as a brief" + assert "focus" not in runner.calls[0], ( + "the daemon is still handed a focus, so the parameter is only half gone" ) - - refused = asyncio.run(BossTool(_Runner()).execute(_injected(focus=["a", "b"]))) - assert refused.error - assert "focus" in refused.content + assert "prompt" not in runner.calls[0] @pytest.mark.parametrize("bad", [True, "600", 0, -1, 1.5]) @@ -223,7 +245,7 @@ def test_the_boss_template_is_a_file_the_host_edits(): original = source.read_text(encoding="utf-8") try: rendered = _get_jinja_env().get_template("boss.j2").render( - parent_history="/tmp/session/history.jsonl", focus="", + parent_history="/tmp/session/history.jsonl", ) assert "/tmp/session/history.jsonl" in rendered, ( "the template does not receive the path to the parent's history, so the " @@ -233,7 +255,7 @@ def test_the_boss_template_is_a_file_the_host_edits(): marker = "TEMPLATE-EDIT-IS-LIVE-4b7d" source.write_text(original + "\n" + marker + "\n", encoding="utf-8") again = _get_jinja_env().get_template("boss.j2").render( - parent_history="/tmp/session/history.jsonl", focus="", + parent_history="/tmp/session/history.jsonl", ) assert marker in again, ( "an edit to boss.j2 did not reach the next render — the wording is the " @@ -289,7 +311,7 @@ def _boss_cwd(tmp_path, session_id: str = "task-1") -> Path: def _ask(server: EmrgServer, tmp_path, session_id: str = "task-1") -> dict: return asyncio.run(server._run_boss_child( - parent_session_id=session_id, parent_cwd=str(tmp_path), focus="", + parent_session_id=session_id, parent_cwd=str(tmp_path), intent="ask", state={"child": None}, parent_tools=["read"], )) @@ -458,3 +480,121 @@ def test_an_owner_of_a_task_inherits_the_turns_tier(tmp_path, monkeypatch): "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" ) + + +# ── the quiet window (issue #2495) ──────────────────────────────────── + + +def test_the_quiet_window_is_four_hours(): + """The number the description advertises and the daemon enforces is one number.""" + assert REASK_WINDOW_S == 4 * 60 * 60, ( + "the window moved: the tool's own description is rendered from this " + "constant, so a second number here would be a second answer" + ) + + +def test_a_second_call_inside_the_window_starts_no_owner_turn(tmp_path, monkeypatch): + server = _server(tmp_path, monkeypatch) + seen = _stub_the_loop(server) + + first = _ask(server, tmp_path) + assert not first.get("error"), first + assert BUSY_MARKER not in first["text"], "the first call was answered as busy" + assert len(seen) == 1, "the first call must actually run the owner" + + second = _ask(server, tmp_path) + assert len(seen) == 1, ( + "a second owner turn started inside the window, which pays twice for a " + "judgement that cannot have changed" + ) + assert second["text"].startswith(BUSY_MARKER), second["text"][:80] + + +def test_the_busy_answer_is_a_result_and_says_no_stance_was_decided( + tmp_path, monkeypatch, +): + """A busy owner is not a failure, and the answer must not read as direction.""" + server = _server(tmp_path, monkeypatch) + _stub_the_loop(server) + _ask(server, tmp_path) + + busy = _ask(server, tmp_path) + assert "error" not in busy, ( + "a busy owner came back as an error, so a caller reads a normal wait as a " + "broken round" + ) + assert busy["text"].startswith(BUSY_MARKER) + assert "remaining:" in busy["text"], "the answer does not say how long the wait is" + assert "no direction was decided" in busy["text"], ( + "the answer does not say in as many words that this round has no stance, " + "and the tool otherwise promises every answer is one" + ) + + +def test_the_owner_is_asked_again_once_the_window_has_passed(tmp_path, monkeypatch): + """The window is a bound, not a permanent refusal at the other end.""" + server = _server(tmp_path, monkeypatch) + seen = _stub_the_loop(server) + _ask(server, tmp_path) + + owner = server._get_or_create_session("boss-task-1", _boss_cwd(tmp_path)) + owner.set_boss_asked_at(time.time() - REASK_WINDOW_S - 1) + + again = _ask(server, tmp_path) + assert len(seen) == 2, "the window refused a call that was outside it" + assert BUSY_MARKER not in again["text"] + + +def test_the_window_survives_a_rebuilt_session(tmp_path, monkeypatch): + """A daemon restart must not reset the clock: the owner session lives on disk.""" + first = _server(tmp_path, monkeypatch) + _stub_the_loop(first) + _ask(first, tmp_path) + + reopened = _server(tmp_path, monkeypatch) + seen = _stub_the_loop(reopened) + out = _ask(reopened, tmp_path) + + assert seen == [], "a rebuilt session forgot it had been asked, so it paid again" + assert out["text"].startswith(BUSY_MARKER) + + +def test_a_turn_that_died_still_leaves_the_clock(tmp_path, monkeypatch): + """The stamp is written before the turn runs, never after it.""" + server = _server(tmp_path, monkeypatch) + + async def dead_loop(*args, **kwargs): + raise RuntimeError("the owner died mid-turn") + + server._run_tool_loop_locked = dead_loop + with pytest.raises(RuntimeError): + _ask(server, tmp_path) + + seen = _stub_the_loop(server) + out = _ask(server, tmp_path) + assert seen == [], ( + "the clock was written only once the turn finished, so a turn that died " + "left no trace and the next call bought the same judgement again" + ) + assert out["text"].startswith(BUSY_MARKER) + + +def test_the_clock_is_written_by_its_own_setter(tmp_path): + """The value on disk is the setter's, not a neighbour's save carrying it along. + + Without this arm, deleting the setter's `_save_meta()` survives every test + above: `set_depth` saves the same `meta.json` immediately afterwards and takes + the new key with it, so the call path looks correct while the writer is broken. + Asserted on the file, because the file is the thing the restart reads. + """ + session = Session.create_with_id("boss-clock", tmp_path) + assert session.boss_asked_at is None, "a fresh session claims it was asked" + + session.set_boss_asked_at(1761000000.0) + + meta_path = tmp_path / ".emrg" / "sessions" / "boss-clock" / "meta.json" + meta = json.loads(meta_path.read_text(encoding="utf-8")) + assert meta.get("boss_asked_at") == 1761000000.0, ( + "the setter did not write the clock: on disk is where a restart reads it" + ) + assert session.boss_asked_at == 1761000000.0