remove tools/ — now at repo root

This commit is contained in:
Levi Neely 2026-07-19 12:37:15 +02:00
parent 65154790bd
commit 51af575e48
42 changed files with 0 additions and 3956 deletions

View File

View File

@ -1,22 +0,0 @@
"""Adapter registry: detect language from file path, return appropriate adapter."""
import os
from .gopls import Gopls
from .clangd import Clangd
from .intelephense import Intelephense
_ADAPTERS = [Gopls, Clangd, Intelephense]
_EXT_MAP: dict[str, type] = {}
for _a in _ADAPTERS:
for _ext in _a.extensions:
_EXT_MAP[_ext] = _a
def detect(filepath: str):
"""Return an adapter class for the given file, or None."""
ext = os.path.splitext(filepath)[1]
cls = _EXT_MAP.get(ext)
if cls:
return cls()
return None

View File

@ -1,19 +0,0 @@
"""clangd adapter."""
from ..workspace import find_root
class Clangd:
name = "clangd"
extensions = [".c", ".h", ".cc", ".cpp", ".cxx", ".hpp", ".hxx"]
root_markers = ["compile_commands.json", "compile_flags.txt", ".clangd", "CMakeLists.txt", "Makefile"]
cmd = ["clangd", "--background-index"]
def initialization_options(self):
return {}
def workspace_root(self, filepath: str, fallback: str | None = None) -> str:
return find_root(filepath, self.root_markers, fallback)
def prepare_open(self, filepath: str) -> dict | None:
return None

View File

@ -1,24 +0,0 @@
"""gopls adapter."""
from ..workspace import find_root
class Gopls:
name = "gopls"
extensions = [".go"]
root_markers = ["go.work", "go.mod"]
cmd = ["gopls", "serve"]
def initialization_options(self):
return {
"usePlaceholders": True,
"staticcheck": True,
"gofumpt": True,
}
def workspace_root(self, filepath: str, fallback: str | None = None) -> str:
return find_root(filepath, self.root_markers, fallback)
def prepare_open(self, filepath: str) -> dict | None:
"""Return extra didOpen params if needed. gopls needs didOpen before queries."""
return None # handled generically by the bridge

View File

@ -1,19 +0,0 @@
"""intelephense adapter (PHP)."""
from ..workspace import find_root
class Intelephense:
name = "intelephense"
extensions = [".php"]
root_markers = ["composer.json", "composer.lock", ".git"]
cmd = ["intelephense", "--stdio"]
def initialization_options(self):
return {}
def workspace_root(self, filepath: str, fallback: str | None = None) -> str:
return find_root(filepath, self.root_markers, fallback)
def prepare_open(self, filepath: str) -> dict | None:
return None

View File

@ -1,418 +0,0 @@
"""LSP bridge daemon: manages LSP server processes, accepts requests over a Unix socket."""
import asyncio
import json
import os
import signal
import sys
import time
from pathlib import Path
from urllib.parse import quote as url_quote
# Ensure _lib is importable when running as a standalone daemon
_tools_dir = str(Path(__file__).resolve().parent.parent.parent)
if _tools_dir not in sys.path:
sys.path.insert(0, _tools_dir)
SOCK_PATH = os.path.join(os.environ.get("XDG_RUNTIME_DIR", "/tmp"), "ollie-lsp.sock")
PID_PATH = SOCK_PATH + ".pid"
IDLE_TIMEOUT = 300 # 5 minutes
def file_uri(path: str) -> str:
return "file://" + os.path.abspath(path)
def uri_to_path(uri: str) -> str:
if uri.startswith("file://"):
return uri[7:]
return uri
class LspServer:
"""A running LSP server process."""
def __init__(self, adapter, root: str):
self.adapter = adapter
self.root = root
self.proc: asyncio.subprocess.Process | None = None
self.req_id = 0
self.pending: dict[int, asyncio.Future] = {}
self.initialized = False
self.open_files: set[str] = set()
self._read_task: asyncio.Task | None = None
self._buf = b""
async def start(self):
self.proc = await asyncio.create_subprocess_exec(
*self.adapter.cmd,
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.DEVNULL,
)
self._read_task = asyncio.create_task(self._read_loop())
await self._initialize()
async def stop(self):
if self.proc and self.proc.returncode is None:
await self._request("shutdown", {})
self._notify("exit", None)
try:
await asyncio.wait_for(self.proc.wait(), timeout=5)
except asyncio.TimeoutError:
self.proc.kill()
if self._read_task:
self._read_task.cancel()
async def _initialize(self):
result = await self._request("initialize", {
"processId": os.getpid(),
"rootUri": file_uri(self.root),
"rootPath": self.root,
"capabilities": {
"textDocument": {
"definition": {"linkSupport": True},
"references": {},
"hover": {"contentFormat": ["plaintext", "markdown"]},
"rename": {"prepareSupport": True},
"documentSymbol": {"hierarchicalDocumentSymbolSupport": True},
"completion": {"completionItem": {"snippetSupport": False}},
"publishDiagnostics": {},
},
"workspace": {
"workspaceFolders": True,
"symbol": {},
},
},
"initializationOptions": self.adapter.initialization_options(),
"workspaceFolders": [{"uri": file_uri(self.root), "name": os.path.basename(self.root)}],
})
self._notify("initialized", {})
self.initialized = True
return result
async def ensure_open(self, filepath: str):
if filepath in self.open_files:
return
try:
with open(filepath, "r") as f:
text = f.read()
except (OSError, UnicodeDecodeError):
text = ""
ext = os.path.splitext(filepath)[1]
lang_map = {
".go": "go", ".c": "c", ".h": "c", ".cc": "cpp", ".cpp": "cpp",
".cxx": "cpp", ".hpp": "cpp", ".hxx": "cpp", ".php": "php",
}
self._notify("textDocument/didOpen", {
"textDocument": {
"uri": file_uri(filepath),
"languageId": lang_map.get(ext, "plaintext"),
"version": 1,
"text": text,
}
})
self.open_files.add(filepath)
# Give the server a moment to index the file
await asyncio.sleep(0.1)
async def definition(self, filepath: str, line: int, col: int):
await self.ensure_open(filepath)
return await self._request("textDocument/definition", {
"textDocument": {"uri": file_uri(filepath)},
"position": {"line": line, "character": col},
})
async def references(self, filepath: str, line: int, col: int):
await self.ensure_open(filepath)
return await self._request("textDocument/references", {
"textDocument": {"uri": file_uri(filepath)},
"position": {"line": line, "character": col},
"context": {"includeDeclaration": True},
})
async def hover(self, filepath: str, line: int, col: int):
await self.ensure_open(filepath)
return await self._request("textDocument/hover", {
"textDocument": {"uri": file_uri(filepath)},
"position": {"line": line, "character": col},
})
async def completion(self, filepath: str, line: int, col: int):
await self.ensure_open(filepath)
return await self._request("textDocument/completion", {
"textDocument": {"uri": file_uri(filepath)},
"position": {"line": line, "character": col},
})
async def rename(self, filepath: str, line: int, col: int, new_name: str):
await self.ensure_open(filepath)
return await self._request("textDocument/rename", {
"textDocument": {"uri": file_uri(filepath)},
"position": {"line": line, "character": col},
"newName": new_name,
})
async def document_symbols(self, filepath: str):
await self.ensure_open(filepath)
return await self._request("textDocument/documentSymbol", {
"textDocument": {"uri": file_uri(filepath)},
})
async def workspace_symbols(self, query: str):
return await self._request("workspace/symbol", {"query": query})
async def diagnostics(self, filepath: str):
await self.ensure_open(filepath)
# Most servers push diagnostics via notification; we wait briefly then return cached
await asyncio.sleep(1.0)
return self._cached_diagnostics.get(file_uri(filepath), [])
_cached_diagnostics: dict[str, list] = {}
def _notify(self, method: str, params):
msg = {"jsonrpc": "2.0", "method": method}
if params is not None:
msg["params"] = params
self._send(msg)
async def _request(self, method: str, params: dict, timeout: float = 30.0) -> dict | list | None:
self.req_id += 1
rid = self.req_id
msg = {"jsonrpc": "2.0", "id": rid, "method": method, "params": params}
fut = asyncio.get_event_loop().create_future()
self.pending[rid] = fut
self._send(msg)
try:
return await asyncio.wait_for(fut, timeout=timeout)
except asyncio.TimeoutError:
self.pending.pop(rid, None)
raise
def _send(self, msg: dict):
body = json.dumps(msg).encode()
header = f"Content-Length: {len(body)}\r\n\r\n".encode()
if self.proc and self.proc.stdin:
self.proc.stdin.write(header + body)
async def _read_loop(self):
try:
while self.proc and self.proc.returncode is None:
# Read header
header_line = await self.proc.stdout.readline()
if not header_line:
break
content_length = 0
while header_line and header_line.strip():
if header_line.startswith(b"Content-Length:"):
content_length = int(header_line.split(b":")[1].strip())
header_line = await self.proc.stdout.readline()
if content_length == 0:
continue
body = await self.proc.stdout.readexactly(content_length)
msg = json.loads(body)
self._handle_message(msg)
except (asyncio.CancelledError, asyncio.IncompleteReadError):
pass
def _handle_message(self, msg: dict):
if "id" in msg and "method" not in msg:
# Response
rid = msg["id"]
fut = self.pending.pop(rid, None)
if fut and not fut.done():
if "error" in msg:
fut.set_exception(RuntimeError(msg["error"].get("message", str(msg["error"]))))
else:
fut.set_result(msg.get("result"))
elif "method" in msg and "id" not in msg:
# Notification
if msg["method"] == "textDocument/publishDiagnostics":
params = msg.get("params", {})
self._cached_diagnostics[params.get("uri", "")] = params.get("diagnostics", [])
class Bridge:
"""The bridge daemon: manages LSP servers, handles tool requests."""
def __init__(self):
self.servers: dict[str, LspServer] = {} # key: "adapter_name:root"
self.last_activity = time.time()
async def handle_request(self, data: dict) -> dict:
self.last_activity = time.time()
method = data.get("method")
filepath = data.get("file")
if not filepath:
return {"error": "missing 'file' parameter"}
filepath = os.path.abspath(filepath)
from _lib.lsp.adapters import detect
adapter = detect(filepath)
if not adapter:
return {"error": f"no adapter for file: {filepath}"}
fallback = data.get("cwd")
root = adapter.workspace_root(filepath, fallback)
server = await self._get_server(adapter, root)
try:
if method == "definition":
result = await server.definition(filepath, data["line"], data["col"])
elif method == "references":
result = await server.references(filepath, data["line"], data["col"])
elif method == "hover":
result = await server.hover(filepath, data["line"], data["col"])
elif method == "completion":
result = await server.completion(filepath, data["line"], data["col"])
elif method == "rename":
result = await server.rename(filepath, data["line"], data["col"], data["new_name"])
if result:
applied = apply_workspace_edit(result)
return {"result": applied}
return {"result": None}
elif method == "symbols":
if data.get("workspace"):
result = await server.workspace_symbols(data.get("query", ""))
else:
result = await server.document_symbols(filepath)
elif method == "diagnostics":
result = await server.diagnostics(filepath)
else:
return {"error": f"unknown method: {method}"}
return {"result": result}
except Exception as e:
return {"error": str(e)}
async def _get_server(self, adapter, root: str) -> LspServer:
key = f"{adapter.name}:{root}"
if key not in self.servers:
server = LspServer(adapter, root)
await server.start()
self.servers[key] = server
return self.servers[key]
async def shutdown(self):
for server in self.servers.values():
await server.stop()
self.servers.clear()
def apply_workspace_edit(edit: dict) -> dict:
"""Apply a WorkspaceEdit to disk. Returns summary of changes."""
changes = {}
# Handle documentChanges (versioned) or changes (simple)
if "documentChanges" in edit:
for doc_change in edit["documentChanges"]:
if doc_change.get("kind") in ("create", "rename", "delete"):
continue # skip resource operations for now
uri = doc_change["textDocument"]["uri"]
edits = doc_change.get("edits", [])
path = uri_to_path(uri)
changes.setdefault(path, []).extend(edits)
elif "changes" in edit:
for uri, edits in edit["changes"].items():
path = uri_to_path(uri)
changes[path] = edits
summary = {}
for path, edits in changes.items():
try:
with open(path, "r") as f:
lines = f.readlines()
except OSError:
continue
# Apply edits in reverse order (bottom-up) to preserve positions
sorted_edits = sorted(edits, key=lambda e: (
e["range"]["start"]["line"],
e["range"]["start"]["character"]
), reverse=True)
for e in sorted_edits:
start = e["range"]["start"]
end = e["range"]["end"]
new_text = e["newText"]
# Convert line/char positions to string manipulation
start_line, start_char = start["line"], start["character"]
end_line, end_char = end["line"], end["character"]
# Build the new content
before = "".join(lines[:start_line]) + lines[start_line][:start_char] if start_line < len(lines) else "".join(lines)
after = lines[end_line][end_char:] + "".join(lines[end_line + 1:]) if end_line < len(lines) else ""
content = before + new_text + after
lines = content.splitlines(True)
with open(path, "w") as f:
f.writelines(lines)
summary[path] = len(sorted_edits)
return {"files_changed": summary, "total_edits": sum(summary.values())}
async def run_bridge():
bridge = Bridge()
async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
try:
data = await asyncio.wait_for(reader.read(1024 * 1024), timeout=5.0)
if not data:
return
request = json.loads(data)
if request.get("method") == "__shutdown":
response = {"result": "ok"}
writer.write(json.dumps(response).encode())
await writer.drain()
writer.close()
raise SystemExit(0)
response = await bridge.handle_request(request)
writer.write(json.dumps(response).encode())
await writer.drain()
except (json.JSONDecodeError, asyncio.TimeoutError) as e:
writer.write(json.dumps({"error": str(e)}).encode())
await writer.drain()
finally:
writer.close()
# Clean up stale socket
if os.path.exists(SOCK_PATH):
os.unlink(SOCK_PATH)
server = await asyncio.start_unix_server(handle_client, path=SOCK_PATH)
os.chmod(SOCK_PATH, 0o600)
# Write PID
with open(PID_PATH, "w") as f:
f.write(str(os.getpid()))
# Idle timeout loop
async def idle_watchdog():
while True:
await asyncio.sleep(30)
if time.time() - bridge.last_activity > IDLE_TIMEOUT:
await bridge.shutdown()
server.close()
return
watchdog = asyncio.create_task(idle_watchdog())
try:
await server.serve_forever()
except (asyncio.CancelledError, SystemExit):
pass
finally:
watchdog.cancel()
await bridge.shutdown()
if os.path.exists(SOCK_PATH):
os.unlink(SOCK_PATH)
if os.path.exists(PID_PATH):
os.unlink(PID_PATH)
def main():
asyncio.run(run_bridge())
if __name__ == "__main__":
main()

View File

