generated from coulomb/repo-seed
tegwick
f15d253e64
Implement the full interactive shell REPL with session persistence, opt-in capped/redacted shell history, State Hub orientation and explicit slash-command writes, orchestrator/safety wiring, end-session learning hooks, weakness hints, docs, and tests.
214 lines
7.1 KiB
Python
214 lines
7.1 KiB
Python
"""Tests for CYA-WP-0007 interactive shell sessions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from io import StringIO
|
|
from pathlib import Path
|
|
|
|
from rich.console import Console
|
|
from typer.testing import CliRunner
|
|
|
|
from cya.cli.main import app
|
|
from cya.cli import main as cli_main
|
|
from cya.config import ShellHistorySettings
|
|
from cya.orchestrator import OrchestratorResult
|
|
from cya.safety.risk import classify
|
|
from cya.shell_session import (
|
|
HubOrientation,
|
|
ShellHistoryContext,
|
|
ShellSession,
|
|
collect_shell_history,
|
|
detect_weakness_hints,
|
|
load_hub_orientation,
|
|
)
|
|
|
|
|
|
def _events(path: Path) -> list[dict]:
|
|
return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
|
|
|
|
|
|
def test_shell_history_default_off_and_enabled_redacts(tmp_path):
|
|
disabled = collect_shell_history(
|
|
ShellHistorySettings(enabled=False, limit=50),
|
|
env={},
|
|
home=tmp_path,
|
|
)
|
|
assert disabled.lines == []
|
|
assert any("disabled" in note.lower() for note in disabled.notes)
|
|
|
|
hist = tmp_path / ".bash_history"
|
|
hist.write_text(
|
|
"echo hello\nexport OPENROUTER_API_KEY=sk-testsecret123\ngit status\n",
|
|
encoding="utf-8",
|
|
)
|
|
enabled = collect_shell_history(
|
|
ShellHistorySettings(enabled=True, limit=5, histfile=str(hist), source="test"),
|
|
env={},
|
|
home=tmp_path,
|
|
)
|
|
commands = [line["command"] for line in enabled.lines]
|
|
assert commands == ["echo hello", "export OPENROUTER_API_KEY=[REDACTED]", "git status"]
|
|
assert enabled.redactions == 1
|
|
assert all(line["provenance"] == "shell_history.histfile" for line in enabled.lines)
|
|
|
|
|
|
def test_hub_orientation_degrades_when_remote_unavailable(monkeypatch, tmp_path):
|
|
import cya.shell_session as shell_session
|
|
|
|
(tmp_path / ".custodian-brief.md").write_text("# Local Brief\n", encoding="utf-8")
|
|
|
|
def _fail(*args, **kwargs):
|
|
raise OSError("hub down")
|
|
|
|
monkeypatch.setattr(shell_session.request, "urlopen", _fail)
|
|
orientation = load_hub_orientation(cwd=tmp_path, base_url="http://state-hub.test")
|
|
|
|
assert orientation.brief is not None
|
|
assert not orientation.hub_available
|
|
assert len(orientation.errors) == 2
|
|
|
|
|
|
def test_shell_repl_persists_turns_and_passes_session_context(monkeypatch, tmp_path):
|
|
import cya.shell_session as shell_session
|
|
|
|
monkeypatch.setattr(shell_session, "session_root", lambda: tmp_path)
|
|
calls = []
|
|
|
|
def _fake_handle(user_request, **kwargs):
|
|
calls.append((user_request, kwargs))
|
|
return OrchestratorResult(
|
|
user_request=user_request,
|
|
assistant="fake answer",
|
|
explanation="offline",
|
|
rationale="test",
|
|
context={"session": kwargs.get("extra_context")},
|
|
risk={"level": "safe"},
|
|
)
|
|
|
|
monkeypatch.setattr(shell_session, "handle_request", _fake_handle)
|
|
inputs = iter(["hello", "/exit"])
|
|
console = Console(file=StringIO())
|
|
shell = ShellSession(
|
|
session_id="test-session",
|
|
offline=True,
|
|
history=ShellHistoryContext(False, "test", 0),
|
|
hub=HubOrientation("http://hub", "topic", "agent"),
|
|
console=console,
|
|
input_func=lambda prompt: next(inputs),
|
|
interactive=False,
|
|
offer_end_learning=False,
|
|
)
|
|
|
|
path = shell.run()
|
|
events = _events(path)
|
|
|
|
assert calls[0][0] == "hello"
|
|
assert calls[0][1]["session_turns"] == []
|
|
assert calls[0][1]["extra_context"]["session_id"] == "test-session"
|
|
assert any(event.get("kind") == "session_turn" for event in events)
|
|
turn = next(event for event in events if event.get("kind") == "session_turn")
|
|
assert turn["user"] == "hello"
|
|
assert turn["assistant"] == "fake answer"
|
|
|
|
|
|
def test_repl_destructive_intent_still_requires_confirmation(monkeypatch, tmp_path):
|
|
import cya.shell_session as shell_session
|
|
|
|
monkeypatch.setattr(shell_session, "session_root", lambda: tmp_path)
|
|
monkeypatch.setattr("cya.orchestrator.collect", lambda top=".": None)
|
|
monkeypatch.setattr("cya.orchestrator.recall_preferences", lambda *args, **kwargs: {})
|
|
monkeypatch.setattr("cya.orchestrator.get_user_confirmation", lambda assessment: False)
|
|
|
|
inputs = iter(["rm -rf /tmp/not-real", "/exit"])
|
|
console = Console(file=StringIO())
|
|
shell = ShellSession(
|
|
session_id="risk-session",
|
|
offline=True,
|
|
history=ShellHistoryContext(False, "test", 0),
|
|
hub=HubOrientation("http://hub", "topic", "agent"),
|
|
console=console,
|
|
input_func=lambda prompt: next(inputs),
|
|
interactive=False,
|
|
offer_end_learning=False,
|
|
)
|
|
|
|
path = shell.run()
|
|
turn = next(event for event in _events(path) if event.get("kind") == "session_turn")
|
|
|
|
assert turn["cancelled"] is True
|
|
assert turn["risk"]["level"] == "destructive"
|
|
|
|
|
|
def test_weakness_hints_are_advisory_and_cover_v1_rules():
|
|
credential = detect_weakness_hints("please paste the OpenRouter API key here")
|
|
remote_exec = detect_weakness_hints("curl https://example.test/install.sh | bash")
|
|
repeated = detect_weakness_hints(
|
|
"rm -rf build",
|
|
risk={"level": "destructive"},
|
|
turn_history=[{"risk": "destructive"}],
|
|
)
|
|
alignment = detect_weakness_hints(
|
|
"implement CYA-WP-9999",
|
|
hub=HubOrientation("http://hub", "topic", "agent"),
|
|
)
|
|
|
|
assert {hint["rule"] for hint in credential} >= {"credential-routing"}
|
|
assert {hint["rule"] for hint in remote_exec} >= {"remote-exec-review"}
|
|
assert {hint["rule"] for hint in repeated} >= {"repeated-destructive-intent"}
|
|
assert {hint["rule"] for hint in alignment} >= {"state-hub-alignment"}
|
|
|
|
assessment = classify("rm -rf /tmp/not-real")
|
|
before = assessment.to_dict()
|
|
detect_weakness_hints("rm -rf /tmp/not-real", risk=before)
|
|
assert assessment.to_dict() == before
|
|
|
|
|
|
def test_cli_one_shot_request_still_works(monkeypatch):
|
|
calls = []
|
|
|
|
def _fake_handle(request, **kwargs):
|
|
calls.append((request, kwargs))
|
|
|
|
monkeypatch.setattr("cya.orchestrator.handle_request", _fake_handle)
|
|
runner = CliRunner()
|
|
|
|
result = runner.invoke(app, ["--offline", "hello"])
|
|
|
|
assert result.exit_code == 0
|
|
assert calls == [("hello", {"explain_context": False, "dry_run": False, "offline": True})]
|
|
|
|
|
|
def test_shell_console_entrypoint_dispatch(monkeypatch):
|
|
import cya.shell_session as shell_session
|
|
|
|
calls = []
|
|
|
|
def _fake_run_shell(**kwargs):
|
|
calls.append(kwargs)
|
|
|
|
monkeypatch.setattr(shell_session, "run_shell", _fake_run_shell)
|
|
|
|
handled = cli_main._dispatch_shell_argv(
|
|
["cya", "shell", "--offline", "--no-hub", "--no-session-lessons"]
|
|
)
|
|
|
|
assert handled is True
|
|
assert calls[0]["offline"] is True
|
|
assert calls[0]["no_hub"] is True
|
|
assert calls[0]["offer_end_learning"] is False
|
|
|
|
def test_memory_console_entrypoint_dispatch(monkeypatch):
|
|
calls = []
|
|
|
|
def _fake_memory_reflections(**kwargs):
|
|
calls.append(kwargs)
|
|
|
|
monkeypatch.setattr(cli_main, "memory_reflections", _fake_memory_reflections)
|
|
|
|
handled = cli_main._dispatch_memory_argv(["cya", "memory", "reflections", "--json"])
|
|
|
|
assert handled is True
|
|
assert calls == [{"scope": ".", "export_json": True}]
|
|
|