refactor(tools): split tool_implementations.py into src/tools/ package (#4423)

* test(tools): add shim protection test for tool_implementations split

Covers all 48 top-level functions (33 do_* + 15 _helpers) extracted from
the original module. Guards the upcoming split: the shim must re-export
every symbol so existing 'from src.tool_implementations import X' imports
keep working. Passes on baseline (pre-split).

* refactor(tools): add src/tools/ package with shared _common

Slice 1 Task 2 (#4082/#4071). Adds the package skeleton and moves the
shared _parse_tool_args helper into src/tools/_common.py. Domain modules
will import from here. tool_implementations.py is untouched at this step.

* refactor(tools): extract system domain into src/tools/system.py

Slice 1 (#4082/#4071), Task 3: move the system-domain tool functions
(do_manage_skills/_skill_dump/do_manage_tasks/do_manage_endpoints/
do_manage_mcp/do_manage_webhooks/do_manage_tokens/do_manage_settings/
do_api_call/do_app_api) and the app_api blocklist constants out of
tool_implementations.py into a new src/tools/system.py module.

tool_implementations.py re-imports all of them so it stays a working
backward-compatible facade (shim test stays green).

- do_manage_mcp resolves get_mcp_manager via a function-local import
  from tool_implementations so the test that patches
  src.tool_implementations.get_mcp_manager still applies post-move.
- do_app_api imports _internal_headers and _INTERNAL_BASE (still in
  tool_implementations) function-locally to avoid a circular import.
- Repoint test_context_budget introspection assertion to the moved
  code's new home in src/tools/system.py.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* refactor(tools): extract cookbook domain into src/tools/cookbook.py

Moves the model-serving (cookbook) tool domain out of tool_implementations.py
into src/tools/cookbook.py as part of slice 1 (#4082/#4071):

- 13 do_* tools: download/serve/list/stop/tail/search/adopt/cached models,
  list downloads/cancel, list cookbook servers, serve presets
- 9 private helpers: _cookbook_servers, _resolve_cookbook_host,
  _cookbook_env_for_host, _infer_serve_{port,host}, _ensure_served_endpoint,
  _cookbook_register_task, _cookbook_apply_retry_suggestion,
  _scan_running_model_processes, _cookbook_kill_session
- _MODEL_PROCESS_PATTERNS constant (used only by _scan_running_model_processes)

tool_implementations.py stays a backward-compatible facade via a re-import
from src.tools.cookbook; src/tools/__init__ re-exports the same symbols.

_internal_headers and _INTERNAL_BASE stay in tool_implementations.py (shared
by system.py's do_app_api and many cookbook funcs). Each cookbook function
that needs them does a function-local import to avoid a top-level circular
dependency, matching the system-domain split.

Verified: compileall clean; shim test green; cookbook-touching suite
(652 passed, 1 skipped); full suite 3587 passed, 2 failed
(pre-existing test_api_chat_security, unrelated).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* refactor(tools): extract search domain into src/tools/search.py

* refactor(tools): extract notes domain into src/tools/notes.py

* refactor(tools): extract calendar domain into src/tools/calendar.py

Repoints tests/test_caldav_bidirectional_sync.py source-introspection
to src/tools/calendar.py (do_manage_calendar moved there).

* refactor(tools): extract image domain into src/tools/image.py

* refactor(tools): extract research domain into src/tools/research.py

* refactor(tools): extract contacts domain into src/tools/contacts.py

* refactor(tools): extract vault domain into src/tools/vault.py

Repoints tests/test_vault_password_not_in_argv.py source-introspection
to src/tools/vault.py (the vault do_* helpers moved there).

* refactor(tools): collapse tool_implementations to clean re-export shim

Move shared _INTERNAL_BASE/_internal_headers to src/tools/_common.py and
drop the duplicate _parse_tool_args (already in _common). tool_implementations.py
is now a pure re-export facade (+ 3 pre-existing email-context helpers, out of
scope). Domain files' function-local imports of these names still resolve via
the facade re-export.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(tools): port upstream cookbook workflow changes to split module

Rebase onto dev dropped c504214 ("Cookbook model workflow fixes") edits
to do_serve_model / do_tail_serve_output: the extraction commit moved
the pre-edit bodies into src/tools/cookbook.py and git auto-accepted the
deletion from tool_implementations.py, losing dev's changes. Restore them
in their post-split home:

- do_serve_model: add where/log_path/next_tools and the expanded
  "Next required check" output message
- do_tail_serve_output: empty-output fallback message replacing
  "(empty pane)"

(do_manage_settings web_fetch alias edit was already applied to
src/tools/system.py during the system-extract conflict resolution.)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(tools): break admin_tools circular import in split facade

After rebasing onto dev (#3629 moved the admin manage_* tools into
src/agent_tools/admin_tools), the facade re-exported them via a top-level
`from src.agent_tools.admin_tools import ...`. But src.agent_tools.__init__
imports this facade at top level, so the eager import re-entered the
partially-initialized agent_tools package and broke collection.

Re-export the admin symbols (do_manage_endpoints/mcp/webhooks/tokens/
settings, _MCP_DENIED_COMMANDS, _validate_mcp_command) lazily through
module __getattr__ instead, and drop them from src/tools/__init__ (they
no longer live in the src.tools package). system.py now holds only the
skills/tasks/api bridges; admin tools live solely in admin_tools.py,
matching upstream.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(tools): re-export dropped helpers through the split shim

Address review finding from #4423: the compatibility facade claimed to
preserve every original top-level symbol but omitted three helpers the
old src.tool_implementations exposed. Re-export them and pin them in
the shim protection test:

- _string_arg, _validate_cookbook_ssh_target <- src/tools/cookbook.py
- _mcp_allowed_commands <- src/agent_tools/admin_tools.py (lazily via
  __getattr__, to keep the agent_tools.__init__ <-> facade import acyclic
  after the #3629 admin-tools migration)

All three added to tests/test_tool_implementations_shim.py _EXPECTED so
the test contract now matches its "every original top-level function"
comment.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* test(tools): self-verify shim re-exports every domain do_*

The hand-maintained _EXPECTED list in the shim protection test can drift
silently when a new tool is added to a domain module but not re-exported
by the facade — exactly the omission a reviewer flagged post-split.
Add an auto-discovering test that enumerates every do_* from the domain
modules (incl. admin_tools) and asserts reachability through the shim,
so a forgotten re-export fails the build automatically.

Uses hasattr (not dir(ti)) because the admin symbols are re-exported
lazily via module __getattr__ and don't appear in dir(ti).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* test(tools): self-verify every in-repo facade import resolves

RaresKeY's P3 on the shim test was a claim-vs-reality gap: the docstring
said it protected "every from src.tool_implementations import X" but the
hand-maintained _EXPECTED list omitted three underscore helpers, so the
claim wasn't enforced. Re-exporting the three (cf1f5e3) fixed the known
gap; this closes the structural one.

Add test_every_facade_import_in_repo_resolves: ast-enumerate every
`from src.tool_implementations import X` site in src/ and tests/ and
assert hasattr(ti, X) for each. A forgotten re-export that anything in
the repo imports now fails the build automatically — including underscore
helpers, which the do_* discovery test does not cover.

Together with test_shim_reexports_every_domain_do_function, the shim
contract is now self-verifying. Demote _EXPECTED in the docstring to the
curated historical/downstream surface (the three helpers have no in-repo
consumer, so they stay manual by necessity) instead of "ground truth".

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(tools): dedupe _parse_tool_args + align shim guard with route consumers

Addresses two P3s from review (RaresKeY, 2026-06-26):

1. maintainability — _common carried a full copy of _parse_tool_args
   alongside the canonical src.tool_utils one; future parser fixes could
   diverge. The two bodies were byte-identical in logic, so _common now
   re-exports from tool_utils (a leaf module, no circular-import risk).
   The single-source test is extended to assert _common._parse_tool_args
   and tool_implementations._parse_tool_args are the same object as
   tool_utils._parse_tool_args.

2. test — the shim guard's import-site scan only walked src/ and tests/,
   missing routes/chat_routes.py's clear_active_email/set_active_email
   imports, and _EXPECTED omitted the active-email facade helpers. The
   scan now walks every first-party Python dir (pruning venvs/caches/data
   in-place), and set/get/clear_active_email are added to _EXPECTED
   (get_active_email has no in-repo importer, so the scan alone can't see
   it).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: yuandonghao <yuandonghao@cohl.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

src/tools/__init__.py

@@ -0,0 +1,32 @@
+"""Tool implementation package, split by domain (slice 1, #4082/#4071).
+
+Public tool functions live in domain modules. ``src.tool_implementations``
+re-exports from here for backward compatibility.
+"""
+from src.tools._common import _parse_tool_args  # noqa: F401
+from src.tools.system import (  # noqa: F401
+    do_manage_skills, _skill_dump, do_manage_tasks,
+    do_api_call, do_app_api,
+)
+from src.tools.cookbook import (  # noqa: F401
+    do_download_model, do_serve_model, do_list_served_models,
+    do_stop_served_model, do_tail_serve_output, do_list_downloads,
+    do_cancel_download, do_search_hf_models, do_adopt_served_model,
+    do_list_cookbook_servers, do_list_serve_presets, do_serve_preset,
+    do_list_cached_models,
+    _cookbook_servers, _resolve_cookbook_host, _cookbook_env_for_host,
+    _infer_serve_port, _infer_serve_host, _ensure_served_endpoint,
+    _cookbook_register_task, _cookbook_apply_retry_suggestion,
+    _scan_running_model_processes, _cookbook_kill_session,
+    _MODEL_PROCESS_PATTERNS,
+)
+from src.tools.search import do_search_chats  # noqa: F401
+from src.tools.notes import do_manage_notes  # noqa: F401
+from src.tools.calendar import do_manage_calendar  # noqa: F401
+from src.tools.image import do_edit_image  # noqa: F401
+from src.tools.research import do_manage_research, do_trigger_research  # noqa: F401
+from src.tools.contacts import do_resolve_contact, do_manage_contact  # noqa: F401
+from src.tools.vault import (  # noqa: F401
+    _load_vault_config, _run_bw,
+    do_vault_search, do_vault_get, do_vault_unlock,
+)

src/tools/_common.py

@@ -0,0 +1,25 @@
+"""Shared helpers used across tool implementation domains.
+
+Extracted from tool_implementations.py as part of slice 1 (#4082/#4071).
+Domain modules under src/tools/ import from here.
+"""
+from typing import Dict, Optional
+
+from core.constants import internal_api_base
+from src.tool_utils import _parse_tool_args  # noqa: F401 — single source of the tool-arg parser; tool_utils is a leaf module (imports nothing from src)
+
+
+# In-process loopback base for agent tools that call Odysseus's own API
+# (cookbook state, model serve, gallery, email, calendar). We ride the
+# per-process internal token so require_admin lets us through. See
+# core/middleware.py. Resolution (override / APP_PORT / 7000) lives in
+# core.constants.internal_api_base().
+_INTERNAL_BASE = internal_api_base()
+
+
+def _internal_headers(owner: Optional[str] = None) -> Dict[str, str]:
+    from core.middleware import INTERNAL_TOOL_HEADER, INTERNAL_TOOL_TOKEN
+    headers = {INTERNAL_TOOL_HEADER: INTERNAL_TOOL_TOKEN}
+    if owner:
+        headers["X-Odysseus-Owner"] = owner
+    return headers

src/tools/calendar.py

@@ -0,0 +1,522 @@
+"""Calendar-domain tool implementations.
+
+Extracted from tool_implementations.py as part of slice 1 (#4082/#4071).
+Holds the manage_calendar tool (CalDAV-backed event CRUD).
+``src.tool_implementations`` re-exports these for backward compatibility.
+"""
+import json
+import logging
+import re
+from typing import Dict, Optional
+
+from src.tools._common import _parse_tool_args
+
+logger = logging.getLogger(__name__)
+
+
+async def do_manage_calendar(content: str, owner: Optional[str] = None) -> Dict:
+    """Handle manage_calendar tool calls: list/create/update/delete calendar events (local SQLite)."""
+    from datetime import datetime, timedelta
+    from core.database import SessionLocal, CalendarCal, CalendarEvent, Note
+    from routes.calendar_routes import (
+        _ensure_default_calendar,
+        _parse_dt,
+        _parse_dt_pair,
+        parse_due_for_user,
+        _resolve_base_uid,
+        _push_caldav_event_after_commit,
+        _record_caldav_delete_tombstone,
+    )
+    import uuid as _uuid
+
+    try:
+        args = _parse_tool_args(content)
+    except ValueError:
+        return {"error": "Invalid JSON arguments", "exit_code": 1}
+
+    # ── Batch normalization ──
+    # Some models (e.g. deepseek-v4-flash) emit {"events": [{...}, ...]}
+    # instead of individual create_event calls. Iterate and create each.
+    if isinstance(args.get("events"), list) and not args.get("action"):
+        results = []
+        for ev in args["events"]:
+            if not isinstance(ev, dict):
+                continue
+            # Normalize start/end from {dateTime: "..."} object to flat string
+            for field, target in [("start", "dtstart"), ("end", "dtend")]:
+                val = ev.pop(field, None)
+                if val and target not in ev:
+                    ev[target] = val.get("dateTime", val) if isinstance(val, dict) else val
+            ev.setdefault("action", "create_event")
+            r = await do_manage_calendar(json.dumps(ev), owner=owner)
+            results.append(r)
+        created = [r for r in results if r.get("exit_code") == 0 and not r.get("error")]
+        failed = [r for r in results if r.get("error")]
+
+        if not results:
+            return {"error": "No events to create", "exit_code": 1}
+
+        # Surface both successes and failures
+        parts = []
+        if created:
+            summaries = [r.get("response", "") for r in created]
+            parts.append(f"Created {len(created)} event(s):\n" + "\n".join(summaries))
+        if failed:
+            first_error = failed[0].get("error", "Unknown error")
+            parts.append(f"Failed to create {len(failed)} event(s). First error: {first_error}")
+
+        response = "\n\n".join(parts)
+        # Non-zero exit code for partial or total failure
+        exit_code = 0 if not failed else 1
+        return {"response": response, "exit_code": exit_code, "created_count": len(created), "failed_count": len(failed)}
+
+    # Normalize action — some models emit hyphens ("list-calendars") instead
+    # of underscores. Treat them as equivalent so we don't bounce a
+    # cosmetic typo back to the model and waste a round-trip. Also accept
+    # short forms (`create`, `update`, `delete`) as aliases for the
+    # full `<verb>_event` names — models keep emitting the short forms.
+    action = (args.get("action") or "list_events").replace("-", "_").strip().lower()
+    _ACTION_ALIASES = {
+        "create": "create_event",
+        "update": "update_event",
+        "delete": "delete_event",
+        "list": "list_events",
+    }
+    action = _ACTION_ALIASES.get(action, action)
+    db = SessionLocal()
+
+    def _calendar_query():
+        q = db.query(CalendarCal)
+        if owner is not None:
+            q = q.filter(CalendarCal.owner == owner)
+        return q
+
+    def _event_query():
+        q = db.query(CalendarEvent).join(CalendarCal)
+        if owner is not None:
+            q = q.filter(CalendarCal.owner == owner)
+        return q
+
+    def _reminder_minutes(raw_args) -> Optional[int]:
+        raw = (
+            raw_args.get("reminder_minutes")
+            or raw_args.get("remind_before_minutes")
+            or raw_args.get("alarm_minutes")
+            or raw_args.get("reminder")
+            or raw_args.get("alarm")
+        )
+        if raw in (None, ""):
+            desc = str(raw_args.get("description") or "")
+            if re.search(r"\b(remind|reminder|alarm)\b", desc, re.I):
+                raw = desc
+        if raw in (None, "", False):
+            return None
+        if raw is True:
+            return 10
+        if isinstance(raw, (int, float)):
+            return max(0, int(raw))
+        text = str(raw).strip().lower()
+        if text in {"none", "no", "off", "false"}:
+            return None
+        m = re.search(r"(\d+)\s*(?:minutes?|mins?|m)\b", text)
+        if m:
+            return max(0, int(m.group(1)))
+        m = re.search(r"(\d+)\s*(?:hours?|hrs?|h)\b", text)
+        if m:
+            return max(0, int(m.group(1)) * 60)
+        if text.isdigit():
+            return max(0, int(text))
+        return None
+
+    def _event_description(raw_args, minutes_before: Optional[int]) -> str:
+        desc = str(raw_args.get("description", "") or "")
+        if minutes_before is None:
+            return desc
+        reminder_only = re.compile(
+            r"^\s*(?:remind(?:er)?|alarm)\s*:?\s*\d+\s*"
+            r"(?:minutes?|mins?|m|hours?|hrs?|h)\b.*$",
+            re.I,
+        )
+        return "" if reminder_only.match(desc) else desc
+
+    def _parse_event_dt(raw: str) -> tuple[datetime, bool]:
+        """Parse agent event datetimes in the user's timezone when available."""
+        return _parse_dt_pair(parse_due_for_user(raw))
+
+    def _first_nonempty_arg(*names: str):
+        for name in names:
+            value = args.get(name)
+            if value not in (None, ""):
+                return value
+        return None
+
+    def _create_calendar_reminder(summary: str, location: str, dtstart: datetime,
+                                  all_day: bool, minutes_before: int,
+                                  is_utc: bool = False) -> tuple[Optional[str], Optional[str]]:
+        remind_at = dtstart - timedelta(minutes=minutes_before)
+        now = datetime.utcnow() if is_utc else datetime.now()
+        if dtstart <= now:
+            return None, "event already passed"
+        if remind_at <= now:
+            # If the requested "before" time already passed but the event is
+            # still upcoming, create an immediate Note reminder instead of
+            # silently dropping it.
+            remind_at = now
+        start_fmt = dtstart.strftime("%a %b %d") if all_day else dtstart.strftime("%a %b %d %H:%M")
+        loc = f" @ {location}" if location else ""
+        text = f"{summary}{loc} — {start_fmt}"
+        due_date = remind_at.isoformat() + ("Z" if is_utc else "")
+        expected_title = f"Reminder: {summary}"
+        existing_q = db.query(Note).filter(
+            Note.archived == False,  # noqa: E712
+            Note.due_date == due_date,
+        )
+        if owner is not None:
+            existing_q = existing_q.filter(Note.owner == owner)
+        target_title = re.sub(r"^\s*reminder\s*:\s*", "", expected_title.strip().lower())
+        for existing in existing_q.limit(25).all():
+            existing_title = re.sub(r"^\s*reminder\s*:\s*", "", (existing.title or "").strip().lower())
+            if existing_title == target_title:
+                return existing.id, "duplicate reminder already exists"
+        note = Note(
+            id=str(_uuid.uuid4()),
+            owner=owner,
+            title=expected_title,
+            items=json.dumps([{"text": text, "done": False, "checked": False}]),
+            note_type="todo",
+            label="calendar",
+            due_date=due_date,
+            source="calendar",
+        )
+        db.add(note)
+        return note.id, None
+
+    try:
+        if action == "list_calendars":
+            _ensure_default_calendar(db, owner)
+            cals = _calendar_query().all()
+            result = [{"name": c.name, "href": c.id} for c in cals]
+            if result:
+                lines = [f"Found {len(result)} calendar(s):"]
+                for c in result:
+                    lines.append(f"- {c['name']} ({c['href'][:8]})")
+                response_text = "\n".join(lines)
+            else:
+                response_text = "No calendars found."
+            return {"response": response_text, "calendars": result, "exit_code": 0}
+
+        elif action == "list_events":
+            try:
+                start_raw = _first_nonempty_arg(
+                    "start", "start_date", "range_start", "from", "dtstart", "since"
+                )
+                end_raw = _first_nonempty_arg(
+                    "end", "end_date", "range_end", "to", "dtend", "until"
+                )
+                if start_raw:
+                    start_dt = _parse_dt(start_raw)
+                else:
+                    start_dt = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)
+                if end_raw:
+                    end_dt = _parse_dt(end_raw)
+                else:
+                    end_dt = start_dt + timedelta(days=14)
+            except ValueError as e:
+                return {"error": f"Invalid date format: {e}", "exit_code": 1}
+
+            if end_dt <= start_dt:
+                end_dt = start_dt + timedelta(days=1)
+
+            q = _event_query().filter(
+                CalendarEvent.dtstart < end_dt,
+                CalendarEvent.dtend > start_dt,
+                CalendarEvent.status != "cancelled",
+            )
+            calendar_filter = args.get("calendar")
+            if calendar_filter:
+                q = q.filter(
+                    (CalendarEvent.calendar_id == calendar_filter) |
+                    (CalendarCal.name == calendar_filter)
+                )
+            rows = q.order_by(CalendarEvent.dtstart).all()
+            events = []
+            for ev in rows:
+                if ev.all_day:
+                    s, e = ev.dtstart.strftime("%Y-%m-%d"), ev.dtend.strftime("%Y-%m-%d")
+                else:
+                    suffix = "Z" if getattr(ev, "is_utc", False) else ""
+                    s, e = ev.dtstart.isoformat() + suffix, ev.dtend.isoformat() + suffix
+                events.append({
+                    "uid": ev.uid, "summary": ev.summary or "", "dtstart": s, "dtend": e,
+                    "all_day": ev.all_day, "description": ev.description or "",
+                    "location": ev.location or "",
+                    "calendar": ev.calendar.name if ev.calendar else "",
+                    "calendar_href": ev.calendar_id,
+                    "event_type": ev.event_type or "",
+                    "importance": ev.importance or "normal",
+                })
+            if not events:
+                response_text = f"No events between {start_dt.date().isoformat()} and {end_dt.date().isoformat()}."
+            else:
+                lines = [f"Found {len(events)} event(s) between {start_dt.date().isoformat()} and {end_dt.date().isoformat()}:"]
+                for ev in events:
+                    when = ev["dtstart"]
+                    when_str = f"{when} (all day)" if ev.get("all_day") else f"{when} -> {ev.get('dtend', '')}"
+                    # Clickable anchor — opens the calendar on the event's day.
+                    line = f"- {when_str}: [{ev['summary']}](#event-{ev['uid']})"
+                    if ev.get("event_type"):
+                        line += f" #{ev['event_type']}"
+                    if ev.get("importance") and ev["importance"] != "normal":
+                        line += f" !{ev['importance']}"
+                    if ev.get("location"):
+                        line += f" @ {ev['location']}"
+                    if ev.get("calendar"):
+                        line += f" ({ev['calendar']})"
+                    if ev.get("description"):
+                        desc = ev["description"].strip().replace("\n", " ")
+                        if len(desc) > 120:
+                            desc = desc[:117] + "..."
+                        line += f"\n    {desc}"
+                    lines.append(line)
+                response_text = "\n".join(lines)
+            return {"response": response_text, "events": events, "exit_code": 0}
+
+        elif action == "create_event":
+            summary = args.get("summary")
+            # Accept the various names models like to use for the start
+            # field: dtstart (canonical), start, start_time, when.
+            dtstart_str = (args.get("dtstart") or args.get("start")
+                           or args.get("start_time") or args.get("when"))
+            if not summary or not dtstart_str:
+                return {"error": "summary and dtstart are required", "exit_code": 1}
+
+            # Accept either an href OR a calendar name/short-id like "Main"
+            # or "62e545d8" — saves the model from having to memorize hrefs
+            # after a `list_calendars` call returned short prefixes.
+            cal_href = args.get("calendar_href") or args.get("calendar")
+            cal = None
+            if cal_href:
+                cal = (_calendar_query()
+                       .filter(CalendarCal.id == cal_href)
+                       .first())
+                if not cal:
+                    # Try by name (case-insensitive) or by short-id prefix
+                    cal = (_calendar_query()
+                           .filter(CalendarCal.name.ilike(cal_href))
+                           .first())
+                if not cal:
+                    cal = (_calendar_query()
+                           .filter(CalendarCal.id.like(f"{cal_href}%"))
+                           .first())
+            if not cal:
+                cal = _ensure_default_calendar(db, owner)
+
+            all_day = bool(args.get("all_day", False))
+            try:
+                dtstart, dtstart_is_utc = _parse_event_dt(dtstart_str)
+            except ValueError as e:
+                return {"error": f"Could not parse dtstart {dtstart_str!r}: {e}", "exit_code": 1}
+            dtend_raw = args.get("dtend") or args.get("end") or args.get("end_time")
+            if dtend_raw:
+                try:
+                    dtend, dtend_is_utc = _parse_event_dt(dtend_raw)
+                    dtstart_is_utc = dtstart_is_utc or dtend_is_utc
+                except ValueError as e:
+                    return {"error": f"Could not parse dtend {dtend_raw!r}: {e}", "exit_code": 1}
+            else:
+                # Support duration: "1h", "30m", "90min", "1hr30m"
+                dur = (args.get("duration") or "").strip().lower()
+                delta = None
+                if dur:
+                    import re as _re_d
+                    h = _re_d.search(r'(\d+)\s*(?:h|hr|hours?)', dur)
+                    m = _re_d.search(r'(\d+)\s*(?:m|min|minutes?)', dur)
+                    secs = (int(h.group(1)) * 3600 if h else 0) + (int(m.group(1)) * 60 if m else 0)
+                    if secs > 0:
+                        delta = timedelta(seconds=secs)
+                if delta is not None:
+                    dtend = dtstart + delta
+                elif all_day:
+                    dtend = dtstart + timedelta(days=1)
+                else:
+                    dtend = dtstart + timedelta(hours=1)
+
+            # Dedup: if a non-cancelled event with the same title + start time already
+            # exists, return its UID instead of creating a fresh copy. Prevents the
+            # email triage from multiplying events when several emails reference the
+            # same meeting. Compare case-insensitively since LLM-extracted titles
+            # can vary in capitalisation.
+            from sqlalchemy import func as _func
+            existing = (
+                _event_query()
+                .filter(
+                    CalendarEvent.dtstart == dtstart,
+                    CalendarEvent.status != "cancelled",
+                    _func.lower(CalendarEvent.summary) == summary.lower(),
+                )
+                .first()
+            )
+            if existing is not None:
+                reminder_note_id = None
+                reminder_skipped_reason = None
+                minutes_before = _reminder_minutes(args)
+                if minutes_before is not None:
+                    reminder_note_id, reminder_skipped_reason = _create_calendar_reminder(
+                        existing.summary or summary,
+                        existing.location or "",
+                        existing.dtstart,
+                        existing.all_day,
+                        minutes_before,
+                        bool(existing.is_utc),
+                    )
+                    if reminder_note_id:
+                        db.commit()
+                reminder_text = ""
+                if minutes_before is not None:
+                    reminder_text = (
+                        f"; reminder set {minutes_before} min before"
+                        if reminder_note_id
+                        else f"; reminder not set ({reminder_skipped_reason or 'reminder time already passed'})"
+                    )
+                return {
+                    "response": (
+                        f"Event already exists: '{summary}' on {dtstart_str}"
+                        + reminder_text
+                    ),
+                    "uid": existing.uid,
+                    "reminder_note_id": reminder_note_id,
+                    "reminder_skipped_reason": reminder_skipped_reason,
+                    "duplicate": True,
+                    "exit_code": 0,
+                }
+
+            # Optional tag/category and importance — friendly aliases.
+            event_type = (args.get("event_type") or args.get("tag")
+                          or args.get("category") or args.get("type") or "") or None
+            importance = args.get("importance") or "normal"
+            minutes_before = _reminder_minutes(args)
+
+            uid = str(_uuid.uuid4())
+            ev = CalendarEvent(
+                uid=uid, calendar_id=cal.id, summary=summary,
+                description=_event_description(args, minutes_before),
+                location=args.get("location", "") or "",
+                dtstart=dtstart, dtend=dtend, all_day=all_day,
+                is_utc=dtstart_is_utc and not all_day,
+                rrule=args.get("rrule", "") or "",
+                event_type=event_type,
+                importance=importance,
+                caldav_sync_pending="create" if cal.source == "caldav" else None,
+            )
+            db.add(ev)
+            reminder_note_id = None
+            reminder_skipped_reason = None
+            if minutes_before is not None:
+                reminder_note_id, reminder_skipped_reason = _create_calendar_reminder(
+                    summary,
+                    args.get("location", "") or "",
+                    dtstart,
+                    all_day,
+                    minutes_before,
+                    dtstart_is_utc and not all_day,
+                )
+            db.commit()
+            if cal.source == "caldav":
+                await _push_caldav_event_after_commit(owner, uid, "create")
+            tag_blurb = f" [{event_type}]" if event_type else ""
+            if minutes_before is None:
+                reminder_blurb = ""
+            elif reminder_note_id:
+                reminder_blurb = f" with reminder {minutes_before} min before"
+            else:
+                reminder_blurb = f" without reminder ({reminder_skipped_reason or 'reminder time already passed'})"
+            # Return a clickable anchor so the agent can surface a link
+            # that opens the calendar on that day. See the markdown
+            # anchor convention ([Name](#event-<uid>)).
+            return {
+                "response": f"Created event [{summary}](#event-{uid}){tag_blurb} on {dtstart_str}{reminder_blurb}",
+                "uid": uid,
+                "anchor": f"[{summary}](#event-{uid})",
+                "reminder_note_id": reminder_note_id,
+                "reminder_skipped_reason": reminder_skipped_reason,
+                "exit_code": 0,
+            }
+
+        elif action == "update_event":
+            uid = args.get("uid")
+            if not uid:
+                return {"error": "uid is required", "exit_code": 1}
+            try:
+                base_uid = _resolve_base_uid(uid)
+            except ValueError as e:
+                return {"error": str(e), "exit_code": 1}
+            ev = _event_query().filter(CalendarEvent.uid == base_uid).first()
+            if not ev:
+                return {"error": f"Event {uid} not found", "exit_code": 1}
+            if args.get("summary") is not None:
+                ev.summary = args["summary"]
+            if args.get("description") is not None:
+                ev.description = args["description"]
+            if args.get("location") is not None:
+                ev.location = args["location"]
+            if args.get("dtstart") is not None:
+                # Anchor naive/natural-language input to the USER's timezone and
+                # refresh is_utc, exactly like create_event. Parsing with the
+                # raw server-local _parse_dt here (and never touching is_utc)
+                # silently shifted an updated event by the user's UTC offset.
+                _eff_all_day = (
+                    args["all_day"] if args.get("all_day") is not None else ev.all_day
+                )
+                ev.dtstart, _su = _parse_event_dt(args["dtstart"])
+                ev.is_utc = bool(_su and not _eff_all_day)
+            if args.get("dtend") is not None:
+                ev.dtend, _eu = _parse_event_dt(args["dtend"])
+            if args.get("all_day") is not None:
+                ev.all_day = args["all_day"]
+            # Tag/category + importance updates (any of these aliases).
+            _tag = (args.get("event_type") or args.get("tag")
+                    or args.get("category") or args.get("type"))
+            if _tag is not None:
+                ev.event_type = _tag or None
+            if args.get("importance") is not None:
+                ev.importance = args["importance"]
+            is_caldav = ev.calendar and ev.calendar.source == "caldav"
+            if is_caldav:
+                ev.caldav_sync_pending = "update"
+            db.commit()
+            if is_caldav:
+                await _push_caldav_event_after_commit(owner, base_uid, "update")
+            return {"response": f"Updated event {uid}", "exit_code": 0}
+
+        elif action == "delete_event":
+            uid = args.get("uid")
+            if not uid:
+                return {"error": "uid is required", "exit_code": 1}
+            try:
+                base_uid = _resolve_base_uid(uid)
+            except ValueError as e:
+                return {"error": str(e), "exit_code": 1}
+            ev = _event_query().filter(CalendarEvent.uid == base_uid).first()
+            if not ev:
+                return {"error": f"Event {uid} not found", "exit_code": 1}
+            is_caldav = ev.calendar and ev.calendar.source == "caldav" and ev.remote_href
+            if is_caldav:
+                _record_caldav_delete_tombstone(db, ev, owner)
+            db.delete(ev)
+            db.commit()
+            if is_caldav:
+                await _push_caldav_event_after_commit(owner, base_uid, "delete")
+            return {"response": f"Deleted event {uid}", "exit_code": 0}
+
+        else:
+            return {
+                "error": f"Unknown action: {action}. Use list_events, create_event, update_event, delete_event, list_calendars",
+                "exit_code": 1,
+            }
+
+    except Exception as e:
+        db.rollback()
+        logger.error(f"manage_calendar error: {e}")
+        return {"error": str(e), "exit_code": 1}
+    finally:
+        db.close()

src/tools/contacts.py

@@ -0,0 +1,148 @@
+"""Contacts-domain tool implementations.
+
+Extracted from tool_implementations.py as part of slice 1 (#4082/#4071).
+Holds the resolve_contact and manage_contact (CardDAV CRUD) tools.
+``src.tool_implementations`` re-exports these for backward compatibility.
+``_INTERNAL_BASE`` still lives in tool_implementations.py and is pulled
+back function-locally where needed.
+"""
+from typing import Dict, Optional
+
+from src.tools._common import _parse_tool_args
+
+
+async def do_resolve_contact(content: str, owner: Optional[str] = None) -> Dict:
+    """Look up a contact by name. Searches: CardDAV -> email history -> memory."""
+    import httpx
+    from src.tool_implementations import _INTERNAL_BASE  # shared constant, still lives in the facade
+    try:
+        args = _parse_tool_args(content)
+    except ValueError:
+        return {"error": "Invalid JSON arguments", "exit_code": 1}
+    name = args.get("name", "")
+    if not name:
+        return {"error": "name is required", "exit_code": 1}
+
+    contacts = {}  # email_or_phone -> {name, source, phone?}
+
+    # 1. CardDAV (Radicale) — structured contacts. Call in-process: a
+    # server-side httpx GET to /api/contacts/search carries no session
+    # cookie and would 401 under require_user.
+    try:
+        import asyncio
+        from routes import contacts_routes as cc
+        all_contacts = await asyncio.to_thread(cc._fetch_contacts)
+        q = name.lower()
+        for c in (all_contacts or []):
+            hay_name = (c.get("name") or "").lower()
+            match = q in hay_name or any(q in (e or "").lower() for e in c.get("emails", []))
+            if not match:
+                continue
+            has_email = False
+            for email in (c.get("emails") or []):
+                email = (email or "").strip().lower()
+                if email and "@" in email:
+                    contacts[email] = {"name": c.get("name") or email, "source": "contacts"}
+                    has_email = True
+            # Fall back to phone numbers when the contact has no email address
+            if not has_email:
+                for phone in (c.get("phones") or []):
+                    phone = (phone or "").strip()
+                    if phone:
+                        contacts[phone] = {"name": c.get("name") or phone, "source": "contacts", "phone": phone}
+    except Exception:
+        pass
+
+    async with httpx.AsyncClient(timeout=30) as client:
+        # 2. Email history (sent/received)
+        try:
+            resp = await client.get(f"{_INTERNAL_BASE}/api/email/resolve-contact", params={"name": name})
+            if resp.status_code == 200:
+                for c in (resp.json().get("contacts") or []):
+                    email = (c.get("email") or "").strip().lower()
+                    if email and email not in contacts:
+                        contacts[email] = {"name": c.get("name") or email, "source": "email history"}
+        except Exception:
+            pass
+
+    if not contacts:
+        return {"output": f"No contacts found matching '{name}'.", "exit_code": 0}
+
+    lines = [f"Contacts matching '{name}':"]
+    for key, info in contacts.items():
+        if info.get("phone"):
+            lines.append(f"- {info['name']} — phone: {info['phone']} ({info['source']})")
+        else:
+            lines.append(f"- {info['name']} <{key}> ({info['source']})")
+    return {"output": "\n".join(lines), "exit_code": 0}
+
+
+async def do_manage_contact(content: str, owner: Optional[str] = None) -> Dict:
+    """Add / update / delete / list CardDAV contacts. Calls the contacts
+    helpers IN-PROCESS rather than over HTTP — a server-side httpx call to
+    /api/contacts/* carries no session cookie and would be rejected by
+    require_user (401), so the tool would see zero contacts even though
+    the browser-side UI works fine."""
+    try:
+        args = _parse_tool_args(content)
+    except ValueError:
+        return {"error": "Invalid JSON arguments", "exit_code": 1}
+    action = (args.get("action") or "").strip().lower()
+    try:
+        from routes import contacts_routes as cc
+    except Exception as e:
+        return {"error": f"Contacts module unavailable: {e}", "exit_code": 1}
+    # The contacts helpers are sync (httpx blocking calls to CardDAV) — run
+    # them in a thread so we don't block the event loop.
+    import asyncio
+    try:
+        if action == "list":
+            rows = await asyncio.to_thread(cc._fetch_contacts, True)
+            if not rows:
+                return {"output": "No contacts.", "exit_code": 0}
+            lines = [f"{len(rows)} contacts:"]
+            for c in rows:
+                em = ", ".join(c.get("emails") or [])
+                lines.append(f"- {c.get('name') or '(no name)'} <{em}>  [uid={c.get('uid','')}]")
+            return {"output": "\n".join(lines), "exit_code": 0}
+
+        if action == "add":
+            email = (args.get("email") or "").strip()
+            if not email:
+                return {"error": "email is required for add", "exit_code": 1}
+            name = (args.get("name") or "").strip() or email.split("@")[0]
+            # Dedupe by email (same as the /add route).
+            existing = await asyncio.to_thread(cc._fetch_contacts)
+            for c in existing:
+                if email.lower() in [e.lower() for e in c.get("emails", [])]:
+                    return {"output": f"{email} is already a contact ({c.get('name','')}).", "exit_code": 0}
+            ok = await asyncio.to_thread(cc._create_contact, name, email)
+            return {"output": f"{'Added' if ok else 'Failed to add'} {name} <{email}>.", "exit_code": 0 if ok else 1}
+
+        if action in ("update", "edit"):
+            uid = (args.get("uid") or "").strip()
+            if not uid:
+                return {"error": "uid is required for update (use action=list to find it)", "exit_code": 1}
+            name = (args.get("name") or "").strip()
+            emails = args.get("emails")
+            if emails is None and args.get("email"):
+                emails = [args["email"]]
+            emails = [e.strip() for e in (emails or []) if e and e.strip()]
+            phones = [p.strip() for p in (args.get("phones") or []) if p and p.strip()]
+            if not name and not emails:
+                return {"error": "Provide a name or emails to update", "exit_code": 1}
+            if not name and emails:
+                name = emails[0].split("@")[0]
+            ok = await asyncio.to_thread(cc._update_contact, uid, name, emails, phones)
+            return {"output": "Contact updated." if ok else "Update failed.", "exit_code": 0 if ok else 1}
+
+        if action == "delete":
+            uid = (args.get("uid") or "").strip()
+            if not uid:
+                return {"error": "uid is required for delete (use action=list to find it)", "exit_code": 1}
+            ok = await asyncio.to_thread(cc._delete_contact, uid)
+            return {"output": "Contact deleted." if ok else "Delete failed.", "exit_code": 0 if ok else 1}
+
+        return {"error": f"Unknown action '{action}'. Use list, add, update, or delete.", "exit_code": 1}
+    except Exception as e:
+        return {"error": f"Contact operation failed: {e}", "exit_code": 1}

src/tools/image.py

@@ -0,0 +1,39 @@
+"""Image-domain tool implementations.
+
+Extracted from tool_implementations.py as part of slice 1 (#4082/#4071).
+Holds the edit_image (gallery) tool.
+``src.tool_implementations`` re-exports these for backward compatibility.
+``_INTERNAL_BASE`` still lives in tool_implementations.py and is pulled back
+function-locally here.
+"""
+from typing import Dict, Optional
+
+from src.tools._common import _parse_tool_args
+
+
+async def do_edit_image(content: str, owner: Optional[str] = None) -> Dict:
+    """Edit a gallery image (upscale, rembg, inpaint, harmonize)."""
+    import httpx
+    from src.tool_implementations import _INTERNAL_BASE  # shared constant, still lives in the facade
+    try:
+        args = _parse_tool_args(content)
+    except ValueError:
+        return {"error": "Invalid JSON arguments", "exit_code": 1}
+    image_id = args.get("image_id", "")
+    action = args.get("action", "")
+    if not image_id or not action:
+        return {"error": "image_id and action are required", "exit_code": 1}
+    payload = {"image_id": image_id}
+    if args.get("prompt"):
+        payload["prompt"] = args["prompt"]
+    if args.get("scale"):
+        payload["scale"] = args["scale"]
+    try:
+        async with httpx.AsyncClient(timeout=120) as client:
+            resp = await client.post(f"{_INTERNAL_BASE}/api/gallery/{action}", json=payload)
+            data = resp.json()
+        if data.get("success") or data.get("id"):
+            return {"output": f"Image edited ({action}). New image ID: {data.get('id', '?')}", "exit_code": 0}
+        return {"error": data.get("error", f"{action} failed"), "exit_code": 1}
+    except Exception as e:
+        return {"error": str(e), "exit_code": 1}

src/tools/notes.py

@@ -0,0 +1,254 @@
+"""Notes-domain tool implementations.
+
+Extracted from tool_implementations.py as part of slice 1 (#4082/#4071).
+Holds the manage_notes tool (notes + checklists CRUD).
+``src.tool_implementations`` re-exports these for backward compatibility.
+"""
+import json
+import logging
+import re
+from typing import Dict, Optional
+
+from src.tools._common import _parse_tool_args
+
+logger = logging.getLogger(__name__)
+
+
+async def do_manage_notes(content: str, owner: Optional[str] = None) -> Dict:
+    """Handle manage_notes tool calls: CRUD on notes and checklists."""
+    import uuid as _uuid
+    from core.database import SessionLocal, Note
+    from sqlalchemy.orm.attributes import flag_modified
+
+    try:
+        args = _parse_tool_args(content)
+    except ValueError:
+        return {"error": "Invalid JSON arguments", "exit_code": 1}
+
+    # Action aliases — match what models actually emit. `create` is the most
+    # common alternative to `add`. Hyphenated forms also accepted.
+    action = (args.get("action") or "").replace("-", "_").strip().lower()
+    _NOTE_ACTION_ALIASES = {
+        "create": "add",
+        "new": "add",
+        "save": "add",
+        "remind": "add",
+        "remove": "delete",
+        "remove_item": "toggle_item",
+    }
+    action = _NOTE_ACTION_ALIASES.get(action, action)
+    db = SessionLocal()
+
+    def _norm_note_title(value: str) -> str:
+        text = (value or "").strip().lower()
+        text = re.sub(r"^\s*reminder\s*:\s*", "", text)
+        return re.sub(r"\s+", " ", text)
+
+    def _note_visible_to_owner(note, owner_value: Optional[str]) -> bool:
+        # Empty owner_value is single-user / auth-disabled mode. A real
+        # authenticated owner must match exactly; null/empty legacy rows are not
+        # shared between accounts.
+        if not owner_value:
+            return True
+        return getattr(note, "owner", None) == owner_value
+
+    def _note_by_prefix(note_id: str):
+        if not note_id:
+            return None
+        q = db.query(Note).filter(Note.id.startswith(note_id))
+        if owner:
+            q = q.filter(Note.owner == owner)
+        return q.first()
+
+    try:
+        if action == "list":
+            q = db.query(Note)
+            if owner is not None:
+                q = q.filter(Note.owner == owner)
+            if args.get("label"):
+                q = q.filter(Note.label == args["label"])
+            show_archived = args.get("archived", False)
+            q = q.filter(Note.archived == show_archived)
+            notes = q.order_by(Note.pinned.desc(), Note.updated_at.desc()).all()
+            if not notes:
+                return {"response": "No notes found.", "exit_code": 0}
+            lines = []
+            for n in notes:
+                pin = " [PINNED]" if n.pinned else ""
+                typ = " [checklist]" if n.note_type == "checklist" else ""
+                lbl = f" #{n.label}" if n.label else ""
+                title = n.title or "(untitled)"
+                lines.append(f"- [{n.id[:8]}] **{title}**{pin}{typ}{lbl}")
+                if n.note_type == "checklist" and n.items:
+                    try:
+                        items = json.loads(n.items)
+                        for i, item in enumerate(items):
+                            mark = "x" if item.get("done") else " "
+                            lines.append(f"  [{mark}] {i}: {item.get('text', '')}")
+                    except (json.JSONDecodeError, TypeError):
+                        pass
+                elif n.content:
+                    snippet = n.content[:80].replace("\n", " ")
+                    lines.append(f"  {snippet}")
+            return {"results": "\n".join(lines)}
+
+        elif action == "add":
+            # Accept the various field names models emit: `text` is the most
+            # common stand-in for "title or body content" when the model
+            # treats the note as a single string. If text was supplied and
+            # neither title nor content, use it as the title.
+            title = (args.get("title") or "").strip()
+            content_raw = args.get("content")
+            text_raw = args.get("text") or args.get("body")
+            if not title and not content_raw and text_raw:
+                title = text_raw.strip()
+            elif not content_raw and text_raw:
+                content_raw = text_raw
+            # Accept both `items` (legacy/internal field) and `checklist_items`
+            # (the schema-exposed name used by native function calls). Models
+            # following the schema emit `checklist_items`; older code paths
+            # and direct API callers still use `items`.
+            items_raw = args.get("checklist_items")
+            if items_raw is None:
+                items_raw = args.get("items")
+            items_json = json.dumps(items_raw) if items_raw is not None else None
+            note_type = args.get("note_type", "checklist" if items_raw else "note")
+            # Accept natural-language due_date ("tomorrow at 1pm") in
+            # addition to ISO. Use the user-tz-aware parser so the LLM's
+            # naive times ("today at 9pm") are anchored to the USER's clock,
+            # not the server's. Returns ISO with explicit offset so frontend
+            # `new Date()` resolves the right absolute moment regardless of
+            # where the user is.
+            due_raw = args.get("due_date")
+            due_iso = None
+            if due_raw:
+                try:
+                    from routes.calendar_routes import parse_due_for_user as _pdt_user
+                    due_iso = _pdt_user(due_raw)
+                except Exception:
+                    due_iso = due_raw  # fall through; trust the model
+            if due_iso and title:
+                # Calendar event reminders are represented as Notes. If the
+                # model creates a calendar event with reminder_minutes and then
+                # also creates a separate note reminder for the same title/time,
+                # keep the existing note so the user gets only one dispatch.
+                existing_q = db.query(Note).filter(
+                    Note.archived == False,  # noqa: E712
+                    Note.due_date == due_iso,
+                )
+                if owner is not None:
+                    existing_q = existing_q.filter(Note.owner == owner)
+                target_title = _norm_note_title(title)
+                for existing in existing_q.limit(25).all():
+                    if _norm_note_title(existing.title or "") == target_title:
+                        return {
+                            "response": f"Reminder already exists: \"{existing.title or title}\" (id: {existing.id[:8]})",
+                            "note_id": existing.id,
+                            "duplicate": True,
+                            "exit_code": 0,
+                        }
+            note = Note(
+                id=str(_uuid.uuid4()),
+                owner=owner,
+                title=title,
+                content=content_raw,
+                items=items_json,
+                note_type=note_type,
+                color=args.get("color"),
+                label=args.get("label"),
+                pinned=args.get("pinned", False),
+                due_date=due_iso,
+                source="agent",
+                session_id=args.get("session_id"),
+            )
+            db.add(note)
+            db.commit()
+            # Return note_id so the chat-side renderer can build a real
+            # "View note" button that opens the notes modal at this id.
+            # Previously the create response only included a prose
+            # confirmation; the model would type "View note" as a markdown
+            # link with no target, leaving the user with a click that
+            # did nothing and uncertainty about whether the note was made.
+            return {
+                "response": f"Note created: \"{title or '(untitled)'}\" (id: {note.id[:8]})",
+                "note_id": note.id,
+                "note_title": title or "",
+                "open_url": f"/#open=notes&note={note.id}",
+                "exit_code": 0,
+            }
+
+        elif action == "update":
+            note_id = args.get("id", "")
+            note = _note_by_prefix(note_id)
+            if not note:
+                return {"error": f"Note '{note_id}' not found", "exit_code": 1}
+            if not _note_visible_to_owner(note, owner):
+                return {"error": "Note not found", "exit_code": 1}
+            for field in ("title", "content", "note_type", "color", "label"):
+                if field in args and args[field] is not None:
+                    setattr(note, field, args[field])
+            # Parse due_date the same way the `add` action does. The schema
+            # advertises natural language ("tomorrow at 9am"), and naive ISO
+            # strings need the user's tz offset attached so the frontend's
+            # `new Date()` resolves the right absolute moment. Storing the raw
+            # value here left updated reminders as unparseable literals that
+            # never fired.
+            if args.get("due_date") is not None:
+                due_raw = args["due_date"]
+                try:
+                    from routes.calendar_routes import parse_due_for_user as _pdt_user
+                    note.due_date = _pdt_user(due_raw)
+                except Exception:
+                    note.due_date = due_raw  # fall through; trust the model
+            new_items = args.get("checklist_items")
+            if new_items is None:
+                new_items = args.get("items")
+            if new_items is not None:
+                note.items = json.dumps(new_items)
+                flag_modified(note, "items")
+            if "pinned" in args:
+                note.pinned = args["pinned"]
+            if "archived" in args:
+                note.archived = args["archived"]
+            db.commit()
+            return {"response": f"Note updated: \"{note.title or '(untitled)'}\"", "exit_code": 0}
+
+        elif action == "delete":
+            note_id = args.get("id", "")
+            note = _note_by_prefix(note_id)
+            if not note:
+                return {"error": f"Note '{note_id}' not found", "exit_code": 1}
+            if not _note_visible_to_owner(note, owner):
+                return {"error": "Note not found", "exit_code": 1}
+            title = note.title
+            db.delete(note)
+            db.commit()
+            return {"response": f"Deleted note: \"{title or '(untitled)'}\"", "exit_code": 0}
+
+        elif action == "toggle_item":
+            note_id = args.get("id", "")
+            index = args.get("index", 0)
+            note = _note_by_prefix(note_id)
+            if not note:
+                return {"error": f"Note '{note_id}' not found", "exit_code": 1}
+            if not _note_visible_to_owner(note, owner):
+                return {"error": "Note not found", "exit_code": 1}
+            if not note.items:
+                return {"error": "Note has no checklist items", "exit_code": 1}
+            items = json.loads(note.items)
+            if index < 0 or index >= len(items):
+                return {"error": f"Item index {index} out of range (0-{len(items)-1})", "exit_code": 1}
+            items[index]["done"] = not items[index].get("done", False)
+            note.items = json.dumps(items)
+            flag_modified(note, "items")
+            db.commit()
+            mark = "done" if items[index]["done"] else "undone"
+            return {"response": f"Item '{items[index].get('text', '')}' marked {mark}", "exit_code": 0}
+
+        else:
+            return {"error": f"Unknown action: {action}. Use list/add/update/delete/toggle_item", "exit_code": 1}
+    except Exception as e:
+        logger.error(f"manage_notes error: {e}")
+        return {"error": str(e), "exit_code": 1}
+    finally:
+        db.close()

src/tools/research.py

@@ -0,0 +1,142 @@
+"""Research-domain tool implementations.
+
+Extracted from tool_implementations.py as part of slice 1 (#4082/#4071).
+Holds the manage_research (library CRUD) and trigger_research (live job)
+tools.
+``src.tool_implementations`` re-exports these for backward compatibility.
+``_internal_headers`` and ``_INTERNAL_BASE`` still live in
+tool_implementations.py and are pulled back function-locally where needed.
+"""
+import re
+from typing import Any, Dict, Optional
+
+from src.constants import DEEP_RESEARCH_DIR
+from src.tools._common import _parse_tool_args
+
+
+async def do_manage_research(content: str, owner: Optional[str] = None) -> Dict:
+    """List, read/open, or delete saved deep-research results from the Library.
+    Args (JSON): {"action": "list|read|delete", "id": "<id>", "search": "..."}.
+    Research is stored as data/deep_research/<id>.json (query, summary, sources)."""
+    import json as _json
+    from pathlib import Path as _Path
+    try:
+        args = _parse_tool_args(content) if content.strip().startswith("{") else {}
+    except ValueError:
+        args = {}
+    if not isinstance(args, dict):
+        args = {}
+    action = (args.get("action") or "list").lower()
+    rid = (args.get("id") or args.get("session_id") or args.get("research_id") or "").strip()
+    data_dir = _Path(DEEP_RESEARCH_DIR)
+
+    # SECURITY: the research id is interpolated straight into a filesystem
+    # path (data/deep_research/<rid>.json) for read AND delete. Without this
+    # gate an agent-supplied id like "../settings" or "../../etc/passwd"
+    # escapes the research dir — reading exfiltrates arbitrary *.json into
+    # chat, deleting unlinks arbitrary *.json on disk. Allow only a bare
+    # token (research session ids are hex/uuid/slug — no separators).
+    if rid and not re.fullmatch(r"[A-Za-z0-9_-]+", rid):
+        return {"error": "Invalid research id."}
+
+    def _load(p):
+        try:
+            return _json.loads(p.read_text(encoding="utf-8"))
+        except Exception:
+            return None
+
+    if action in ("read", "open", "view", "get"):
+        if not rid:
+            return {"error": "Provide the research id (from action='list')."}
+        p = data_dir / f"{rid}.json"
+        if not p.exists():
+            return {"error": f"Research '{rid}' not found."}
+        d = _load(p) or {}
+        summary = d.get("result") or d.get("raw_report") or d.get("summary") or d.get("report") or "(no report body)"
+        srcs = d.get("sources", []) or []
+        out = f"# {d.get('query', '(untitled)')}\n\n{summary}"
+        if srcs:
+            out += "\n\nSources:\n" + "\n".join(
+                f"- {s.get('title') or s.get('url', '')}: {s.get('url', '')}" for s in srcs[:30]
+            )
+        return {"output": out[:16000], "exit_code": 0}
+
+    if action == "delete":
+        if not rid:
+            return {"error": "Provide the research id to delete (from action='list')."}
+        p = data_dir / f"{rid}.json"
+        if p.exists():
+            try:
+                p.unlink()
+            except Exception as e:
+                return {"error": f"Failed to delete: {e}"}
+            return {"output": f"Deleted research '{rid}'.", "exit_code": 0}
+        return {"error": f"Research '{rid}' not found."}
+
+    # default: list — clickable [query](#research-<id>) rows, most-recent first
+    search = (args.get("search") or "").lower()
+    items = []
+    if data_dir.exists():
+        for p in data_dir.glob("*.json"):
+            d = _load(p)
+            if not d:
+                continue
+            q = d.get("query", "")
+            if search and search not in q.lower():
+                continue
+            items.append((d.get("completed_at", 0) or 0, p.stem, q, len(d.get("sources", []) or [])))
+    items.sort(reverse=True)
+    if not items:
+        return {"output": "No research found in the library." + (f" (search: {search})" if search else ""), "exit_code": 0}
+    rows = "\n".join(f"- [{q or '(untitled)'}](#research-{sid}) — {n} sources" for _, sid, q, n in items[:50])
+    return {"output": f"Research library ({len(items)} item{'s' if len(items) != 1 else ''}):\n{rows}", "exit_code": 0}
+
+
+async def do_trigger_research(content: str, owner: Optional[str] = None) -> Dict:
+    """Start a live deep-research job that appears in the Deep Research
+    sidebar. Hits /api/research/start (the same path the sidebar's
+    'Research' button uses) so the session is discoverable + streamable
+    there, rather than creating a scheduled task that never surfaces."""
+    import httpx
+    from src.tool_implementations import _internal_headers, _INTERNAL_BASE  # shared constants, still live in the facade
+    try:
+        args = _parse_tool_args(content)
+    except ValueError:
+        return {"error": "Invalid JSON arguments", "exit_code": 1}
+    topic = args.get("topic", "") or args.get("query", "")
+    if not topic:
+        return {"error": "topic (or query) is required", "exit_code": 1}
+    payload: Dict[str, Any] = {"query": topic}
+    # Optional knobs the research panel supports.
+    if args.get("max_rounds") is not None:
+        try: payload["max_rounds"] = int(args["max_rounds"])
+        except (ValueError, TypeError): pass
+    if args.get("max_time") is not None:
+        try: payload["max_time"] = int(args["max_time"])
+        except (ValueError, TypeError): pass
+    if args.get("category"):
+        payload["category"] = args["category"]
+    if args.get("search_provider"):
+        payload["search_provider"] = args["search_provider"]
+    try:
+        async with httpx.AsyncClient(timeout=30) as client:
+            resp = await client.post(f"{_INTERNAL_BASE}/api/research/start",
+                                     json=payload, headers=_internal_headers(owner))
+        if resp.status_code >= 400:
+            return {"error": f"research/start returned HTTP {resp.status_code}: {resp.text[:200]}", "exit_code": 1}
+        data = resp.json()
+        sid = data.get("session_id", "?")
+        return {
+            "output": (
+                f"Deep research started: [{topic}](#research-{sid}). "
+                "Click to open the Deep Research sidebar and watch progress / read the report."
+            ),
+            "session_id": sid,
+            "anchor": f"[{topic}](#research-{sid})",
+            # UI hint so the frontend can open/refresh the research panel.
+            "ui_event": "research_started",
+            "research_session_id": sid,
+            "exit_code": 0,
+        }
+    except Exception as e:
+        return {"error": str(e), "exit_code": 1}

src/tools/search.py

@@ -0,0 +1,51 @@
+"""Search-domain tool implementations.
+
+Extracted from tool_implementations.py as part of slice 1 (#4082/#4071).
+Holds the search_chats tool.
+``src.tool_implementations`` re-exports these for backward compatibility.
+"""
+import logging
+from typing import Dict
+
+logger = logging.getLogger(__name__)
+
+
+async def do_search_chats(query: str, limit: int = 20, owner: str | None = None) -> Dict:
+    """Search past session transcripts for the calling user's sessions only.
+
+    Without an owner filter this used to leak EVERY user's chat history
+    into the agent's `search_chats` results (v2 review HIGH-11). The
+    caller in `tool_execution.execute_tool_block` now plumbs the owner
+    through; legacy callers without owner pass through as before but
+    will only see legacy/null-owner rows.
+    """
+    try:
+        from src.session_search import search_session_messages
+
+        results = search_session_messages(query, limit=limit, owner=owner)
+        if not results:
+            return {"results": f"No chats found matching \"{query}\"."}
+
+        # Group by session to avoid duplicate links
+        seen_sessions = {}
+        for result in results:
+            if result.session_id not in seen_sessions:
+                seen_sessions[result.session_id] = result
+
+        lines = [f"Found {len(seen_sessions)} session(s) matching \"{query}\":\n"]
+        for sid, result in seen_sessions.items():
+            lines.append(f"- **{result.session_name}** (#{sid})")
+            lines.append(f"  Link: [Open chat](#{sid})")
+            lines.append(f"  Match ({result.role}): {result.content_snippet}")
+            if result.context_before:
+                before = result.context_before[-1]
+                lines.append(f"  Before ({before['role']}): {before['content'][:180]}")
+            if result.context_after:
+                after = result.context_after[0]
+                lines.append(f"  After ({after['role']}): {after['content'][:180]}")
+            lines.append("")
+
+        return {"results": "\n".join(lines)}
+    except Exception as e:
+        logger.error(f"search_chats failed: {e}")
+        return {"error": str(e), "exit_code": 1}

src/tools/system.py

@@ -0,0 +1,700 @@
+"""System-domain tool implementations.
+
+Extracted from tool_implementations.py as part of slice 1 (#4082/#4071).
+Holds the skills/tasks tools plus the generic API bridges (api_call, app_api).
+The admin manage_* tools (endpoints, mcp, webhooks, tokens, settings) live in
+``src.agent_tools.admin_tools`` after the upstream registry migration (#3629);
+``src.tool_implementations`` re-exports both sets for backward compatibility.
+"""
+import json
+import logging
+import os
+import re
+from typing import Any, Dict, List, Optional
+
+from src.tools._common import _parse_tool_args
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Skills management tool
+# ---------------------------------------------------------------------------
+
+async def do_manage_skills(content: str, owner: Optional[str] = None) -> Dict:
+    """Handle manage_skills tool calls.
+
+    SKILL.md-backed CRUD with progressive disclosure (Hermes-style). Actions:
+
+      list / index               — Level 0: name + description summary.
+      view {name}                — Level 1: full SKILL.md.
+      view_ref {name, path}      — Level 2: a sub-file under the skill dir.
+      add  {name, description, when_to_use, procedure[], pitfalls[],
+            verification[], tags[], category, status}
+                                 — Create a new skill (draft by default).
+      patch {name, old_string, new_string}
+                                 — Token-efficient surgical edit on the
+                                   raw SKILL.md text. Fails on ambiguous
+                                   `old_string` (multiple matches).
+      edit  {name, content}      — Replace the entire SKILL.md.
+      publish {name}             — Flip status: draft -> published.
+      delete {name}              — Remove the skill directory.
+      search {query}             — Relevance match on published skills.
+    """
+    try:
+        args = _parse_tool_args(content)
+    except ValueError:
+        return {"error": "Invalid JSON arguments", "exit_code": 1}
+
+    action = (args.get("action") or "").lower()
+    from services.memory.skills import SkillsManager
+    from services.memory.skill_format import Skill, slugify
+    from src.constants import DATA_DIR
+    sm = SkillsManager(DATA_DIR)
+
+    # Accept legacy `skill_id` as an alias for `name`.
+    name = (args.get("name") or args.get("skill_id") or "").strip()
+
+    if action in ("list", "index", ""):
+        all_skills = sm.load(owner=owner)
+        if not all_skills:
+            return {"results": "No skills yet. Create one with action='add'."}
+        published = [s for s in all_skills if s.get("status") == "published"]
+        drafts = [s for s in all_skills if s.get("status") == "draft"]
+        lines = []
+        if published:
+            lines.append("## Published")
+            for s in sorted(published, key=lambda x: x["name"]):
+                lines.append(f"- **{s['name']}** ({s.get('category','general')}): {s.get('description','')}")
+        if drafts:
+            lines.append("\n## Drafts")
+            for s in sorted(drafts, key=lambda x: x["name"]):
+                lines.append(f"- **{s['name']}** [draft]: {s.get('description','')}")
+        return {"results": "\n".join(lines) if lines else "No skills yet."}
+
+    if action == "view":
+        if not name:
+            return {"error": "name is required for view", "exit_code": 1}
+        md = sm.read_skill_md(name, owner=owner)
+        if md is None:
+            return {"error": f"Skill {name!r} not found", "exit_code": 1}
+        return {"results": md}
+
+    if action == "view_ref":
+        if not name:
+            return {"error": "name is required for view_ref", "exit_code": 1}
+        ref = (args.get("path") or "").strip()
+        if not ref:
+            return {"error": "path is required for view_ref", "exit_code": 1}
+        text = sm.read_skill_reference(name, ref, owner=owner)
+        if text is None:
+            return {"error": f"Reference {ref!r} not found under {name!r}", "exit_code": 1}
+        return {"results": text}
+
+    if action == "add":
+        if not name:
+            return {
+                "error": "name is required for add. Provide the exact slug the user should see, then report the returned name.",
+                "exit_code": 1,
+            }
+        proc = args.get("procedure")
+        if proc is None:
+            proc = args.get("steps") or []
+        if not proc and not args.get("body_extra") and not args.get("solution"):
+            return {"error": "procedure (or solution body) is required", "exit_code": 1}
+        # Same auto-publish gate as the extractor path — when the user
+        # has auto_approve_skills on and the caller didn't pin an explicit
+        # status, publish immediately. Audit later demotes/removes on fail.
+        _status_arg = args.get("status")
+        if not _status_arg:
+            try:
+                from routes.prefs_routes import _load_for_user as _load_prefs
+                _prefs = _load_prefs(owner) or {}
+                _status_arg = "published" if _prefs.get("auto_approve_skills", True) else "draft"
+            except Exception:
+                _status_arg = "draft"
+        entry = sm.add_skill(
+            name=args.get("name"),
+            description=(args.get("description") or args.get("title") or "").strip(),
+            category=args.get("category") or "general",
+            tags=args.get("tags") or [],
+            platforms=args.get("platforms") or [],
+            requires_toolsets=args.get("requires_toolsets") or [],
+            fallback_for_toolsets=args.get("fallback_for_toolsets") or [],
+            when_to_use=(args.get("when_to_use") if args.get("when_to_use") is not None
+                         else args.get("problem", "")),
+            procedure=proc,
+            pitfalls=args.get("pitfalls") or [],
+            verification=args.get("verification") or [],
+            status=_status_arg,
+            version=args.get("version") or "1.0.0",
+            confidence=args.get("confidence", 0.8),
+            source=args.get("source", "learned"),
+            teacher_model=args.get("teacher_model"),
+            owner=owner,
+            title=args.get("title", ""),
+            problem=args.get("problem", ""),
+            solution=args.get("solution", ""),
+            steps=args.get("steps") or [],
+        )
+        if entry.get("_deduped"):
+            return {"results": (
+                f"A near-identical skill already exists: `{entry['name']}` — not creating "
+                f"a duplicate. View or edit it with action='view', name='{entry['name']}'."
+            )}
+        try:
+            from src.event_bus import fire_event
+            fire_event("skill_added", owner)
+        except Exception:
+            logger.debug("skill_added event dispatch failed", exc_info=True)
+        verify_hint = ""
+        if entry.get("status") == "draft":
+            verify_hint = (
+                "\n\nThis skill is a DRAFT. Run through the procedure once to verify, "
+                f"then publish with action='publish', name='{entry['name']}'."
+            )
+        return {"results": f"Created skill `{entry['name']}` — {entry.get('description','')}{verify_hint}"}
+
+    if action == "edit":
+        if not name:
+            return {"error": "name is required for edit", "exit_code": 1}
+        new_content = args.get("content")
+        if not isinstance(new_content, str) or not new_content.strip():
+            return {"error": "content (full SKILL.md) is required for edit", "exit_code": 1}
+        try:
+            sk_new = Skill.from_markdown(new_content)
+        except Exception as e:
+            return {"error": f"Could not parse content as SKILL.md: {e}", "exit_code": 1}
+        sk_new.name = slugify(sk_new.name or name)
+        existing = sm.load(owner=owner)
+        match = next((s for s in existing if s.get("name") == name), None)
+        if not match:
+            return {"error": f"Skill {name!r} not found", "exit_code": 1}
+        if not sk_new.owner:
+            sk_new.owner = match.get("owner") or owner
+        ok = sm.update_skill(name, _skill_dump(sk_new), owner=owner)
+        return {"results": f"Edited skill `{sk_new.name}`."} if ok else {"error": "Update failed", "exit_code": 1}
+
+    if action == "patch":
+        if not name:
+            return {"error": "name is required for patch", "exit_code": 1}
+        old = args.get("old_string")
+        new_str = args.get("new_string", "")
+        if not isinstance(old, str) or not old:
+            return {"error": "old_string is required and must be non-empty", "exit_code": 1}
+        md = sm.read_skill_md(name, owner=owner)
+        if md is None:
+            return {"error": f"Skill {name!r} not found", "exit_code": 1}
+        count = md.count(old)
+        if count == 0:
+            return {"error": "old_string not found in SKILL.md", "exit_code": 1}
+        if count > 1:
+            return {"error": f"old_string is ambiguous (appears {count} times). Make it more specific.", "exit_code": 1}
+        new_md = md.replace(old, new_str, 1)
+        try:
+            sk_new = Skill.from_markdown(new_md)
+        except Exception as e:
+            return {"error": f"Patched content is not valid SKILL.md: {e}", "exit_code": 1}
+        sk_new.name = slugify(sk_new.name or name)
+        ok = sm.update_skill(name, _skill_dump(sk_new), owner=owner)
+        return {"results": f"Patched skill `{sk_new.name}`."} if ok else {"error": "Patch update failed", "exit_code": 1}
+
+    if action == "publish":
+        if not name:
+            return {"error": "name is required for publish", "exit_code": 1}
+        all_skills = sm.load(owner=owner)
+        match = next((s for s in all_skills if s.get("name") == name), None)
+        if not match:
+            return {"error": f"Skill {name!r} not found", "exit_code": 1}
+        updates = {"status": "published"}
+        if args.get("confidence") is not None:
+            updates["confidence"] = max(0.0, min(1.0, float(args["confidence"])))
+        sm.update_skill(name, updates, owner=owner)
+        return {"results": f"✅ Published `{name}`. It now appears in the skills index for future turns."}
+
+    if action == "delete":
+        if not name:
+            return {"error": "name is required for delete", "exit_code": 1}
+        ok = sm.delete_skill(name, owner=owner)
+        return {"results": f"Deleted skill `{name}`."} if ok else {"error": f"Skill {name!r} not found", "exit_code": 1}
+
+    if action == "search":
+        query = (args.get("query") or "").strip()
+        if not query:
+            return {"error": "query is required for search", "exit_code": 1}
+        results = sm.get_relevant_skills(query, sm.load(owner=owner), max_items=5)
+        if not results:
+            return {"results": "No matching skills found."}
+        lines = []
+        for sk in results:
+            proc = sk.get("procedure") or sk.get("steps") or []
+            steps_str = " → ".join(proc[:5])
+            lines.append(f"**{sk['name']}**: {sk.get('description','')}\n  When: {sk.get('when_to_use','')}\n  Steps: {steps_str}")
+        return {"results": "\n\n".join(lines)}
+
+    return {
+        "error": (
+            f"Unknown action: {action!r}. "
+            "Use one of: list, view, view_ref, add, edit, patch, publish, delete, search."
+        ),
+        "exit_code": 1,
+    }
+
+
+def _skill_dump(sk) -> Dict:
+    """Translate a parsed Skill back into the kwargs `update_skill` expects."""
+    return {
+        "name": sk.name,
+        "description": sk.description,
+        "version": sk.version,
+        "category": sk.category,
+        "tags": sk.tags,
+        "platforms": sk.platforms,
+        "requires_toolsets": sk.requires_toolsets,
+        "fallback_for_toolsets": sk.fallback_for_toolsets,
+        "status": sk.status,
+        "confidence": sk.confidence,
+        "source": sk.source,
+        "teacher_model": sk.teacher_model,
+        "owner": sk.owner,
+        "when_to_use": sk.when_to_use,
+        "procedure": sk.procedure,
+        "pitfalls": sk.pitfalls,
+        "verification": sk.verification,
+        "body_extra": sk.body_extra,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Task management tool
+# ---------------------------------------------------------------------------
+
+async def do_manage_tasks(content: str, owner: Optional[str] = None) -> Dict:
+    """Handle manage_tasks tool calls: CRUD on scheduled tasks."""
+    import uuid as _uuid
+    from core.database import SessionLocal, ScheduledTask
+    from src.task_scheduler import compute_next_run
+
+    try:
+        args = _parse_tool_args(content)
+    except ValueError:
+        return {"error": "Invalid JSON arguments", "exit_code": 1}
+
+    action = args.get("action", "list")
+    db = SessionLocal()
+    try:
+        if action == "list":
+            q = db.query(ScheduledTask)
+            if owner:
+                q = q.filter(ScheduledTask.owner == owner)
+            tasks = q.order_by(ScheduledTask.created_at.desc()).all()
+            task_list = []
+            for t in tasks:
+                task_list.append({
+                    "id": t.id, "name": t.name, "status": t.status,
+                    "task_type": t.task_type or "llm",
+                    "action": t.action,
+                    "trigger_type": t.trigger_type or "schedule",
+                    "schedule": t.schedule,
+                    "trigger_event": t.trigger_event,
+                    "trigger_count": t.trigger_count,
+                    "next_run": t.next_run.isoformat() + "Z" if t.next_run else None,
+                    "last_run": t.last_run.isoformat() + "Z" if t.last_run else None,
+                    "run_count": t.run_count or 0,
+                })
+            return {"response": f"Found {len(task_list)} tasks", "tasks": task_list, "exit_code": 0}
+
+        elif action == "create":
+            task_type = args.get("task_type", "llm")
+            trigger_type = args.get("trigger_type", "schedule")
+
+            if task_type in ("llm", "research") and not args.get("prompt"):
+                return {"error": "Prompt is required for llm/research tasks", "exit_code": 1}
+            if task_type == "action" and not args.get("action_name"):
+                return {"error": "action_name is required for action tasks", "exit_code": 1}
+
+            # Compute next_run for schedule triggers
+            next_run = None
+            if trigger_type == "schedule":
+                schedule = args.get("schedule", "daily")
+                next_run = compute_next_run(
+                    schedule, args.get("scheduled_time", "09:00"),
+                    args.get("scheduled_day"),
+                )
+
+            task_id = str(_uuid.uuid4())
+            # Guard each fallback with `or`: args.get("prompt", default) returns
+            # None when the key is present but null, and None[:50] raises.
+            name = args.get("name") or (args.get("prompt") or args.get("action_name") or "Task")[:50]
+
+            task = ScheduledTask(
+                id=task_id,
+                owner=owner,
+                name=name,
+                prompt=args.get("prompt"),
+                task_type=task_type,
+                action=args.get("action_name"),
+                schedule=args.get("schedule") if trigger_type == "schedule" else None,
+                scheduled_time=args.get("scheduled_time", "09:00") if trigger_type == "schedule" else None,
+                scheduled_day=args.get("scheduled_day"),
+                trigger_type=trigger_type,
+                trigger_event=args.get("trigger_event"),
+                trigger_count=args.get("trigger_count"),
+                trigger_counter=0,
+                next_run=next_run,
+                status="active",
+                output_target=args.get("output_target", "session"),
+            )
+            db.add(task)
+            db.commit()
+            return {"response": f"Created task '{name}' (id: {task_id})", "task_id": task_id, "exit_code": 0}
+
+        elif action == "edit":
+            task_id = args.get("task_id")
+            if not task_id:
+                return {"error": "task_id is required for edit", "exit_code": 1}
+            task = db.query(ScheduledTask).filter(ScheduledTask.id == task_id).first()
+            if not task:
+                return {"error": f"Task {task_id} not found", "exit_code": 1}
+            if owner and task.owner and task.owner != owner:
+                return {"error": "Access denied", "exit_code": 1}
+
+            changed = []
+            for field in ("name", "prompt", "output_target"):
+                if args.get(field) is not None:
+                    setattr(task, field, args[field])
+                    changed.append(field)
+            if args.get("task_type") is not None:
+                task.task_type = args["task_type"]
+                changed.append("task_type")
+            if args.get("action_name") is not None:
+                task.action = args["action_name"]
+                changed.append("action")
+            if args.get("trigger_type") is not None:
+                task.trigger_type = args["trigger_type"]
+                changed.append("trigger_type")
+            if args.get("trigger_event") is not None:
+                task.trigger_event = args["trigger_event"]
+                changed.append("trigger_event")
+            if args.get("trigger_count") is not None:
+                task.trigger_count = args["trigger_count"]
+                changed.append("trigger_count")
+
+            schedule_changed = False
+            for field in ("schedule", "scheduled_time", "scheduled_day"):
+                if args.get(field) is not None:
+                    setattr(task, field, args[field])
+                    changed.append(field)
+                    schedule_changed = True
+
+            if schedule_changed and (task.trigger_type or "schedule") == "schedule":
+                task.next_run = compute_next_run(
+                    task.schedule, task.scheduled_time, task.scheduled_day,
+                )
+
+            db.commit()
+            return {"response": f"Updated task '{task.name}': {', '.join(changed)}", "exit_code": 0}
+
+        elif action == "delete":
+            task_id = args.get("task_id")
+            if not task_id:
+                return {"error": "task_id is required for delete", "exit_code": 1}
+            task = db.query(ScheduledTask).filter(ScheduledTask.id == task_id).first()
+            if not task:
+                return {"error": f"Task {task_id} not found", "exit_code": 1}
+            if owner and task.owner and task.owner != owner:
+                return {"error": "Access denied", "exit_code": 1}
+            name = task.name
+            db.delete(task)
+            db.commit()
+            return {"response": f"Deleted task '{name}'", "exit_code": 0}
+
+        elif action in ("pause", "resume"):
+            task_id = args.get("task_id")
+            if not task_id:
+                return {"error": f"task_id is required for {action}", "exit_code": 1}
+            task = db.query(ScheduledTask).filter(ScheduledTask.id == task_id).first()
+            if not task:
+                return {"error": f"Task {task_id} not found", "exit_code": 1}
+            if owner and task.owner and task.owner != owner:
+                return {"error": "Access denied", "exit_code": 1}
+
+            if action == "pause":
+                task.status = "paused"
+            else:
+                task.status = "active"
+                if (task.trigger_type or "schedule") == "schedule":
+                    task.next_run = compute_next_run(
+                        task.schedule, task.scheduled_time, task.scheduled_day,
+                    )
+            db.commit()
+            return {"response": f"Task '{task.name}' {action}d", "exit_code": 0}
+
+        elif action == "run":
+            task_id = args.get("task_id")
+            if not task_id:
+                return {"error": "task_id is required for run", "exit_code": 1}
+            task = db.query(ScheduledTask).filter(ScheduledTask.id == task_id).first()
+            if not task:
+                return {"error": f"Task {task_id} not found", "exit_code": 1}
+            if owner and task.owner and task.owner != owner:
+                return {"error": "Access denied", "exit_code": 1}
+
+            from src.event_bus import get_task_scheduler
+            scheduler = get_task_scheduler()
+            if scheduler:
+                started = await scheduler.run_task_now(task_id)
+                if started:
+                    return {"response": f"Task '{task.name}' triggered", "exit_code": 0}
+                else:
+                    return {"error": "Task is already running", "exit_code": 1}
+            return {"error": "Task scheduler not available", "exit_code": 1}
+
+        else:
+            return {"error": f"Unknown action: {action}", "exit_code": 1}
+
+    except Exception as e:
+        logger.error(f"manage_tasks error: {e}")
+        return {"error": str(e), "exit_code": 1}
+    finally:
+        db.close()
+
+
+# ---------------------------------------------------------------------------
+# API call tool
+# ---------------------------------------------------------------------------
+
+async def do_api_call(content: str) -> Dict:
+    """Execute an API call to a registered integration."""
+    from src.integrations import execute_api_call, load_integrations
+    try:
+        args = json.loads(content)
+    except json.JSONDecodeError:
+        # Try line-based format: integration\nmethod path\nbody
+        lines = content.strip().split("\n")
+        args = {"integration": lines[0].strip() if lines else ""}
+        if len(lines) > 1:
+            parts = lines[1].strip().split(" ", 1)
+            args["method"] = parts[0] if parts else "GET"
+            args["path"] = parts[1] if len(parts) > 1 else "/"
+        if len(lines) > 2:
+            try:
+                args["body"] = json.loads("\n".join(lines[2:]))
+            except json.JSONDecodeError:
+                pass
+
+    integration_name = args.get("integration", "")
+    integrations = load_integrations()
+    intg = next((i for i in integrations if i["id"] == integration_name
+                 or i["name"].lower() == integration_name.lower()), None)
+    if not intg:
+        available = ", ".join(i["name"] for i in integrations if i.get("enabled", True))
+        return {"error": f"No integration matching '{integration_name}'. Available: {available or 'none configured'}", "exit_code": 1}
+
+    return await execute_api_call(
+        intg["id"],
+        args.get("method", "GET"),
+        args.get("path", "/"),
+        params=args.get("params"),
+        body=args.get("body"),
+        extra_headers=args.get("headers"),
+    )
+
+
+# Paths the generic `app_api` tool will refuse to call. Auth/token/user
+# administration and host shell execution are too risky to route through an
+# agent surface even when the agent is admin-context; accidental account or
+# command mistakes have permanent blast radius.
+_APP_API_BLOCKLIST_PREFIXES = (
+    "/api/auth",           # login/logout/password
+    "/api/users",          # user CRUD (bare /api/users list+create+delete must also block)
+    "/api/tokens",         # api token mgmt (bare /api/tokens list+create must also block)
+    "/api/admin",          # admin one-shots (wipe etc.)
+    "/api/shell",          # host shell execution must stay behind named command tooling
+    "/api/backup/restore", # destructive restore
+)
+
+# (method, prefix) pairs to refuse specifically. Used for endpoints
+# where GET is fine but writes are destructive or host-control shaped.
+# Saw the agent wipe cookbook_state.json (presets + tasks) by POSTing
+# {"tasks": []} to /api/cookbook/state, which overwrote the whole file.
+# Use dedicated tools or UI flows instead.
+_APP_API_BLOCKLIST_METHOD_PATH = (
+    ("GET",    "/api/email/accounts"),  # owner-filtered in tool context; use list_email_accounts MCP tool
+    ("POST",   "/api/cookbook/state"),   # whole-file overwrite — agent must use serve_preset/serve_model instead
+    ("DELETE", "/api/cookbook/state"),
+    # Host-control routes: package install, engine rebuild, and process
+    # signalling should not be reachable through the generic API bridge.
+    ("POST",   "/api/cookbook/packages/install"),
+    ("POST",   "/api/cookbook/rebuild-engine"),
+    ("POST",   "/api/cookbook/kill-pid"),
+    # Use the named tools (download_model / serve_model) — they handle
+    # host-name resolution, per-host env_prefix, AND register the task
+    # in cookbook state so it shows in the UI + list_downloads. Hitting
+    # the raw endpoint via app_api skips all of that → orphan task.
+    ("POST",   "/api/model/download"),
+    ("POST",   "/api/model/serve"),
+    # Use trigger_research — it returns a UI hint so the Deep Research
+    # sidebar surfaces the session. Raw start works but the agent
+    # fumbles the payload + the session doesn't reliably show up.
+    ("POST",   "/api/research/start"),
+    # Use the named tools — they handle owner attribution, natural-
+    # language due_date parsing, timezone, dedup, and tag/category
+    # normalization. Hitting the raw endpoint via app_api saves a
+    # note/event with the wrong fields, no reminder, or the wrong tz.
+    ("POST",   "/api/notes"),
+    ("PUT",    "/api/notes"),
+    ("DELETE", "/api/notes"),
+    ("POST",   "/api/calendar/events"),
+    ("PUT",    "/api/calendar/events"),
+    ("DELETE", "/api/calendar/events"),
+)
+
+
+async def do_app_api(content: str, owner: Optional[str] = None) -> Dict:
+    """Generic loopback to allowed internal Odysseus API endpoints. Lets the
+    agent reach the full UI-button surface (cookbook, email, notes,
+    calendar, skills, sessions, gallery, research, etc.) without us
+    landing a named tool wrapper for every one.
+
+    Args (JSON):
+      action: "call" (default) | "endpoints"
+      path:   "/api/cookbook/gpus"     # required for call
+      method: "GET" | "POST" | "PUT" | "PATCH" | "DELETE" (default GET)
+      body:   <object>                 # JSON body for POST/PUT/PATCH
+      query:  <object>                 # querystring params
+
+    The `endpoints` action returns the OpenAPI surface (method + path +
+    summary) so the agent can discover what's reachable. A blocklist
+    refuses sensitive auth/user/admin/shell paths and method-specific
+    host-control routes to keep blast radius bounded.
+    """
+    # `_internal_headers` and `_INTERNAL_BASE` still live in
+    # tool_implementations.py (shared by many domain tools). Function-local
+    # import avoids a top-level circular dependency until a later task
+    # relocates them.
+    from src.tool_implementations import _internal_headers, _INTERNAL_BASE
+
+    import httpx
+    try:
+        args = _parse_tool_args(content) if content.strip() else {}
+    except ValueError:
+        return {"error": "Invalid JSON arguments", "exit_code": 1}
+
+    action = (args.get("action") or "call").lower()
+    base = _INTERNAL_BASE
+
+    if action == "endpoints":
+        # Fetch FastAPI's OpenAPI schema so the agent can discover any
+        # endpoint without us pre-listing them. Filter by an optional
+        # `filter` keyword (substring match on path or summary).
+        kw = (args.get("filter") or "").lower()
+        try:
+            async with httpx.AsyncClient(timeout=15) as client:
+                resp = await client.get(f"{base}/openapi.json",
+                                        headers=_internal_headers())
+                data = resp.json()
+        except Exception as e:
+            return {"error": f"OpenAPI fetch failed: {e}", "exit_code": 1}
+        rows: List[Dict[str, Any]] = []
+        for path, methods in (data.get("paths") or {}).items():
+            if not isinstance(methods, dict):
+                continue
+            if any(path.startswith(p) for p in _APP_API_BLOCKLIST_PREFIXES):
+                continue
+            for method, op in methods.items():
+                if method.lower() not in ("get", "post", "put", "patch", "delete"):
+                    continue
+                if any(method.upper() == m and path.startswith(p) for m, p in _APP_API_BLOCKLIST_METHOD_PATH):
+                    continue
+                summary = (op or {}).get("summary") or (op or {}).get("description") or ""
+                if isinstance(summary, str):
+                    summary = summary.strip().split("\n")[0][:140]
+                if kw and kw not in path.lower() and kw not in (summary or "").lower():
+                    continue
+                rows.append({"method": method.upper(), "path": path, "summary": summary})
+        rows.sort(key=lambda r: (r["path"], r["method"]))
+        if not rows:
+            return {"output": f"No endpoints match filter {kw!r}." if kw else "No endpoints found.", "exit_code": 0}
+        lines = [f"{len(rows)} endpoint(s)" + (f" matching {kw!r}" if kw else "") + ":"]
+        for r in rows[:200]:
+            line = f"  {r['method']:6s} {r['path']}"
+            if r["summary"]:
+                line += f"  — {r['summary']}"
+            lines.append(line)
+        if len(rows) > 200:
+            lines.append(f"  ...({len(rows) - 200} more — filter to narrow)")
+        return {"output": "\n".join(lines), "endpoints": rows, "exit_code": 0}
+
+    # action == "call"
+    path = args.get("path") or ""
+    if not path:
+        return {"error": "path is required (e.g. '/api/cookbook/gpus')", "exit_code": 1}
+    if not path.startswith("/"):
+        path = "/" + path
+    if any(path.startswith(p) for p in _APP_API_BLOCKLIST_PREFIXES):
+        return {"error": f"Path blocked for safety: {path}. Sensitive endpoints are off-limits via app_api.", "exit_code": 1}
+
+    method = (args.get("method") or "GET").upper()
+    if method not in ("GET", "POST", "PUT", "PATCH", "DELETE"):
+        return {"error": f"Unsupported method: {method}", "exit_code": 1}
+    if any(method == m and path.startswith(p) for m, p in _APP_API_BLOCKLIST_METHOD_PATH):
+        if "/api/email/accounts" in path:
+            return {"error": "Don't use /api/email/accounts via app_api — it is owner-filtered in tool context and may return empty. Use the `list_email_accounts` email tool, then pass `account` to list_emails/read_email.", "exit_code": 1}
+        if "/api/cookbook/packages/install" in path:
+            return {"error": "Don't POST /api/cookbook/packages/install via app_api — package installation is host code execution. Use the dedicated Cookbook dependency UI/flow instead.", "exit_code": 1}
+        if "/api/cookbook/rebuild-engine" in path:
+            return {"error": "Don't POST /api/cookbook/rebuild-engine via app_api — engine rebuild mutates local or remote host state. Use the dedicated Cookbook UI/flow instead.", "exit_code": 1}
+        if "/api/cookbook/kill-pid" in path:
+            return {"error": "Don't POST /api/cookbook/kill-pid via app_api — process signalling is host control. Use the dedicated Cookbook stop/diagnostic flow instead.", "exit_code": 1}
+        if "/api/model/download" in path:
+            return {"error": "Don't POST /api/model/download directly — use the `download_model` tool (it resolves the server name, sets the venv env_prefix, and registers the task so it shows in the UI).", "exit_code": 1}
+        if "/api/model/serve" in path:
+            return {"error": "Don't POST /api/model/serve directly — use the `serve_model` or `serve_preset` tool (handles host resolution, env_prefix, and cookbook tracking).", "exit_code": 1}
+        if "/api/research/start" in path:
+            return {"error": "Don't POST /api/research/start directly — use the `trigger_research` tool (it surfaces the session in the Deep Research sidebar).", "exit_code": 1}
+        if "/api/notes" in path:
+            return {"error": "Don't hit /api/notes via app_api — use the `manage_notes` tool. It accepts natural-language due_date ('11pm today', 'tomorrow at 9am'), fires reminders from the due_date itself (no separate calendar event), and uses the caller's timezone. The raw endpoint requires ISO-UTC + a separate calendar event, both of which the agent tends to get wrong.", "exit_code": 1}
+        if "/api/calendar/events" in path:
+            return {"error": "Don't hit /api/calendar/events via app_api — use the `manage_calendar` tool. It handles tz-aware natural-language datetimes and reminder_minutes correctly. If the user wants a note + reminder, prefer `manage_notes` with due_date — it bundles both.", "exit_code": 1}
+        return {"error": f"{method} {path} is blocked — it overwrites the whole cookbook state file. Use list_serve_presets / serve_preset / serve_model instead.", "exit_code": 1}
+
+    body = args.get("body")
+    query = args.get("query") or None
+    # Pass owner so the backend impersonates the user — without this,
+    # POSTs (notes, calendar, todos, ...) get owner="internal-tool"
+    # and the user that asked for them can't see the result.
+    headers = {**_internal_headers(owner=owner), "Content-Type": "application/json"}
+
+    try:
+        async with httpx.AsyncClient(timeout=60) as client:
+            resp = await client.request(
+                method, f"{base}{path}",
+                json=body if body is not None and method in ("POST", "PUT", "PATCH") else None,
+                params=query,
+                headers=headers,
+            )
+        # Try to parse JSON; fall back to raw text.
+        try:
+            payload = resp.json()
+            preview = json.dumps(payload, indent=2, default=str)
+            if len(preview) > 4000:
+                preview = preview[:4000] + "\n... (truncated)"
+        except Exception:
+            payload = None
+            preview = (resp.text or "")[:4000]
+        if resp.status_code >= 400:
+            return {
+                "error": f"{method} {path} -> HTTP {resp.status_code}",
+                "status_code": resp.status_code,
+                "body": preview,
+                "exit_code": 1,
+            }
+        return {
+            "output": f"{method} {path} -> {resp.status_code}\n{preview}",
+            "status_code": resp.status_code,
+            "json": payload,
+            "exit_code": 0,
+        }
+    except Exception as e:
+        return {"error": f"{method} {path} failed: {e}", "exit_code": 1}

src/tools/vault.py

@@ -0,0 +1,189 @@
+"""Vault-domain tool implementations.
+
+Extracted from tool_implementations.py as part of slice 1 (#4082/#4071).
+Holds the Bitwarden CLI wrappers (vault_search / vault_get / vault_unlock)
+and their helpers (_load_vault_config, _run_bw).
+``src.tool_implementations`` re-exports these for backward compatibility.
+"""
+import json
+from typing import Dict, Optional
+
+from src.constants import VAULT_FILE
+from src.tools._common import _parse_tool_args
+
+
+def _load_vault_config() -> Dict:
+    """Load Vaultwarden config from data/vault.json."""
+    from pathlib import Path
+    p = Path(VAULT_FILE)
+    if p.exists():
+        try:
+            return json.loads(p.read_text(encoding="utf-8"))
+        except Exception:
+            pass
+    return {}
+
+
+async def _run_bw(args: list, session: Optional[str] = None, input_text: Optional[str] = None) -> tuple:
+    """Run a bw CLI command with optional session + stdin. Returns (stdout, stderr, returncode)."""
+    import asyncio
+    env = {}
+    import os as _os
+    env.update(_os.environ)
+    if session:
+        env["BW_SESSION"] = session
+
+    proc = await asyncio.create_subprocess_exec(
+        "bw", *args,
+        stdin=asyncio.subprocess.PIPE if input_text else None,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+        env=env,
+    )
+    stdout, stderr = await proc.communicate(input=input_text.encode() if input_text else None)
+    return stdout.decode(errors="replace").strip(), stderr.decode(errors="replace").strip(), proc.returncode
+
+
+async def do_vault_search(content: str, owner: Optional[str] = None) -> Dict:
+    """Search the vault by keyword. Returns matching item names + URLs, NO passwords."""
+    try:
+        args = _parse_tool_args(content)
+    except ValueError:
+        return {"error": "Invalid JSON arguments", "exit_code": 1}
+    query = args.get("query", "").strip()
+    if not query:
+        return {"error": "query is required", "exit_code": 1}
+
+    cfg = _load_vault_config()
+    session = cfg.get("session")
+    if not session:
+        return {"error": "Vault is locked. Run vault_unlock or provide session key in settings.", "exit_code": 1}
+
+    stdout, stderr, rc = await _run_bw(["list", "items", "--search", query], session=session)
+    if rc != 0:
+        return {"error": f"bw failed: {stderr[:300]}", "exit_code": 1}
+
+    try:
+        items = json.loads(stdout)
+    except json.JSONDecodeError:
+        return {"error": "Failed to parse bw output", "exit_code": 1}
+
+    if not items:
+        return {"output": f"No vault items match '{query}'.", "exit_code": 0}
+
+    lines = [f"Found {len(items)} item(s) matching '{query}':"]
+    for it in items[:20]:
+        item_id = it.get("id", "?")
+        name = it.get("name", "?")
+        login = it.get("login") or {}
+        username = login.get("username", "")
+        uris = login.get("uris") or []
+        url = uris[0].get("uri", "") if uris else ""
+        parts = [f"[{item_id[:8]}] {name}"]
+        if username:
+            parts.append(f"user: {username}")
+        if url:
+            parts.append(f"url: {url}")
+        lines.append("- " + " · ".join(parts))
+    lines.append("\nUse vault_get(item_id, reason) to retrieve the password.")
+    return {"output": "\n".join(lines), "exit_code": 0}
+
+
+async def do_vault_get(content: str, owner: Optional[str] = None) -> Dict:
+    """Retrieve a full vault entry (including password) by item ID. Logs access to assistant chat."""
+    try:
+        args = _parse_tool_args(content)
+    except ValueError:
+        return {"error": "Invalid JSON arguments", "exit_code": 1}
+    item_id = args.get("item_id", "").strip()
+    reason = args.get("reason", "").strip()
+    if not item_id:
+        return {"error": "item_id is required", "exit_code": 1}
+    if not reason:
+        return {"error": "reason is required — explain WHY you need this password", "exit_code": 1}
+
+    cfg = _load_vault_config()
+    session = cfg.get("session")
+    if not session:
+        return {"error": "Vault is locked. Unlock first.", "exit_code": 1}
+
+    stdout, stderr, rc = await _run_bw(["get", "item", item_id], session=session)
+    if rc != 0:
+        return {"error": f"bw failed: {stderr[:300]}", "exit_code": 1}
+
+    try:
+        item = json.loads(stdout)
+    except json.JSONDecodeError:
+        return {"error": "Failed to parse bw output", "exit_code": 1}
+
+    login = item.get("login") or {}
+    name = item.get("name", "?")
+
+    # Audit log to assistant chat
+    try:
+        from src.assistant_log import log_to_assistant
+        if owner:
+            log_to_assistant(
+                owner,
+                f"Retrieved password for **{name}** — reason: {reason}",
+                category="Vault",
+            )
+    except Exception:
+        pass
+
+    output = [
+        f"Vault item: {name}",
+        f"Username: {login.get('username', '(none)')}",
+        f"Password: {login.get('password', '(none)')}",
+    ]
+    if login.get("totp"):
+        output.append(f"TOTP secret: {login['totp']}")
+    uris = login.get("uris") or []
+    if uris:
+        output.append("URLs: " + ", ".join(u.get("uri", "") for u in uris))
+    if item.get("notes"):
+        output.append(f"Notes: {item['notes']}")
+
+    return {"output": "\n".join(output), "exit_code": 0}
+
+
+async def do_vault_unlock(content: str, owner: Optional[str] = None) -> Dict:
+    """Unlock the vault using a master password. Stores the resulting session key."""
+    try:
+        args = _parse_tool_args(content)
+    except ValueError:
+        return {"error": "Invalid JSON arguments", "exit_code": 1}
+    master_password = args.get("master_password", "")
+    if not master_password:
+        return {"error": "master_password is required", "exit_code": 1}
+
+    # Do not pass the master password as an argv element. Local process lists
+    # can expose argv to other users; stdin keeps the secret out of `ps`.
+    stdout, stderr, rc = await _run_bw(["unlock", "--raw"], input_text=master_password + "\n")
+    if rc != 0:
+        return {"error": f"Unlock failed: {stderr[:300]}", "exit_code": 1}
+
+    session = stdout.strip()
+    if not session:
+        return {"error": "bw returned empty session", "exit_code": 1}
+
+    # Save session to vault.json
+    from pathlib import Path
+    p = Path(VAULT_FILE)
+    cfg = {}
+    if p.exists():
+        try:
+            cfg = json.loads(p.read_text(encoding="utf-8"))
+        except Exception:
+            pass
+    cfg["session"] = session
+    from datetime import datetime as _dt
+    cfg["unlocked_at"] = _dt.utcnow().isoformat()
+    p.write_text(json.dumps(cfg, indent=2), encoding="utf-8")
+    try:
+        import os as _os
+        _os.chmod(str(p), 0o600)
+    except Exception:
+        pass
+
+    return {"output": "Vault unlocked. Session saved.", "exit_code": 0}

tests/test_admin_tools_registry.py

@@ -58,12 +58,17 @@ def test_owner_adapter_defaults_owner_to_none():
 
 
 def test_parse_tool_args_lives_in_tool_utils_single_source():
-    # The helper was de-duplicated into tool_utils; admin_tools imports it
+    # The helper was de-duplicated into tool_utils; every consumer imports it
-    # from there rather than carrying its own copy.
+    # from there rather than carrying its own copy. After the tool_implementations
+    # split, _common and the facade must also re-export the same object.
     from src.tool_utils import _parse_tool_args
     from src.agent_tools import admin_tools, document_tools
+    from src.tools import _common
+    import src.tool_implementations as ti
     assert admin_tools._parse_tool_args is _parse_tool_args
     assert document_tools._parse_tool_args is _parse_tool_args
+    assert _common._parse_tool_args is _parse_tool_args
+    assert ti._parse_tool_args is _parse_tool_args
     assert _parse_tool_args('{"action":"add"}') == {"action": "add"}
     # body-envelope unwrap still works
     assert _parse_tool_args('{"body":{"action":"x"}}') == {"action": "x"}

tests/test_caldav_bidirectional_sync.py

@@ -53,7 +53,7 @@ def test_http_calendar_writes_mark_pending_and_push_after_commit():
 
 
 def test_agent_calendar_writes_share_caldav_push_path():
-    source = Path("src/tool_implementations.py").read_text()
+    source = Path("src/tools/calendar.py").read_text()
 
     assert "_push_caldav_event_after_commit" in source
     assert 'caldav_sync_pending="create" if cal.source == "caldav" else None' in source

tests/test_internal_api_base.py

@@ -38,7 +38,7 @@ def test_no_hardcoded_loopback_left_in_call_sites():
     # Regression guard: the converted files must not reintroduce the literal.
     root = pathlib.Path(__file__).resolve().parent.parent
     for rel in (
-        "src/tool_implementations.py",
+        "src/tools/_common.py",
         "src/cookbook_serve_lifecycle.py",
         "src/builtin_actions.py",
         "routes/task_routes.py",

tests/test_tool_implementations_shim.py

@@ -0,0 +1,165 @@
+"""Protection test: the tool_implementations compatibility shim must keep
+re-exporting every symbol importers depend on.
+
+Guards the slice-1 split (tool_implementations.py -> src/tools/*) from
+accidentally dropping a symbol. The contract is enforced by two
+self-verifying tests, not by the hand-maintained list below:
+
+* ``test_shim_reexports_every_domain_do_function`` discovers every ``do_*``
+  from the domain modules and asserts reachability through the shim.
+* ``test_every_facade_import_in_repo_resolves`` discovers every
+  ``from src.tool_implementations import X`` site across first-party Python
+  dirs (src/, tests/, routes/, ...) and asserts ``X`` resolves through the
+  shim.
+
+Both fail automatically if a re-export is forgotten (the do_* discovery
+covers the tool surface; the import-site scan covers underscore helpers a
+reviewer's P3 finding showed could otherwise slip through the list). The
+``_EXPECTED`` list below is the curated historical surface (the original
+module's top-level names), kept as a belt-and-suspenders check and as the
+async-shape contract for ``do_*``; it is not the ground truth.
+"""
+
+import inspect
+
+import src.tool_implementations as ti
+
+# 33 do_* tool functions
+_EXPECTED = [
+    "do_adopt_served_model", "do_api_call", "do_app_api", "do_cancel_download",
+    "do_download_model", "do_edit_image", "do_list_cached_models",
+    "do_list_cookbook_servers", "do_list_downloads", "do_list_served_models",
+    "do_list_serve_presets", "do_manage_calendar", "do_manage_contact",
+    "do_manage_endpoints", "do_manage_mcp", "do_manage_notes",
+    "do_manage_research", "do_manage_settings", "do_manage_skills",
+    "do_manage_tasks", "do_manage_tokens", "do_manage_webhooks",
+    "do_resolve_contact", "do_search_chats", "do_search_hf_models",
+    "do_serve_model", "do_serve_preset", "do_stop_served_model",
+    "do_tail_serve_output", "do_trigger_research", "do_vault_get",
+    "do_vault_search", "do_vault_unlock",
+    # module-private helpers (importable by name too)
+    "_cookbook_apply_retry_suggestion", "_cookbook_env_for_host",
+    "_cookbook_kill_session", "_cookbook_register_task", "_cookbook_servers",
+    "_ensure_served_endpoint", "_infer_serve_host", "_infer_serve_port",
+    "_internal_headers", "_load_vault_config", "_mcp_allowed_commands",
+    "_parse_tool_args", "_resolve_cookbook_host", "_run_bw",
+    "_scan_running_model_processes", "_skill_dump", "_string_arg",
+    "_validate_cookbook_ssh_target",
+    # active-email facade helpers (no do_* prefix); consumed by
+    # routes/chat_routes.py — listed here because get_active_email has no
+    # in-repo importer, so the import-site scan below can't see it alone.
+    "set_active_email", "get_active_email", "clear_active_email",
+]
+
+
+def test_shim_reexports_all_top_level_symbols():
+    """Every original top-level function must remain importable via the module."""
+    missing = [name for name in _EXPECTED if not hasattr(ti, name)]
+    assert not missing, f"shim dropped symbols: {missing}"
+
+
+def test_do_functions_remain_async_through_shim():
+    """Every do_* must remain a coroutine function through the shim."""
+    for name in _EXPECTED:
+        if name.startswith("do_"):
+            obj = getattr(ti, name)
+            assert inspect.iscoroutinefunction(obj), (
+                f"{name} is not async via shim (got {type(obj).__name__})"
+            )
+
+
+# Domain modules that own tool implementations after the slice-1 split.
+# The shim must re-export every public do_* from each so existing
+# `from src.tool_implementations import do_X` imports keep resolving.
+_DOMAIN_MODULES = (
+    "src.tools.system",
+    "src.tools.cookbook",
+    "src.tools.search",
+    "src.tools.notes",
+    "src.tools.calendar",
+    "src.tools.image",
+    "src.tools.research",
+    "src.tools.contacts",
+    "src.tools.vault",
+    "src.agent_tools.admin_tools",  # admin manage_* tools migrated here (#3629)
+)
+
+
+def test_shim_reexports_every_domain_do_function():
+    """Auto-discovered guard: every do_* defined in a domain module must be
+    reachable through the shim.
+
+    The hand-maintained ``_EXPECTED`` list above can drift silently when a
+    new tool is added to a domain module but not re-exported by the facade
+    (exactly the omission a reviewer found post-split). This test discovers
+    the ground truth from the domain modules themselves, so a forgotten
+    re-export fails the build automatically. ``hasattr`` is used (not
+    ``dir(ti)``) because the admin symbols are re-exported lazily via
+    module ``__getattr__`` and therefore do not appear in ``dir(ti)``.
+    """
+    import importlib
+
+    dropped = []
+    for mod_name in _DOMAIN_MODULES:
+        mod = importlib.import_module(mod_name)
+        for name in dir(mod):
+            if not name.startswith("do_"):
+                continue
+            if not inspect.iscoroutinefunction(getattr(mod, name, None)):
+                continue
+            if not hasattr(ti, name):
+                dropped.append(f"{mod_name}.{name}")
+    assert not dropped, f"shim dropped domain do_* (re-export forgotten): {dropped}"
+
+
+def test_every_facade_import_in_repo_resolves():
+    """Every ``from src.tool_implementations import X`` in any first-party
+    Python dir (src/, tests/, routes/, ...) must resolve through the shim.
+
+    This makes the module-docstring contract ("existing ``from
+    src.tool_implementations import X`` imports keep working") self-verifying
+    instead of reliant on the hand-maintained ``_EXPECTED`` list, which
+    omitted three underscore helpers in a reviewer's P3 finding and can drift
+    again. The import sites are enumerated with ``ast`` rather than checked
+    at runtime because the invariant is *which names the rest of the
+    codebase asks the facade for* — no runtime hook enumerates that set,
+    only the import statements do (the narrow source-scanning exception to
+    the behavioral-first rule). The per-name assertion is runtime
+    (``hasattr``), so any forgotten re-export — helper or ``do_*`` — fails
+    here automatically.
+    """
+    import ast
+    import os
+    from pathlib import Path
+
+    repo = Path(__file__).resolve().parents[1]
+    # Walk every first-party Python dir so route-level (and any future)
+    # facade consumers are covered, not just src/ and tests/. Prune
+    # non-source trees (venvs, caches, data, build artifacts) in-place.
+    _SKIP_DIRS = {
+        "__pycache__", "venv", "node_modules", "data", "logs",
+        "odysseus.egg-info", "static", "specs", "licenses", "docker",
+    }
+    names = set()
+    for root, _dirs, files in os.walk(repo):
+        _dirs[:] = [d for d in _dirs if not (d.startswith(".") or d in _SKIP_DIRS)]
+        for fn in files:
+            if not fn.endswith(".py"):
+                continue
+            path = Path(root) / fn
+            text = path.read_text(encoding="utf-8")
+            if "src.tool_implementations" not in text:
+                continue
+            try:
+                tree = ast.parse(text, filename=str(path))
+            except SyntaxError:
+                continue  # unrelated to the facade contract
+            for node in ast.walk(tree):
+                if isinstance(node, ast.ImportFrom) and node.module == "src.tool_implementations":
+                    for alias in node.names:
+                        if alias.name != "*":
+                            names.add(alias.name)
+    unresolved = sorted(n for n in names if not hasattr(ti, n))
+    assert not unresolved, (
+        f"facade consumers import names the shim does not re-export: {unresolved}"
+    )

tests/test_vault_password_not_in_argv.py

@@ -102,7 +102,7 @@ def test_unlock_handler_feeds_password_on_stdin_not_argv():
 
 
 def test_tool_vault_unlock_feeds_password_on_stdin_not_argv():
-    text = open("src/tool_implementations.py", encoding="utf-8").read()
+    text = open("src/tools/vault.py", encoding="utf-8").read()
 
     assert '["unlock", master_password, "--raw"]' not in text
     assert '_run_bw(["unlock", master_password' not in text