@ -1,77 +0,0 @@
"""Client library for LSP tool scripts. Connects to the bridge daemon, starting it if needed."""
import json
import os
import socket
import subprocess
import sys
import time
SOCK_PATH = os.path.join(os.environ.get("XDG_RUNTIME_DIR", "/tmp"), "ollie-lsp.sock")
PID_PATH = SOCK_PATH + ".pid"
def _bridge_running() -> bool:
if not os.path.exists(PID_PATH):
return False
try:
pid = int(open(PID_PATH).read().strip())
os.kill(pid, 0)
return True
except (OSError, ValueError):
return False
def _start_bridge():
"""Start the bridge daemon as a background process."""
lib_dir = os.path.dirname(os.path.abspath(__file__))
bridge_script = os.path.join(lib_dir, "bridge.py")
subprocess.Popen(
[sys.executable, bridge_script],
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
start_new_session=True,
)
# Wait for socket to appear
for _ in range(50):
if os.path.exists(SOCK_PATH):
return
time.sleep(0.1)
print("error: bridge daemon failed to start", file=sys.stderr)
sys.exit(1)
def request(method: str, **kwargs) -> dict:
"""Send a request to the bridge and return the response."""
if not _bridge_running():
_start_bridge()
payload = {"method": method, **kwargs}
data = json.dumps(payload).encode()
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
try:
sock.connect(SOCK_PATH)
sock.sendall(data)
sock.shutdown(socket.SHUT_WR)
chunks = []
while True:
chunk = sock.recv(4096)
if not chunk:
break
chunks.append(chunk)
response = json.loads(b"".join(chunks))
except ConnectionRefusedError:
# Stale socket, restart
if os.path.exists(SOCK_PATH):
os.unlink(SOCK_PATH)
_start_bridge()
return request(method, **kwargs)
finally:
sock.close()
if "error" in response:
print(f"error: {response['error']}", file=sys.stderr)
sys.exit(1)
return response.get("result")

View File

@ -1,20 +0,0 @@
"""Workspace root detection: walk up from a file to find language-specific markers."""
import os
from pathlib import Path
def find_root(filepath: str, markers: list[str], fallback: str | None = None) -> str:
"""Walk up from filepath looking for any marker file/dir. Return the directory containing it."""
p = Path(filepath).resolve()
if p.is_file():
p = p.parent
while True:
for marker in markers:
if (p / marker).exists():
return str(p)
parent = p.parent
if parent == p:
break
p = parent
return fallback or os.getcwd()

View File

@ -1,66 +0,0 @@
#!/usr/bin/env bash
# capabilities: web
# description: Capture browser window showing a specific URL: capture_browser.sh [output_path] [url_pattern]
set -e
# Verify running under landrun (test filesystem restriction)
OUTPUT="${1:-${TMPDIR:-/tmp}/browser_capture.png}"
URL_PATTERN="${2:-localhost}"
case "$(uname -s)" in
Darwin)
# macOS: use osascript + screencapture
osascript -e 'tell application "Google Chrome" to activate' 2>/dev/null
sleep 0.3
osascript -e "tell application \"Google Chrome\" to set index of (first window whose name contains \"$URL_PATTERN\") to 1" 2>/dev/null
sleep 0.3
screencapture -x "$OUTPUT"
;;
Linux)
if [ -z "$DISPLAY" ]; then
echo "ERROR: DISPLAY not set" >&2
exit 1
fi
# Linux: use brotab + wmctrl + import
TAB_INFO=$(brotab list | grep -i "$URL_PATTERN" | head -1)
if [ -z "$TAB_INFO" ]; then
echo "ERROR: No tab matching '$URL_PATTERN'" >&2
exit 1
fi
TAB_ID=$(echo "$TAB_INFO" | cut -f1)
TAB_TITLE=$(echo "$TAB_INFO" | cut -f2)
echo "Found tab: $TAB_ID ($TAB_TITLE)" >&2
brotab activate "$TAB_ID"
sleep 0.5
WIN_ID=$(wmctrl -l | grep -i "$TAB_TITLE" | head -1 | awk '{print $1}')
if [ -z "$WIN_ID" ]; then
WIN_ID=$(wmctrl -l | grep -iE "firefox|chrome|chromium" | head -1 | awk '{print $1}')
fi
if [ -z "$WIN_ID" ]; then
echo "ERROR: Could not find browser window" >&2
exit 1
fi
echo "Capturing window: $WIN_ID" >&2
wmctrl -i -a "$WIN_ID"
sleep 0.3
import -window "$WIN_ID" "$OUTPUT"
;;
*)
echo "ERROR: Unsupported OS: $(uname -s)" >&2
exit 1
;;
esac
if [ -f "$OUTPUT" ]; then
echo "$OUTPUT"
else
echo "ERROR: Failed to capture" >&2
exit 1
fi

View File

@ -1,31 +0,0 @@
#!/bin/bash
# ollie:parallel read
# capabilities: agent-kb
# description: View a denote document by identifier
# Usage: view_denote.sh --id <denote:identifier|identifier>
set -euo pipefail
DENOTE="${DENOTE_9MOUNT:-$HOME/mnt/denote}"
ID=""
while [[ $# -gt 0 ]]; do
case "$1" in
--id) ID="$2"; shift 2 ;;
*) echo "unknown argument: $1" >&2; exit 1 ;;
esac
done
if [ -z "$ID" ]; then
echo "usage: view_denote.sh --id <denote:identifier|identifier>" >&2
exit 1
fi
ID="${ID#denote:}"
if [ -d "$DENOTE/n/$ID" ]; then
cat "$DENOTE/n/$ID/body"
else
echo "Error: no denote document found for identifier $ID" >&2
exit 1
fi

View File

@ -1,132 +0,0 @@
#!/usr/bin/env python3
# description: Replace text in a file; old_string not found is the guard
# Args: <file_path> <old_string> <new_string> [replace_all=true|1|yes]
import sys
import os
import difflib
def format_diff(path, old, new):
old_lines = old.splitlines(keepends=True)
new_lines = new.splitlines(keepends=True)
diff = ''.join(difflib.unified_diff(old_lines, new_lines, fromfile=path, tofile=path))
if not diff:
return None
return f"```diff\n{diff}```"
def find_exact(content, find):
return find if find in content else None
def find_ws_normalized(content, find):
nf = ' '.join(find.split())
find_lines = find.split('\n')
n = len(find_lines)
lines = content.split('\n')
for i in range(len(lines) - n + 1):
block = '\n'.join(lines[i:i+n])
if ' '.join(block.split()) == nf:
return block
return None
def find_indent_flexible(content, find):
def deindent(text):
ls = text.split('\n')
min_indent = float('inf')
for l in ls:
stripped = l.lstrip(' \t')
if stripped:
min_indent = min(min_indent, len(l) - len(stripped))
if min_indent == float('inf'):
return text
return '\n'.join(l[min_indent:] if len(l) > min_indent else l.lstrip(' \t') for l in ls)
nf = deindent(find)
find_lines = find.split('\n')
n = len(find_lines)
lines = content.split('\n')
for i in range(len(lines) - n + 1):
block = '\n'.join(lines[i:i+n])
if deindent(block) == nf:
return block
return None
def find_trimmed_boundary(content, find):
trimmed = find.strip()
if trimmed == find:
return None
if trimmed in content:
return trimmed
find_lines = find.split('\n')
n = len(find_lines)
lines = content.split('\n')
for i in range(len(lines) - n + 1):
block = '\n'.join(lines[i:i+n])
if block.strip() == trimmed:
return block
return None
def robust_replace(content, old_string, new_string, replace_all):
match = (find_exact(content, old_string) or
find_ws_normalized(content, old_string) or
find_indent_flexible(content, old_string) or
find_trimmed_boundary(content, old_string))
if match is None:
return None, "old_string not found in file", False
if replace_all:
return content.replace(match, new_string), None, False
n = content.count(match)
if n > 1:
return None, f"old_string is not unique ({n} occurrences); use replace_all=true or provide more context", True
return content.replace(match, new_string, 1), None, False
if len(sys.argv) < 4:
print("usage: file_edit <file_path> <old_string> <new_string> [replace_all=true|1|yes]")
sys.exit(1)
file_path = sys.argv[1]
old_string = sys.argv[2]
new_string = sys.argv[3]
_ra = sys.argv[4].lower() if len(sys.argv) > 4 else ''
replace_all = _ra.split('=')[-1] in ('true', '1', 'yes')
if not os.path.isabs(file_path):
print(f"STATUS=error STATE_CHANGED=false RETRYABLE=false")
print(f"error: path must be absolute, got: {file_path}")
sys.exit(1)
if old_string == new_string:
print("STATUS=no_change STATE_CHANGED=false RETRYABLE=false")
print("error: old_string and new_string must be different")
sys.exit(1)
if not os.path.exists(file_path):
print(f"STATUS=error STATE_CHANGED=false RETRYABLE=false")
print(f"error: file not found: {file_path}")
sys.exit(1)
try:
with open(file_path, 'r', encoding='utf-8') as f:
original = f.read()
except UnicodeDecodeError:
print(f"STATUS=error STATE_CHANGED=false RETRYABLE=false")
print(f"error: file is not valid UTF-8: {file_path}")
sys.exit(1)
new_content, err, retryable = robust_replace(original, old_string, new_string, replace_all)
if err:
print(f"STATUS=error STATE_CHANGED=false RETRYABLE={'true' if retryable else 'false'}")
print(f"error: {err}")
sys.exit(1)
try:
with open(file_path, 'w', encoding='utf-8') as f:
f.write(new_content)
except OSError as e:
print(f"STATUS=error STATE_CHANGED=false RETRYABLE=false")
print(f"error: {e}")
sys.exit(1)
diff = format_diff(file_path, original, new_content)
if diff is None:
print("STATUS=no_change STATE_CHANGED=false RETRYABLE=false")
print("(no changes)")
else:
print("STATUS=done STATE_CHANGED=true")
print(diff)

View File

@ -1,38 +0,0 @@
#!/usr/bin/env bash
# ollie:parallel read
# description: Find files by name/path pattern, sorted by modification time (newest first)
# Args: <pattern> [path]
pattern="$1"
search_path="${2:-$PWD}"
[ -z "$pattern" ] && echo "usage: file_glob <pattern> [path]" && exit 1
if [ ! -d "$search_path" ]; then
echo "STATUS=error"
echo "error: path is not a directory: $search_path"
exit 1
fi
# collect matches and sort by mtime descending
rg_out=$(rg --no-config --files -g "$pattern" "$search_path" 2>&1)
rg_exit=$?
if [ $rg_exit -gt 1 ]; then
echo "STATUS=error"
echo "error: $rg_out"
exit 1
fi
if [ -z "$rg_out" ]; then
echo "STATUS=no_matches MATCH_COUNT=0"
echo "no matches"
exit 0
fi
sorted=$(echo "$rg_out" \
| xargs -d '\n' stat --printf '%Y\t%n\n' 2>&1 \
| sort -rn \
| cut -f2-)
count=$(echo "$sorted" | wc -l | tr -d ' ')
echo "STATUS=ok MATCH_COUNT=$count"
echo "$sorted"

View File

@ -1,97 +0,0 @@
#!/usr/bin/env python3
# ollie:parallel read
# description: Search file contents using ripgrep. Default output: file:lineno| content (one match per line).
# Args: <pattern> [--path=<dir>] [--mode=files_with_matches|content|count] [--glob=<pattern>] [--type=<type>] [-i] [-A <n>] [-B <n>] [-C <n>] [--multiline] [--head=<n>] [--offset=<n>]
import sys
import os
import subprocess
import argparse
parser = argparse.ArgumentParser(prog='file_grep', add_help=False)
parser.add_argument('pattern')
parser.add_argument('--path', default='')
parser.add_argument('--mode', default='content',
choices=['files_with_matches', 'content', 'count'])
parser.add_argument('--glob', default='')
parser.add_argument('--type', default='')
parser.add_argument('-i', action='store_true')
parser.add_argument('-n', action='store_true', default=True)
parser.add_argument('-A', type=int, default=None)
parser.add_argument('-B', type=int, default=None)
parser.add_argument('-C', type=int, default=None)
parser.add_argument('--multiline', action='store_true')
parser.add_argument('--head', type=int, default=0)
parser.add_argument('--offset', type=int, default=0)
args = parser.parse_args()
search_path = args.path or os.getcwd()
cmd = ['rg', '--no-config', '--color=never']
if args.mode == 'files_with_matches':
cmd.append('-l')
elif args.mode == 'count':
cmd.append('--count')
elif args.mode == 'content':
if args.n:
cmd.append('-n')
if args.i:
cmd.append('-i')
if args.multiline:
cmd += ['-U', '--multiline-dotall']
if args.glob:
cmd += ['-g', args.glob]
if args.type:
cmd += ['-t', args.type]
if args.C is not None:
cmd += ['-C', str(args.C)]
if args.A is not None:
cmd += ['-A', str(args.A)]
if args.B is not None:
cmd += ['-B', str(args.B)]
cmd += ['-e', args.pattern, search_path]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 1:
print("STATUS=no_matches MATCH_COUNT=0")
print("no matches")
sys.exit(0)
if result.returncode > 1:
print(f"STATUS=error")
print(f"error: {result.stderr.strip() or f'rg exited with {result.returncode}'}")
sys.exit(1)
lines = result.stdout.splitlines()
total = len(lines)
if args.offset:
lines = lines[args.offset:]
max_matches = args.head if args.head else 50
capped = len(lines) > max_matches
if capped:
lines = lines[:max_matches]
if args.mode == 'content':
formatted = []
for line in lines:
if line == '--':
formatted.append(line)
continue
for sep in (':', '-'):
parts = line.split(sep, 2)
if len(parts) == 3 and parts[1].isdigit():
formatted.append(f"{parts[0]}:{parts[1]:>5}| {parts[2]}")
break
else:
formatted.append(line)
lines = formatted
print(f"STATUS=ok MATCH_COUNT={total}")
print('\n'.join(lines))
if capped:
print(f"\n(showing {max_matches} of {total} matches; use --head=N or --offset=N for more)")

View File

