toolprune: protect Codex packages, and recheck protection when a list is used

plugin/README.md

@@ -730,7 +730,7 @@ and are shared by every agent working in it. The examples use Claude Code's form
 | `toolprune_min_sessions` | `2` | sessions those prompts must span; `1` lets a single session's prompts decide |
 | `toolprune_sessions` | `5` | or this many sessions, however few prompts they ran |
 | `toolprune_quiet_minutes` | `60` | how long the repository's sessions must have been idle before `auto` writes a deny list, so that no conversation resumed while its prompt cache is warm loses it (a list is written at the latest a day after it formed); `0` writes at once |
-| `toolprune_protect` | none | tools never left out however long they go unused, by name or pattern (`mcp__headroom__*`), separated by commas; a deny rule the plugin added for one is removed at the next fold, and rules you wrote are never touched |
+| `toolprune_protect` | none | tools never left out however long they go unused, by name or pattern (`mcp__headroom__*`), separated by commas (on Codex, capability packages: `collaboration`, `goals`, `image_media`); a deny rule the plugin added for one is removed at the next fold, and rules you wrote are never touched |
 | `loopguard_aggr` | `default` | the guard's threshold preset: `conservative`, `default` or `aggressive` |
 | `loopguard_nudges` | `1` | redirects per turn before the guard only tells you |
 | `loopguard.<name>` | from the preset | one guard threshold, set by hand (below) |

tests/test_live_hooks.py

@@ -726,6 +726,47 @@ def start(sid):
     assert "removed 3 tool(s)" in start("n2")["systemMessage"]
 
 
+def test_a_protected_codex_package_in_a_committed_or_unreadable_config_stays(
+        env, repo, tmp_path, codex_accepts, monkeypatch):
+    """Giving a package back never edits a config the repository
+    commits, and a config it cannot read leaves the fold to finish."""
+    st = store_for(env, repo, "codex")
+    denylist.apply(repo, ["collaboration"], st, host="codex")
+    path = denylist.settings_path(repo, "codex")
+    before = open(path).read()
+    monkeypatch.setattr(denylist, "_git_tracked", lambda root, rel: True)   # committed since
+    env = dict(env, TE_TOOLPRUNE="propose", TE_TOOLPRUNE_PROTECT="collaboration")
+    run_session(env, repo, "cx0", codex_rollout(str(tmp_path / "rollout-0.jsonl"), repo, 0),
+                host="codex")
+    assert open(path).read() == before
+    monkeypatch.setattr(denylist, "_git_tracked", lambda root, rel: False)
+
+    def unreadable(*a, **k):
+        raise PermissionError("denied")
+    monkeypatch.setattr(denylist, "restore", unreadable)
+    run_session(env, repo, "cx1", codex_rollout(str(tmp_path / "rollout-1.jsonl"), repo, 1),
+                host="codex")
+    assert st.read("untool.json")["sessions"] == 2                 # the fold finished
+
+
+def test_copilot_server_lists_are_not_filtered_as_tool_names(env, repo):
+    """`sql` is a protected Copilot tool and also a valid MCP server name in its lists."""
+    ctx = LiveContext.build(hosts.Copilot(), repo, env)
+    assert fold.unprotected(ctx, ["sql", "github-mcp-server"]) == ["sql", "github-mcp-server"]
+
+
+def test_a_tool_protected_after_the_list_formed_is_not_written(env, repo, tmp_path):
+    """A pending list, and the proposal `toolprune apply` reads, are
+    checked against the protection in force when they are used, not when they formed."""
+    from tokeneconomizer.plugin import cli
+    env, ctx = _pending_repo(env, repo, tmp_path)
+    assert ctx.config.set("toolprune_protect", "Workflow,Share*")
+    assert cli._proposal(ctx, {"proposal": ["CronCreate", "Workflow"]}) == ["CronCreate"]
+    assert fold.apply_pending(ctx, now=time.time() + 3700) is None
+    assert _deny(repo) == ["CronCreate"]
+    assert ctx.store.read("untool.json")["proposal"] == []      # applied, or protected now
+
+
 def test_a_session_still_writing_its_log_keeps_the_deny_list_waiting(env, repo, tmp_path):
     """Activity is a hook call or a line in the session's own log: a session can go an hour
     without a hook (one long answer, or the loop guard off) and still have a warm cache."""
