yusufipek/dikte
1
0
Fork 0
mirror of https://github.com/yusufipk/dikte.git synced 2026-09-12 03:16:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

Let the command go to Codex or OpenRouter, not only Claude Code

Browse Source 
Everything the last commit built assumed one agent was installed, which is a
poor assumption to bake into a dictation tool. So the provider is a setting, and
what it selects is one of three quite different things.

Claude Code and Codex are the same shape: a CLI, streaming JSONL, a session id
to resume, tools that reach the machine and whatever is connected to it. They
share the runner. What differs is spelled out where it differs, which is more
than the flag names: Codex has no system prompt to append, so the instruction
rides in front of the command with a rule between them; it confines its commands
in a sandbox rather than asking about them, so the permission setting is a
sandbox mode; and `-s` is not accepted by `exec resume`, so both settings go
through `-c` overrides, which are.

OpenRouter is the odd one and is meant to be. No tools, no files, no calendar:
it can say what the capital of Peru is and not what is in your diary, and the
settings box says so rather than letting it be discovered. It also has no
session to resume, so the conversation is kept here and resent, capped at 24
messages.

A stored conversation names the provider that made it, and is ignored by any
other: none of them can pick up another's thread, and a stale id would otherwise
fail every command until the timeout cleared it.

The interface calls the thing by its name, which in Turkish means the suffix has
to agree with it: Claude'a but Codex'e, Claude'u but Codex'i. A name dropped
into a sentence through t() cannot be inflected by that sentence, so it arrives
inflected, from a small table in i18n. English takes the name as it is and keeps
the preposition in the sentence.
This commit is contained in:
yusufipk
2026-07-28 17:08:14 +07:00
parent 0206303751
commit 01eea7a363
9 changed files with 527 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files
assistant.py
+266 -105
View File
@@ -1,14 +1,25 @@
"""Handing a dictation to Claude Code as a command, and pasting back its answer.
"""Handing a dictation to an agent as a command, and pasting back its answer.
`claude -p` runs the same session an interactive window would open: the same
skills, the same MCP servers, the same account. So a dictation does not have to
end as text on the screen. It can be a question to answer or a job to carry out,
and what comes back is pasted exactly where the transcript would have been.
Three of them, because not everyone has the same one installed:
Its output is read as it arrives rather than waited out. A command that reaches
for the calendar or the web takes long enough that a still indicator is
indistinguishable from a hang, so every tool it picks up is named in the corner
while it works.
Claude Code `claude -p`, the session you would have opened yourself
Codex `codex exec`, the same idea from the other shop
OpenRouter a plain chat request, over the key that is already configured
The first two are the whole machine: they run commands, read files, and reach
whatever skills and services you have connected, which is what makes "put that
in my calendar on Thursday" a thing you can say. OpenRouter cannot touch any of
that, and is there so that a question still gets an answer on a machine with
neither CLI installed.
Whichever it is, the reply is pasted exactly where the transcript would have
been, and the conversation carries across dictations so that "and move that to
Friday" knows what "that" is.
The two CLIs are read as they stream rather than waited out. A command that
reaches for the calendar or the web takes long enough that a still indicator is
indistinguishable from a hang, so every tool they pick up is named in the corner
while they work.
"""
import json
@@ -18,15 +29,22 @@ import subprocess
import threading
import time
import api
import config as cfg
from i18n import t
SESSION_FILE = cfg.DATA_DIR / "assistant.json"
PROVIDERS = ("claude", "codex", "openrouter")
# What to say in the indicator for a tool, keyed by name. Anything unlisted is
# named as it comes, which is better than a generic "working" for the tools that
# arrive from an MCP server nobody wrote this table for.
TOOL_LABELS = {
# How many messages of an OpenRouter conversation are carried forward. The two
# CLIs keep their own history and need no such number; here every turn is resent
# in full, so the window has to end somewhere.
MAX_HISTORY = 24
# What to say in the indicator for a tool, keyed by the name the CLI uses.
# Anything unlisted is named as it comes, which beats a generic "working" for
# tools arriving from an MCP server nobody wrote this table for.
CLAUDE_TOOLS = {
"Bash": "Running a command…",
"BashOutput": "Running a command…",
"Read": "Reading…",
@@ -40,6 +58,14 @@ TOOL_LABELS = {
"Task": "Handing it to a subagent…",
"TodoWrite": "Planning…",
}
CODEX_ITEMS = {
"command_execution": "Running a command…",
"reasoning": "Thinking…",
"web_search": "Searching the web…",
"file_change": "Editing a file…",
"patch_apply": "Editing a file…",
"todo_list": "Planning…",
}
class AssistantError(Exception):
@@ -50,34 +76,61 @@ class Cancelled(Exception):
pass
def provider(conf):
chosen = conf["assistant_provider"]
return chosen if chosen in PROVIDERS else "claude"
def executable(name):
"""The CLI a provider runs, or "" when it needs none."""
return {"claude": "claude", "codex": "codex"}.get(name, "")
def display_name(conf):
"""What to call the thing being asked, in the tray and in the corner."""
return {"claude": "Claude", "codex": "Codex"}.get(provider(conf), "OpenRouter")
# --- the conversation -----------------------------------------------------
#
# One conversation is kept across dictations, so "and move that to tomorrow"
# means something. It is dropped once it has sat unused for long enough: an
# hour later the next command is almost certainly a new subject, and dragging
# the old one along costs tokens and invites the model to answer the wrong
# question.
# means something. It is dropped once it has sat unused for long enough: an hour
# later the next command is almost certainly a new subject, and dragging the old
# one along costs tokens and invites an answer to the wrong question. Switching
# provider drops it too, since none of them can pick up another's thread.
def read_session(max_age_seconds):
"""The session to continue, or "" when there is none worth continuing."""
def _read_row(name, max_age_seconds):
try:
with open(SESSION_FILE, encoding="utf-8") as fh:
row = json.load(fh)
except (OSError, json.JSONDecodeError, ValueError):
return ""
session = str(row.get("session", ""))
if not session:
return ""
return {}
if not isinstance(row, dict) or row.get("provider") != name:
return {}
if max_age_seconds and time.time() - row.get("ts", 0) > max_age_seconds:
return ""
return session
return {}
return row
def write_session(session):
def read_session(name, max_age_seconds):
"""The id to resume, for the providers that keep their own history."""
return str(_read_row(name, max_age_seconds).get("session", ""))
def read_messages(name, max_age_seconds):
"""The conversation so far, for the provider that does not."""
messages = _read_row(name, max_age_seconds).get("messages")
return messages if isinstance(messages, list) else []
def write_session(name, session="", messages=None):
row = {"provider": name, "session": session, "ts": time.time()}
if messages is not None:
row["messages"] = messages[-MAX_HISTORY:]
try:
cfg.DATA_DIR.mkdir(parents=True, exist_ok=True)
with open(SESSION_FILE, "w", encoding="utf-8") as fh:
json.dump({"session": session, "ts": time.time()}, fh)
json.dump(row, fh, ensure_ascii=False)
except OSError:
pass
@@ -96,7 +149,9 @@ def session_age():
row = json.load(fh)
except (OSError, json.JSONDecodeError, ValueError):
return None
return time.time() - row.get("ts", 0) if row.get("session") else None
if not isinstance(row, dict) or not (row.get("session") or row.get("messages")):
return None
return time.time() - row.get("ts", 0)
# --- the call -------------------------------------------------------------
@@ -109,34 +164,42 @@ def working_dir(conf):
def ask(prompt, conf, on_stage=None, should_stop=None):
"""Run the prompt through Claude Code. Returns (answer, warning).
"""Run the prompt through the configured agent. Returns (answer, warning).
`warning` is set when the answer arrived but something about the run should
be seen anyway, a denied tool above all: the reply still reads like a normal
one, and only the denial explains why it did not do what it was asked to.
"""
if not shutil.which("claude"):
name = provider(conf)
if name == "openrouter":
return _ask_openrouter(prompt, conf, on_stage)
binary = executable(name)
if not shutil.which(binary):
raise AssistantError(t(
"claude not found. Install Claude Code and make sure `claude` is on "
"your PATH."
"{binary} not found. Install it, or pick another provider under "
"Settings → Claude.", binary=binary,
))
session = read_session(conf["assistant_session_minutes"] * 60)
run = _ask_claude if name == "claude" else _ask_codex
session = read_session(name, conf["assistant_session_minutes"] * 60)
try:
return _run(prompt, conf, session, on_stage, should_stop)
return run(prompt, conf, session, on_stage, should_stop)
except _SessionGone:
# The conversation it pointed at is not there any more: the history was
# cleared, or it was started somewhere else. Say nothing and start over,
# because from the outside this is just the first command of the day.
clear_session()
return _run(prompt, conf, "", on_stage, should_stop)
return run(prompt, conf, "", on_stage, should_stop)
class _SessionGone(Exception):
pass
def _run(prompt, conf, session, on_stage, should_stop):
# --- Claude Code ----------------------------------------------------------
def _ask_claude(prompt, conf, session, on_stage, should_stop):
cmd = [
"claude", "-p", prompt,
"--output-format", "stream-json", "--verbose",
@@ -147,6 +210,137 @@ def _run(prompt, conf, session, on_stage, should_stop):
if session:
cmd += ["--resume", session]
found = {"answer": "", "warning": "", "session": "", "failure": ""}
def on_event(event):
kind = event.get("type")
if kind == "system" and event.get("subtype") == "init":
found["session"] = event.get("session_id") or found["session"]
elif kind == "assistant" and on_stage:
for block in event.get("message", {}).get("content", []) or []:
if isinstance(block, dict) and block.get("type") == "tool_use":
on_stage(_claude_label(block))
elif kind == "result":
found["session"] = event.get("session_id") or found["session"]
answer = (event.get("result") or "").strip()
if event.get("is_error"):
found["failure"] = answer or t("Claude ended with an error.")
else:
found["answer"] = answer
found["warning"] = _denial_warning(event)
code, stderr = _stream(cmd, conf, on_event, should_stop)
return _conclude(found, code, stderr, session, "Claude")
def _claude_label(block):
name = block.get("name", "")
if name in CLAUDE_TOOLS:
return t(CLAUDE_TOOLS[name])
if name == "Skill":
return t("Using {name}…", name=(block.get("input") or {}).get("skill") or "a skill")
if name.startswith("mcp__"):
parts = name.split("__")
return t("Using {name}…", name=parts[1] if len(parts) > 1 else name)
return t("Using {name}…", name=name or "a tool")
def _denial_warning(event):
denials = event.get("permission_denials") or []
names = []
for denial in denials:
name = denial.get("tool_name") if isinstance(denial, dict) else str(denial)
if name and name not in names:
names.append(name)
return t("It was not allowed to use: {tools}", tools=", ".join(names)) if names else ""
# --- Codex ----------------------------------------------------------------
def _ask_codex(prompt, conf, session, on_stage, should_stop):
# Codex takes no system prompt of its own, so the instruction rides along in
# front of the command, kept apart from it so the two are not read as one.
body = f"{conf.assistant_prompt()}\n\n---\n\n{prompt}"
settings = [
"-c", f'sandbox_mode="{conf["assistant_codex_sandbox"]}"',
"-c", 'approval_policy="never"', # there is nobody here to approve
"--skip-git-repo-check",
"--json",
]
if conf["assistant_codex_model"].strip():
settings += ["-m", conf["assistant_codex_model"].strip()]
cmd = (["codex", "exec", "resume", session] if session else ["codex", "exec"])
cmd += settings + [body]
found = {"answer": "", "warning": "", "session": "", "failure": ""}
def on_event(event):
kind = event.get("type")
if kind == "thread.started":
found["session"] = event.get("thread_id") or found["session"]
elif kind in ("item.started", "item.completed"):
item = event.get("item") or {}
item_type = item.get("type")
# Every message is kept rather than only the last: a run can say
# something, use a tool and speak again, and the closing one is the
# answer.
if item_type == "agent_message" and kind == "item.completed":
found["answer"] = (item.get("text") or "").strip() or found["answer"]
elif on_stage and kind == "item.started":
on_stage(_codex_label(item))
elif kind in ("turn.failed", "error"):
error = event.get("error") or {}
found["failure"] = (error.get("message") if isinstance(error, dict)
else str(error)) or t("Codex ended with an error.")
code, stderr = _stream(cmd, conf, on_event, should_stop)
return _conclude(found, code, stderr, session, "Codex")
def _codex_label(item):
item_type = item.get("type", "")
if item_type in CODEX_ITEMS:
return t(CODEX_ITEMS[item_type])
if item_type == "mcp_tool_call":
return t("Using {name}…", name=item.get("server") or item.get("tool") or "a tool")
return t("Using {name}…", name=item_type or "a tool")
# --- OpenRouter -----------------------------------------------------------
def _ask_openrouter(prompt, conf, on_stage):
"""No tools, no files, no calendar: a question and an answer.
It is the fallback for a machine with neither CLI on it, so it says what it
knows and nothing else. The conversation is ours to keep here, since there
is no session on the other end to resume.
"""
if on_stage:
on_stage(t("Thinking…"))
history = read_messages("openrouter", conf["assistant_session_minutes"] * 60)
messages = history + [{"role": "user", "content": prompt}]
try:
answer = api.chat(
messages, conf.openrouter_key(), conf["assistant_openrouter_model"],
conf.assistant_prompt(), base_url=conf["openrouter_base_url"],
timeout=conf["assistant_timeout"],
)
except api.ApiError as exc:
raise AssistantError(str(exc)) from exc
write_session("openrouter",
messages=messages + [{"role": "assistant", "content": answer}])
return answer, ""
# --- running a CLI --------------------------------------------------------
def _stream(cmd, conf, on_event, should_stop):
"""Run cmd, hand every JSON line it prints to on_event.
Returns (exit code, stderr). Raises Cancelled when the stop was asked for,
and AssistantError when the clock ran out.
"""
try:
proc = subprocess.Popen(
cmd, cwd=working_dir(conf), stdin=subprocess.DEVNULL,
@@ -154,9 +348,9 @@ def _run(prompt, conf, session, on_stage, should_stop):
text=True, encoding="utf-8", errors="replace", bufsize=1,
)
except OSError as exc:
raise AssistantError(t("Could not run claude: {error}", error=exc)) from exc
raise AssistantError(t("Could not run {binary}: {error}",
binary=cmd[0], error=exc)) from exc
answer, warning, new_session, failure = "", "", "", ""
# Reading the stream blocks between lines, and a model that thinks for a
# minute sends none. So the clock and the stop button are watched from the
# side, and they end the run by killing the process: that closes the stream
@@ -171,23 +365,16 @@ def _run(prompt, conf, session, on_stage, should_stop):
try:
for line in proc.stdout:
line = line.strip()
# Both CLIs print the odd unstructured line among the JSON.
if not line.startswith("{"):
continue
try:
event = json.loads(line)
except (json.JSONDecodeError, ValueError):
continue
kind = event.get("type")
if kind == "system" and event.get("subtype") == "init":
new_session = event.get("session_id", "") or new_session
elif kind == "assistant" and on_stage:
for label in _labels(event):
on_stage(label)
elif kind == "result":
new_session = event.get("session_id", "") or new_session
answer = (event.get("result") or "").strip()
if event.get("is_error"):
failure = answer or t("Claude ended with an error.")
answer = ""
warning = _denial_warning(event)
if isinstance(event, dict):
on_event(event)
finally:
stderr = _finish(proc)
watchdog.join(timeout=1)
@@ -195,24 +382,33 @@ def _run(prompt, conf, session, on_stage, should_stop):
if ended["cancelled"]:
raise Cancelled()
if ended["timed_out"]:
raise AssistantError(t(
"Claude did not finish within {seconds} seconds.",
seconds=conf["assistant_timeout"],
))
if proc.returncode != 0 and not answer:
raise AssistantError(t("It did not finish within {seconds} seconds.",
seconds=conf["assistant_timeout"]))
return proc.returncode, stderr
def _conclude(found, code, stderr, session, service):
"""Turn what the stream said into an answer, or into the reason there is none."""
if code != 0 and not found["answer"]:
if session and _session_missing(stderr):
raise _SessionGone()
raise AssistantError(_first_line(stderr) or failure or t(
"claude exited with code {code}.", code=proc.returncode
))
if failure:
raise AssistantError(failure)
if not answer:
raise AssistantError(t("Claude answered with nothing."))
raise AssistantError(_last_line(stderr) or found["failure"] or t(
"{service} exited with code {code}.", service=service, code=code))
if found["failure"] and not found["answer"]:
raise AssistantError(found["failure"])
if not found["answer"]:
raise AssistantError(t("{service} answered with nothing.", service=service))
if found["session"]:
write_session("claude" if service == "Claude" else "codex", found["session"])
return found["answer"], found["warning"]
if new_session:
write_session(new_session)
return answer, warning
def _session_missing(stderr):
lowered = (stderr or "").lower()
if "session" in lowered or "thread" in lowered or "conversation" in lowered:
return any(word in lowered for word in ("not found", "no such", "unknown",
"does not exist", "no conversation"))
return False
def _watch(proc, deadline, should_stop, ended):
@@ -228,49 +424,14 @@ def _watch(proc, deadline, should_stop, ended):
_kill(proc)
def _labels(event):
"""The indicator lines for one assistant message, in the order they happen."""
out = []
for block in event.get("message", {}).get("content", []) or []:
if not isinstance(block, dict) or block.get("type") != "tool_use":
continue
name = block.get("name", "")
if name in TOOL_LABELS:
out.append(t(TOOL_LABELS[name]))
elif name == "Skill":
skill = (block.get("input") or {}).get("skill", "")
out.append(t("Using {name}…", name=skill or "a skill"))
elif name.startswith("mcp__"):
parts = name.split("__")
out.append(t("Using {name}…", name=parts[1] if len(parts) > 1 else name))
elif name:
out.append(t("Using {name}…", name=name))
return out
def _denial_warning(event):
denials = event.get("permission_denials") or []
if not denials:
return ""
names = []
for denial in denials:
name = denial.get("tool_name") if isinstance(denial, dict) else str(denial)
if name and name not in names:
names.append(name)
return t("Claude was not allowed to use: {tools}", tools=", ".join(names))
def _session_missing(stderr):
lowered = stderr.lower()
return "session" in lowered and ("not found" in lowered or "no conversation" in lowered)
def _kill(proc):
proc.terminate()
try:
proc.terminate()
proc.wait(timeout=3)
except subprocess.TimeoutExpired:
proc.kill()
except OSError:
pass
def _finish(proc):
@@ -290,6 +451,6 @@ def _finish(proc):
return stderr
def _first_line(text):
def _last_line(text):
lines = [line for line in (text or "").splitlines() if line.strip()]
return lines[-1].strip() if lines else ""