@ -1,111 +0,0 @@
#!/usr/bin/env python3
# ollie:parallel read
# ollie:tier cold
# description: Read a file with accurate line numbers. Prefer segments (start/end) over full-file reads.
# Args: <file_path> [start] [end] [--head=<n>]
import sys
import os
from itertools import islice
MAX_LINES = 500 # default window for whole-file reads (no start/end)
MAX_SEGMENT = 2000 # max window when start/end are given
MAX_LINE_LEN = 2000
# Parse --head=N from args
head = None
args = []
for a in sys.argv[1:]:
if a.startswith('--head='):
head = int(a.split('=', 1)[1])
else:
args.append(a)
if not args:
print("usage: file_read <file_path> [start] [end] [--head=N]")
sys.exit(1)
file_path = args[0]
start = int(args[1]) if len(args) > 1 else 1
end = int(args[2]) if len(args) > 2 else None
if not os.path.isabs(file_path):
print(f"STATUS=error")
print(f"error: path must be absolute, got: {file_path}")
sys.exit(1)
if not os.path.exists(file_path):
print("STATUS=not_found")
print("(new file)")
sys.exit(0)
if not os.path.isfile(file_path):
print(f"STATUS=error")
print(f"error: not a regular file: {file_path}")
sys.exit(1)
with open(file_path, 'rb') as f:
raw_head = f.read(8192)
if b'\x00' in raw_head:
# Binary file -- try to detect image type
IMAGE_SIGS = [
b'\x89PNG\r\n\x1a\n', b'\xff\xd8\xff',
b'GIF87a', b'GIF89a', b'RIFF',
]
is_image = any(raw_head.startswith(s) for s in IMAGE_SIGS)
if is_image:
print("STATUS=ok BINARY=true IMAGE=true")
print("(image file — use image_read for visual content)")
else:
print("STATUS=ok BINARY=true")
print("(binary file)")
sys.exit(0)
# Count total lines
try:
with open(file_path, 'r', encoding='utf-8') as f:
total_lines = sum(1 for _ in f)
except UnicodeDecodeError:
with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
total_lines = sum(1 for _ in f)
start = max(1, start)
if head:
end = start + head - 1
elif end is None:
end = start + MAX_LINES - 1
else:
end = min(end, start + MAX_SEGMENT - 1)
try:
with open(file_path, 'r', encoding='utf-8') as f:
for _ in islice(f, start - 1):
pass
segment_lines = list(islice(f, end - start + 1))
except UnicodeDecodeError:
with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
for _ in islice(f, start - 1):
pass
segment_lines = list(islice(f, end - start + 1))
if not segment_lines:
if start == 1:
print(f"STATUS=ok LINES=0 TOTAL_LINES={total_lines} TRUNCATED=false EMPTY=true")
print("(empty file)")
else:
print(f"STATUS=ok LINES=0 TOTAL_LINES={total_lines} TRUNCATED=false EMPTY=true")
print("(no lines in range)")
sys.exit(0)
out = []
for i, line in enumerate(segment_lines):
lineno = start + i
text = line.rstrip('\n')
if len(text) > MAX_LINE_LEN:
text = text[:MAX_LINE_LEN] + '...'
out.append(f"{lineno:>6}\t{text}")
last_line_shown = start + len(out) - 1
truncated = last_line_shown < total_lines
print(f"STATUS=ok LINES={len(out)} TOTAL_LINES={total_lines} TRUNCATED={str(truncated).lower()}")
print('\n'.join(out))

View File

@ -1,58 +0,0 @@
#!/usr/bin/env python3
# description: Write or overwrite a file
# Args: <file_path> <content>
import sys
import os
import difflib
def format_diff(path, old, new):
old_lines = old.splitlines(keepends=True)
new_lines = new.splitlines(keepends=True)
diff = ''.join(difflib.unified_diff(old_lines, new_lines, fromfile=path, tofile=path))
if not diff:
return None
return f"```diff\n{diff}```"
if len(sys.argv) < 3:
print("usage: file_write <file_path> <content>")
sys.exit(1)
file_path = sys.argv[1]
content = sys.argv[2]
if not os.path.isabs(file_path):
print(f"STATUS=error STATE_CHANGED=false RETRYABLE=false")
print(f"error: path must be absolute, got: {file_path}")
sys.exit(1)
parent = os.path.dirname(file_path)
if parent and not os.path.isdir(parent):
print(f"STATUS=error STATE_CHANGED=false RETRYABLE=false")
print(f"error: parent directory does not exist: {parent}")
sys.exit(1)
old_content = ""
if os.path.exists(file_path):
try:
with open(file_path, 'r', encoding='utf-8') as f:
old_content = f.read()
except UnicodeDecodeError:
with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
old_content = f.read()
try:
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
except OSError as e:
print(f"STATUS=error STATE_CHANGED=false RETRYABLE=false")
print(f"error: {e}")
sys.exit(1)
diff = format_diff(file_path, old_content, content)
if diff is None:
print("STATUS=no_change STATE_CHANGED=false RETRYABLE=false")
print("(no changes)")
else:
print("STATUS=done STATE_CHANGED=true")
print(diff)

View File

@ -1,464 +0,0 @@
#!/usr/bin/env python3
# description: GUI automation via AT-SPI2 accessibility tree. Args: <inspect|find|click|type|read|tree|action> [app] [query...]
"""Interact with any GUI application through the accessibility (AT-SPI2) interface.
This uses the D-Bus-based accessibility tree to inspect, navigate, and interact
with UI elements in any application that supports AT-SPI (virtually all GTK, Qt,
and Electron apps).
Commands:
tree <app> - dump full widget tree
inspect <app> [depth] - brief overview of top-level widgets
find <app> <role> [name] - find elements by role and optional name
read <app> <path> - read text/value of element
click <app> <path> - click/activate element
type <app> <path> <text> - type text into element
action <app> <path> <action_name> - invoke named action on element
apps - list accessible applications
Paths are slash-separated indices (e.g., "0/2/1" = first child, third child, second child)
or you can use "role:name" patterns (e.g., "push button:OK").
Requires: python3 with gi (PyGObject) and Atspi typelib, OR the 'accerciser' package.
"""
import sys
import os
import time
def check_deps():
"""Check for AT-SPI2 dependencies."""
try:
import gi
gi.require_version('Atspi', '2.0')
from gi.repository import Atspi
return True
except (ImportError, ValueError):
return False
def get_atspi():
import gi
gi.require_version('Atspi', '2.0')
from gi.repository import Atspi
return Atspi
def list_apps():
"""List all accessible applications."""
Atspi = get_atspi()
desktop = Atspi.get_desktop(0)
count = desktop.get_child_count()
print(f"{'INDEX':<6} {'NAME':<30} {'CHILDREN'}")
for i in range(count):
app = desktop.get_child_at_index(i)
if app:
name = app.get_name() or "(unnamed)"
children = app.get_child_count()
print(f"{i:<6} {name:<30} {children}")
def find_app(name_substr):
"""Find application by name substring."""
Atspi = get_atspi()
desktop = Atspi.get_desktop(0)
count = desktop.get_child_count()
for i in range(count):
app = desktop.get_child_at_index(i)
if app:
app_name = app.get_name() or ""
if name_substr.lower() in app_name.lower():
return app
return None
def role_name(accessible):
"""Get human-readable role name."""
Atspi = get_atspi()
try:
role = accessible.get_role()
return accessible.get_role_name()
except Exception:
return "unknown"
def get_states(accessible):
"""Get state set as list of strings."""
Atspi = get_atspi()
try:
state_set = accessible.get_state_set()
states = []
for s in dir(Atspi.StateType):
if s.startswith('_'):
continue
try:
st = getattr(Atspi.StateType, s)
if state_set.contains(st):
states.append(s.lower())
except Exception:
pass
return states
except Exception:
return []
def resolve_path(root, path_str):
"""Resolve a path string to an accessible element.
Path can be:
- Numeric indices: "0/2/1"
- Role:name pattern: "push button:OK"
- Mixed: "0/frame:Main/push button:OK"
"""
if not path_str:
return root
parts = path_str.split('/')
current = root
for part in parts:
if ':' in part:
# Role:name pattern
target_role, target_name = part.split(':', 1)
found = False
count = current.get_child_count()
for i in range(count):
child = current.get_child_at_index(i)
if child:
r = role_name(child)
n = child.get_name() or ""
if (target_role.lower() in r.lower() and
target_name.lower() in n.lower()):
current = child
found = True
break
if not found:
print(f"ERROR: no child matching '{part}' under {role_name(current)}:{current.get_name()}",
file=sys.stderr)
sys.exit(1)
else:
# Numeric index
try:
idx = int(part)
child = current.get_child_at_index(idx)
if not child:
print(f"ERROR: no child at index {idx}", file=sys.stderr)
sys.exit(1)
current = child
except ValueError:
# Try as name match
found = False
count = current.get_child_count()
for i in range(count):
child = current.get_child_at_index(i)
if child and part.lower() in (child.get_name() or "").lower():
current = child
found = True
break
if not found:
print(f"ERROR: no child matching '{part}'", file=sys.stderr)
sys.exit(1)
return current
def dump_tree(accessible, indent=0, max_depth=4, path=""):
"""Dump accessibility tree."""
if indent > max_depth:
return
name = accessible.get_name() or ""
role = role_name(accessible)
children = accessible.get_child_count()
# Get text content if available
text = ""
try:
ti = accessible.get_text()
if ti:
text_content = ti.get_text(0, min(ti.get_character_count(), 50))
if text_content and text_content != name:
text = f' text="{text_content}"'
except Exception:
pass
# Get value if available
value = ""
try:
vi = accessible.get_value()
if vi:
value = f' value={vi.get_current_value()}'
except Exception:
pass
prefix = " " * indent
path_display = path or "/"
# States of interest
states = get_states(accessible)
state_flags = ""
if "focused" in states:
state_flags += " [focused]"
if "checked" in states:
state_flags += " [checked]"
if "selected" in states:
state_flags += " [selected]"
if not ("visible" in states and "showing" in states):
state_flags += " [hidden]"
print(f"{prefix}[{path_display}] {role}: \"{name}\"{text}{value}{state_flags} ({children} children)")
for i in range(children):
child = accessible.get_child_at_index(i)
if child:
child_path = f"{path}/{i}" if path else str(i)
dump_tree(child, indent + 1, max_depth, child_path)
def find_elements(accessible, target_role, target_name="", results=None, path=""):
"""Find elements matching role and optional name."""
if results is None:
results = []
r = role_name(accessible)
n = accessible.get_name() or ""
role_match = target_role.lower() in r.lower() if target_role else True
name_match = target_name.lower() in n.lower() if target_name else True
if role_match and name_match:
results.append((path or "/", r, n, accessible))
count = accessible.get_child_count()
for i in range(min(count, 200)): # cap to avoid infinite trees
child = accessible.get_child_at_index(i)
if child:
child_path = f"{path}/{i}" if path else str(i)
find_elements(child, target_role, target_name, results, child_path)
return results
def read_element(accessible):
"""Read text/value from element."""
# Try text interface
try:
ti = accessible.get_text()
if ti:
count = ti.get_character_count()
if count > 0:
text = ti.get_text(0, count)
print(text)
return
except Exception:
pass
# Try value interface
try:
vi = accessible.get_value()
if vi:
print(vi.get_current_value())
return
except Exception:
pass
# Fall back to name
name = accessible.get_name()
if name:
print(name)
else:
print("(no readable content)")
def click_element(accessible):
"""Activate/click an element."""
Atspi = get_atspi()
# Try action interface
try:
ai = accessible.get_action()
if ai:
n_actions = ai.get_n_actions()
for i in range(n_actions):
action_name = ai.get_action_name(i)
if action_name in ('click', 'activate', 'press', 'invoke'):
ai.do_action(i)
print(f"performed: {action_name}")
return
# If no preferred action found, do the first one
if n_actions > 0:
action_name = ai.get_action_name(0)
ai.do_action(0)
print(f"performed: {action_name}")
return
except Exception as e:
pass
# Try generating a click event via component interface
try:
ci = accessible.get_component()
if ci:
pos = ci.get_position(Atspi.CoordType.SCREEN)
size = ci.get_size()
# Click center
x = pos.x + size.x // 2
y = pos.y + size.y // 2
Atspi.generate_mouse_event(x, y, 'b1c')
print(f"clicked at: {x},{y}")
return
except Exception:
pass
print("ERROR: no way to click this element", file=sys.stderr)
sys.exit(1)
def type_text(accessible, text):
"""Type text into an element."""
Atspi = get_atspi()
# Try editable text interface
try:
ei = accessible.get_editable_text()
if ei:
# Clear existing and insert
ti = accessible.get_text()
if ti:
count = ti.get_character_count()
if count > 0:
ei.delete_text(0, count)
ei.insert_text(0, text, len(text))
print(f"typed: {len(text)} chars")
return
except Exception:
pass
# Fallback: focus and use keyboard events
try:
ci = accessible.get_component()
if ci:
ci.grab_focus()
time.sleep(0.1)
for ch in text:
Atspi.generate_keyboard_event(0, ch, Atspi.KeySynthType.STRING)
time.sleep(0.01)
print(f"typed via keyboard: {len(text)} chars")
return
except Exception:
pass
print("ERROR: cannot type into this element", file=sys.stderr)
sys.exit(1)
def do_action(accessible, action_name):
"""Invoke a named action on element."""
try:
ai = accessible.get_action()
if ai:
n_actions = ai.get_n_actions()
for i in range(n_actions):
if ai.get_action_name(i) == action_name:
ai.do_action(i)
print(f"performed: {action_name}")
return
# List available actions
available = [ai.get_action_name(i) for i in range(n_actions)]
print(f"ERROR: action '{action_name}' not found. Available: {available}", file=sys.stderr)
sys.exit(1)
except Exception as e:
print(f"ERROR: {e}", file=sys.stderr)
sys.exit(1)
def main():
if not check_deps():
print("ERROR: AT-SPI2 dependencies not available", file=sys.stderr)
print("Install: pip install PyGObject (or system package python3-gobject)", file=sys.stderr)
print("Also need: gir1.2-atspi-2.0 (Debian/Ubuntu) or gobject-introspection with Atspi typelib", file=sys.stderr)
sys.exit(1)
if len(sys.argv) < 2:
print("usage: gui_accessibility <apps|tree|inspect|find|read|click|type|action> [app] [args...]", file=sys.stderr)
sys.exit(1)
cmd = sys.argv[1]
if cmd == "apps":
list_apps()
return
if len(sys.argv) < 3:
print(f"usage: gui_accessibility {cmd} <app_name> [args...]", file=sys.stderr)
sys.exit(1)
app_name = sys.argv[2]
app = find_app(app_name)
if not app:
print(f"ERROR: no accessible application matching '{app_name}'", file=sys.stderr)
print("Tip: run 'gui_accessibility apps' to see available applications", file=sys.stderr)
sys.exit(1)
if cmd == "tree":
depth = int(sys.argv[3]) if len(sys.argv) > 3 else 4
dump_tree(app, max_depth=depth)
elif cmd == "inspect":
depth = int(sys.argv[3]) if len(sys.argv) > 3 else 2
dump_tree(app, max_depth=depth)
elif cmd == "find":
if len(sys.argv) < 4:
print("usage: gui_accessibility find <app> <role> [name]", file=sys.stderr)
sys.exit(1)
target_role = sys.argv[3]
target_name = sys.argv[4] if len(sys.argv) > 4 else ""
results = find_elements(app, target_role, target_name)
if not results:
print(f"no elements found matching role='{target_role}' name='{target_name}'")
else:
print(f"{'PATH':<20} {'ROLE':<20} {'NAME'}")
for path, role, name, _ in results[:50]:
print(f"{path:<20} {role:<20} {name}")
elif cmd == "read":
if len(sys.argv) < 4:
print("usage: gui_accessibility read <app> <path>", file=sys.stderr)
sys.exit(1)
path = sys.argv[3]
element = resolve_path(app, path)
read_element(element)
elif cmd == "click":
if len(sys.argv) < 4:
print("usage: gui_accessibility click <app> <path>", file=sys.stderr)
sys.exit(1)
path = sys.argv[3]
element = resolve_path(app, path)
click_element(element)
elif cmd == "type":
if len(sys.argv) < 5:
print("usage: gui_accessibility type <app> <path> <text>", file=sys.stderr)
sys.exit(1)
path = sys.argv[3]
text = sys.argv[4]
element = resolve_path(app, path)
type_text(element, text)
elif cmd == "action":
if len(sys.argv) < 5:
print("usage: gui_accessibility action <app> <path> <action_name>", file=sys.stderr)
sys.exit(1)
path = sys.argv[3]
action_name = sys.argv[4]
element = resolve_path(app, path)
do_action(element, action_name)
else:
print(f"unknown command: {cmd}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()

View File

@ -1,137 +0,0 @@
#!/usr/bin/env bash
# description: Manage KDE Plasma Activities via D-Bus. Args: <list|current|switch|create|remove> [name_or_id]
set -euo pipefail
SERVICE="org.kde.ActivityManager"
OBJ_PATH="/ActivityManager/Activities"
IFACE="org.kde.ActivityManager.Activities"
cmd="${1:-list}"
case "$cmd" in
list)
# List all activities with their names and states
ids=$(gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.ListActivities" 2>/dev/null | \
sed "s/^(\[//;s/\],)$//;s/'//g;s/, /\n/g")
if [ -z "$ids" ]; then
echo "ERROR: cannot list activities (ActivityManager not running?)" >&2
exit 1
fi
current=$(gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.CurrentActivity" 2>/dev/null | \
sed "s/^('//;s/',)$//")
echo "ID STATE NAME"
while IFS= read -r id; do
[ -z "$id" ] && continue
name=$(gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.ActivityName" "$id" 2>/dev/null | \
sed "s/^('//;s/',)$//")
if [ "$id" = "$current" ]; then
state="active"
else
state="stopped"
fi
echo "$id $state $name"
done <<< "$ids"
;;
current)
id=$(gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.CurrentActivity" 2>/dev/null | \
sed "s/^('//;s/',)$//")
name=$(gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.ActivityName" "$id" 2>/dev/null | \
sed "s/^('//;s/',)$//")
echo "$id $name"
;;
switch)
target="${2:?usage: gui_activities switch <name_or_id>}"
# Try as ID first
result=$(gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.SetCurrentActivity" "$target" 2>/dev/null) && {
echo "switched to: $target"
exit 0
}
# Search by name
ids=$(gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.ListActivities" 2>/dev/null | \
sed "s/^(\[//;s/\],)$//;s/'//g;s/, /\n/g")
while IFS= read -r id; do
[ -z "$id" ] && continue
name=$(gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.ActivityName" "$id" 2>/dev/null | \
sed "s/^('//;s/',)$//")
if echo "$name" | grep -qi "$target"; then
gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.SetCurrentActivity" "$id" >/dev/null
echo "switched to: $name ($id)"
exit 0
fi
done <<< "$ids"
echo "ERROR: no activity matching '$target'" >&2
exit 1
;;
create)
name="${2:?usage: gui_activities create <name>}"
id=$(gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.AddActivity" "$name" 2>/dev/null | \
sed "s/^('//;s/',)$//")
echo "created: $id ($name)"
;;
remove)
target="${2:?usage: gui_activities remove <name_or_id>}"
# Try as ID directly
gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.RemoveActivity" "$target" 2>/dev/null && {
echo "removed: $target"
exit 0
}
# Search by name
ids=$(gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.ListActivities" 2>/dev/null | \
sed "s/^(\[//;s/\],)$//;s/'//g;s/, /\n/g")
while IFS= read -r id; do
[ -z "$id" ] && continue
name=$(gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.ActivityName" "$id" 2>/dev/null | \
sed "s/^('//;s/',)$//")
if echo "$name" | grep -qi "$target"; then
gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.RemoveActivity" "$id" >/dev/null
echo "removed: $name ($id)"
exit 0
fi
done <<< "$ids"
echo "ERROR: no activity matching '$target'" >&2
exit 1
;;
*)
echo "usage: gui_activities <list|current|switch|create|remove> [name_or_id]" >&2
exit 1
;;
esac

View File

@ -1,104 +0,0 @@
#!/usr/bin/env bash
# description: Launch/list/activate applications via D-Bus and KService. Args: <list|launch|activate> [app_name_or_desktop_file]
set -euo pipefail
cmd="${1:-list}"
case "$cmd" in
list)
# List running applications via KWin
if command -v wmctrl &>/dev/null; then
echo "CLASS TITLE"
wmctrl -l -x | while read -r wid desktop class host title; do
echo "$class $title"
done
elif command -v xdotool &>/dev/null; then
echo "WID NAME"
xdotool search --onlyvisible --name '' 2>/dev/null | while read -r wid; do
name=$(xdotool getwindowname "$wid" 2>/dev/null || echo '?')
echo "$wid $name"
done
else
# D-Bus fallback: list org.freedesktop.DBus registered names
echo "ERROR: install wmctrl or xdotool for window listing" >&2
exit 1
fi
;;
launch)
app="${2:?usage: gui_apps launch <app_name_or_desktop_file>}"
# Try gtk-launch first (handles .desktop files)
if command -v gtk-launch &>/dev/null; then
gtk-launch "$app" 2>/dev/null && { echo "launched: $app"; exit 0; }
fi
# Try kioclient
if command -v kioclient5 &>/dev/null || command -v kioclient &>/dev/null; then
kio=$(command -v kioclient5 || command -v kioclient)
"$kio" exec "applications:$app.desktop" 2>/dev/null && { echo "launched: $app"; exit 0; }
fi
# Try gio
if command -v gio &>/dev/null; then
# Find .desktop file
desktop_file=$(find /usr/share/applications ~/.local/share/applications \
-name "*${app}*.desktop" 2>/dev/null | head -1)
if [ -n "$desktop_file" ]; then
gio launch "$desktop_file" && { echo "launched: $app"; exit 0; }
fi
fi
# Try D-Bus activation
gdbus call --session \
--dest org.freedesktop.DBus \
--object-path /org/freedesktop/DBus \
--method org.freedesktop.DBus.StartServiceByName \
"$app" 0 2>/dev/null && { echo "activated: $app"; exit 0; }
# Last resort: just exec it
if command -v "$app" &>/dev/null; then
setsid "$app" &>/dev/null &
echo "launched: $app (pid $!)"
else
echo "ERROR: cannot find or launch '$app'" >&2
exit 1
fi
;;
activate)
app="${2:?usage: gui_apps activate <title_or_class>}"
# Focus existing window
if command -v wmctrl &>/dev/null; then
# Try by class first
wid=$(wmctrl -l -x | grep -i "$app" | head -1 | awk '{print $1}')
if [ -n "$wid" ]; then
wmctrl -i -a "$wid"
echo "activated: $wid"
exit 0
fi
fi
if command -v xdotool &>/dev/null; then
wid=$(xdotool search --name "$app" 2>/dev/null | head -1)
if [ -n "$wid" ]; then
xdotool windowactivate "$wid"
echo "activated: $wid"
exit 0
fi
fi
echo "ERROR: no window matching '$app'" >&2
exit 1
;;
installed)
# List installed .desktop files
find /usr/share/applications ~/.local/share/applications \
-name '*.desktop' 2>/dev/null | \
xargs grep -l '^Name=' | \
while read -r f; do
name=$(grep '^Name=' "$f" | head -1 | cut -d= -f2)
echo "$(basename "$f" .desktop) $name"
done | sort
;;
*)
echo "usage: gui_apps <list|launch|activate|installed> [app]" >&2
exit 1
;;
esac

View File

@ -1,84 +0,0 @@
#!/usr/bin/env bash
# description: Get/set screen brightness via PowerDevil D-Bus. Args: [get|set <percent>|up|down]
# NOTE: PowerDevil uses absolute values (0 to brightnessMax, typically 10000).
# This tool accepts PERCENTAGES (0-100) and converts internally.
set -euo pipefail
SERVICE="org.kde.Solid.PowerManagement"
OBJ_PATH="/org/kde/Solid/PowerManagement/Actions/BrightnessControl"
IFACE="org.kde.Solid.PowerManagement.Actions.BrightnessControl"
cmd="${1:-get}"
# Get max brightness (needed for percentage conversion)
get_max() {
gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.brightnessMax" | \
sed 's/(//;s/,)//'
}
# Get current brightness (absolute)
get_current() {
gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.brightness" | \
sed 's/(//;s/,)//'
}
case "$cmd" in
get)
max=$(get_max)
current=$(get_current)
percent=$(( current * 100 / max ))
echo "${percent}% ($current/$max)"
;;
set)
percent="${2:?usage: gui_brightness set <percent 0-100>}"
if [ "$percent" -lt 0 ] || [ "$percent" -gt 100 ]; then
echo "ERROR: percent must be 0-100" >&2
exit 1
fi
max=$(get_max)
value=$(( percent * max / 100 ))
gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.setBrightness" "$value" >/dev/null
echo "brightness: ${percent}% ($value/$max)"
;;
up)
step="${2:-10}"
max=$(get_max)
current=$(get_current)
current_pct=$(( current * 100 / max ))
new_pct=$(( current_pct + step ))
[ "$new_pct" -gt 100 ] && new_pct=100
value=$(( new_pct * max / 100 ))
gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.setBrightness" "$value" >/dev/null
echo "brightness: ${new_pct}% ($value/$max)"
;;
down)
step="${2:-10}"
max=$(get_max)
current=$(get_current)
current_pct=$(( current * 100 / max ))
new_pct=$(( current_pct - step ))
[ "$new_pct" -lt 0 ] && new_pct=0
value=$(( new_pct * max / 100 ))
gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.setBrightness" "$value" >/dev/null
echo "brightness: ${new_pct}% ($value/$max)"
;;
*)
echo "usage: gui_brightness [get|set <percent>|up [step]|down [step]]" >&2
echo "" >&2
echo " get - show current brightness as percentage" >&2
echo " set <0-100> - set brightness to percentage" >&2
echo " up [step] - increase by step% (default 10)" >&2
echo " down [step] - decrease by step% (default 10)" >&2
exit 1
;;
esac