@@ -1269,6 +1310,41 @@ def test_codex_proposes_capabilities_it_never_saw_used(env, repo, tmp_path):
     assert not os.path.exists(denylist.settings_path(repo, "codex"))      # nothing written yet
 
 
+def test_codex_never_proposes_a_protected_capability(env, repo, tmp_path):
+    """`toolprune_protect` covers Codex's capability packages too, by
+    name or pattern, as it covers single tools on the other agents, from the next fold on."""
+    env = dict(env, TE_TOOLPRUNE="propose")
+    for i in range(5):                   # thirty prompts: enough for image_media as well
+        t = codex_rollout(str(tmp_path / f"rollout-{i}.jsonl"), repo, i, prompts=6)
+        run_session(env, repo, f"cx{i}", t, host="codex")
+    unt = store_for(env, repo, "codex").read
+    assert unt("untool.json")["proposal"] == ["collaboration", "goals", "image_media"]
+    env = dict(env, TE_TOOLPRUNE_PROTECT="collab*,image_media")
+    run_session(env, repo, "cx5", codex_rollout(str(tmp_path / "rollout-5.jsonl"), repo, 5),
+                host="codex")
+    assert unt("untool.json")["proposal"] == ["goals"]
+
+
+def test_a_codex_package_applied_before_it_was_protected_comes_back(env, repo, tmp_path,
+                                                                     codex_accepts):
+    """The plugin's own line for a newly protected package is removed at
+    the next fold; the other packages it wrote stay."""
+    st = store_for(env, repo, "codex")
+    assert denylist.apply(repo, ["collaboration", "goals"], st, host="codex") == [
+        "collaboration", "goals"]
+    env = dict(env, TE_TOOLPRUNE="propose", TE_TOOLPRUNE_PROTECT="collab*")
+    run_session(env, repo, "cx0", codex_rollout(str(tmp_path / "rollout-0.jsonl"), repo, 0),
+                host="codex")
+    assert denylist.applied_for(st, repo, "codex") == ["goals"]
+    text = open(denylist.settings_path(repo, "codex")).read()
+    assert "[agents]" not in text and "include_collaboration_mode_instructions" not in text
+    assert "goals = false  # tokeneconomizer:goals" in text
+    # The session ran with it switched off, so it is no evidence that the package went unused.
+    caps = st.read("untool.json")["capabilities"]
+    assert caps["collaboration"]["loaded_sessions"] == 0
+    assert caps["image_media"]["loaded_sessions"] == 1
+
+
 def test_codex_needs_the_full_evidence_before_proposing(env, repo, tmp_path):
     """The experiments' controller demotes collaboration after ONE unused task; a person's
     sessions need `toolprune_evidence` prompts of it, however they split into sessions."""

tokeneconomizer/plugin/cli.py

