Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
77f10fa611 |
@@ -1,4 +1,4 @@
|
||||
"""Shared auxiliary client router for side tasks.
|
||||
from agent.telemetry_logger import log_token_usage\n"""Shared auxiliary client router for side tasks.
|
||||
|
||||
Provides a single resolution chain so every consumer (context compression,
|
||||
session search, web extraction, vision analysis, browser vision) picks up
|
||||
@@ -34,8 +34,6 @@ Payment / credit exhaustion fallback:
|
||||
their OpenRouter balance but has Codex OAuth or another provider available.
|
||||
"""
|
||||
|
||||
from agent.telemetry_logger import log_token_usage
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
@@ -398,8 +396,7 @@ class _CodexCompletionsAdapter:
|
||||
prompt_tokens=getattr(resp_usage, "input_tokens", 0),
|
||||
completion_tokens=getattr(resp_usage, "output_tokens", 0),
|
||||
total_tokens=getattr(resp_usage, "total_tokens", 0),
|
||||
)
|
||||
log_token_usage(usage.prompt_tokens, usage.completion_tokens, model)
|
||||
)\n log_token_usage(usage.prompt_tokens, usage.completion_tokens, model)
|
||||
except Exception as exc:
|
||||
logger.debug("Codex auxiliary Responses API call failed: %s", exc)
|
||||
raise
|
||||
@@ -532,8 +529,7 @@ class _AnthropicCompletionsAdapter:
|
||||
prompt_tokens=prompt_tokens,
|
||||
completion_tokens=completion_tokens,
|
||||
total_tokens=total_tokens,
|
||||
)
|
||||
log_token_usage(usage.prompt_tokens, usage.completion_tokens, model)
|
||||
)\n log_token_usage(usage.prompt_tokens, usage.completion_tokens, model)
|
||||
|
||||
choice = SimpleNamespace(
|
||||
index=0,
|
||||
|
||||
240
cli.py
240
cli.py
@@ -7254,40 +7254,6 @@ class HermesCLI:
|
||||
"Use your best judgement to make the choice and proceed."
|
||||
)
|
||||
|
||||
def _handle_clarify_selection(self) -> None:
|
||||
"""Process the currently selected clarify choice."""
|
||||
state = self._clarify_state
|
||||
if not state or self._clarify_freetext:
|
||||
return
|
||||
|
||||
selected = state.get("selected", 0)
|
||||
choices = state.get("choices") or []
|
||||
if selected < len(choices):
|
||||
state["response_queue"].put(choices[selected])
|
||||
self._clarify_state = None
|
||||
self._clarify_freetext = False
|
||||
self._invalidate()
|
||||
return
|
||||
|
||||
if selected == len(choices):
|
||||
self._clarify_freetext = True
|
||||
self._invalidate()
|
||||
|
||||
def _handle_clarify_number_shortcut(self, number: int) -> bool:
|
||||
"""Select a clarify option by number key."""
|
||||
state = self._clarify_state
|
||||
if not state or self._clarify_freetext:
|
||||
return False
|
||||
|
||||
choices = state.get("choices") or []
|
||||
max_option = len(choices) + 1
|
||||
if number < 1 or number > max_option:
|
||||
return False
|
||||
|
||||
state["selected"] = number - 1
|
||||
self._handle_clarify_selection()
|
||||
return True
|
||||
|
||||
def _sudo_password_callback(self) -> str:
|
||||
"""
|
||||
Prompt for sudo password through the prompt_toolkit UI.
|
||||
@@ -7396,20 +7362,6 @@ class HermesCLI:
|
||||
choices.append("view")
|
||||
return choices
|
||||
|
||||
def _handle_approval_number_shortcut(self, number: int) -> bool:
|
||||
"""Select an approval option by number key."""
|
||||
state = self._approval_state
|
||||
if not state:
|
||||
return False
|
||||
|
||||
choices = state.get("choices") or []
|
||||
if number < 1 or number > len(choices):
|
||||
return False
|
||||
|
||||
state["selected"] = number - 1
|
||||
self._handle_approval_selection()
|
||||
return True
|
||||
|
||||
def _handle_approval_selection(self) -> None:
|
||||
"""Process the currently selected dangerous-command approval choice."""
|
||||
state = self._approval_state
|
||||
@@ -7485,9 +7437,8 @@ class HermesCLI:
|
||||
preview_lines.extend(_wrap_panel_text(cmd_display, 60))
|
||||
for i, choice in enumerate(choices):
|
||||
prefix = '❯ ' if i == selected else ' '
|
||||
label = f"{i + 1}. {choice_labels.get(choice, choice)}"
|
||||
preview_lines.extend(_wrap_panel_text(
|
||||
f"{prefix}{label}",
|
||||
f"{prefix}{choice_labels.get(choice, choice)}",
|
||||
60,
|
||||
subsequent_indent=" ",
|
||||
))
|
||||
@@ -7505,7 +7456,7 @@ class HermesCLI:
|
||||
_append_panel_line(lines, 'class:approval-border', 'class:approval-cmd', wrapped, box_width)
|
||||
_append_blank_panel_line(lines, 'class:approval-border', box_width)
|
||||
for i, choice in enumerate(choices):
|
||||
label = f"{i + 1}. {choice_labels.get(choice, choice)}"
|
||||
label = choice_labels.get(choice, choice)
|
||||
style = 'class:approval-selected' if i == selected else 'class:approval-choice'
|
||||
prefix = '❯ ' if i == selected else ' '
|
||||
for wrapped in _wrap_panel_text(f"{prefix}{label}", inner_text_width, subsequent_indent=" "):
|
||||
@@ -7514,97 +7465,6 @@ class HermesCLI:
|
||||
lines.append(('class:approval-border', '╰' + ('─' * box_width) + '╯\n'))
|
||||
return lines
|
||||
|
||||
def _get_clarify_display_fragments(self):
|
||||
"""Render the clarify panel for the prompt_toolkit UI."""
|
||||
state = self._clarify_state
|
||||
if not state:
|
||||
return []
|
||||
|
||||
def _panel_box_width(title: str, content_lines: list[str], min_width: int = 46, max_width: int = 76) -> int:
|
||||
term_cols = shutil.get_terminal_size((100, 20)).columns
|
||||
longest = max([len(title)] + [len(line) for line in content_lines] + [min_width - 4])
|
||||
inner = min(max(longest + 4, min_width - 2), max_width - 2, max(24, term_cols - 6))
|
||||
return inner + 2
|
||||
|
||||
def _wrap_panel_text(text: str, width: int, subsequent_indent: str = "") -> list[str]:
|
||||
wrapped = textwrap.wrap(
|
||||
text,
|
||||
width=max(8, width),
|
||||
break_long_words=False,
|
||||
break_on_hyphens=False,
|
||||
subsequent_indent=subsequent_indent,
|
||||
)
|
||||
return wrapped or [""]
|
||||
|
||||
def _append_panel_line(lines, border_style: str, content_style: str, text: str, box_width: int) -> None:
|
||||
inner_width = max(0, box_width - 2)
|
||||
lines.append((border_style, "│ "))
|
||||
lines.append((content_style, text.ljust(inner_width)))
|
||||
lines.append((border_style, " │\n"))
|
||||
|
||||
def _append_blank_panel_line(lines, border_style: str, box_width: int) -> None:
|
||||
lines.append((border_style, "│" + (" " * box_width) + "│\n"))
|
||||
|
||||
question = state["question"]
|
||||
choices = state.get("choices") or []
|
||||
selected = state.get("selected", 0)
|
||||
preview_lines = _wrap_panel_text(question, 60)
|
||||
for i, choice in enumerate(choices):
|
||||
prefix = "❯ " if i == selected and not self._clarify_freetext else " "
|
||||
label = f"{i + 1}. {choice}"
|
||||
preview_lines.extend(_wrap_panel_text(f"{prefix}{label}", 60, subsequent_indent=" "))
|
||||
other_number = len(choices) + 1
|
||||
other_label = (
|
||||
f"❯ {other_number}. Other (type below)" if self._clarify_freetext
|
||||
else f"❯ {other_number}. Other (type your answer)" if selected == len(choices)
|
||||
else f" {other_number}. Other (type your answer)"
|
||||
)
|
||||
preview_lines.extend(_wrap_panel_text(other_label, 60, subsequent_indent=" "))
|
||||
box_width = _panel_box_width("Hermes needs your input", preview_lines)
|
||||
inner_text_width = max(8, box_width - 2)
|
||||
|
||||
lines = []
|
||||
lines.append(('class:clarify-border', '╭─ '))
|
||||
lines.append(('class:clarify-title', 'Hermes needs your input'))
|
||||
lines.append(('class:clarify-border', ' ' + ('─' * max(0, box_width - len("Hermes needs your input") - 3)) + '╮\n'))
|
||||
_append_blank_panel_line(lines, 'class:clarify-border', box_width)
|
||||
|
||||
for wrapped in _wrap_panel_text(question, inner_text_width):
|
||||
_append_panel_line(lines, 'class:clarify-border', 'class:clarify-question', wrapped, box_width)
|
||||
_append_blank_panel_line(lines, 'class:clarify-border', box_width)
|
||||
|
||||
if self._clarify_freetext and not choices:
|
||||
guidance = "Type your answer in the prompt below, then press Enter."
|
||||
for wrapped in _wrap_panel_text(guidance, inner_text_width):
|
||||
_append_panel_line(lines, 'class:clarify-border', 'class:clarify-choice', wrapped, box_width)
|
||||
_append_blank_panel_line(lines, 'class:clarify-border', box_width)
|
||||
|
||||
if choices:
|
||||
for i, choice in enumerate(choices):
|
||||
style = 'class:clarify-selected' if i == selected and not self._clarify_freetext else 'class:clarify-choice'
|
||||
prefix = '❯ ' if i == selected and not self._clarify_freetext else ' '
|
||||
label = f"{i + 1}. {choice}"
|
||||
wrapped_lines = _wrap_panel_text(f"{prefix}{label}", inner_text_width, subsequent_indent=" ")
|
||||
for wrapped in wrapped_lines:
|
||||
_append_panel_line(lines, 'class:clarify-border', style, wrapped, box_width)
|
||||
|
||||
other_idx = len(choices)
|
||||
if selected == other_idx and not self._clarify_freetext:
|
||||
other_style = 'class:clarify-selected'
|
||||
other_label = f'❯ {other_number}. Other (type your answer)'
|
||||
elif self._clarify_freetext:
|
||||
other_style = 'class:clarify-active-other'
|
||||
other_label = f'❯ {other_number}. Other (type below)'
|
||||
else:
|
||||
other_style = 'class:clarify-choice'
|
||||
other_label = f' {other_number}. Other (type your answer)'
|
||||
for wrapped in _wrap_panel_text(other_label, inner_text_width, subsequent_indent=" "):
|
||||
_append_panel_line(lines, 'class:clarify-border', other_style, wrapped, box_width)
|
||||
|
||||
_append_blank_panel_line(lines, 'class:clarify-border', box_width)
|
||||
lines.append(('class:clarify-border', '╰' + ('─' * box_width) + '╯\n'))
|
||||
return lines
|
||||
|
||||
def _secret_capture_callback(self, var_name: str, prompt: str, metadata=None) -> dict:
|
||||
return prompt_for_secret(self, var_name, prompt, metadata)
|
||||
|
||||
@@ -8511,8 +8371,17 @@ class HermesCLI:
|
||||
|
||||
# --- Clarify choice mode: confirm the highlighted selection ---
|
||||
if self._clarify_state and not self._clarify_freetext:
|
||||
self._handle_clarify_selection()
|
||||
event.app.invalidate()
|
||||
state = self._clarify_state
|
||||
selected = state["selected"]
|
||||
choices = state.get("choices") or []
|
||||
if selected < len(choices):
|
||||
state["response_queue"].put(choices[selected])
|
||||
self._clarify_state = None
|
||||
event.app.invalidate()
|
||||
else:
|
||||
# "Other" selected → switch to freetext
|
||||
self._clarify_freetext = True
|
||||
event.app.invalidate()
|
||||
return
|
||||
|
||||
# --- Normal input routing ---
|
||||
@@ -8632,19 +8501,6 @@ class HermesCLI:
|
||||
self._approval_state["selected"] = min(max_idx, self._approval_state["selected"] + 1)
|
||||
event.app.invalidate()
|
||||
|
||||
# --- Numbered shortcuts for clarify / approval modal prompts ---
|
||||
for _digit in '123456789':
|
||||
@kb.add(_digit, filter=Condition(lambda: bool(self._approval_state) or (bool(self._clarify_state) and not self._clarify_freetext)))
|
||||
def _handle_modal_number(event, digit=_digit):
|
||||
number = int(digit)
|
||||
handled = False
|
||||
if self._approval_state:
|
||||
handled = self._handle_approval_number_shortcut(number)
|
||||
elif self._clarify_state and not self._clarify_freetext:
|
||||
handled = self._handle_clarify_number_shortcut(number)
|
||||
if handled:
|
||||
event.app.invalidate()
|
||||
|
||||
# --- /model picker: arrow-key navigation ---
|
||||
@kb.add('up', filter=Condition(lambda: bool(self._model_picker_state)))
|
||||
def model_picker_up(event):
|
||||
@@ -9139,7 +8995,7 @@ class HermesCLI:
|
||||
if cli_ref._approval_state:
|
||||
remaining = max(0, int(cli_ref._approval_deadline - _time.monotonic()))
|
||||
return [
|
||||
('class:hint', ' 1-9 or ↑/↓ to select, Enter to confirm'),
|
||||
('class:hint', ' ↑/↓ to select, Enter to confirm'),
|
||||
('class:clarify-countdown', f' ({remaining}s)'),
|
||||
]
|
||||
|
||||
@@ -9152,7 +9008,7 @@ class HermesCLI:
|
||||
('class:clarify-countdown', countdown),
|
||||
]
|
||||
return [
|
||||
('class:hint', ' 1-9 or ↑/↓ to select, Enter to confirm'),
|
||||
('class:hint', ' ↑/↓ to select, Enter to confirm'),
|
||||
('class:clarify-countdown', countdown),
|
||||
]
|
||||
|
||||
@@ -9230,7 +9086,71 @@ class HermesCLI:
|
||||
lines.append((border_style, "│" + (" " * box_width) + "│\n"))
|
||||
|
||||
def _get_clarify_display():
|
||||
return cli_ref._get_clarify_display_fragments()
|
||||
"""Build styled text for the clarify question/choices panel."""
|
||||
state = cli_ref._clarify_state
|
||||
if not state:
|
||||
return []
|
||||
|
||||
question = state["question"]
|
||||
choices = state.get("choices") or []
|
||||
selected = state.get("selected", 0)
|
||||
preview_lines = _wrap_panel_text(question, 60)
|
||||
for i, choice in enumerate(choices):
|
||||
prefix = "❯ " if i == selected and not cli_ref._clarify_freetext else " "
|
||||
preview_lines.extend(_wrap_panel_text(f"{prefix}{choice}", 60, subsequent_indent=" "))
|
||||
other_label = (
|
||||
"❯ Other (type below)" if cli_ref._clarify_freetext
|
||||
else "❯ Other (type your answer)" if selected == len(choices)
|
||||
else " Other (type your answer)"
|
||||
)
|
||||
preview_lines.extend(_wrap_panel_text(other_label, 60, subsequent_indent=" "))
|
||||
box_width = _panel_box_width("Hermes needs your input", preview_lines)
|
||||
inner_text_width = max(8, box_width - 2)
|
||||
|
||||
lines = []
|
||||
# Box top border
|
||||
lines.append(('class:clarify-border', '╭─ '))
|
||||
lines.append(('class:clarify-title', 'Hermes needs your input'))
|
||||
lines.append(('class:clarify-border', ' ' + ('─' * max(0, box_width - len("Hermes needs your input") - 3)) + '╮\n'))
|
||||
_append_blank_panel_line(lines, 'class:clarify-border', box_width)
|
||||
|
||||
# Question text
|
||||
for wrapped in _wrap_panel_text(question, inner_text_width):
|
||||
_append_panel_line(lines, 'class:clarify-border', 'class:clarify-question', wrapped, box_width)
|
||||
_append_blank_panel_line(lines, 'class:clarify-border', box_width)
|
||||
|
||||
if cli_ref._clarify_freetext and not choices:
|
||||
guidance = "Type your answer in the prompt below, then press Enter."
|
||||
for wrapped in _wrap_panel_text(guidance, inner_text_width):
|
||||
_append_panel_line(lines, 'class:clarify-border', 'class:clarify-choice', wrapped, box_width)
|
||||
_append_blank_panel_line(lines, 'class:clarify-border', box_width)
|
||||
|
||||
if choices:
|
||||
# Multiple-choice mode: show selectable options
|
||||
for i, choice in enumerate(choices):
|
||||
style = 'class:clarify-selected' if i == selected and not cli_ref._clarify_freetext else 'class:clarify-choice'
|
||||
prefix = '❯ ' if i == selected and not cli_ref._clarify_freetext else ' '
|
||||
wrapped_lines = _wrap_panel_text(f"{prefix}{choice}", inner_text_width, subsequent_indent=" ")
|
||||
for wrapped in wrapped_lines:
|
||||
_append_panel_line(lines, 'class:clarify-border', style, wrapped, box_width)
|
||||
|
||||
# "Other" option (5th line, only shown when choices exist)
|
||||
other_idx = len(choices)
|
||||
if selected == other_idx and not cli_ref._clarify_freetext:
|
||||
other_style = 'class:clarify-selected'
|
||||
other_label = '❯ Other (type your answer)'
|
||||
elif cli_ref._clarify_freetext:
|
||||
other_style = 'class:clarify-active-other'
|
||||
other_label = '❯ Other (type below)'
|
||||
else:
|
||||
other_style = 'class:clarify-choice'
|
||||
other_label = ' Other (type your answer)'
|
||||
for wrapped in _wrap_panel_text(other_label, inner_text_width, subsequent_indent=" "):
|
||||
_append_panel_line(lines, 'class:clarify-border', other_style, wrapped, box_width)
|
||||
|
||||
_append_blank_panel_line(lines, 'class:clarify-border', box_width)
|
||||
lines.append(('class:clarify-border', '╰' + ('─' * box_width) + '╯\n'))
|
||||
return lines
|
||||
|
||||
clarify_widget = ConditionalContainer(
|
||||
Window(
|
||||
|
||||
@@ -38,6 +38,18 @@ def _cron_api(**kwargs):
|
||||
return json.loads(cronjob_tool(**kwargs))
|
||||
|
||||
|
||||
def _print_runtime_overrides(job: dict) -> None:
|
||||
model = job.get("model")
|
||||
provider = job.get("provider")
|
||||
base_url = job.get("base_url")
|
||||
if model:
|
||||
print(f" Model: {model}")
|
||||
if provider:
|
||||
print(f" Provider: {provider}")
|
||||
if base_url:
|
||||
print(f" Base URL: {base_url}")
|
||||
|
||||
|
||||
def cron_list(show_all: bool = False):
|
||||
"""List all scheduled jobs."""
|
||||
from cron.jobs import list_jobs
|
||||
@@ -93,6 +105,7 @@ def cron_list(show_all: bool = False):
|
||||
script = job.get("script")
|
||||
if script:
|
||||
print(f" Script: {script}")
|
||||
_print_runtime_overrides(job)
|
||||
|
||||
# Execution history
|
||||
last_status = job.get("last_status")
|
||||
@@ -167,6 +180,9 @@ def cron_create(args):
|
||||
repeat=getattr(args, "repeat", None),
|
||||
skill=getattr(args, "skill", None),
|
||||
skills=_normalize_skills(getattr(args, "skill", None), getattr(args, "skills", None)),
|
||||
model=getattr(args, "model", None),
|
||||
provider=getattr(args, "provider", None),
|
||||
base_url=getattr(args, "base_url", None),
|
||||
script=getattr(args, "script", None),
|
||||
)
|
||||
if not result.get("success"):
|
||||
@@ -180,6 +196,8 @@ def cron_create(args):
|
||||
job_data = result.get("job", {})
|
||||
if job_data.get("script"):
|
||||
print(f" Script: {job_data['script']}")
|
||||
if job_data:
|
||||
_print_runtime_overrides(job_data)
|
||||
print(f" Next run: {result['next_run_at']}")
|
||||
return 0
|
||||
|
||||
@@ -217,6 +235,9 @@ def cron_edit(args):
|
||||
deliver=getattr(args, "deliver", None),
|
||||
repeat=getattr(args, "repeat", None),
|
||||
skills=final_skills,
|
||||
model=getattr(args, "model", None),
|
||||
provider=getattr(args, "provider", None),
|
||||
base_url=getattr(args, "base_url", None),
|
||||
script=getattr(args, "script", None),
|
||||
)
|
||||
if not result.get("success"):
|
||||
@@ -233,6 +254,7 @@ def cron_edit(args):
|
||||
print(" Skills: none")
|
||||
if updated.get("script"):
|
||||
print(f" Script: {updated['script']}")
|
||||
_print_runtime_overrides(updated)
|
||||
return 0
|
||||
|
||||
|
||||
|
||||
@@ -4958,6 +4958,9 @@ For more help on a command:
|
||||
cron_create.add_argument("--deliver", help="Delivery target: origin, local, telegram, discord, signal, or platform:chat_id")
|
||||
cron_create.add_argument("--repeat", type=int, help="Optional repeat count")
|
||||
cron_create.add_argument("--skill", dest="skills", action="append", help="Attach a skill. Repeat to add multiple skills.")
|
||||
cron_create.add_argument("--model", help="Pin this job to a specific model (for example: google/gemma-4-31b-it)")
|
||||
cron_create.add_argument("--provider", help="Pin this job to a specific provider (for example: openrouter)")
|
||||
cron_create.add_argument("--base-url", dest="base_url", help="Optional base URL override for the job's runtime provider")
|
||||
cron_create.add_argument("--script", help="Path to a Python script whose stdout is injected into the prompt each run")
|
||||
|
||||
# cron edit
|
||||
@@ -4972,6 +4975,9 @@ For more help on a command:
|
||||
cron_edit.add_argument("--add-skill", dest="add_skills", action="append", help="Append a skill without replacing the existing list. Repeatable.")
|
||||
cron_edit.add_argument("--remove-skill", dest="remove_skills", action="append", help="Remove a specific attached skill. Repeatable.")
|
||||
cron_edit.add_argument("--clear-skills", action="store_true", help="Remove all attached skills from the job")
|
||||
cron_edit.add_argument("--model", help="Update the job's pinned model")
|
||||
cron_edit.add_argument("--provider", help="Update the job's pinned provider")
|
||||
cron_edit.add_argument("--base-url", dest="base_url", help="Update the job's pinned base URL. Pass an empty string to clear it.")
|
||||
cron_edit.add_argument("--script", help="Path to a Python script whose stdout is injected into the prompt each run. Pass empty string to clear.")
|
||||
|
||||
# lifecycle actions
|
||||
|
||||
@@ -1,172 +0,0 @@
|
||||
import queue
|
||||
import threading
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from cli import HermesCLI
|
||||
|
||||
|
||||
class _FakeBuffer:
|
||||
def __init__(self, text="", cursor_position=None):
|
||||
self.text = text
|
||||
self.cursor_position = len(text) if cursor_position is None else cursor_position
|
||||
|
||||
def reset(self, append_to_history=False):
|
||||
self.text = ""
|
||||
self.cursor_position = 0
|
||||
|
||||
|
||||
def _make_cli_stub():
|
||||
cli = HermesCLI.__new__(HermesCLI)
|
||||
cli._approval_state = None
|
||||
cli._approval_deadline = 0
|
||||
cli._approval_lock = threading.Lock()
|
||||
cli._clarify_state = None
|
||||
cli._clarify_freetext = False
|
||||
cli._clarify_deadline = 0
|
||||
cli._sudo_state = None
|
||||
cli._sudo_deadline = 0
|
||||
cli._secret_state = None
|
||||
cli._secret_deadline = 0
|
||||
cli._modal_input_snapshot = None
|
||||
cli._invalidate = MagicMock()
|
||||
cli._app = SimpleNamespace(invalidate=MagicMock(), current_buffer=_FakeBuffer())
|
||||
return cli
|
||||
|
||||
|
||||
def test_approval_display_numbers_choices():
|
||||
cli = _make_cli_stub()
|
||||
cli._approval_state = {
|
||||
"command": "sudo rm -rf /tmp/example",
|
||||
"description": "dangerous command",
|
||||
"choices": ["once", "session", "always", "deny"],
|
||||
"selected": 0,
|
||||
"response_queue": queue.Queue(),
|
||||
}
|
||||
|
||||
rendered = "".join(text for _style, text in cli._get_approval_display_fragments())
|
||||
|
||||
assert "❯ 1. Allow once" in rendered
|
||||
assert "2. Allow for this session" in rendered
|
||||
assert "3. Add to permanent allowlist" in rendered
|
||||
assert "4. Deny" in rendered
|
||||
|
||||
|
||||
def test_approval_number_shortcut_submits_choice():
|
||||
cli = _make_cli_stub()
|
||||
response_queue = queue.Queue()
|
||||
cli._approval_state = {
|
||||
"command": "sudo rm -rf /tmp/example",
|
||||
"description": "dangerous command",
|
||||
"choices": ["once", "session", "always", "deny"],
|
||||
"selected": 0,
|
||||
"response_queue": response_queue,
|
||||
}
|
||||
|
||||
assert cli._handle_approval_number_shortcut(2) is True
|
||||
assert response_queue.get_nowait() == "session"
|
||||
assert cli._approval_state is None
|
||||
|
||||
|
||||
def test_approval_selection_still_submits_selected_choice():
|
||||
cli = _make_cli_stub()
|
||||
response_queue = queue.Queue()
|
||||
cli._approval_state = {
|
||||
"command": "sudo rm -rf /tmp/example",
|
||||
"description": "dangerous command",
|
||||
"choices": ["once", "session", "always", "deny"],
|
||||
"selected": 1,
|
||||
"response_queue": response_queue,
|
||||
}
|
||||
|
||||
cli._handle_approval_selection()
|
||||
|
||||
assert response_queue.get_nowait() == "session"
|
||||
assert cli._approval_state is None
|
||||
|
||||
|
||||
def test_approval_number_shortcut_handles_view_in_place():
|
||||
cli = _make_cli_stub()
|
||||
response_queue = queue.Queue()
|
||||
cli._approval_state = {
|
||||
"command": "sudo dd if=/tmp/in of=/usr/share/keyrings/githubcli-archive-keyring.gpg bs=4M status=progress",
|
||||
"description": "disk copy",
|
||||
"choices": ["once", "session", "always", "deny", "view"],
|
||||
"selected": 0,
|
||||
"response_queue": response_queue,
|
||||
}
|
||||
|
||||
assert cli._handle_approval_number_shortcut(5) is True
|
||||
assert cli._approval_state is not None
|
||||
assert cli._approval_state["show_full"] is True
|
||||
assert "view" not in cli._approval_state["choices"]
|
||||
assert cli._approval_state["selected"] == 3
|
||||
assert response_queue.empty()
|
||||
|
||||
|
||||
def test_clarify_display_numbers_choices_and_other():
|
||||
cli = _make_cli_stub()
|
||||
cli._clarify_state = {
|
||||
"question": "Pick the best option",
|
||||
"choices": ["Alpha", "Beta", "Gamma", "Delta"],
|
||||
"selected": 1,
|
||||
"response_queue": queue.Queue(),
|
||||
}
|
||||
|
||||
rendered = "".join(text for _style, text in cli._get_clarify_display_fragments())
|
||||
|
||||
assert "1. Alpha" in rendered
|
||||
assert "❯ 2. Beta" in rendered
|
||||
assert "3. Gamma" in rendered
|
||||
assert "4. Delta" in rendered
|
||||
assert "5. Other (type your answer)" in rendered
|
||||
|
||||
|
||||
def test_clarify_number_shortcut_submits_choice():
|
||||
cli = _make_cli_stub()
|
||||
response_queue = queue.Queue()
|
||||
cli._clarify_state = {
|
||||
"question": "Pick the best option",
|
||||
"choices": ["Alpha", "Beta", "Gamma"],
|
||||
"selected": 0,
|
||||
"response_queue": response_queue,
|
||||
}
|
||||
|
||||
assert cli._handle_clarify_number_shortcut(3) is True
|
||||
assert response_queue.get_nowait() == "Gamma"
|
||||
assert cli._clarify_state is None
|
||||
assert cli._clarify_freetext is False
|
||||
|
||||
|
||||
def test_clarify_selection_still_submits_selected_choice():
|
||||
cli = _make_cli_stub()
|
||||
response_queue = queue.Queue()
|
||||
cli._clarify_state = {
|
||||
"question": "Pick the best option",
|
||||
"choices": ["Alpha", "Beta", "Gamma"],
|
||||
"selected": 1,
|
||||
"response_queue": response_queue,
|
||||
}
|
||||
|
||||
cli._handle_clarify_selection()
|
||||
|
||||
assert response_queue.get_nowait() == "Beta"
|
||||
assert cli._clarify_state is None
|
||||
assert cli._clarify_freetext is False
|
||||
|
||||
|
||||
def test_clarify_number_shortcut_activates_other_freetext():
|
||||
cli = _make_cli_stub()
|
||||
response_queue = queue.Queue()
|
||||
cli._clarify_state = {
|
||||
"question": "Pick the best option",
|
||||
"choices": ["Alpha", "Beta", "Gamma"],
|
||||
"selected": 0,
|
||||
"response_queue": response_queue,
|
||||
}
|
||||
|
||||
assert cli._handle_clarify_number_shortcut(4) is True
|
||||
assert cli._clarify_state is not None
|
||||
assert cli._clarify_state["selected"] == 3
|
||||
assert cli._clarify_freetext is True
|
||||
assert response_queue.empty()
|
||||
@@ -1,6 +1,7 @@
|
||||
"""Tests for hermes_cli.cron command handling."""
|
||||
|
||||
from argparse import Namespace
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -105,3 +106,135 @@ class TestCronCommandLifecycle:
|
||||
assert len(jobs) == 1
|
||||
assert jobs[0]["skills"] == ["blogwatcher", "find-nearby"]
|
||||
assert jobs[0]["name"] == "Skill combo"
|
||||
|
||||
def test_create_can_pin_runtime_model_provider_and_base_url(self, tmp_cron_dir, capsys):
|
||||
cron_command(
|
||||
Namespace(
|
||||
cron_command="create",
|
||||
schedule="every 1h",
|
||||
prompt="Run the burn loop",
|
||||
name="Gemma burn",
|
||||
deliver=None,
|
||||
repeat=None,
|
||||
skill=None,
|
||||
skills=None,
|
||||
script=None,
|
||||
model="google/gemma-4-31b-it",
|
||||
provider="openrouter",
|
||||
base_url="https://openrouter.ai/api/v1",
|
||||
)
|
||||
)
|
||||
|
||||
job = list_jobs()[0]
|
||||
assert job["model"] == "google/gemma-4-31b-it"
|
||||
assert job["provider"] == "openrouter"
|
||||
assert job["base_url"] == "https://openrouter.ai/api/v1"
|
||||
|
||||
out = capsys.readouterr().out
|
||||
assert "Created job" in out
|
||||
assert "Model: google/gemma-4-31b-it" in out
|
||||
assert "Provider: openrouter" in out
|
||||
|
||||
def test_edit_can_update_runtime_model_provider_and_clear_base_url(self, tmp_cron_dir, capsys):
|
||||
job = create_job(prompt="Check server status", schedule="every 1h")
|
||||
|
||||
cron_command(
|
||||
Namespace(
|
||||
cron_command="edit",
|
||||
job_id=job["id"],
|
||||
schedule=None,
|
||||
prompt=None,
|
||||
name=None,
|
||||
deliver=None,
|
||||
repeat=None,
|
||||
skill=None,
|
||||
skills=None,
|
||||
add_skills=None,
|
||||
remove_skills=None,
|
||||
clear_skills=False,
|
||||
script=None,
|
||||
model="google/gemma-4-31b-it",
|
||||
provider="openrouter",
|
||||
base_url="",
|
||||
)
|
||||
)
|
||||
|
||||
updated = get_job(job["id"])
|
||||
assert updated["model"] == "google/gemma-4-31b-it"
|
||||
assert updated["provider"] == "openrouter"
|
||||
assert updated["base_url"] is None
|
||||
|
||||
out = capsys.readouterr().out
|
||||
assert "Updated job" in out
|
||||
assert "Model: google/gemma-4-31b-it" in out
|
||||
assert "Provider: openrouter" in out
|
||||
|
||||
|
||||
class TestCronParserRuntimeOverrideFlags:
|
||||
def test_main_parses_create_runtime_override_flags(self, monkeypatch):
|
||||
from hermes_cli import main as main_mod
|
||||
|
||||
captured = {}
|
||||
|
||||
def fake_cmd_cron(args):
|
||||
captured["args"] = args
|
||||
|
||||
monkeypatch.setattr(main_mod, "cmd_cron", fake_cmd_cron)
|
||||
monkeypatch.setattr(
|
||||
"sys.argv",
|
||||
[
|
||||
"hermes",
|
||||
"cron",
|
||||
"create",
|
||||
"every 1h",
|
||||
"Run the burn loop",
|
||||
"--model",
|
||||
"google/gemma-4-31b-it",
|
||||
"--provider",
|
||||
"openrouter",
|
||||
"--base-url",
|
||||
"https://openrouter.ai/api/v1",
|
||||
],
|
||||
)
|
||||
|
||||
main_mod.main()
|
||||
|
||||
args = captured["args"]
|
||||
assert args.cron_command == "create"
|
||||
assert args.model == "google/gemma-4-31b-it"
|
||||
assert args.provider == "openrouter"
|
||||
assert args.base_url == "https://openrouter.ai/api/v1"
|
||||
|
||||
def test_main_parses_edit_runtime_override_flags(self, monkeypatch):
|
||||
from hermes_cli import main as main_mod
|
||||
|
||||
captured = {}
|
||||
|
||||
def fake_cmd_cron(args):
|
||||
captured["args"] = args
|
||||
|
||||
monkeypatch.setattr(main_mod, "cmd_cron", fake_cmd_cron)
|
||||
monkeypatch.setattr(
|
||||
"sys.argv",
|
||||
[
|
||||
"hermes",
|
||||
"cron",
|
||||
"edit",
|
||||
"job123",
|
||||
"--model",
|
||||
"google/gemma-4-31b-it",
|
||||
"--provider",
|
||||
"openrouter",
|
||||
"--base-url",
|
||||
"",
|
||||
],
|
||||
)
|
||||
|
||||
main_mod.main()
|
||||
|
||||
args = captured["args"]
|
||||
assert args.cron_command == "edit"
|
||||
assert args.job_id == "job123"
|
||||
assert args.model == "google/gemma-4-31b-it"
|
||||
assert args.provider == "openrouter"
|
||||
assert args.base_url == ""
|
||||
|
||||
Reference in New Issue
Block a user