View File

@ -1,43 +0,0 @@
#!/usr/bin/env bash
# ollie:parallel read
# description: Read/write KDE clipboard via Klipper D-Bus. Args: [get|set|history] [text|index]
set -euo pipefail
SERVICE="org.kde.klipper"
PATH_OBJ="/klipper"
IFACE="org.kde.klipper.klipper"
cmd="${1:-get}"
case "$cmd" in
get)
gdbus call --session --dest "$SERVICE" \
--object-path "$PATH_OBJ" \
--method "$IFACE.getClipboardContents" | \
sed "s/^('//;s/')$//;s/\\\\n/\n/g"
;;
set)
text="${2:?usage: gui_clipboard set <text>}"
gdbus call --session --dest "$SERVICE" \
--object-path "$PATH_OBJ" \
--method "$IFACE.setClipboardContents" "$text" >/dev/null
echo "OK: clipboard set"
;;
history)
index="${2:-0}"
gdbus call --session --dest "$SERVICE" \
--object-path "$PATH_OBJ" \
--method "$IFACE.getClipboardHistoryItem" "$index" | \
sed "s/^('//;s/')$//;s/\\\\n/\n/g"
;;
clear)
gdbus call --session --dest "$SERVICE" \
--object-path "$PATH_OBJ" \
--method "$IFACE.clearClipboardHistory" >/dev/null
echo "OK: clipboard history cleared"
;;
*)
echo "usage: gui_clipboard [get|set|history|clear] [text|index]" >&2
exit 1
;;
esac

View File

@ -1,83 +0,0 @@
#!/usr/bin/env bash
# ollie:parallel read
# description: Search files using Baloo (KDE file indexer). Args: <query> [--type=document|image|video|audio] [--path=dir] [--limit=n]
set -euo pipefail
query="${1:?usage: gui_files <query> [--type=TYPE] [--path=DIR] [--limit=N]}"
shift
# Parse flags
file_type=""
path_filter=""
limit="50"
while [ $# -gt 0 ]; do
case "$1" in
--type=*) file_type="${1#--type=}" ;;
--path=*) path_filter="${1#--path=}" ;;
--limit=*) limit="${1#--limit=}" ;;
*) ;;
esac
shift
done
# Build baloosearch command
cmd=(baloosearch)
# Add type filter
case "$file_type" in
document|doc) cmd+=(-t Document) ;;
image|img) cmd+=(-t Image) ;;
video|vid) cmd+=(-t Video) ;;
audio|snd) cmd+=(-t Audio) ;;
folder|dir) cmd+=(-t Folder) ;;
"")
;;
*)
cmd+=(-t "$file_type")
;;
esac
# Add path filter
if [ -n "$path_filter" ]; then
cmd+=(-d "$path_filter")
fi
# Add limit
cmd+=(-l "$limit")
# Add query
cmd+=("$query")
# Try baloosearch first
if command -v baloosearch &>/dev/null; then
"${cmd[@]}" 2>/dev/null
exit $?
fi
# Fallback: try baloosearch6
if command -v baloosearch6 &>/dev/null; then
cmd[0]="baloosearch6"
"${cmd[@]}" 2>/dev/null
exit $?
fi
# Fallback: D-Bus query to Baloo
if gdbus introspect --session --dest org.kde.baloo --object-path /org/kde/baloo &>/dev/null 2>&1; then
result=$(gdbus call --session \
--dest org.kde.baloo \
--object-path /org/kde/baloo \
--method org.kde.baloo.main.search "$query" "$limit" 2>/dev/null)
echo "$result" | sed "s/^(\[//;s/\],)$//;s/'//g;s/, /\n/g"
exit 0
fi
# Last fallback: locate
if command -v locate &>/dev/null; then
echo "# (using locate fallback - results may be stale)" >&2
locate -i "$query" | head -n "$limit"
exit 0
fi
echo "ERROR: Baloo not available and no fallback found (install baloo-utils or mlocate)" >&2
exit 1

View File

@ -1,116 +0,0 @@
#!/usr/bin/env bash
# description: Synthetic keyboard/mouse input via ydotool (Wayland) or xdotool (X11). Args: <key|type|click|move|scroll> [args...]
set -euo pipefail
cmd="${1:?usage: gui_input <key|type|click|move|scroll> [args...]}"
shift
# Detect input method: prefer ydotool (Wayland-safe), fall back to xdotool
if command -v ydotool &>/dev/null; then
INPUT_TOOL="ydotool"
elif command -v xdotool &>/dev/null; then
INPUT_TOOL="xdotool"
else
echo "ERROR: install ydotool (Wayland) or xdotool (X11)" >&2
exit 1
fi
case "$cmd" in
key)
# Simulate key press. Examples: "Return", "ctrl+c", "alt+F4", "super"
keys="${1:?usage: gui_input key <key_combo>}"
if [ "$INPUT_TOOL" = "ydotool" ]; then
# ydotool uses key codes; map common names
ydotool key "$keys"
else
xdotool key "$keys"
fi
echo "key: $keys"
;;
type)
# Type a string
text="${1:?usage: gui_input type <text>}"
if [ "$INPUT_TOOL" = "ydotool" ]; then
ydotool type -- "$text"
else
xdotool type --clearmodifiers -- "$text"
fi
echo "typed: ${#text} chars"
;;
click)
# Mouse click. Args: [button] [x y]
button="${1:-1}" # 1=left, 2=middle, 3=right
if [ $# -ge 3 ]; then
x="$2"; y="$3"
if [ "$INPUT_TOOL" = "ydotool" ]; then
ydotool mousemove --absolute "$x" "$y"
sleep 0.05
ydotool click "$button"
else
xdotool mousemove "$x" "$y"
sleep 0.05
xdotool click "$button"
fi
echo "click: button=$button at $x,$y"
else
if [ "$INPUT_TOOL" = "ydotool" ]; then
ydotool click "$button"
else
xdotool click "$button"
fi
echo "click: button=$button at cursor"
fi
;;
move)
# Move mouse. Args: <x> <y> [--relative]
x="${1:?usage: gui_input move <x> <y> [--relative]}"
y="${2:?usage: gui_input move <x> <y> [--relative]}"
relative="${3:-}"
if [ "$INPUT_TOOL" = "ydotool" ]; then
if [ "$relative" = "--relative" ]; then
ydotool mousemove -- "$x" "$y"
else
ydotool mousemove --absolute "$x" "$y"
fi
else
if [ "$relative" = "--relative" ]; then
xdotool mousemove_relative "$x" "$y"
else
xdotool mousemove "$x" "$y"
fi
fi
echo "moved: $x,$y${relative:+ (relative)}"
;;
scroll)
# Scroll. Args: <up|down|left|right> [amount]
direction="${1:?usage: gui_input scroll <up|down|left|right> [amount]}"
amount="${2:-3}"
if [ "$INPUT_TOOL" = "ydotool" ]; then
case "$direction" in
up) ydotool mousemove --wheel -- 0 "-$amount" ;;
down) ydotool mousemove --wheel -- 0 "$amount" ;;
left) ydotool mousemove --wheel -- "-$amount" 0 ;;
right) ydotool mousemove --wheel -- "$amount" 0 ;;
esac
else
case "$direction" in
up) xdotool click --repeat "$amount" 4 ;;
down) xdotool click --repeat "$amount" 5 ;;
left) xdotool click --repeat "$amount" 6 ;;
right) xdotool click --repeat "$amount" 7 ;;
esac
fi
echo "scroll: $direction $amount"
;;
*)
echo "usage: gui_input <key|type|click|move|scroll> [args...]" >&2
echo "" >&2
echo "Commands:" >&2
echo " key <combo> - press key combo (e.g. ctrl+c, Return, alt+F4)" >&2
echo " type <text> - type a string" >&2
echo " click [btn] [x y] - mouse click (1=left 2=mid 3=right)" >&2
echo " move <x> <y> [--relative] - move cursor" >&2
echo " scroll <dir> [amount] - scroll (up/down/left/right)" >&2
exit 1
;;
esac