@@ -419,7 +419,7 @@ def _proposal(ctx, unt):
     repository share its evidence, and one of them may have applied the list before another."""
     mine = set(denylist.applied_for(ctx.store, ctx.repo.root, ctx.host.name))
     mine |= set(denylist.kept_for(ctx.store, ctx.repo.root, ctx.host.name))
-    return [t for t in unt.get("proposal", []) if t not in mine]
+    return fold.unprotected(ctx, [t for t in unt.get("proposal", []) if t not in mine])
 
 
 def cmd_toolprune(args):

tokeneconomizer/plugin/fold.py

@@ -284,7 +284,22 @@ def _fold_toolprune_codex(ctx, adapter, session_id, transcript, workdir, prompts
     stage.build_tools({"instance_id": session_id}, 0)
     applied = denylist.applied_for(ctx.store, ctx.repo.root, ctx.host.name)
     kept = set(denylist.kept_for(ctx.store, ctx.repo.root, ctx.host.name))   # given back
+    # The packages this session ran without, read before any is given back below.
     stage.codex_profile._pending["logical_capabilities"] = sorted(applied)
+    # A package that has since become protected comes back, as a tool does on the other
+    # agents: only the plugin's own lines, and only toward giving it back.
+    protected = adapter.protected_tools()
+    restored = []
+    # Never in a config the repository commits.
+    if (any(c in protected for c in applied) and not ctx.config.shadow
+            and not denylist._git_tracked(ctx.repo.root, denylist.CODEX_CONFIG_REL)):
+        try:
+            restored = denylist.restore(ctx.repo.root, ctx.store,
+                                        [c for c in applied if c in protected], host="codex")
+        except (RuntimeError, OSError, ValueError):     # unreadable or redirected: leave it
+            restored = []
+    if restored:
+        applied = denylist.applied_for(ctx.store, ctx.repo.root, ctx.host.name)
     loaded_before = dict(stage.codex_profile.state.get("loaded_exposures") or {})
     srow = {}
     stage.after_task({"instance_id": session_id}, {}, run_dir, srow)
@@ -301,7 +316,7 @@ def _fold_toolprune_codex(ctx, adapter, session_id, transcript, workdir, prompts
                                                                  else prompts)
     proposal = [c for c in denylist.CODEX_LIVE_CAPABILITIES
                 if c in (state.get("current_pruned") or ()) and c not in ever_used
-                and c not in applied and c not in kept
+                and c not in applied and c not in kept and c not in protected
                 and enough_evidence(int(loaded.get(c, 0)), int(loaded_prompts.get(c, 0)),
                                     ctx.config, high_risk=c in CODEX_HIGH_RISK)]
     doc = {"sessions": int(state.get("tasks_completed") or 0), "catalog_source": "capabilities",
@@ -319,7 +334,17 @@ def _fold_toolprune_codex(ctx, adapter, session_id, transcript, workdir, prompts
         _hold(doc, prior, proposal, ctx.repo.root)
     ctx.store.write("untool.json", doc)
     return {"proposal": doc["proposal"], "pending": bool(doc.get("auto_pending")),
-            "restored": []}
+            "restored": restored}
+
+
+def unprotected(ctx, items):
+    """`items` less what is protected now: a list stored before a tool or package was
+    protected must not write it later."""
+    if ctx.host.name == "copilot":      # its lists name MCP servers, not tools
+        return list(items)
+    protected = LiveAdapter(ctx.host, ctx.repo.root,
+                            protect=ctx.config.toolprune_protect).protected_tools()
+    return [t for t in items if t not in protected]
 
 
 def _hold(doc, prior, proposal, root):
@@ -368,7 +393,7 @@ def apply_pending(ctx, now=None):
         root = pend.get("root") or ctx.repo.root
         done = set(denylist.applied_for(ctx.store, root, ctx.host.name))
         done |= set(denylist.kept_for(ctx.store, root, ctx.host.name))
-        items = [t for t in pend.get("items") or [] if t not in done]
+        items = unprotected(ctx, [t for t in pend.get("items") or [] if t not in done])
         if items and ctx.config.toolprune == "auto" and not ctx.config.shadow:
             idle = now - last_activity(ctx)
             quiet_s = ctx.config.toolprune_quiet_minutes * 60
@@ -381,6 +406,9 @@ def apply_pending(ctx, now=None):
                 unt.pop("auto_apply_refused", None)
             except RuntimeError as e:      # an unreadable, committed or redirected file
                 unt["auto_apply_refused"] = str(e)
+        # A tool protected while the list waited leaves the proposal too.
+        if "proposal" in unt:
+            unt["proposal"] = unprotected(ctx, unt["proposal"] or [])
         unt.pop("auto_pending", None)
         ctx.store.write("untool.json", unt)
         return None