@-routing: leading tags select provider/model/bot with per-dir DB defaults

Leading @tags on any turn now route and pin: @provider pins the
backend, @model-id pins a model, @botname routes one turn (bare
@bot switches sticky), tags combine freely, @auto clears the pins,
and @path references a file by absolute path. Selections save into
the dir_defaults table keyed by directory and re-apply on next boot
there. Completion for all four kinds in the REPL (readline) and in
TUI textboxes (mentions= source on TuiTextBox/TuiPromptDialog).

Bump 1.22.0 -> 1.22.1.
This commit is contained in:
2026-10-11 02:28:01 +02:00
parent bd1c400a27
commit 2e0e0547ed
4 changed files with 629 additions and 34 deletions
+3
View File
@@ -250,12 +250,15 @@ A **profile** is a complete identity: system message plus session history under
A **bot** is a lighter-weight thing inside one profile: just a system message plus its own history, with the whole vault shared. Every profile starts with `main`; `create_bot` adds named bots from a description, rules, behavior, and optional nicknames (lowercased, with a short prefix auto-registering). `/bot` switches roles and resumes exactly where that bot left off; `@name` routes a single turn to another bot and files the question and answer into *both* histories, so the bot you're actually talking to stays aware a detour even happened.
Leading `@tags` also select the backend: `@provider` pins a provider (`muse`, `opencode`, `ollama`, ...), `@model-id` pins a model, and tags combine freely (`@opencode @coder fix this`). A bare tag with no prompt just switches (`@coder` moves to that bot). Every selection is saved as the default for the current directory in the vault (`dir_defaults` table) and re-applied on the next boot there; `@auto` clears the pins. `@path` anywhere in the text references a file by absolute path for the agent to read. All of it Tab-completes in the REPL and in TUI boxes.
/profile show current profile
/profile [name] switch profile, creating it when missing
/profiles list all profiles
/bots list bots with nicknames
/bot coder switch to the coder bot
@coder fix this one turn via coder, logged in both
@opencode @coder fix this same turn pinned to opencode
`remember` merges an instruction into the current profile's system message *through the model itself* — adding facts, updating behavior, or removing forgotten items while leaving the rest intact — and fires by default on new passwords or behavior changes. `recall` searches the per-profile episodic log in `~/.tai/memory.db`. Credentials never enter memory this way: secrets go through `store_secret` (or `/secret set`), and any profile written before that rule existed gets migrated to it automatically on load.
+362 -28
View File
@@ -93,7 +93,7 @@ def _fromisoformat(text):
return parsed.replace(tzinfo=tzinfo)
raise ValueError("invalid datetime, use ISO like 2026-10-08T09:00")
VERSION = "1.22.0"
VERSION = "1.22.1"
WORKER_STEPS = 12
CREATE_SKILL_STEPS = 40
FORK_MAX_DEPTH = 2
@@ -141,6 +141,9 @@ ZEN_FREE_MODELS = ("mimo-v2.6-flash-free", "nemotron-3.5-lightning-free", "nemot
POLLINATIONS_LABEL = "pollinations"
POLLINATIONS_BASE = "https://text.pollinations.ai/openai/v1"
POLLINATIONS_MODELS = ("openai",)
PROVIDER_LABELS = (MUSE_LABEL, GROK_LABEL, CLAUDE_LABEL, CODEX_LABEL, GEMINI_LABEL, OPENCODE_LABEL, OLLAMA_LABEL, OPENROUTER_LABEL, POLLINATIONS_LABEL, ZEN_LABEL, DEVPLACE_LABEL)
MENTION_AUTO = "auto"
DIR_DEFAULTS_CAP = 512
OPENCODE_USER_AGENT = "opencode/1.18.29 ai-sdk/provider-utils/4.0.23 runtime/bun/1.3.15"
OPENCODE_CLIENT_NAME = "cli"
MODEL_HEALTH_FILE = "model_health.json"
@@ -289,6 +292,114 @@ EDGE_CHROMIUM = "143.0.3650.75"
PROFILE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,31}$")
BOT_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,31}$")
BOT_MENTION_RE = re.compile(r"^@([A-Za-z0-9][A-Za-z0-9_-]{0,31})\s+(.*)$", re.DOTALL)
LEADING_TAG_RE = re.compile(r"@(\S+)\s*")
FILE_MENTION_RE = re.compile(r"(?<!\S)@(\S+)")
def dir_key(directory=None):
try:
return os.path.realpath(directory or os.getcwd())
except OSError:
return ""
def split_leading_tags(text):
tags = []
rest = text or ""
while rest.startswith("@"):
match = LEADING_TAG_RE.match(rest)
if not match:
break
tags.append(match.group(1))
rest = rest[match.end():]
return tags, rest
def known_model_ids(config, health=None):
ids = set()
ids.update(ZEN_FREE_MODELS)
ids.update(POLLINATIONS_MODELS)
ids.update(DEVPLACE_FREE_ROUTES)
ids.add(OPENROUTER_MODEL)
for attr in ("openrouter_models", "devplace_models", "pollinations_models", "zen_models", "opencode_models", "ollama_models"):
for item in getattr(config, attr, None) or []:
if item:
ids.add(item)
for backend in getattr(config, "backends", None) or []:
if (backend or {}).get("model"):
ids.add(backend["model"])
if getattr(config, "explicit_model", ""):
ids.add(config.explicit_model)
if health is not None:
for name in ("openrouter:free", "opencode:free", "ollama:local", "devplace:free", "opencode:box:free"):
try:
cached = health.cached_list(name)
except Exception:
cached = None
for item in cached or []:
ids.add(item)
return ids
def resolve_tag(tag, agent, directory):
word = (tag or "").strip()
lowered = word.lower()
target = agent.store.resolve_bot(agent.profile, lowered)
if target is not None:
return ("bot", target)
if lowered in PROVIDER_LABELS:
return ("provider", lowered)
for backend in agent.config.backends or []:
if ((backend or {}).get("label") or "").strip().lower() == lowered:
return ("provider", lowered)
if word in known_model_ids(agent.config, getattr(agent.chat, "health", None)):
return ("model", word)
if lowered == MENTION_AUTO:
return ("auto", MENTION_AUTO)
core = word.rstrip(",;:!?)")
expanded = os.path.expanduser(core)
path = expanded if os.path.isabs(expanded) else os.path.join(directory, expanded)
if core and os.path.exists(path):
return ("file", os.path.abspath(path))
return ("unknown", word)
def expand_file_mentions(text, directory):
def replace(match):
core = match.group(1).rstrip(",;:!?)")
punct = match.group(1)[len(core):]
expanded = os.path.expanduser(core)
path = expanded if os.path.isabs(expanded) else os.path.join(directory, expanded)
if core and os.path.exists(path):
return os.path.abspath(path) + punct
return match.group(0)
return FILE_MENTION_RE.sub(replace, text or "")
def complete_files(prefix, directory, cap=100):
if "/" in prefix:
sub, _, part = prefix.rpartition("/")
base = os.path.expanduser(os.path.join(directory, sub))
else:
base, part = os.path.expanduser(directory), prefix
try:
entries = sorted(os.listdir(base))
except OSError:
return []
found = []
for entry in entries:
if entry.startswith(".") and not part.startswith("."):
continue
if not entry.lower().startswith(part.lower()):
continue
rel = entry if "/" not in prefix else sub + "/" + entry
if os.path.isdir(os.path.join(base, entry)):
rel += "/"
found.append(rel)
if len(found) >= cap:
break
return found
SECRET_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$")
LEGACY_PASSWORD_NOTE = "especially passwords (collect and keep every password the user shares), preferences, or standing behavior changes"
VAULT_MEMORY_NOTE = "preferences, or standing behavior changes. Store passwords, tokens, and secrets with store_secret and reference them by name; never write secret values into memory"
@@ -361,6 +472,7 @@ HELP_TEXT = (
" /demo fullscreen TUI framework showcase\n"
" /help this text\n"
" /quit exit\n"
" @bot @provider @model route this turn and save dir defaults (@auto clears pins, @path references a file)\n"
"anything else is sent to the agent."
)
@@ -1988,6 +2100,7 @@ SCHEMA = (
("edges", ("src TEXT", "dst TEXT", "relation TEXT", "profile TEXT", "created TEXT"), ("src", "dst", "relation", "profile")),
("audit", ("id INTEGER PRIMARY KEY", "profile TEXT", "ts TEXT", "actor TEXT", "action TEXT", "path TEXT", "message TEXT", "old_size INTEGER", "new_size INTEGER", "old TEXT", "new TEXT", "tags TEXT"), ()),
("chunks", ("id INTEGER PRIMARY KEY", "parent_kind TEXT", "parent_id TEXT", "idx INTEGER", "text TEXT", "profile TEXT", "created TEXT"), ()),
("dir_defaults", ("dir TEXT PRIMARY KEY", "provider TEXT", "model TEXT", "bot TEXT", "updated TEXT"), ()),
)
SCHEMA_INDEXES = (
@@ -2736,6 +2849,38 @@ class Store:
except OSError:
pass
def load_dir_defaults(self, directory):
try:
row = self.db.execute("SELECT provider, model, bot FROM dir_defaults WHERE dir = ?", (directory,)).fetchone()
except sqlite3.Error:
return {}
if not row:
return {}
return {key: value for key, value in zip(("provider", "model", "bot"), row) if value}
def save_dir_defaults(self, directory, provider=None, model=None, bot=None, clear=()):
if not directory:
return
current = self.load_dir_defaults(directory)
for key, value in (("provider", provider), ("model", model), ("bot", bot)):
if value:
current[key] = value
for key in clear or ():
current.pop(key, None)
try:
if not current:
self.db.execute("DELETE FROM dir_defaults WHERE dir = ?", (directory,))
else:
self.db.execute(
"INSERT OR REPLACE INTO dir_defaults (dir, provider, model, bot, updated) VALUES (?, ?, ?, ?, ?)",
(directory, current.get("provider", ""), current.get("model", ""), current.get("bot", ""), now_iso()))
self.db.execute(
"DELETE FROM dir_defaults WHERE dir NOT IN (SELECT dir FROM dir_defaults ORDER BY updated DESC LIMIT ?)",
(DIR_DEFAULTS_CAP,))
self.db.commit()
except sqlite3.Error:
pass
def turn_state_path(self, profile, bot="main"):
if bot == "main":
return os.path.join(self.config.profiles_dir, "%s.turn.json" % profile)
@@ -3840,6 +3985,8 @@ class ChatClient:
self.config = config
self.scope = ("default", "main")
self.fixed_key = None
self.provider_pin = None
self.model_pin = None
self.health = ModelHealth(getattr(config, "home", "") or "")
def muse_key(self):
@@ -3948,7 +4095,22 @@ class ChatClient:
cli = [cand for cand in self.candidates() if cand["chat_url"] in CLI_CHAT_URLS]
ranked = sorted(enumerate(http), key=lambda pair: (openrouter_free_tier(pair[1]), -self.health.weight(pair[1]["provider"], pair[1]["model"]), pair[0]))
usable = [cand for _, cand in ranked if not self.health.circuit_open(cand["provider"], cand["model"])]
return (usable or [cand for _, cand in ranked]) + cli
return self._apply_pins((usable or [cand for _, cand in ranked]) + cli)
def _apply_pins(self, cands):
if self.provider_pin:
cands = [cand for cand in cands if (cand["provider"] or "").lower() == self.provider_pin]
if self.model_pin:
cands = [cand for cand in cands if cand["model"] == self.model_pin]
return cands
def pin_note(self):
bits = []
if self.provider_pin:
bits.append("provider '%s'" % self.provider_pin)
if self.model_pin:
bits.append("model '%s'" % self.model_pin)
return "no backend offers " + " and ".join(bits) if bits else "no backends available"
def backends(self):
rows = []
@@ -4007,6 +4169,8 @@ class ChatClient:
while True:
errors = []
cands = self.ordered_candidates() if round_no == 0 else self.all_candidates()
if not cands:
raise BackendError(self.pin_note())
for cand in cands:
reply = self.try_candidate(cand, messages, tools, payload, stream_sink, ephemeral, errors)
if reply is not None:
@@ -4035,7 +4199,7 @@ class ChatClient:
http = [cand for cand in self.candidates() if cand["chat_url"] not in CLI_CHAT_URLS]
cli = [cand for cand in self.candidates() if cand["chat_url"] in CLI_CHAT_URLS]
ranked = sorted(enumerate(http), key=lambda pair: (openrouter_free_tier(pair[1]), -self.health.weight(pair[1]["provider"], pair[1]["model"]), pair[0]))
return [cand for _, cand in ranked] + cli + self.box_opencode_candidates()
return self._apply_pins([cand for _, cand in ranked] + cli + self.box_opencode_candidates())
def box_opencode_candidates(self):
"""Last-resort only -- complete() reaches this via all_candidates(),
@@ -7118,16 +7282,81 @@ class Agent:
if self._turn_goal:
self.store.save_turn_state(self.profile, self.bot, {"open": False, "goal": self._turn_goal, "steps": self._turn_steps, "tools": self._turn_tools, "updated": now_iso()})
def mention_error(self, tag):
bots = ", ".join(item["name"] for item in self.store.list_bots(self.profile))
return ("unknown bot @%s (available: %s; providers: %s; @path references a file)"
% (tag, bots, ", ".join(PROVIDER_LABELS)))
def mention_confirm(self, selectors, cleared, directory):
bits = []
if cleared and selectors["provider"] is None and selectors["model"] is None:
bits.append("pins cleared")
if selectors["provider"] is not None:
bits.append("provider pinned to %s" % selectors["provider"])
if selectors["model"] is not None:
bits.append("model pinned to %s" % selectors["model"])
if selectors["bot"] is not None:
bits.append("switched to bot '%s'" % selectors["bot"])
if not bits:
bits.append("nothing to select")
suffix = "saved as default for %s" % directory if self.persist else "not saved (persist off)"
return "%s (%s)" % ("; ".join(bits), suffix)
def apply_mentions(self, text, capture=False, max_steps=MAX_STEPS):
"""Parse leading @tags. Returns (text, result): result is not None
when the turn is fully handled (unknown-tag error, bare-selector
confirm, or a routed bot turn)."""
tags, rest = split_leading_tags(text)
directory = dir_key()
selectors = {"provider": None, "model": None, "bot": None}
cleared = False
files = []
loud = not capture and not self.quiet
for tag in tags:
kind, value = resolve_tag(tag, self, directory)
if kind == "unknown":
error = self.mention_error(tag)
if loud:
print(error)
return text, error
if kind == "auto":
cleared = True
selectors["provider"] = None
selectors["model"] = None
continue
if kind == "file":
files.append(value)
continue
selectors[kind] = value
if cleared or selectors["provider"] is not None:
self.chat.provider_pin = selectors["provider"]
if cleared or selectors["model"] is not None:
self.chat.model_pin = selectors["model"]
if (cleared or any(selectors.values())) and self.persist:
clear = [key for key in ("provider", "model") if cleared and selectors[key] is None]
self.store.save_dir_defaults(directory, provider=selectors["provider"], model=selectors["model"],
bot=selectors["bot"], clear=clear)
prompt = expand_file_mentions(rest, directory)
if files:
prompt = " ".join(files + ([prompt] if prompt.strip() else []))
if not tags:
return prompt, None
if not prompt.strip():
if selectors["bot"] is not None and selectors["bot"] != self.bot:
self.switch_bot(selectors["bot"], silent=True)
confirm = self.mention_confirm(selectors, cleared, directory)
if loud:
print(confirm)
return None, confirm
if selectors["bot"] is not None and selectors["bot"] != self.bot:
return None, self.run_mention(selectors["bot"], prompt, text, capture, max_steps)
return prompt, None
def run_turn(self, text, capture=False, max_steps=MAX_STEPS, _routed=False):
if not _routed:
mentioned = BOT_MENTION_RE.match(text or "")
if mentioned:
target = self.store.resolve_bot(self.profile, mentioned.group(1))
if target is None:
return "unknown bot @%s (available: %s)" % (mentioned.group(1), ", ".join(item["name"] for item in self.store.list_bots(self.profile)))
if target != self.bot:
return self.run_mention(target, mentioned.group(2), text, capture, max_steps)
text = mentioned.group(2)
text, result = self.apply_mentions(text, capture, max_steps)
if result is not None:
return result
self.messages.append({"role": "user", "content": text})
self.store.log_event(self.profile, "user", "message", self.store.redact(text, self.profile))
self._turn_open = True
@@ -7520,6 +7749,54 @@ def complete_command(text, state):
return options[state] if state < len(options) else None
def mention_options(prefix, agent, directory=None, cap=200):
found = []
seen = set()
low = (prefix or "").lower()
try:
bots = agent.store.list_bots(agent.profile)
except Exception:
bots = []
for item in bots:
for name in [item["name"]] + list(item.get("nicknames") or []):
if name.lower().startswith(low) and name not in seen:
seen.add(name)
found.append("@" + name)
labels = list(PROVIDER_LABELS) + [MENTION_AUTO]
for backend in agent.config.backends or []:
label = ((backend or {}).get("label") or "").strip()
if label:
labels.append(label)
for label in labels:
if label.lower().startswith(low) and label not in seen:
seen.add(label)
found.append("@" + label)
try:
models = sorted(known_model_ids(agent.config, getattr(agent.chat, "health", None)))
except Exception:
models = []
for model in models:
if model.lower().startswith(low) and model not in seen:
seen.add(model)
found.append("@" + model)
for rel in complete_files(prefix or "", directory or dir_key()):
if rel not in seen:
seen.add(rel)
found.append("@" + rel)
return found[:cap]
def complete_mention(text, state, agent):
options = mention_options(text[1:], agent) if (text or "").startswith("@") else []
return options[state] if state < len(options) else None
def complete_input(text, state, agent=None):
if (text or "").startswith("@") and agent is not None:
return complete_mention(text, state, agent)
return complete_command(text, state)
def handle_command(agent, text):
parts = text[1:].split(None, 1)
name = parts[0].lower()
@@ -8523,6 +8800,27 @@ def install_self():
return 0
def apply_dir_defaults(agent, directory=None):
key = dir_key(directory)
saved = agent.store.load_dir_defaults(key)
if not saved:
return ""
bits = []
if saved.get("provider"):
agent.chat.provider_pin = saved["provider"]
bits.append("provider=%s" % saved["provider"])
if saved.get("model") and not getattr(agent.config, "explicit_model", ""):
agent.chat.model_pin = saved["model"]
bits.append("model=%s" % saved["model"])
if saved.get("bot") and saved["bot"] != agent.bot:
switched = agent.switch_bot(saved["bot"], silent=True)
if switched.startswith("switched"):
bits.append("bot=%s" % saved["bot"])
if not bits:
return ""
return "dir defaults for %s: %s" % (key, " ".join(bits))
def boot(args):
args.yes = args.yes or args.yolo or args.auto
config = Config(args)
@@ -8552,7 +8850,14 @@ def boot(args):
backed_up = None
if backed_up is not None:
print("self-backup: %s" % os.path.basename(backed_up))
return Agent(config, store, yolo=args.yolo or args.auto, auto=args.auto)
agent = Agent(config, store, yolo=args.yolo or args.auto, auto=args.auto)
try:
loaded = apply_dir_defaults(agent)
except OSError:
loaded = ""
if loaded:
print(paint(loaded, Ansi.DIM))
return agent
# --------------------------------------------------------------------------- #
@@ -10555,7 +10860,7 @@ class TuiOptions(TuiPanel):
self.move(1)
COMPLETE_WORD_RE = re.compile(r"(?:^|\s)/([A-Za-z0-9_-]*)$")
COMPLETE_WORD_RE = re.compile(r"(?:^|\s)([/@])([A-Za-z0-9_./-]*)$")
class TuiAutocomplete(TuiOptions):
@@ -10595,6 +10900,29 @@ def tui_command_help():
return TUI_COMMAND_HELP
def tui_complete_mention(prefix, directory=None, bots=(), models=()):
"""Completion source for @mentions: [(label, value)] filtered by
prefix across bots, providers, known models, and files."""
found = []
low = (prefix or "").lower()
for name in bots:
if name.lower().startswith(low):
found.append(("@%s [bot]" % name, name))
for label in PROVIDER_LABELS:
if label.lower().startswith(low):
found.append(("@%s [provider]" % label, label))
if MENTION_AUTO.startswith(low):
found.append(("@%s [reset pins]" % MENTION_AUTO, MENTION_AUTO))
if not models:
models = sorted(set(ZEN_FREE_MODELS) | set(POLLINATIONS_MODELS) | set(DEVPLACE_FREE_ROUTES) | {OPENROUTER_MODEL})
for mid in models:
if mid.lower().startswith(low):
found.append(("@%s [model]" % mid, mid))
for rel in complete_files(prefix or "", directory or dir_key()):
found.append(("@%s [file]" % rel, rel))
return found[:100]
def tui_complete_command(prefix):
"""Completion source for slash commands: [(label, name)] filtered by
prefix. Labels carry the HELP_TEXT description when one exists."""
@@ -10614,9 +10942,10 @@ class TuiTextBox(TuiPanel):
lines, Backspace/Delete join them, clicks position the cursor. Escape
is left alone so dialogs can close on it, unless a completion popup
is open: then Esc dismisses the popup first. complete= wires a
prefix-to-items callable driving a TuiAutocomplete popup."""
prefix-to-items callable for /words and mentions= the same for
@words, both driving a TuiAutocomplete popup."""
def __init__(self, text="", complete=None, **kwargs):
def __init__(self, text="", complete=None, mentions=None, **kwargs):
kwargs.setdefault("focusable", True)
kwargs.setdefault("border", True)
super().__init__(**kwargs)
@@ -10625,6 +10954,7 @@ class TuiTextBox(TuiPanel):
self.cy = len(self.lines) - 1
self.top = 0
self.complete_fn = complete
self.mention_fn = mentions
self.complete_popup = None
@property
@@ -10694,13 +11024,16 @@ class TuiTextBox(TuiPanel):
def _complete_refresh(self):
popup = self.complete_popup
if not callable(self.complete_fn) or self.app is None:
if self.app is None:
if popup is not None:
self._complete_hide()
return
match = COMPLETE_WORD_RE.search(self.lines[self.cy][:self.cx])
fn = None
if match:
fn = self.complete_fn if match.group(1) == "/" else self.mention_fn
try:
items = self.complete_fn(match.group(1)) if match else []
items = fn(match.group(2)) if (match and callable(fn)) else []
except Exception:
items = []
if not items:
@@ -10779,17 +11112,17 @@ class TuiTextBox(TuiPanel):
def _complete_accept(self, value):
line = self.lines[self.cy]
start = self.cx
while start > 0 and (line[start - 1].isalnum() or line[start - 1] in "_-"):
start -= 1
if start <= 0 or line[start - 1] != "/":
match = COMPLETE_WORD_RE.search(line[:self.cx])
if not match:
self._complete_hide()
return
word_start = start - 1
trigger = match.group(1)
word_start = match.start(1)
chars = "_-" if trigger == "/" else "_-./"
end = self.cx
while end < len(line) and (line[end].isalnum() or line[end] in "_-"):
while end < len(line) and (line[end].isalnum() or line[end] in chars):
end += 1
replacement = "/" + str(value)
replacement = trigger + str(value)
self.lines[self.cy] = line[:word_start] + replacement + line[end:]
self.cx = word_start + len(replacement)
if self.cx >= len(self.lines[self.cy]):
@@ -10972,13 +11305,13 @@ class TuiPromptDialog(TuiDialog):
"""Pretty prompt box: label, multiline editor, OK/Cancel. OK runs
on_execute(value) then closes; Cancel and Esc just close."""
def __init__(self, label="Enter prompt:", box_h=3, on_execute=None, complete=None, **kwargs):
def __init__(self, label="Enter prompt:", box_h=3, on_execute=None, complete=None, mentions=None, **kwargs):
kwargs.setdefault("title", "Prompt")
kwargs.setdefault("w", 56)
super().__init__(**kwargs)
self.on_execute = on_execute
self.add(TuiText(label, style=TuiStyle(dim=True)))
self.box = TuiTextBox(h=box_h, style=TuiStyle(bg="black", fg="white"), complete=complete)
self.box = TuiTextBox(h=box_h, style=TuiStyle(bg="black", fg="white"), complete=complete, mentions=mentions)
self.add(self.box)
self.buttons(("OK", lambda _event: self._run()), ("Cancel", lambda _event: self.close()))
@@ -11822,7 +12155,8 @@ def tui_build_demo(app, live=True):
toast("executing: %s" % first)
log.append("prompt: %s" % text.split("\n")[0][:60])
root.add(TuiPromptDialog(on_execute=execute, id="prompt", complete=tui_complete_command,
label="Enter prompt (/ for commands):"))
mentions=tui_complete_mention,
label="Enter prompt (/ commands, @ mentions):"))
return True
def on_double_tap(event):
@@ -12004,7 +12338,7 @@ def repl(agent):
readline.read_history_file(agent.config.history_path)
except OSError:
pass
readline.set_completer(complete_command)
readline.set_completer(lambda text, state: complete_input(text, state, agent))
readline.parse_and_bind("tab: complete")
scheduler_stop = start_scheduler(agent.config, agent.store.seal)
try:
+256
View File
@@ -1972,6 +1972,236 @@ class BotsTests(unittest.TestCase):
self.assertEqual(self.agent.bot, "coder")
class MentionTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
os.environ["TAI_HOME"] = self.tmp.name
self.workdir = tempfile.TemporaryDirectory()
self.saved_cwd = os.getcwd()
os.chdir(self.workdir.name)
class FakeArgs:
profile = "t"
yes = True
self.config = tai.Config(FakeArgs())
self.store = tai.Store(self.config, tai.Seal(self.config.home, "mention-test-1"))
self.agent = tai.Agent(self.config, self.store, persist=True, quiet=True)
def tearDown(self):
os.chdir(self.saved_cwd)
self.store.close()
self.workdir.cleanup()
self.tmp.cleanup()
os.environ.pop("TAI_HOME", None)
def fake_chat(self, seen):
reply = {"role": "assistant", "content": "done-1", "reasoning": "", "tool_calls": [], "backend": "x"}
def fake_complete(messages, tools, stream_sink=None, deadline=None):
seen.append([item.get("content", "") for item in messages if item.get("role") == "user"])
seen.append(self.agent.chat.provider_pin)
return dict(reply)
return mock.patch.object(self.agent.chat, "complete", side_effect=fake_complete)
def tools_dispatch_create_bot(self):
self.agent.tools.dispatch("create_bot", json.dumps({"name": "coder", "description": "writes code"}))
def test_split_leading_tags(self):
self.assertEqual(tai.split_leading_tags("@a @b rest"), (["a", "b"], "rest"))
self.assertEqual(tai.split_leading_tags("@a"), (["a"], ""))
self.assertEqual(tai.split_leading_tags("hi @a"), ([], "hi @a"))
self.assertEqual(tai.split_leading_tags("@"), ([], "@"))
self.assertEqual(tai.split_leading_tags(""), ([], ""))
def test_resolve_precedence(self):
self.tools_dispatch_create_bot()
with open("coder", "w") as handle:
handle.write("x")
with open("note.txt", "w") as handle:
handle.write("x")
self.assertEqual(tai.resolve_tag("coder", self.agent, os.getcwd()), ("bot", "coder"))
self.assertEqual(tai.resolve_tag("opencode", self.agent, os.getcwd()), ("provider", "opencode"))
self.assertEqual(tai.resolve_tag("mimo-v2.6-flash-free", self.agent, os.getcwd()), ("model", "mimo-v2.6-flash-free"))
self.assertEqual(tai.resolve_tag("auto", self.agent, os.getcwd()), ("auto", "auto"))
kind, value = tai.resolve_tag("note.txt", self.agent, os.getcwd())
self.assertEqual(kind, "file")
self.assertTrue(value.endswith("note.txt"))
self.assertEqual(tai.resolve_tag("nope-zzz", self.agent, os.getcwd()), ("unknown", "nope-zzz"))
def test_combined_tags_route_and_pin(self):
self.tools_dispatch_create_bot()
seen = []
with self.fake_chat(seen):
answer = self.agent.run_turn("@opencode @coder write frob", capture=True)
self.assertEqual(answer, "done-1")
self.assertEqual(self.agent.chat.provider_pin, "opencode")
self.assertEqual(self.agent.bot, "main")
self.assertEqual(seen[1], "opencode")
saved = self.store.load_dir_defaults(tai.dir_key())
self.assertEqual(saved, {"provider": "opencode", "bot": "coder"})
def test_last_tag_wins(self):
answer = self.agent.run_turn("@ollama @opencode", capture=True)
self.assertIn("opencode", answer)
self.assertEqual(self.agent.chat.provider_pin, "opencode")
def test_unknown_tag_errors(self):
answer = self.agent.run_turn("@ghost-zzz hi", capture=True)
self.assertIn("unknown bot", answer)
self.assertIn("@ghost-zzz", answer)
self.assertIsNone(self.agent.chat.provider_pin)
self.assertEqual(self.store.load_dir_defaults(tai.dir_key()), {})
def test_bare_selectors_confirm_and_save(self):
seen = []
with self.fake_chat(seen):
answer = self.agent.run_turn("@opencode", capture=True)
self.assertIn("provider pinned to opencode", answer)
self.assertIn("saved as default", answer)
self.assertEqual(seen, [])
self.assertEqual(self.store.load_dir_defaults(tai.dir_key()), {"provider": "opencode"})
def test_bare_bot_switches_sticky(self):
self.tools_dispatch_create_bot()
answer = self.agent.run_turn("@coder", capture=True)
self.assertIn("switched to bot", answer)
self.assertEqual(self.agent.bot, "coder")
self.assertEqual(self.store.load_dir_defaults(tai.dir_key()), {"bot": "coder"})
def test_auto_clears(self):
self.agent.chat.provider_pin = "opencode"
self.agent.chat.model_pin = "mimo-v2.6-flash-free"
self.store.save_dir_defaults(tai.dir_key(), provider="opencode", model="mimo-v2.6-flash-free", bot="main")
answer = self.agent.run_turn("@auto", capture=True)
self.assertIn("pins cleared", answer)
self.assertIsNone(self.agent.chat.provider_pin)
self.assertIsNone(self.agent.chat.model_pin)
self.assertEqual(self.store.load_dir_defaults(tai.dir_key()), {"bot": "main"})
def test_model_tag_pins(self):
seen = []
with self.fake_chat(seen):
self.agent.run_turn("@mimo-v2.6-flash-free hi", capture=True)
self.assertEqual(self.agent.chat.model_pin, "mimo-v2.6-flash-free")
self.assertEqual(self.store.load_dir_defaults(tai.dir_key()), {"model": "mimo-v2.6-flash-free"})
def test_midtext_file_expands(self):
with open("doc.txt", "w") as handle:
handle.write("x")
seen = []
with self.fake_chat(seen):
self.agent.run_turn("explain @doc.txt briefly", capture=True)
user_text = seen[0][-1]
self.assertIn(os.path.realpath("doc.txt"), user_text)
self.assertNotIn("@doc.txt", user_text)
def test_midtext_prose_untouched(self):
seen = []
with self.fake_chat(seen):
self.agent.run_turn("ping me @home base", capture=True)
self.assertEqual(seen[0][-1], "ping me @home base")
def test_missing_midtext_path_stays_literal(self):
seen = []
with self.fake_chat(seen):
self.agent.run_turn("read @nope-missing-1.txt now", capture=True)
self.assertIn("@nope-missing-1.txt", seen[0][-1])
def test_leading_file_tag(self):
with open("doc.txt", "w") as handle:
handle.write("x")
seen = []
with self.fake_chat(seen):
self.agent.run_turn("@doc.txt summarize", capture=True)
self.assertTrue(seen[0][-1].startswith(os.path.realpath("doc.txt")))
def test_pins_filter_candidates(self):
cands = [
{"provider": "opencode", "chat_url": "opencode://run", "models_url": "x", "key": "", "model": "a", "profile": ""},
{"provider": "ollama", "chat_url": "http://x", "models_url": "y", "key": "", "model": "b", "profile": ""},
]
with mock.patch.object(self.agent.chat, "candidates", return_value=cands):
self.agent.chat.provider_pin = "opencode"
got = self.agent.chat.ordered_candidates()
self.assertEqual([cand["provider"] for cand in got], ["opencode"])
self.agent.chat.provider_pin = None
self.agent.chat.model_pin = "b"
got = self.agent.chat.all_candidates()
self.assertEqual([cand["model"] for cand in got], ["b"])
def test_empty_candidates_raise(self):
cands = [{"provider": "opencode", "chat_url": "opencode://run", "models_url": "x", "key": "", "model": "a", "profile": ""}]
with mock.patch.object(self.agent.chat, "candidates", return_value=cands):
self.agent.chat.provider_pin = "no-such-provider"
with self.assertRaises(tai.BackendError) as raised:
self.agent.chat.complete([{"role": "user", "content": "hi"}])
self.assertIn("no-such-provider", str(raised.exception))
def test_dir_defaults_roundtrip(self):
self.store.save_dir_defaults("d1", provider="opencode")
self.store.save_dir_defaults("d1", bot="coder")
self.assertEqual(self.store.load_dir_defaults("d1"), {"provider": "opencode", "bot": "coder"})
self.store.save_dir_defaults("d1", clear=("provider",))
self.assertEqual(self.store.load_dir_defaults("d1"), {"bot": "coder"})
self.store.save_dir_defaults("d1", clear=("bot",))
self.assertEqual(self.store.load_dir_defaults("d1"), {})
def test_apply_dir_defaults(self):
self.tools_dispatch_create_bot()
self.store.save_dir_defaults(tai.dir_key(), provider="opencode", model="mimo-v2.6-flash-free", bot="coder")
note = tai.apply_dir_defaults(self.agent)
self.assertIn("dir defaults", note)
self.assertEqual(self.agent.chat.provider_pin, "opencode")
self.assertEqual(self.agent.chat.model_pin, "mimo-v2.6-flash-free")
self.assertEqual(self.agent.bot, "coder")
def test_apply_dir_defaults_skips_unknown_bot(self):
self.store.save_dir_defaults(tai.dir_key(), bot="ghost-zzz")
note = tai.apply_dir_defaults(self.agent)
self.assertEqual(note, "")
self.assertEqual(self.agent.bot, "main")
def test_apply_dir_defaults_env_model_wins(self):
self.config.explicit_model = "env-model-1"
self.store.save_dir_defaults(tai.dir_key(), model="mimo-v2.6-flash-free")
tai.apply_dir_defaults(self.agent)
self.assertIsNone(self.agent.chat.model_pin)
def test_complete_mention(self):
self.tools_dispatch_create_bot()
with open("doc.txt", "w") as handle:
handle.write("x")
self.assertEqual(tai.complete_mention("@op", 0, self.agent), "@opencode")
self.assertEqual(tai.complete_mention("@cod", 0, self.agent), "@coder")
self.assertEqual(tai.complete_mention("@doc", 0, self.agent), "@doc.txt")
self.assertIsNone(tai.complete_mention("@doc", 5, self.agent))
self.assertIsNone(tai.complete_mention("xx", 0, self.agent))
self.assertEqual(tai.complete_input("/he", 0, None), "/help")
self.assertEqual(tai.complete_input("@op", 0, self.agent), "@opencode")
def test_loud_paths_print(self):
self.agent.quiet = False
with mock.patch("builtins.print") as shown:
self.agent.run_turn("@ghost-zzz hi", capture=False)
self.agent.run_turn("@opencode", capture=False)
printed = "\n".join(str(call.args[0]) for call in shown.call_args_list if call.args)
self.assertIn("unknown bot @ghost-zzz", printed)
self.assertIn("provider pinned to opencode", printed)
def test_complete_files(self):
os.makedirs("src/nest", exist_ok=True)
with open("src/main.py", "w") as handle:
handle.write("x")
with open(".hidden", "w") as handle:
handle.write("x")
self.assertEqual(tai.complete_files("src/ma", os.getcwd()), ["src/main.py"])
self.assertEqual(tai.complete_files("src/", os.getcwd()), ["src/main.py", "src/nest/"])
self.assertNotIn(".hidden", tai.complete_files("", os.getcwd()))
self.assertIn(".hidden", tai.complete_files(".", os.getcwd()))
class FTSSearchTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
@@ -4498,6 +4728,32 @@ class TuiAutocompleteTests(unittest.TestCase):
plain = [label for label, value in tai.tui_complete_command("exi") if value == "exit"]
self.assertEqual(plain, ["/exit"])
def test_mention_provider_accepts(self):
app, box = self.make_box(mentions=tai.tui_complete_mention)
self.type(app, "@op")
popup = box.complete_popup
self.assertIsNotNone(popup)
self.assertEqual(popup.options[0][1], "opencode")
self.key(app, "enter", "\n")
self.assertEqual(box.value, "@opencode ")
def test_mention_file_accepts(self):
with tempfile.TemporaryDirectory() as tmp:
with open(os.path.join(tmp, "doc.txt"), "w") as handle:
handle.write("x")
app, box = self.make_box(mentions=lambda prefix: tai.tui_complete_mention(prefix, directory=tmp))
self.type(app, "@do")
popup = box.complete_popup
self.assertIsNotNone(popup)
self.assertEqual([value for _label, value in popup.options], ["doc.txt"])
self.key(app, "enter", "\n")
self.assertEqual(box.value, "@doc.txt ")
def test_mention_needs_fn(self):
app, box = self.make_box(mentions=None, complete=None)
self.type(app, "@op")
self.assertIsNone(box.complete_popup)
class TuiPromptDialogTests(unittest.TestCase):
def make_demo(self):
+8 -6
View File
@@ -182,12 +182,14 @@ callbacks; exceptions in callbacks never break the loop.
Press event); `on_close(event)` fires on close.
- `TuiQuestion(prompt, options, on_done, multi)` - prompt plus
options; single-select answers on first pick, multi-select via OK.
- `TuiTextBox(text, complete)` - multiline editor; arrows/home/end/
pgup/pgdn move (no line wrap), Enter splits, Backspace/Delete join,
click positions the cursor, Escape bubbles for dialogs;
`.value` joins lines. Bracketed pastes insert literally with
CR/CRLF normalized to LF. `complete(prefix)` wires a
`TuiAutocomplete` popup for `/words` before the cursor.
- `TuiTextBox(text, complete, mentions)` - multiline editor; arrows/
home/end/pgup/pgdn move (no line wrap), Enter splits,
Backspace/Delete join, click positions the cursor, Escape bubbles
for dialogs; `.value` joins lines. Bracketed pastes insert
literally with CR/CRLF normalized to LF. `complete(prefix)` wires
a `TuiAutocomplete` popup for `/words` and `mentions(prefix)` the
same for `@words` (bots, providers, models, files) before the
cursor.
- `TuiAutocomplete` - non-focusable `TuiOptions` popup owned by a
textbox; typing filters, arrows/home/end/pgup/pgdn move, Enter
accepts, Esc dismisses (a second Esc still bubbles); flips above