View File

@ -1,284 +0,0 @@
#!/usr/bin/env python3
# description: Control MPRIS players and build local playlists. Args: <list|status|open|playlist|play|pause|stop|next|previous> [target] [options]
"""Reliable MPRIS control, including Elisa's Application.Open requirement."""
import argparse
import hashlib
import json
import pathlib
import random
import re
import subprocess
import sys
import time
BUS = "org.freedesktop.DBus"
OBJ = "/org/mpris/MediaPlayer2"
PLAYER = "org.mpris.MediaPlayer2.Player"
PROPS = "org.freedesktop.DBus.Properties"
def call(dest, path, method, *args, check=True):
p = subprocess.run(
["gdbus", "call", "--session", "--dest", dest,
"--object-path", path, "--method", method, *args],
text=True, capture_output=True,
)
if check and p.returncode:
raise RuntimeError(p.stderr.strip() or p.stdout.strip())
return p.stdout.strip() if p.returncode == 0 else ""
def players():
out = call(BUS, "/org/freedesktop/DBus", BUS + ".ListNames")
return sorted(set(re.findall(r"org\.mpris\.MediaPlayer2[\w.-]+", out)))
def choose(hint=None, activate=False):
names = players()
if hint:
matches = [n for n in names if hint.lower() in n.lower()]
if matches:
return matches[0]
if activate and hint.lower() == "elisa":
call(BUS, "/org/freedesktop/DBus", BUS + ".StartServiceByName",
"org.kde.elisa", "0")
time.sleep(0.5)
matches = [n for n in players() if "elisa" in n.lower()]
if matches:
return matches[0]
raise RuntimeError(f"no MPRIS player matching {hint!r}")
if names:
playing = [n for n in names if prop(n, "PlaybackStatus") == "Playing"]
return (playing or names)[0]
if activate:
call(BUS, "/org/freedesktop/DBus", BUS + ".StartServiceByName",
"org.kde.elisa", "0")
time.sleep(0.5)
names = players()
if names:
return names[0]
raise RuntimeError("no MPRIS media player is available")
def prop(dest, name):
out = call(dest, OBJ, PROPS + ".Get", PLAYER, name, check=False)
m = re.search(r"<'([^']*)'>", out)
return m.group(1) if m else out
def metadata(dest):
return call(dest, OBJ, PROPS + ".Get", PLAYER, "Metadata", check=False)
def metadata_has_uri(dest, uri):
# Elisa reports file URLs with decoded spaces, while pathlib.as_uri()
# correctly sends percent-encoded spaces to Application.Open.
from urllib.parse import unquote
return uri in metadata(dest) or unquote(uri) in metadata(dest)
def uri_for(value):
if re.match(r"^[A-Za-z][A-Za-z0-9+.-]*://", value):
return value
path = pathlib.Path(value).expanduser().resolve()
if not path.is_file():
raise RuntimeError(f"media file not found: {path}")
return path.as_uri()
def clear_playlist(dest):
call(dest, OBJ, PLAYER + ".Stop", check=False)
if "elisa" not in dest.lower():
return
call(dest, "/org/kde/elisa", "org.freedesktop.Application.ActivateAction",
"clear-playlist", "[]", "{}")
# Elisa resets currentTrack to zero when the queue is empty. Metadata may
# remain stale briefly, so the index is the reliable completion signal.
for _ in range(20):
current = call(dest, OBJ, PROPS + ".Get", PLAYER, "currentTrack", check=False)
if re.search(r"<0>", current):
return
time.sleep(0.1)
raise RuntimeError("Elisa did not clear its playlist")
def open_uris(dest, uris, clear=False):
if clear:
clear_playlist(dest)
else:
call(dest, OBJ, PLAYER + ".Stop", check=False)
opened = False
if "elisa" in dest.lower():
# Elisa advertises MPRIS OpenUri but may ignore it. Its actual file-open
# path is org.freedesktop.Application.Open. JSON array syntax is also
# valid GVariant text and safely handles quotes in paths.
opened = bool(call(dest, "/org/kde/elisa",
"org.freedesktop.Application.Open",
json.dumps(uris), "{}", check=False) == "()")
if not opened:
if len(uris) != 1:
raise RuntimeError("this player cannot load a playlist through D-Bus")
call(dest, OBJ, PLAYER + ".OpenUri", uris[0])
def start_verified(dest, uri):
for _ in range(40):
if metadata_has_uri(dest, uri):
break
time.sleep(0.25)
else:
raise RuntimeError("player did not load the requested URI; playback left stopped")
call(dest, OBJ, PLAYER + ".Play")
for attempt in range(20):
if prop(dest, "PlaybackStatus") == "Playing" and metadata_has_uri(dest, uri):
return
# Elisa can expose a newly opened track as Stopped while ignoring the
# first Play. Toggling once initializes its playback state; Play then
# reliably starts the already verified track.
if attempt == 3 and "elisa" in dest.lower():
call(dest, OBJ, PLAYER + ".PlayPause", check=False)
call(dest, OBJ, PLAYER + ".Play", check=False)
time.sleep(0.25)
raise RuntimeError("requested URI loaded but did not start playing")
def open_media(dest, value):
uri = uri_for(value)
open_uris(dest, [uri])
start_verified(dest, uri)
print(f"player={dest}\nstatus=Playing\nuri={uri}")
AUDIO_EXTS = {".flac", ".mp3", ".ogg", ".opus", ".m4a", ".wav"}
MOODS = {
"chill": ({"ambient", "dream", "dreamy", "soft", "slow", "quiet", "calm", "mellow", "warm", "night", "moon", "ocean", "sea", "wave", "waves", "air", "shine", "memory", "sleep", "blue", "acoustic", "instrumental", "home"},
{"metal", "death", "war", "fire", "burn", "rage", "angry", "hate", "kill", "blood", "violent", "blast", "bore", "rancorous"}),
"focus": ({"ambient", "instrumental", "minimal", "piano", "acoustic", "memory", "space", "air"},
{"party", "dance", "rage", "chaos", "scream", "live"}),
"energetic": ({"dance", "fire", "fast", "power", "run", "alive", "electric", "party", "drive", "exalter"},
{"sleep", "slow", "quiet", "ambient", "lament"}),
"melancholy": ({"blue", "rain", "memory", "alone", "lost", "goodbye", "dead", "dream", "night", "winter", "ghost", "undertow"},
{"party", "dance", "happy", "sunshine"}),
}
def words(text):
return set(re.findall(r"[a-z0-9]+", text.lower()))
def library_tracks(root):
root = pathlib.Path(root).expanduser().resolve()
if not root.is_dir():
raise RuntimeError(f"music library not found: {root}")
tracks = []
for path in root.rglob("*"):
if path.is_file() and path.suffix.lower() in AUDIO_EXTS:
parent = path.parent.name
artist, _, album = parent.partition(" - ")
name = path.stem
title = re.sub(r"^.*? - \d+\s*", "", name)
tracks.append({"path": path, "artist": artist or parent,
"album": album, "title": title,
"text": " ".join((artist, album, title)).lower()})
if not tracks:
raise RuntimeError(f"no audio files found under {root}")
return tracks
def rank_playlist(description, tracks, count):
query = words(description)
positive, negative = set(), set()
for mood, (good, bad) in MOODS.items():
if mood in query or (mood == "chill" and query & {"relax", "relaxed", "mellow", "calm"}):
positive |= good
negative |= bad
positive |= query - {"make", "playlist", "music", "songs", "tracks", "some", "me", "a", "an", "the"}
seed = int(hashlib.sha256(description.lower().encode()).hexdigest()[:16], 16)
rng = random.Random(seed)
ranked = []
for track in tracks:
tokens = words(track["text"])
score = 4 * len(tokens & positive) - 5 * len(tokens & negative)
# Album/artist matches influence the whole selection; a tiny seeded
# jitter makes equally ranked results varied but reproducible.
score += 2 * len(words(track["album"]) & positive)
score += len(words(track["artist"]) & positive)
ranked.append((score + rng.random() * 0.2, track))
ranked.sort(key=lambda item: item[0], reverse=True)
selected, artists, albums = [], {}, {}
for _, track in ranked:
artist = track["artist"].lower()
album = (artist, track["album"].lower())
if artists.get(artist, 0) >= 3 or albums.get(album, 0) >= 2:
continue
selected.append(track)
artists[artist] = artists.get(artist, 0) + 1
albums[album] = albums.get(album, 0) + 1
if len(selected) == count:
break
return selected
def build_playlist(dest, description, root, count, output=None):
if count < 1 or count > 100:
raise RuntimeError("playlist count must be between 1 and 100")
selected = rank_playlist(description, library_tracks(root), count)
if not selected:
raise RuntimeError("no tracks matched the playlist description")
if output:
playlist = pathlib.Path(output).expanduser().resolve()
else:
slug = re.sub(r"[^a-z0-9]+", "-", description.lower()).strip("-")[:50] or "playlist"
playlist = pathlib.Path.home() / ".local/share/ollie/playlists" / f"{slug}.m3u8"
playlist.parent.mkdir(parents=True, exist_ok=True)
playlist.write_text("#EXTM3U\n" + "\n".join(str(t["path"]) for t in selected) + "\n",
encoding="utf-8")
uris = [t["path"].as_uri() for t in selected]
open_uris(dest, uris, clear=True)
start_verified(dest, uris[0])
print(f"player={dest}\nstatus=Playing\nplaylist={playlist}\ntracks={len(selected)}")
for index, track in enumerate(selected, 1):
print(f"{index:02d}. {track['artist']} — {track['title']}")
def main():
ap = argparse.ArgumentParser()
ap.add_argument("command", choices=["list", "status", "open", "playlist",
"play", "pause", "stop", "next", "previous"])
ap.add_argument("target", nargs="?", help="path, URI, or playlist description")
ap.add_argument("--player")
ap.add_argument("--library", default=str(pathlib.Path.home() / "music"))
ap.add_argument("--count", type=int, default=15)
ap.add_argument("--output", help="M3U8 output path")
args = ap.parse_args()
if args.command == "list":
for name in players():
print(f"{name}\t{prop(name, 'PlaybackStatus')}")
return
dest = choose(args.player, activate=args.command in {"open", "playlist"})
if args.command == "open":
if not args.target:
ap.error("open requires PATH_OR_URI")
open_media(dest, args.target)
elif args.command == "playlist":
if not args.target:
ap.error("playlist requires a description")
build_playlist(dest, args.target, args.library, args.count, args.output)
elif args.command == "status":
print(f"player={dest}\nstatus={prop(dest, 'PlaybackStatus')}\nmetadata={metadata(dest)}")
else:
call(dest, OBJ, PLAYER + "." + args.command.capitalize())
print(f"player={dest}\nstatus={prop(dest, 'PlaybackStatus')}")
if __name__ == "__main__":
try:
main()
except RuntimeError as e:
print(f"ERROR: {e}", file=sys.stderr)
sys.exit(1)

View File

@ -1,119 +0,0 @@
#!/usr/bin/env python3
# description: Show desktop notifications with optional actions. Args: <summary> [body] [--actions=label1,label2,...] [--timeout=ms] [--icon=name]
"""Send interactive notifications via org.freedesktop.Notifications D-Bus interface.
If --actions are specified, waits for user response and prints chosen action label.
Timeout defaults to 10000ms (10s) for action notifications, 5000ms otherwise.
"""
import sys
import argparse
import subprocess
import json
def parse_args():
p = argparse.ArgumentParser(description="Desktop notification with optional actions")
p.add_argument("summary", help="Notification title")
p.add_argument("body", nargs="?", default="", help="Notification body text")
p.add_argument("--actions", default="", help="Comma-separated action labels")
p.add_argument("--timeout", type=int, default=0, help="Timeout in ms (0=default)")
p.add_argument("--icon", default="ollie", help="Icon name (default: ollie)")
p.add_argument("--urgency", choices=["low", "normal", "critical"], default="normal")
return p.parse_args()
def main():
args = parse_args()
# Build actions array: ["id1", "label1", "id2", "label2", ...]
actions_list = []
if args.actions:
labels = [a.strip() for a in args.actions.split(",") if a.strip()]
for i, label in enumerate(labels):
actions_list.append(str(i))
actions_list.append(label)
timeout = args.timeout
if timeout == 0:
timeout = 10000 if actions_list else 5000
urgency_map = {"low": 0, "normal": 1, "critical": 2}
# Build gdbus call
# Notify(app_name, replaces_id, icon, summary, body, actions, hints, timeout)
actions_str = "[" + ", ".join(f"'{a}'" for a in actions_list) + "]"
hints_str = "{'urgency': <byte %d>}" % urgency_map[args.urgency]
cmd = [
"gdbus", "call", "--session",
"--dest", "org.freedesktop.Notifications",
"--object-path", "/org/freedesktop/Notifications",
"--method", "org.freedesktop.Notifications.Notify",
"ollie", # app_name
"0", # replaces_id (uint32)
args.icon, # icon
args.summary, # summary
args.body, # body
actions_str, # actions
hints_str, # hints
str(timeout), # timeout
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
print(f"ERROR: {result.stderr.strip()}", file=sys.stderr)
sys.exit(1)
# Parse notification ID from "(uint32 N,)"
nid = result.stdout.strip().strip("()").split(",")[0].replace("uint32 ", "")
if not actions_list:
print(f"notification:{nid}")
return
# Wait for ActionInvoked or NotificationClosed signal
labels = [actions_list[i+1] for i in range(0, len(actions_list), 2)]
monitor_cmd = [
"gdbus", "monitor", "--session",
"--dest", "org.freedesktop.Notifications",
"--object-path", "/org/freedesktop/Notifications",
]
import signal
class TimeoutError(Exception):
pass
def alarm_handler(signum, frame):
raise TimeoutError()
signal.signal(signal.SIGALRM, alarm_handler)
signal.alarm((timeout // 1000) + 2) # extra 2s grace
try:
proc = subprocess.Popen(monitor_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
for line in proc.stdout:
if "ActionInvoked" in line and nid in line:
# Parse: /org/freedesktop/Notifications: org.freedesktop.Notifications.ActionInvoked (uint32 N, 'action_id')
parts = line.split("'")
if len(parts) >= 2:
action_id = parts[1]
try:
idx = int(action_id)
print(labels[idx])
except (ValueError, IndexError):
print(action_id)
proc.terminate()
return
if "NotificationClosed" in line and nid in line:
print("dismissed")
proc.terminate()
return
except TimeoutError:
proc.terminate()
print("timeout")
except Exception as e:
print(f"ERROR: {e}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()

View File

@ -1,158 +0,0 @@
#!/usr/bin/env python3
# description: Capture screenshots via KWin ScreenShot2 D-Bus (Wayland-safe). Args: [active|screen|area|window] [--output=path] [--screen=n]
"""KDE screenshot tool using org.kde.KWin.ScreenShot2.
Modes:
active - capture the currently active window (default)
screen - capture entire screen (or specific with --screen=N)
area - capture a region (interactive picker)
window - capture window by title substring match
Output defaults to /tmp/ollie_screenshot_<timestamp>.png
"""
import sys
import os
import time
import argparse
import subprocess
import tempfile
SERVICE = "org.kde.KWin"
OBJ_PATH = "/org/kde/KWin/ScreenShot2"
IFACE = "org.kde.KWin.ScreenShot2"
def parse_args():
p = argparse.ArgumentParser(description="KDE screenshot via KWin D-Bus")
p.add_argument("mode", nargs="?", default="active",
choices=["active", "screen", "area", "window"],
help="Capture mode (default: active)")
p.add_argument("--output", "-o", default="",
help="Output file path (default: /tmp/ollie_screenshot_<ts>.png)")
p.add_argument("--screen", type=int, default=0,
help="Screen index for 'screen' mode")
p.add_argument("--title", default="",
help="Window title substring for 'window' mode")
p.add_argument("--include-cursor", action="store_true", default=False)
p.add_argument("--include-decoration", action="store_true", default=True)
return p.parse_args()
def get_output_path(args):
if args.output:
return args.output
ts = int(time.time())
return f"/tmp/ollie_screenshot_{ts}.png"
def capture_with_spectacle(mode, output, title=""):
"""Fallback: use spectacle CLI if ScreenShot2 is unavailable."""
cmd = ["spectacle", "-b", "-n", "-o", output]
if mode == "active":
cmd.append("-a")
elif mode == "screen":
cmd.append("-f")
elif mode == "area":
cmd.append("-r")
elif mode == "window":
cmd.append("-a") # fall back to active window
result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
if result.returncode != 0:
return None
return output
def find_window_id_by_title(title):
"""Use KWin scripting to find window UUID by title."""
script = f"""
const clients = workspace.windowList();
for (let i = 0; i < clients.length; i++) {{
if (clients[i].caption.toLowerCase().includes('{title.lower()}')) {{
print(clients[i].internalId);
break;
}}
}}
"""
# Load script via D-Bus
load_cmd = [
"gdbus", "call", "--session",
"--dest", "org.kde.KWin",
"--object-path", "/Scripting",
"--method", "org.kde.kwin.Scripting.loadScript", "", script
]
# This is complex; for now fall back to xdotool/wmctrl
try:
result = subprocess.run(
["xdotool", "search", "--name", title],
capture_output=True, text=True, timeout=5
)
if result.stdout.strip():
return result.stdout.strip().split("\n")[0]
except (FileNotFoundError, subprocess.TimeoutExpired):
pass
return None
def capture_via_dbus(mode, output, args):
"""Use KWin ScreenShot2 portal-style capture with pipe fd."""
hints = {
"include-cursor": args.include_cursor,
"include-decoration": args.include_decoration,
}
# ScreenShot2 methods use file descriptors which gdbus can't handle directly.
# Use the portal interface instead, or fall back to spectacle.
# Try the portal first:
portal_cmd = [
"gdbus", "call", "--session",
"--dest", "org.freedesktop.portal.Desktop",
"--object-path", "/org/freedesktop/portal/desktop",
"--method", "org.freedesktop.portal.Screenshot.Screenshot",
"", # parent_window
"{'interactive': <false>}",
]
# Spectacle is more reliable for non-interactive capture
result = capture_with_spectacle(mode, output, args.title)
if result and os.path.isfile(result):
return result
# Last resort: try import (ImageMagick) for X11
if os.environ.get("DISPLAY"):
if mode == "active":
wid = subprocess.run(
["xdotool", "getactivewindow"],
capture_output=True, text=True
).stdout.strip()
if wid:
r = subprocess.run(["import", "-window", wid, output],
capture_output=True, timeout=10)
if r.returncode == 0:
return output
elif mode == "screen":
r = subprocess.run(["import", "-window", "root", output],
capture_output=True, timeout=10)
if r.returncode == 0:
return output
return None
def main():
args = parse_args()
output = get_output_path(args)
result = capture_via_dbus(args.mode, output, args)
if result and os.path.isfile(result):
size = os.path.getsize(result)
print(f"{result} ({size} bytes)")
else:
print("ERROR: screenshot capture failed", file=sys.stderr)
print("Ensure spectacle or imagemagick is installed", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()

View File

@ -1,95 +0,0 @@
#!/usr/bin/env bash
# description: Manage KDE global shortcuts via KGlobalAccel D-Bus. Args: <list|trigger|get|set> [component] [shortcut] [keys]
set -euo pipefail
SERVICE="org.kde.kglobalaccel"
OBJ_PATH="/kglobalaccel"
IFACE="org.kde.KGlobalAccel"
# Also component interface
COMP_OBJ="/component"
COMP_IFACE="org.kde.kglobalaccel.Component"
cmd="${1:-list}"
case "$cmd" in
list)
# List all registered shortcut components
components=$(gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.allComponents" 2>/dev/null | \
sed "s/^(\[//;s/\],)$//;s/'//g;s/, /\n/g")
if [ -z "$components" ]; then
echo "ERROR: KGlobalAccel not responding" >&2
exit 1
fi
if [ -n "${2:-}" ]; then
# List shortcuts for specific component
comp="$2"
echo "# Shortcuts for: $comp"
gdbus call --session --dest "$SERVICE" \
--object-path "/component/$comp" \
--method "$COMP_IFACE.shortcutNames" 2>/dev/null | \
sed "s/^(\[//;s/\],)$//;s/'//g;s/, /\n/g"
else
echo "$components"
fi
;;
get)
# Get key binding for a shortcut
comp="${2:?usage: gui_shortcuts get <component> <shortcut_name>}"
name="${3:?usage: gui_shortcuts get <component> <shortcut_name>}"
gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.shortcut" \
"['$comp', '$name', '', '']" 2>/dev/null
;;
trigger)
# Trigger (invoke) a global shortcut by component/name
comp="${2:?usage: gui_shortcuts trigger <component> <shortcut_name>}"
name="${3:?usage: gui_shortcuts trigger <component> <shortcut_name>}"
# First get the shortcut keys, then simulate them
# Or use the invokeShortcut method if available
gdbus call --session --dest "$SERVICE" \
--object-path "/component/$comp" \
--method "$COMP_IFACE.invokeShortcut" "$name" 2>/dev/null && {
echo "triggered: $comp/$name"
exit 0
}
echo "ERROR: could not trigger $comp/$name" >&2
exit 1
;;
set)
# Set key binding for a shortcut
comp="${2:?usage: gui_shortcuts set <component> <shortcut_name> <keys>}"
name="${3:?usage: gui_shortcuts set <component> <shortcut_name> <keys>}"
keys="${4:?usage: gui_shortcuts set <component> <shortcut_name> <keys>}"
gdbus call --session --dest "$SERVICE" \
--object-path "$OBJ_PATH" \
--method "$IFACE.setShortcut" \
"['$comp', '$name', '', '']" \
"[['$keys']]" 0 2>/dev/null && {
echo "set: $comp/$name = $keys"
exit 0
}
echo "ERROR: could not set shortcut" >&2
exit 1
;;
*)
echo "usage: gui_shortcuts <list|get|trigger|set> [component] [shortcut] [keys]" >&2
echo "" >&2
echo "Examples:" >&2
echo " gui_shortcuts list - list all components" >&2
echo " gui_shortcuts list kwin - list kwin shortcuts" >&2
echo " gui_shortcuts trigger kwin 'Walk Through Windows' - trigger Alt+Tab" >&2
echo " gui_shortcuts get kwin 'Walk Through Windows'" >&2
exit 1
;;
esac

View File

@ -1,414 +0,0 @@
#!/usr/bin/env python3
# description: KWin window management via D-Bus scripting. Args: <list|focus|close|minimize|maximize|move|resize|desktops> [options]
"""Manage KDE windows via KWin D-Bus scripting interface.
Commands:
list - list all windows (title, class, geometry, desktop, state)
focus <title_substr> - raise and focus window matching title
close <title_substr> - close window matching title
minimize <title_substr> - minimize window
maximize <title_substr> - toggle maximize on window
move <title_substr> <x> <y> - move window to position
resize <title_substr> <w> <h> - resize window
desktops - list virtual desktops
desktop <number> - switch to virtual desktop
tile <left|right|title_substr> - tile window to half-screen
"""
import sys
import subprocess
import json
import tempfile
import os
import time
def kwin_script(js_code):
"""Execute a KWin script and return output."""
# Write script to temp file
with tempfile.NamedTemporaryFile(mode='w', suffix='.js', delete=False) as f:
f.write(js_code)
script_path = f.name
try:
# Load the script
load_result = subprocess.run([
"gdbus", "call", "--session",
"--dest", "org.kde.KWin",
"--object-path", "/Scripting",
"--method", "org.kde.kwin.Scripting.loadScript",
script_path, "ollie_gui_windows"
], capture_output=True, text=True, timeout=5)
if load_result.returncode != 0:
print(f"ERROR: loadScript failed: {load_result.stderr}", file=sys.stderr)
return None
# Parse script ID from "(int32 N,)"
script_id = load_result.stdout.strip().strip("()").split(",")[0].replace("int32 ", "")
# Run the script
run_result = subprocess.run([
"gdbus", "call", "--session",
"--dest", "org.kde.KWin",
"--object-path", f"/Scripting/Script{script_id}",
"--method", "org.kde.kwin.Script.run"
], capture_output=True, text=True, timeout=5)
# Give script time to execute (async)
time.sleep(0.3)
# Read output from KWin's journal/stdout - scripts use print()
# Unfortunately KWin script output goes to systemd journal, not easily capturable.
# For listing we'll use wmctrl/qdbus instead.
# Unload
subprocess.run([
"gdbus", "call", "--session",
"--dest", "org.kde.KWin",
"--object-path", "/Scripting",
"--method", "org.kde.kwin.Scripting.unloadScript",
"ollie_gui_windows"
], capture_output=True, text=True, timeout=5)
return True
finally:
os.unlink(script_path)
def list_windows_wmctrl():
"""List windows using wmctrl (works on X11 and XWayland)."""
try:
result = subprocess.run(
["wmctrl", "-l", "-G", "-p"],
capture_output=True, text=True, timeout=5
)
if result.returncode == 0:
return result.stdout
except FileNotFoundError:
pass
return None
def list_windows_qdbus():
"""List windows using KWin's D-Bus interface for Wayland."""
# Use KWin scripting to dump window list to a known file
out_file = "/tmp/ollie_winlist.txt"
js = f"""
var clients = workspace.windowList();
var lines = [];
for (var i = 0; i < clients.length; i++) {{
var c = clients[i];
if (c.skipTaskbar || c.specialWindow) continue;
var geo = c.frameGeometry;
var state = [];
if (c.minimized) state.push('minimized');
if (c.maximized) state.push('maximized');
if (c.fullScreen) state.push('fullscreen');
if (c.active) state.push('active');
lines.push(
c.caption + '\\t' +
c.resourceClass + '\\t' +
geo.x + ',' + geo.y + ',' + geo.width + ',' + geo.height + '\\t' +
'desktop:' + c.desktop + '\\t' +
state.join(',')
);
}}
callDBus('org.ollie.SessionManager', '/org/ollie/SessionManager',
'org.ollie.SessionManager', 'SetConfig',
'winlist', lines.join('\\n'));
"""
# Alternative: write to file via script. But KWin scripts are sandboxed.
# Use qdbus to get window list if available
try:
result = subprocess.run(
["qdbus6", "org.kde.KWin", "/KWin", "org.kde.KWin.queryWindowInfo"],
capture_output=True, text=True, timeout=5
)
if result.returncode == 0:
return result.stdout
except FileNotFoundError:
try:
result = subprocess.run(
["qdbus", "org.kde.KWin", "/KWin", "org.kde.KWin.queryWindowInfo"],
capture_output=True, text=True, timeout=5
)
if result.returncode == 0:
return result.stdout
except FileNotFoundError:
pass
return None
def list_windows():
"""List windows using best available method."""
# Try wmctrl first (most parseable)
result = list_windows_wmctrl()
if result:
print("ID\tDESKTOP\tPID\tX\tY\tW\tH\tTITLE")
for line in result.strip().split("\n"):
parts = line.split(None, 7)
if len(parts) >= 8:
wid, desktop, pid, x, y, w, h = parts[0], parts[1], parts[2], parts[3], parts[4], parts[5], parts[6]
title = parts[7] if len(parts) > 7 else ""
# Strip hostname prefix from title
if " " in title:
title = title.split(" ", 1)[1] if title.split(" ", 1)[0].isalpha() else title
print(f"{wid}\t{desktop}\t{pid}\t{x}\t{y}\t{w}\t{h}\t{title}")
return
# Try xdotool
try:
result = subprocess.run(
["xdotool", "search", "--onlyvisible", "--name", ""],
capture_output=True, text=True, timeout=5
)
if result.returncode == 0 and result.stdout.strip():
print("ID\tTITLE")
for wid in result.stdout.strip().split("\n"):
name_result = subprocess.run(
["xdotool", "getwindowname", wid],
capture_output=True, text=True, timeout=2
)
name = name_result.stdout.strip() if name_result.returncode == 0 else "?"
print(f"{wid}\t{name}")
return
except FileNotFoundError:
pass
print("ERROR: no window listing tool available (install wmctrl or xdotool)", file=sys.stderr)
sys.exit(1)
def find_window(title_substr):
"""Find window ID by title substring."""
try:
result = subprocess.run(
["wmctrl", "-l"],
capture_output=True, text=True, timeout=5
)
if result.returncode == 0:
for line in result.stdout.strip().split("\n"):
if title_substr.lower() in line.lower():
return line.split()[0]
except FileNotFoundError:
pass
try:
result = subprocess.run(
["xdotool", "search", "--name", title_substr],
capture_output=True, text=True, timeout=5
)
if result.returncode == 0 and result.stdout.strip():
return result.stdout.strip().split("\n")[0]
except FileNotFoundError:
pass
return None
def focus_window(title_substr):
wid = find_window(title_substr)
if not wid:
print(f"ERROR: no window matching '{title_substr}'", file=sys.stderr)
sys.exit(1)
try:
subprocess.run(["wmctrl", "-i", "-a", wid], check=True, timeout=5)
except (FileNotFoundError, subprocess.CalledProcessError):
try:
subprocess.run(["xdotool", "windowactivate", wid], check=True, timeout=5)
except (FileNotFoundError, subprocess.CalledProcessError) as e:
print(f"ERROR: {e}", file=sys.stderr)
sys.exit(1)
print(f"focused: {wid}")
def close_window(title_substr):
wid = find_window(title_substr)
if not wid:
print(f"ERROR: no window matching '{title_substr}'", file=sys.stderr)
sys.exit(1)
try:
subprocess.run(["wmctrl", "-i", "-c", wid], check=True, timeout=5)
except (FileNotFoundError, subprocess.CalledProcessError):
try:
subprocess.run(["xdotool", "windowclose", wid], check=True, timeout=5)
except (FileNotFoundError, subprocess.CalledProcessError) as e:
print(f"ERROR: {e}", file=sys.stderr)
sys.exit(1)
print(f"closed: {wid}")
def minimize_window(title_substr):
wid = find_window(title_substr)
if not wid:
print(f"ERROR: no window matching '{title_substr}'", file=sys.stderr)
sys.exit(1)
try:
subprocess.run(["xdotool", "windowminimize", wid], check=True, timeout=5)
except (FileNotFoundError, subprocess.CalledProcessError):
# KWin script fallback
kwin_script(f"""
var clients = workspace.windowList();
for (var i = 0; i < clients.length; i++) {{
if (clients[i].caption.toLowerCase().includes('{title_substr.lower()}')) {{
clients[i].minimized = true;
break;
}}
}}
""")
print(f"minimized: {wid}")
def maximize_window(title_substr):
# KWin script is the clean way
kwin_script(f"""
var clients = workspace.windowList();
for (var i = 0; i < clients.length; i++) {{
if (clients[i].caption.toLowerCase().includes('{title_substr.lower()}')) {{
clients[i].setMaximize(!clients[i].maximized, !clients[i].maximized);
break;
}}
}}
""")
print(f"toggled maximize: {title_substr}")
def move_window(title_substr, x, y):
wid = find_window(title_substr)
if not wid:
print(f"ERROR: no window matching '{title_substr}'", file=sys.stderr)
sys.exit(1)
try:
subprocess.run(["wmctrl", "-i", "-r", wid, "-e", f"0,{x},{y},-1,-1"],
check=True, timeout=5)
except (FileNotFoundError, subprocess.CalledProcessError):
try:
subprocess.run(["xdotool", "windowmove", wid, str(x), str(y)],
check=True, timeout=5)
except (FileNotFoundError, subprocess.CalledProcessError) as e:
print(f"ERROR: {e}", file=sys.stderr)
sys.exit(1)
print(f"moved {wid} to {x},{y}")
def resize_window(title_substr, w, h):
wid = find_window(title_substr)
if not wid:
print(f"ERROR: no window matching '{title_substr}'", file=sys.stderr)
sys.exit(1)
try:
subprocess.run(["wmctrl", "-i", "-r", wid, "-e", f"0,-1,-1,{w},{h}"],
check=True, timeout=5)
except (FileNotFoundError, subprocess.CalledProcessError):
try:
subprocess.run(["xdotool", "windowsize", wid, str(w), str(h)],
check=True, timeout=5)
except (FileNotFoundError, subprocess.CalledProcessError) as e:
print(f"ERROR: {e}", file=sys.stderr)
sys.exit(1)
print(f"resized {wid} to {w}x{h}")
def list_desktops():
try:
result = subprocess.run(
["wmctrl", "-d"],
capture_output=True, text=True, timeout=5
)
if result.returncode == 0:
print(result.stdout.rstrip())
return
except FileNotFoundError:
pass
# Fallback via KWin D-Bus
try:
result = subprocess.run([
"gdbus", "call", "--session",
"--dest", "org.kde.KWin",
"--object-path", "/VirtualDesktopManager",
"--method", "org.freedesktop.DBus.Properties.Get",
"org.kde.KWin.VirtualDesktopManager", "desktops"
], capture_output=True, text=True, timeout=5)
if result.returncode == 0:
print(result.stdout.rstrip())
return
except FileNotFoundError:
pass
print("ERROR: cannot list desktops", file=sys.stderr)
sys.exit(1)
def switch_desktop(num):
try:
subprocess.run(["wmctrl", "-s", str(int(num) - 1)], check=True, timeout=5)
print(f"switched to desktop {num}")
except (FileNotFoundError, subprocess.CalledProcessError):
try:
subprocess.run([
"gdbus", "call", "--session",
"--dest", "org.kde.KWin",
"--object-path", "/VirtualDesktopManager",
"--method", "org.kde.KWin.VirtualDesktopManager.setCurrent",
str(num)
], check=True, capture_output=True, timeout=5)
print(f"switched to desktop {num}")
except Exception as e:
print(f"ERROR: {e}", file=sys.stderr)
sys.exit(1)
def main():
if len(sys.argv) < 2:
print("usage: gui_windows <list|focus|close|minimize|maximize|move|resize|desktops|desktop> [args...]", file=sys.stderr)
sys.exit(1)
cmd = sys.argv[1]
if cmd == "list":
list_windows()
elif cmd == "focus":
if len(sys.argv) < 3:
print("usage: gui_windows focus <title_substr>", file=sys.stderr)
sys.exit(1)
focus_window(sys.argv[2])
elif cmd == "close":
if len(sys.argv) < 3:
print("usage: gui_windows close <title_substr>", file=sys.stderr)
sys.exit(1)
close_window(sys.argv[2])
elif cmd == "minimize":
if len(sys.argv) < 3:
print("usage: gui_windows minimize <title_substr>", file=sys.stderr)
sys.exit(1)
minimize_window(sys.argv[2])
elif cmd == "maximize":
if len(sys.argv) < 3:
print("usage: gui_windows maximize <title_substr>", file=sys.stderr)
sys.exit(1)
maximize_window(sys.argv[2])
elif cmd == "move":
if len(sys.argv) < 5:
print("usage: gui_windows move <title_substr> <x> <y>", file=sys.stderr)
sys.exit(1)
move_window(sys.argv[2], sys.argv[3], sys.argv[4])
elif cmd == "resize":
if len(sys.argv) < 5:
print("usage: gui_windows resize <title_substr> <w> <h>", file=sys.stderr)
sys.exit(1)
resize_window(sys.argv[2], sys.argv[3], sys.argv[4])
elif cmd == "desktops":
list_desktops()
elif cmd == "desktop":
if len(sys.argv) < 3:
print("usage: gui_windows desktop <number>", file=sys.stderr)
sys.exit(1)
switch_desktop(sys.argv[2])
else:
print(f"unknown command: {cmd}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()

View File

@ -1,62 +0,0 @@
#!/usr/bin/env python3
# ollie:parallel read
# ollie:tier cold
# description: Read an image file, recompressing/resizing to fit context budget. Returns an image content block.
# args: <file_path> [--region=x,y,w,h] [--scale=N] [--quality=N]
import sys
import os
import base64
import json
import mimetypes
file_path = None
for a in sys.argv[1:]:
if not a.startswith('--'):
file_path = a
if not file_path:
print("usage: image_read <file_path>")
sys.exit(1)
if not os.path.isabs(file_path):
print("STATUS=error")
print("error: path must be absolute")
sys.exit(1)
if not os.path.exists(file_path):
print("STATUS=not_found")
sys.exit(0)
mime, _ = mimetypes.guess_type(file_path)
if not mime:
mime = 'application/octet-stream'
with open(file_path, 'rb') as f:
data = f.read()
MAX_SIZE = 128 * 1024
if mime == 'image/png' and len(data) > MAX_SIZE:
from PIL import Image
import io
img = Image.open(io.BytesIO(data))
# progressively scale down until under budget
for scale in [1.0, 0.75, 0.5, 0.35, 0.25]:
buf = io.BytesIO()
if scale < 1.0:
new_size = (int(img.width * scale), int(img.height * scale))
resized = img.resize(new_size, Image.LANCZOS)
else:
resized = img
resized.save(buf, format='PNG', optimize=True)
data = buf.getvalue()
if len(data) <= MAX_SIZE:
break
b64 = base64.b64encode(data).decode('ascii')
print(json.dumps({"isError": False, "content": [
{"type": "image", "media_type": mime, "data": b64},
{"type": "text", "text": f"[image: {os.path.basename(file_path)} {len(data)} bytes {mime}]"}
]}))

View File

@ -1,35 +0,0 @@
#!/usr/bin/env python3
# ollie:parallel read
# description: Get completions at cursor position from the language server.
# Args: <file> <line> <col>
import os
import sys
sys.path.insert(0, os.path.join(os.environ['OLLIE'], 's', os.environ['OLLIE_SESSION_ID'], 't'))
from _lib.lsp.client import request
if len(sys.argv) < 4:
print("usage: lsp_completion <file> <line> <col>")
sys.exit(1)
filepath = os.path.abspath(sys.argv[1])
line = int(sys.argv[2]) - 1
col = int(sys.argv[3]) - 1
result = request("completion", file=filepath, line=line, col=col)
if not result:
sys.exit(0)
items = result.get("items", result) if isinstance(result, dict) else result
if not isinstance(items, list):
sys.exit(0)
for item in items[:20]:
label = item.get("label", "")
detail = item.get("detail", "")
if detail:
print(f"{label}\t{detail}")
else:
print(label)

View File

@ -1,34 +0,0 @@
#!/usr/bin/env python3
# ollie:parallel read
# description: Go to definition of symbol at position. Returns file:line:col.
# Args: <file> <line> <col>
import os
import sys
sys.path.insert(0, os.path.join(os.environ['OLLIE'], 's', os.environ['OLLIE_SESSION_ID'], 't'))
from _lib.lsp.client import request
if len(sys.argv) < 4:
print("usage: lsp_definition <file> <line> <col>")
sys.exit(1)
filepath = os.path.abspath(sys.argv[1])
line = int(sys.argv[2]) - 1 # convert 1-indexed to 0-indexed
col = int(sys.argv[3]) - 1
result = request("definition", file=filepath, line=line, col=col)
if not result:
print("(no definition found)")
sys.exit(0)
# Normalize: result can be a Location or list of Locations
locations = result if isinstance(result, list) else [result]
for loc in locations:
uri = loc.get("uri", loc.get("targetUri", ""))
path = uri[7:] if uri.startswith("file://") else uri
pos = loc.get("range", loc.get("targetRange", {})).get("start", {})
ln = pos.get("line", 0) + 1
ch = pos.get("character", 0) + 1
print(f"{path}:{ln}:{ch}")

View File

@ -1,35 +0,0 @@
#!/usr/bin/env python3
# ollie:parallel read
# description: Get diagnostics (errors, warnings) for a file.
# Args: <file>
import os
import sys
sys.path.insert(0, os.path.join(os.environ['OLLIE'], 's', os.environ['OLLIE_SESSION_ID'], 't'))
from _lib.lsp.client import request
if len(sys.argv) < 2:
print("usage: lsp_diagnostics <file>")
sys.exit(1)
filepath = os.path.abspath(sys.argv[1])
result = request("diagnostics", file=filepath)
if not result:
print("(no diagnostics)")
sys.exit(0)
SEVERITY = {1: "error", 2: "warning", 3: "info", 4: "hint"}
for diag in result:
sev = SEVERITY.get(diag.get("severity", 3), "info")
ln = diag.get("range", {}).get("start", {}).get("line", 0) + 1
col = diag.get("range", {}).get("start", {}).get("character", 0) + 1
msg = diag.get("message", "")
src = diag.get("source", "")
prefix = f"{src}: " if src else ""
print(f"{ln}:{col} {sev}: {prefix}{msg}")
print(f"\n({len(result)} diagnostics)")

View File

@ -1,37 +0,0 @@
#!/usr/bin/env python3
# ollie:parallel read
# description: Get type/documentation info for symbol at position.
# Args: <file> <line> <col>
import os
import sys
sys.path.insert(0, os.path.join(os.environ['OLLIE'], 's', os.environ['OLLIE_SESSION_ID'], 't'))
from _lib.lsp.client import request
if len(sys.argv) < 4:
print("usage: lsp_hover <file> <line> <col>")
sys.exit(1)
filepath = os.path.abspath(sys.argv[1])
line = int(sys.argv[2]) - 1
col = int(sys.argv[3]) - 1
result = request("hover", file=filepath, line=line, col=col)
if not result:
print("(no hover info)")
sys.exit(0)
contents = result.get("contents", "")
if isinstance(contents, dict):
# MarkedString or MarkupContent
print(contents.get("value", str(contents)))
elif isinstance(contents, list):
for item in contents:
if isinstance(item, dict):
print(item.get("value", str(item)))
else:
print(item)
else:
print(contents)

View File

@ -1,41 +0,0 @@
#!/usr/bin/env python3
# ollie:parallel read
# description: Find all references to symbol at position. Returns file:line:col per reference.
# Args: <file> <line> <col>
import os
import sys
sys.path.insert(0, os.path.join(os.environ['OLLIE'], 's', os.environ['OLLIE_SESSION_ID'], 't'))
from _lib.lsp.client import request
if len(sys.argv) < 4:
print("usage: lsp_references <file> <line> <col>")
sys.exit(1)
filepath = os.path.abspath(sys.argv[1])
line = int(sys.argv[2]) - 1
col = int(sys.argv[3]) - 1
result = request("references", file=filepath, line=line, col=col)
if not result:
print("(no references found)")
sys.exit(0)
total = len(result)
cap = 50
refs = result[:cap] if total > cap else result
for loc in refs:
uri = loc.get("uri", "")
path = uri[7:] if uri.startswith("file://") else uri
pos = loc.get("range", {}).get("start", {})
ln = pos.get("line", 0) + 1
ch = pos.get("character", 0) + 1
print(f"{path}:{ln}:{ch}")
if total > cap:
print(f"\n(showing {cap} of {total} references)")
else:
print(f"\n({total} references)")

View File

@ -1,32 +0,0 @@
#!/usr/bin/env python3
# description: Rename symbol at position across the workspace. Applies edits directly to files.
# Args: <file> <line> <col> <new_name>
import os
import sys
sys.path.insert(0, os.path.join(os.environ['OLLIE'], 's', os.environ['OLLIE_SESSION_ID'], 't'))
from _lib.lsp.client import request
if len(sys.argv) < 5:
print("usage: lsp_rename <file> <line> <col> <new_name>")
sys.exit(1)
filepath = os.path.abspath(sys.argv[1])
line = int(sys.argv[2]) - 1
col = int(sys.argv[3]) - 1
new_name = sys.argv[4]
result = request("rename", file=filepath, line=line, col=col, new_name=new_name)
if not result:
print("(rename returned no edits)")
sys.exit(0)
files_changed = result.get("files_changed", {})
total = result.get("total_edits", 0)
for path, count in files_changed.items():
print(f" {path} ({count} edits)")
print(f"\nRenamed to '{new_name}': {total} edits across {len(files_changed)} files.")

View File

@ -1,56 +0,0 @@
#!/usr/bin/env python3
# ollie:parallel read
# description: List symbols in a file or search workspace symbols.
# Args: <file_or_query> [--workspace]
import os
import sys
sys.path.insert(0, os.path.join(os.environ['OLLIE'], 's', os.environ['OLLIE_SESSION_ID'], 't'))
from _lib.lsp.client import request
if len(sys.argv) < 2:
print("usage: lsp_symbols <file_or_query> [--workspace]")
sys.exit(1)
workspace = "--workspace" in sys.argv
arg = sys.argv[1]
SYMBOL_KINDS = {
1: "File", 2: "Module", 3: "Namespace", 4: "Package", 5: "Class",
6: "Method", 7: "Property", 8: "Field", 9: "Constructor", 10: "Enum",
11: "Interface", 12: "Function", 13: "Variable", 14: "Constant",
15: "String", 16: "Number", 17: "Boolean", 18: "Array", 19: "Object",
20: "Key", 21: "Null", 22: "EnumMember", 23: "Struct", 24: "Event",
25: "Operator", 26: "TypeParameter",
}
if workspace:
result = request("symbols", file=arg, workspace=True, query=arg)
else:
filepath = os.path.abspath(arg)
result = request("symbols", file=filepath, workspace=False)
if not result:
print("(no symbols found)")
sys.exit(0)
def print_symbols(symbols, indent=0):
for sym in symbols:
kind = SYMBOL_KINDS.get(sym.get("kind", 0), "?")
name = sym.get("name", "")
# DocumentSymbol has range; SymbolInformation has location
if "range" in sym:
ln = sym["range"]["start"]["line"] + 1
print(f"{' ' * indent}{kind:12} {name} (line {ln})")
if "children" in sym:
print_symbols(sym["children"], indent + 1)
elif "location" in sym:
uri = sym["location"].get("uri", "")
path = uri[7:] if uri.startswith("file://") else uri
ln = sym["location"]["range"]["start"]["line"] + 1
print(f"{kind:12} {name} {path}:{ln}")
print_symbols(result)

View File

@ -1,45 +0,0 @@
#!/usr/bin/env bash
# ollie:parallel read
# ollie:tier cold
# description: Search memories by keyword, tag, or title
# Args: <query>
[ -z "$*" ] && echo "usage: memory_recall <query>" && exit 1
mem_dir="${OLLIE_MEMORY_PATH:-$HOME/.config/ollie/memory}"
[ ! -d "$mem_dir" ] && echo "error: memory directory does not exist: $mem_dir" && exit 1
query="$*"
max_files=10
max_lines=50
stale_days=14
now=$(date +%s)
# Search filenames and bodies; discard I/O errors on stale inodes
matches=$(grep -ril --include='*.md' "$query" "$mem_dir" 2>/dev/null)
[ -z "$matches" ] && echo "no matches for: $query" && exit 0
total=$(echo "$matches" | wc -l)
count=0
while IFS= read -r file; do
count=$((count + 1))
[ $count -gt $max_files ] && echo "(showing $max_files of $total matched files)" && break
fname=$(basename "$file")
# Extract date from filename instead of stat
ts=${fname%%__*}
if [[ $ts =~ ^([0-9]{4})([0-9]{2})([0-9]{2})T ]]; then
fdate=$(date -d "${BASH_REMATCH[1]}-${BASH_REMATCH[2]}-${BASH_REMATCH[3]}" +%s 2>/dev/null)
[ -n "$fdate" ] && age_days=$(( (now - fdate) / 86400 )) || age_days=0
else
age_days=0
fi
[ $age_days -ge $stale_days ] && status="STALE (${age_days}d)" || status="fresh"
echo "=== $fname [$status] ==="
awk -v max="$max_lines" 'NR<=max{print} NR>max{over++} END{if(over) printf "... (%d lines total, showing first %d)\n", NR, max}' "$file" 2>/dev/null || echo "error: could not read $file"
echo
done <<< "$matches"

View File

@ -1,59 +0,0 @@
#!/usr/bin/env python3
# description: Save a memory as a denote-formatted flat file
# Args: <title> <tags_csv> <body>
import sys
import os
import re
from datetime import datetime
if len(sys.argv) < 4:
print("usage: memory_remember <title> <tags_csv> <body>")
sys.exit(1)
title = sys.argv[1]
tags_csv = sys.argv[2]
body = sys.argv[3]
mem_dir = os.environ.get("OLLIE_MEMORY_PATH", os.path.expanduser("~/.config/ollie/memory"))
if not os.path.isdir(mem_dir):
os.makedirs(mem_dir, exist_ok=True)
now = datetime.now()
ident = now.strftime("%Y%m%dT%H%M%S")
date = now.strftime("%Y-%m-%d %a %H:%M")
tags = [t.strip() for t in tags_csv.split(",") if t.strip()]
def slugify(s):
s = s.lower()
s = re.sub(r"[^a-z0-9]+", "-", s)
return s.strip("-")
slug = slugify(title)
tag_slug = "-".join(slugify(t) for t in tags)
filename = f"{ident}__{slug}--{tag_slug}.md" if tag_slug else f"{ident}__{slug}.md"
path = os.path.join(mem_dir, filename)
tags_yaml = "[" + ", ".join(f'"{t}"' for t in tags) + "]"
content = f"""---
title: {title}
date: {date}
tags: {tags_yaml}
identifier: {ident}
signature:
---
{body}
"""
try:
with open(path, "w") as f:
f.write(content)
except OSError as e:
print(f"error: failed to write {path}: {e}")
sys.exit(1)
print(f"saved: {filename}")

View File

@ -1,7 +0,0 @@
#!/usr/bin/env bash
# ollie:parallel read
# description: Internal reasoning scratchpad; think through complex problems before acting
# Args: <thought>
[ -z "$*" ] && echo "usage: reasoning_think <thought>" && exit 1
exit 0

View File

@ -1,107 +0,0 @@
#!/usr/bin/env bash
# ollie:parallel read
# description: Route a subtask to an appropriate model tier. Returns immediately; the agent reports back to your prompt when done.
# Args: <instruction> [--tier=worker|expert] [--backend=<name>] [id]
set -euo pipefail
SERVICE="org.ollie.SessionManager"
OBJ="/org/ollie/SessionManager"
IFACE="org.ollie.SessionManager"
die() { printf 'route: %s\n' "$*" >&2; exit 1; }
dbus_call() {
dbus-send --session --dest=$SERVICE --type=method_call --print-reply \
$OBJ "$IFACE.$1" "${@:2}" 2>/dev/null
}
dbus_string() {
dbus_call "$@" | awk -F'"' '/string "/{print $2; exit}'
}
[ -z "${OLLIE_SESSION_ID:-}" ] && die "OLLIE_SESSION_ID not set (route must run inside a session)"
# Parse args
tier="worker"
backend_override=""
instruction=""
id=""
for arg in "$@"; do
case "$arg" in
--tier=*) tier="${arg#--tier=}" ;;
--backend=*) backend_override="${arg#--backend=}" ;;
*) [ -z "$instruction" ] && instruction="$arg" || id="$arg" ;;
esac
done
[ -z "$instruction" ] && die "instruction required"
case "$tier" in
worker|expert) ;;
*) die "unknown tier: $tier (use worker or expert)" ;;
esac
# Resolve backend: explicit flag > parent session's config > OLLIE_BACKEND
if [ -n "$backend_override" ]; then
BACKEND="$backend_override"
else
parent_config=$(dbus_string GetConfig string:"$OLLIE_SESSION_ID")
BACKEND=$(printf '%s\n' "$parent_config" | grep -m1 '^backend=' | cut -d= -f2-)
BACKEND="${BACKEND:-${OLLIE_BACKEND:-openrouter}}"
fi
# Resolve model: OLLIE_TIER_{TIER}_{BACKEND} > legacy flat vars > hardcoded defaults
tier_upper=$(echo "$tier" | tr '[:lower:]' '[:upper:]')
backend_lower=$(echo "$BACKEND" | tr '[:upper:]' '[:lower:]')
tier_var="OLLIE_TIER_${tier_upper}_${backend_lower}"
MODEL="${!tier_var:-}"
if [ -z "$MODEL" ]; then
case "$tier" in
worker) MODEL="${OLLIE_CASCADE_MODEL:-mistralai/devstral-small}" ;;
expert) MODEL="${OLLIE_ESCALATE_MODEL:-anthropic/claude-sonnet-4-20250514}" ;;
esac
fi
# Agent config
case "$tier" in
worker) agent="worker" ;;
expert) agent="expert" ;;
esac
id="${id:-route-${tier}-$(date +%s%N)-$$}"
[ -n "$OLLIE_SESSION_ID" ] && id="${OLLIE_SESSION_ID}__${id}"
# Create session via D-Bus
session_id=$(dbus_string CreateSession \
string:"$(pwd)" string:"$BACKEND" string:"$MODEL" string:"$agent")
[ -z "$session_id" ] && die "failed to create session"
# Rename to our chosen ID
dbus_call RenameSession string:"$session_id" string:"$id" > /dev/null 2>&1
prompt="[parent_session_id=${OLLIE_SESSION_ID}]
[session_id=${id}]
Complete the following task. Read code, reason about implementation, and stay within scope.
When finished, report your result back to the parent session using PeerSubmit.
Format:
[from=${id}]
STATUS: done
SUMMARY: <what was accomplished>
On failure:
[from=${id}]
STATUS: error
REASON: <why>
After reporting the result, STOP. One report, then stop.
TASK:
${instruction}"
dbus_call Submit string:"$session_id" string:"$prompt" > /dev/null 2>&1 || die "failed to submit prompt"
printf 'session=%s\nbackend=%s\nmodel=%s\ntier=%s\n' "$id" "$BACKEND" "$MODEL" "$tier"

View File

@ -1,99 +0,0 @@
#!/usr/bin/env python3
# description: Generate a JIT subagent config. No args prints the JSON template. With JSON as first arg or on stdin, creates the config and prints the agent name.
# Args: [json] — JSON config as first arg, or piped via stdin
# Output: template (no input) or agent name
import json
import os
import secrets
import sys
AGENTS_DIR = os.path.expanduser("~/.config/ollie/agents")
BASE_CONFIG = os.path.join(AGENTS_DIR, "default.json")
TEMPLATE = {
"name": "",
"role": "",
"roster": "",
"shared_task": "",
"protocol": "",
"constraints": [],
"prompt": "",
}
def die(msg):
print(f"subagent_generate: {msg}", file=sys.stderr)
sys.exit(1)
def main():
# No input: print template.
if len(sys.argv) < 2 and sys.stdin.isatty():
json.dump(TEMPLATE, sys.stdout, indent=2)
print()
return
# Read config from arg or stdin.
if len(sys.argv) > 1:
text = sys.argv[1]
else:
text = sys.stdin.read()
if not text.strip():
die("empty input")
try:
raw = json.loads(text)
except json.JSONDecodeError as e:
die(f"invalid JSON: {e}")
name = raw.get("name", "") or f"jit-{secrets.token_hex(8)}"
role = raw.get("role", "")
roster = raw.get("roster", "")
shared_task = raw.get("shared_task", "")
protocol = raw.get("protocol", "")
constraints = raw.get("constraints", [])
if not isinstance(constraints, list):
die("constraints must be an array of strings")
legacy = raw.get("prompt", "")
if not role and not legacy:
die("role or prompt required")
# Build identity text.
if role:
parts = ["## Role", role]
if roster:
parts += ["## Roster", roster]
if shared_task:
parts += ["## Shared Task", shared_task]
if protocol:
parts += ["## Coordination Protocol", protocol]
if constraints:
parts += ["## Constraints", "\n".join(f"- {c}" for c in constraints)]
identity = "\n\n".join(parts)
else:
identity = legacy
# Load base config.
try:
with open(BASE_CONFIG) as f:
cfg = json.load(f)
except (OSError, json.JSONDecodeError) as e:
die(f"failed to read base config: {e}")
cfg["prompt"] = ["cat ~/.config/ollie/prompts/SYSTEM_PROMPT.md", f"echo {json.dumps(identity)}"]
# Write JIT config.
out_path = os.path.join(AGENTS_DIR, f"{name}.json")
try:
with open(out_path, "w") as f:
json.dump(cfg, f, indent=2)
except OSError as e:
die(f"failed to write config: {e}")
print(name)
if __name__ == "__main__":
main()

View File

@ -1,74 +0,0 @@
#!/usr/bin/env bash
# description: Spawn a subagent session, submit a prompt, and return immediately. The subagent runs independently; use PeerSubmit for coordination.
# Args: [-name NAME] [-agent A] [-backend B] [-model M] [-cwd DIR] <prompt...>
# Output: structured KV (session, backend, model)
set -euo pipefail
SERVICE="org.ollie.SessionManager"
OBJ="/org/ollie/SessionManager"
IFACE="org.ollie.SessionManager"
die() { printf 'subagent_spawn: %s\n' "$*" >&2; exit 1; }
dbus_call() {
dbus-send --session --dest=$SERVICE --type=method_call --print-reply \
$OBJ "$IFACE.$1" "${@:2}" 2>/dev/null
}
dbus_string() {
dbus_call "$@" | awk -F'"' '/string "/{print $2; exit}'
}
name=; agent=; backend=; model=; cwd=
while [ $# -gt 0 ]; do
case $1 in
-name) name=$2; shift ;;
-agent) agent=$2; shift ;;
-backend) backend=$2; shift ;;
-model) model=$2; shift ;;
-cwd) cwd=$2; shift ;;
--) shift; break ;;
-*) die "unknown flag: $1" ;;
*) break ;;
esac
shift
done
[ $# -eq 0 ] && die "prompt required"
# Read agent name from stdin if not provided via flag and stdin is piped.
if [ -z "$agent" ] && ! [ -t 0 ]; then
agent=$(head -1)
fi
name=${name:-$(/bin/date +%s%N)-$$}
# Prefix with parent session ID for global uniqueness.
[ -n "${OLLIE_SESSION_ID:-}" ] && name="${OLLIE_SESSION_ID}__${name}"
cwd=${cwd:-$(pwd)}
agent=${agent:-default}
prompt="[parent_session_id=${OLLIE_SESSION_ID:-}]
[session_id=${name}]
$*"
# Create session via D-Bus
session_id=$(dbus_string CreateSession \
string:"$cwd" string:"${backend:-}" string:"${model:-}" string:"$agent")
[ -z "$session_id" ] && die "failed to create session"
# Rename to our chosen name
dbus_call RenameSession string:"$session_id" string:"$name" > /dev/null 2>&1
# Submit prompt
dbus_call Submit string:"$session_id" string:"$prompt" > /dev/null 2>&1 || die "failed to submit prompt"
# Clean up JIT agent config now that the server has loaded it.
[[ "${agent:-}" == jit-* ]] && rm -f "$HOME/.config/ollie/agents/${agent}.json" 2>/dev/null || true
# Report resolved values from the live session.
config=$(dbus_string GetConfig string:"$session_id")
resolved_backend=$(printf '%s\n' "$config" | grep -m1 '^backend=' | cut -d= -f2- || echo "${backend:-default}")
resolved_model=$(printf '%s\n' "$config" | grep -m1 '^model=' | cut -d= -f2- || echo "${model:-default}")
printf 'session=%s\nbackend=%s\nmodel=%s\n' "$name" "$resolved_backend" "$resolved_model"