diff --git a/src/opensquilla/cli/agents_cmd.py b/src/opensquilla/cli/agents_cmd.py index 7e511bdb6e..fbd461548a 100644 --- a/src/opensquilla/cli/agents_cmd.py +++ b/src/opensquilla/cli/agents_cmd.py @@ -62,14 +62,14 @@ def agents_list( _print_json(agents) return - console = Console(width=200, force_terminal=False) + console = Console(force_terminal=False) table = Table(title=f"Agents in {target}") - table.add_column("id", no_wrap=True) - table.add_column("name", no_wrap=True) - table.add_column("type", no_wrap=True) - table.add_column("enabled", no_wrap=True) - table.add_column("model", no_wrap=True) - table.add_column("workspace") + table.add_column("id", overflow="fold") + table.add_column("name", overflow="fold") + table.add_column("type", overflow="fold") + table.add_column("enabled", overflow="fold") + table.add_column("model", overflow="fold") + table.add_column("workspace", overflow="fold") for agent in agents: table.add_row( str(agent.get("id", "")), diff --git a/src/opensquilla/cli/channels_cmd.py b/src/opensquilla/cli/channels_cmd.py index dfa3ba4eaf..26293b8522 100644 --- a/src/opensquilla/cli/channels_cmd.py +++ b/src/opensquilla/cli/channels_cmd.py @@ -70,13 +70,13 @@ def _render_channels_table(entries: list[dict[str, Any]], *, title: str) -> None if not entries: typer.echo("0 channels configured.") return - console = Console(width=200, force_terminal=False) + console = Console(force_terminal=False) table = Table(title=title) - table.add_column("name", no_wrap=True) - table.add_column("type", no_wrap=True) - table.add_column("enabled", no_wrap=True) - table.add_column("agent_id", no_wrap=True) - table.add_column("details") + table.add_column("name", overflow="fold") + table.add_column("type", overflow="fold") + table.add_column("enabled", overflow="fold") + table.add_column("agent_id", overflow="fold") + table.add_column("details", overflow="fold") for e in entries: details = ", ".join( f"{k}={v}" @@ -110,14 +110,14 @@ def _status_diagnostic_text(row: dict[str, Any]) -> str: def _render_status_table(payload: dict[str, Any], *, name: str | None = None) -> None: rows = _filter_status_rows(payload, name) table = Table(title="Channel status", show_header=True, header_style=ACCENT_HEADER) - table.add_column("Name") - table.add_column("Type") - table.add_column("Status") - table.add_column("Connected") - table.add_column("Enabled") - table.add_column("Configured") - table.add_column("Restart attempts", justify="right") - table.add_column("Diagnostic") + table.add_column("Name", overflow="fold") + table.add_column("Type", overflow="fold") + table.add_column("Status", overflow="fold") + table.add_column("Connected", overflow="fold") + table.add_column("Enabled", overflow="fold") + table.add_column("Configured", overflow="fold") + table.add_column("Restart attempts", justify="right", overflow="fold") + table.add_column("Diagnostic", overflow="fold") for row in rows: table.add_row( str(row.get("name") or ""), @@ -129,7 +129,7 @@ def _render_status_table(payload: dict[str, Any], *, name: str | None = None) -> str(row.get("restart_attempts") or 0), _status_diagnostic_text(row), ) - Console(width=180, force_terminal=False).print(table) + Console(force_terminal=False).print(table) def _filter_status_rows(payload: dict[str, Any], name: str | None) -> list[dict[str, Any]]: @@ -400,18 +400,18 @@ def channels_types( ]) return table = Table(title="Supported channel types") - table.add_column("type", no_wrap=True) - table.add_column("label") - table.add_column("transport", no_wrap=True) - table.add_column("public URL", no_wrap=True) - table.add_column("extras", no_wrap=True) + table.add_column("type", overflow="fold") + table.add_column("label", overflow="fold") + table.add_column("transport", overflow="fold") + table.add_column("public URL", overflow="fold") + table.add_column("extras", overflow="fold") for s in specs: table.add_row( s.type, s.label, s.transport, "yes" if s.requires_public_url else "no", s.dependency_extra or "—", ) - Console(width=140, force_terminal=False).print(table) + Console(force_terminal=False).print(table) @channels_app.command("describe") @@ -448,7 +448,7 @@ def channels_describe( }) return - console = Console(width=160, force_terminal=False) + console = Console(force_terminal=False) typer.echo(f"{spec.label} ({spec.type})") typer.echo(spec.description) typer.echo( @@ -458,12 +458,12 @@ def channels_describe( f"docs={spec.docs_hint}" ) table = Table(title="Fields") - table.add_column("name", no_wrap=True) - table.add_column("type", no_wrap=True) - table.add_column("required", no_wrap=True) - table.add_column("secret", no_wrap=True) - table.add_column("default") - table.add_column("choices") + table.add_column("name", overflow="fold") + table.add_column("type", overflow="fold") + table.add_column("required", overflow="fold") + table.add_column("secret", overflow="fold") + table.add_column("default", overflow="fold") + table.add_column("choices", overflow="fold") for f in spec.fields: table.add_row( f.name, @@ -548,12 +548,12 @@ def channels_certify( print_json(evidence) else: table = Table(title="Channel certification", show_header=True) - table.add_column("Provider") - table.add_column("Operation") - table.add_column("Status") - table.add_column("Authenticated") - table.add_column("Latency", justify="right") - table.add_column("Detail") + table.add_column("Provider", overflow="fold") + table.add_column("Operation", overflow="fold") + table.add_column("Status", overflow="fold") + table.add_column("Authenticated", overflow="fold") + table.add_column("Latency", justify="right", overflow="fold") + table.add_column("Detail", overflow="fold") for row in evidence["providers"]: detail = str(row.get("detail") or "") missing = row.get("missingEnvironment") @@ -567,7 +567,7 @@ def channels_certify( f"{row.get('latencyMs', 0)} ms", detail, ) - Console(width=180, force_terminal=False).print(table) + Console(force_terminal=False).print(table) if not evidence_passed(evidence): raise typer.Exit(code=1) @@ -579,11 +579,11 @@ def channels_certify( def _render_pairings_table(records: list[dict[str, Any]], *, channel: str) -> None: table = Table(title=f"Pairing requests: {channel}", header_style=ACCENT_HEADER) - table.add_column("Code") - table.add_column("Sender") - table.add_column("Status") - table.add_column("Requested") - table.add_column("Approved") + table.add_column("Code", overflow="fold") + table.add_column("Sender", overflow="fold") + table.add_column("Status", overflow="fold") + table.add_column("Requested", overflow="fold") + table.add_column("Approved", overflow="fold") for record in records: sender = str(record.get("senderName") or record.get("senderId") or "") table.add_row( @@ -593,7 +593,7 @@ def _render_pairings_table(records: list[dict[str, Any]], *, channel: str) -> No str(record.get("createdAt") or ""), str(record.get("approvedAt") or ""), ) - Console(width=140, force_terminal=False).print(table) + Console(force_terminal=False).print(table) if not records: typer.echo("No pairing requests.") diff --git a/src/opensquilla/cli/doctor_cmd.py b/src/opensquilla/cli/doctor_cmd.py index a3887bb85e..c46a0b7e16 100644 --- a/src/opensquilla/cli/doctor_cmd.py +++ b/src/opensquilla/cli/doctor_cmd.py @@ -672,7 +672,7 @@ def _report_context(report: dict[str, Any]) -> str: def _render_report(report: dict[str, Any]) -> None: - console = Console(width=180, force_terminal=False) + console = Console(force_terminal=False) console.print(f"[bold]OpenSquilla Doctor[/bold] - {report.get('summary', '')}") if context := _report_context(report): console.print(context) @@ -683,11 +683,11 @@ def _render_report(report: dict[str, Any]) -> None: return step_column = _step_column(visible_findings) table = Table(show_header=True) - table.add_column("Severity", no_wrap=True) - table.add_column("Impact", no_wrap=True) - table.add_column("Surface", no_wrap=True) - table.add_column("Finding") - table.add_column(step_column) + table.add_column("Severity", overflow="fold") + table.add_column("Impact", overflow="fold") + table.add_column("Surface", overflow="fold") + table.add_column("Finding", overflow="fold") + table.add_column(step_column, overflow="fold") for finding in visible_findings: recovery = "\n".join(_fix_step_text(step) for step in finding.get("fixSteps") or []) restart = " Recovery requires restart." if finding.get("restartRequired") else "" diff --git a/src/opensquilla/cli/providers_cmd.py b/src/opensquilla/cli/providers_cmd.py index c06d41fc70..39c1b34327 100644 --- a/src/opensquilla/cli/providers_cmd.py +++ b/src/opensquilla/cli/providers_cmd.py @@ -33,14 +33,14 @@ def providers_list( print_json(provider_catalog_payload()) return - console = Console(width=200, force_terminal=False) + console = Console(force_terminal=False) table = Table(title="Providers") - table.add_column("provider", no_wrap=True) - table.add_column("label", no_wrap=True) - table.add_column("runtime", no_wrap=True) - table.add_column("requires key", no_wrap=True) - table.add_column("requires base url", no_wrap=True) - table.add_column("default base url") + table.add_column("provider", overflow="fold") + table.add_column("label", overflow="fold") + table.add_column("runtime", overflow="fold") + table.add_column("requires key", overflow="fold") + table.add_column("requires base url", overflow="fold") + table.add_column("default base url", overflow="fold") for s in list_provider_setup_specs(): table.add_row( s.provider_id, @@ -77,16 +77,16 @@ async def _run(client): print_json(payload) return - console = Console(width=180, force_terminal=False) + console = Console(force_terminal=False) table = Table(title="Provider status") - table.add_column("provider", no_wrap=True) - table.add_column("active", no_wrap=True) - table.add_column("configured", no_wrap=True) - table.add_column("buildable", no_wrap=True) - table.add_column("model") - table.add_column("error") + table.add_column("provider", overflow="fold") + table.add_column("active", overflow="fold") + table.add_column("configured", overflow="fold") + table.add_column("buildable", overflow="fold") + table.add_column("model", overflow="fold") + table.add_column("error", overflow="fold") if probe_models: - table.add_column("probe") + table.add_column("probe", overflow="fold") for row in payload.get("providers", []): values = [ str(row.get("providerId") or ""), diff --git a/src/opensquilla/cli/search_cmd.py b/src/opensquilla/cli/search_cmd.py index 16af9299d3..b752528c1e 100644 --- a/src/opensquilla/cli/search_cmd.py +++ b/src/opensquilla/cli/search_cmd.py @@ -65,13 +65,13 @@ def search_list( print_json(search_provider_catalog_payload()) return - console = Console(width=160, force_terminal=False) + console = Console(force_terminal=False) table = Table(title="Search providers") - table.add_column("provider", no_wrap=True) - table.add_column("label", no_wrap=True) - table.add_column("runtime", no_wrap=True) - table.add_column("requires key", no_wrap=True) - table.add_column("env key") + table.add_column("provider", overflow="fold") + table.add_column("label", overflow="fold") + table.add_column("runtime", overflow="fold") + table.add_column("requires key", overflow="fold") + table.add_column("env key", overflow="fold") for spec in list_search_provider_setup_specs(): table.add_row( spec.provider_id, @@ -102,15 +102,15 @@ async def _run(client): print_json(payload) return - console = Console(width=140, force_terminal=False) + console = Console(force_terminal=False) table = Table(title="Search status") - table.add_column("provider", no_wrap=True) - table.add_column("active", no_wrap=True) - table.add_column("configured", no_wrap=True) - table.add_column("buildable", no_wrap=True) - table.add_column("network", no_wrap=True) - table.add_column("fallback") - table.add_column("error") + table.add_column("provider", overflow="fold") + table.add_column("active", overflow="fold") + table.add_column("configured", overflow="fold") + table.add_column("buildable", overflow="fold") + table.add_column("network", overflow="fold") + table.add_column("fallback", overflow="fold") + table.add_column("error", overflow="fold") network_blocked = str(payload.get("networkBlockedReason") or "") table.add_row( str(payload.get("provider") or ""), @@ -218,11 +218,11 @@ async def _run(client): typer.secho(f"Search failed: {message}", fg=typer.colors.RED, err=True) raise typer.Exit(1) - console = Console(width=160, force_terminal=False) + console = Console(force_terminal=False) table = Table(title=f"Search: {query}") - table.add_column("Title") - table.add_column("URL") - table.add_column("Snippet") + table.add_column("Title", overflow="fold") + table.add_column("URL", overflow="fold") + table.add_column("Snippet", overflow="fold") for row in payload.get("results", []): table.add_row( str(row.get("title") or ""), @@ -288,15 +288,15 @@ def _emit_invalid_request(message: str, *, json_output: bool) -> NoReturn: def _print_research_results(query: str, payload: dict[str, Any]) -> None: - console = Console(width=180, force_terminal=False) + console = Console(force_terminal=False) table = Table(title=f"Research search: {query}") - table.add_column("Rank", no_wrap=True) - table.add_column("Title") - table.add_column("Domain", no_wrap=True) - table.add_column("Provider", no_wrap=True) - table.add_column("Fetched", no_wrap=True) - table.add_column("URL") - table.add_column("Excerpt") + table.add_column("Rank", overflow="fold") + table.add_column("Title", overflow="fold") + table.add_column("Domain", overflow="fold") + table.add_column("Provider", overflow="fold") + table.add_column("Fetched", overflow="fold") + table.add_column("URL", overflow="fold") + table.add_column("Excerpt", overflow="fold") for index, row in enumerate(payload.get("results", []), start=1): if not isinstance(row, dict): continue @@ -335,11 +335,11 @@ def search_benchmark( print_json(payload) return - console = Console(width=160, force_terminal=False) + console = Console(force_terminal=False) table = Table(title=f"Search benchmark: {profile}") - table.add_column("Metric") - table.add_column("v1", justify="right") - table.add_column("v2", justify="right") + table.add_column("Metric", overflow="fold") + table.add_column("v1", justify="right", overflow="fold") + table.add_column("v2", justify="right", overflow="fold") for metric in ( "p50_latency_ms", "p95_latency_ms", @@ -465,7 +465,7 @@ def search_configure( ) typer.echo(f"Search provider configured: {provider}") typer.echo(f"Config: {persist.path}") - warning_console = Console(file=sys.stdout, width=160, force_terminal=False) + warning_console = Console(file=sys.stdout, force_terminal=False) for warning in env_reference_warnings(result.config): warning_console.print(warning_panel(warning)) if persist.backup_path: diff --git a/tests/test_cli/test_cli_product_completeness.py b/tests/test_cli/test_cli_product_completeness.py index 0691bb9a34..59759b658d 100644 --- a/tests/test_cli/test_cli_product_completeness.py +++ b/tests/test_cli/test_cli_product_completeness.py @@ -1461,6 +1461,8 @@ def test_channels_status_and_logout_use_existing_rpcs(monkeypatch): def test_channels_status_table_shows_channel_diagnostic(monkeypatch): + # Wide terminal so content assertions are not split by wrapping. + monkeypatch.setenv("COLUMNS", "300") fake = _install_fake_gateway(monkeypatch) fake.rpc_payloads = { "channels.status": { diff --git a/tests/test_cli/test_cli_table_width.py b/tests/test_cli/test_cli_table_width.py new file mode 100644 index 0000000000..d65eace198 --- /dev/null +++ b/tests/test_cli/test_cli_table_width.py @@ -0,0 +1,168 @@ +"""CLI tests for terminal-width-aware table output. + +Regression tests for the hardcoded ``Console(width=...)`` pins that made +every CLI table render wider than the user's terminal regardless of +``COLUMNS``. The tables must follow the terminal width instead. +""" + +from __future__ import annotations + +import re +from functools import partial +from pathlib import Path +from types import SimpleNamespace + +import pytest +from rich.cells import cell_len +from rich.console import Console +from typer.testing import CliRunner + +from opensquilla.cli import agents_cmd, channels_cmd, doctor_cmd, providers_cmd, search_cmd +from opensquilla.cli.main import app +from opensquilla.onboarding.provider_specs import list_provider_setup_specs + +runner = CliRunner() + +_HARDCODED_WIDTH = re.compile(r"Console\([^)]*width=\d+") + + +@pytest.fixture(params=[False, True], ids=["modern", "legacy-windows"]) +def terminal_style(monkeypatch, request) -> None: + # Exercise Windows' safe-box fallback even when running on Linux or macOS. + for module in (agents_cmd, channels_cmd, doctor_cmd, providers_cmd, search_cmd): + monkeypatch.setattr(module, "Console", partial(Console, legacy_windows=request.param)) + + +def _assert_fits(output: str, width: int) -> None: + assert output + assert max(cell_len(line) for line in output.splitlines()) <= width + + +def _column_text(output: str, index: int) -> str: + """Reassemble a displayed column across folded lines, ignoring whitespace.""" + cells = [] + in_body = False + for line in output.splitlines(): + # Windows' safe box uses the same vertical border for header and body. + # Start after the header divider in either box style. + if line.startswith(("┡", "├")): + in_body = True + elif in_body and line.startswith("│"): + cells.append("".join(line.split("│")[index + 1].split())) + assert in_body, output + return "".join(cells) + + +@pytest.mark.parametrize("width", [40, 80]) +def test_providers_list_preserves_ids_within_terminal_width( + tmp_path: Path, monkeypatch, terminal_style, width: int, +) -> None: + monkeypatch.setenv("OPENSQUILLA_GATEWAY_CONFIG_PATH", str(tmp_path / "c.toml")) + monkeypatch.setenv("COLUMNS", str(width)) + result = runner.invoke(app, ["providers", "list"]) + assert result.exit_code == 0 + _assert_fits(result.stdout, width) + assert _column_text(result.stdout, 0) == "".join( + spec.provider_id for spec in list_provider_setup_specs() + ) + + +@pytest.mark.parametrize("width", [40, 80]) +def test_agents_list_preserves_identifiers_and_workspace( + monkeypatch, capsys, terminal_style, width: int, +) -> None: + agent = { + "id": "support-assistant", + "name": "演示助手", + "type": "custom", + "enabled": True, + "model": "example/model-with-a-long-name", + "workspace": "/workspace/project-with-a-long-name", + } + + async def list_agents(**_kwargs): + return [agent] + + monkeypatch.setenv("COLUMNS", str(width)) + monkeypatch.setattr( + agents_cmd, "_load_registry", + lambda _path: (Path("config.toml"), None, SimpleNamespace(list_agents=list_agents)), + ) + agents_cmd.agents_list(json_output=False, config_path=None) + output = capsys.readouterr().out + _assert_fits(output, width) + for index, field in [(0, "id"), (1, "name"), (4, "model"), (5, "workspace")]: + assert _column_text(output, index) == agent[field] + + +@pytest.mark.parametrize("width", [40, 80]) +def test_pairings_preserve_complete_code(monkeypatch, capsys, terminal_style, width: int) -> None: + monkeypatch.setenv("COLUMNS", str(width)) + channels_cmd._render_pairings_table([{ + "pairingCode": "abcdef12", + "senderName": "Example User", + "status": "pending", + "createdAt": "2026-01-01T09:00:00+00:00", + "approvedAt": None, + }], channel="example") + output = capsys.readouterr().out + _assert_fits(output, width) + assert _column_text(output, 0) == "abcdef12" + assert _column_text(output, 1) == "ExampleUser" + assert _column_text(output, 3) == "2026-01-01T09:00:00+00:00" + + +@pytest.mark.parametrize("width", [40, 80]) +def test_doctor_preserves_complete_recovery_command( + monkeypatch, capsys, terminal_style, width: int, +) -> None: + command = ( + "opensquilla providers configure example --api-key YOUR_API_KEY " + "--config /workspace/project-with-a-long-name/config.toml" + ) + monkeypatch.setenv("COLUMNS", str(width)) + doctor_cmd._render_report({ + "summary": "Needs setup", + "findings": [{ + "severity": "error", + "surface": "provider", + "impact": "blocks_ready", + "title": "Provider is not configured", + "detail": "The configured provider requires a key.", + "fixSteps": [{"command": command}], + }], + }) + output = capsys.readouterr().out + _assert_fits(output, width) + assert _column_text(output, 4) == "".join(command.split()) + + +@pytest.mark.parametrize("width", [40, 80]) +def test_research_results_preserve_complete_url( + monkeypatch, capsys, terminal_style, width: int, +) -> None: + url = "https://example.com/research/a-long-result-path" + monkeypatch.setenv("COLUMNS", str(width)) + search_cmd._print_research_results("example", {"results": [{ + "rank": 1, + "title": "Example research result", + "domain": "example.com", + "provider": "example-search", + "fetched": False, + "url": url, + "excerpt": "Example result summary.", + }]}) + output = capsys.readouterr().out + _assert_fits(output, width) + assert _column_text(output, 3) == "example-search" + assert _column_text(output, 5) == url + + +def test_cli_modules_have_no_hardcoded_console_width() -> None: + cli_dir = Path(__file__).resolve().parents[2] / "src" / "opensquilla" / "cli" + offenders = [ + str(path) + for path in sorted(cli_dir.rglob("*.py")) + if _HARDCODED_WIDTH.search(path.read_text(encoding="utf-8")) + ] + assert offenders == [] diff --git a/tests/test_cli/test_doctor_cmd.py b/tests/test_cli/test_doctor_cmd.py index 472877a718..790a8b0c6d 100644 --- a/tests/test_cli/test_doctor_cmd.py +++ b/tests/test_cli/test_doctor_cmd.py @@ -531,6 +531,8 @@ def test_doctor_ready_table_summarizes_without_dumping_ok_findings(monkeypatch) def test_doctor_table_shows_gateway_config_and_agent_context(tmp_path, monkeypatch) -> None: + # Keep this content assertion independent of wrapping inside long temporary paths. + monkeypatch.setenv("COLUMNS", "300") _ReadyGatewayClient.calls = [] target = tmp_path / "custom.toml" target.write_text('host = "127.0.0.1"\nport = 20003\n', encoding="utf-8") @@ -668,6 +670,8 @@ def test_doctor_config_mismatch_keeps_running_findings_scoped_to_running_config( def test_doctor_table_prints_fix_steps(monkeypatch) -> None: + # Wide terminal so content assertions are not split by wrapping. + monkeypatch.setenv("COLUMNS", "300") _FakeGatewayClient.calls = [] monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", _FakeGatewayClient) @@ -679,6 +683,8 @@ def test_doctor_table_prints_fix_steps(monkeypatch) -> None: def test_doctor_degraded_table_exits_zero(monkeypatch) -> None: + # Wide terminal so content assertions are not split by wrapping. + monkeypatch.setenv("COLUMNS", "300") _FakeGatewayClient.calls = [] original_payload = _FakeGatewayClient.payload _FakeGatewayClient.payload = { @@ -721,6 +727,8 @@ def test_doctor_degraded_table_exits_zero(monkeypatch) -> None: def test_doctor_table_explains_non_blocking_error_impact(monkeypatch) -> None: + # Wide terminal so content assertions are not split by wrapping. + monkeypatch.setenv("COLUMNS", "300") _FakeGatewayClient.calls = [] original_payload = _FakeGatewayClient.payload _FakeGatewayClient.payload = { @@ -760,6 +768,8 @@ def test_doctor_table_explains_non_blocking_error_impact(monkeypatch) -> None: def test_doctor_table_uses_readiness_impact_to_select_action_findings(monkeypatch) -> None: + # Wide terminal so content assertions are not split by wrapping. + monkeypatch.setenv("COLUMNS", "300") _FakeGatewayClient.calls = [] original_payload = _FakeGatewayClient.payload _FakeGatewayClient.payload = { @@ -821,6 +831,8 @@ def test_doctor_table_uses_readiness_impact_to_select_action_findings(monkeypatc def test_doctor_table_uses_readiness_impact_to_label_optional_steps(monkeypatch) -> None: + # Wide terminal so content assertions are not split by wrapping. + monkeypatch.setenv("COLUMNS", "300") _FakeGatewayClient.calls = [] original_payload = _FakeGatewayClient.payload _FakeGatewayClient.payload = { @@ -860,6 +872,8 @@ def test_doctor_table_uses_readiness_impact_to_label_optional_steps(monkeypatch) def test_doctor_table_hides_non_actionable_findings_when_attention_is_needed(monkeypatch) -> None: + # Wide terminal so content assertions are not split by wrapping. + monkeypatch.setenv("COLUMNS", "300") _FakeGatewayClient.calls = [] original_payload = _FakeGatewayClient.payload _FakeGatewayClient.payload = { @@ -910,6 +924,8 @@ def test_doctor_table_hides_non_actionable_findings_when_attention_is_needed(mon def test_doctor_table_labels_info_only_steps_as_optional(monkeypatch) -> None: + # Wide terminal so content assertions are not split by wrapping. + monkeypatch.setenv("COLUMNS", "300") _FakeGatewayClient.calls = [] original_payload = _FakeGatewayClient.payload _FakeGatewayClient.payload = { @@ -1294,6 +1310,8 @@ def test_refresh_report_readiness_labels_operator_action_items() -> None: def test_doctor_appends_local_tui_finding_when_opentui_unavailable(monkeypatch) -> None: + # Wide terminal so content assertions are not split by wrapping. + monkeypatch.setenv("COLUMNS", "300") from opensquilla.cli.tui.renderers.selection import ( ChatUiFallback, ChatUiSelection, diff --git a/tests/test_cli/test_providers_cmd.py b/tests/test_cli/test_providers_cmd.py index 1ca217e1f3..63ef4ef9e9 100644 --- a/tests/test_cli/test_providers_cmd.py +++ b/tests/test_cli/test_providers_cmd.py @@ -22,6 +22,8 @@ def test_providers_list_shows_all_supported(tmp_path: Path, monkeypatch): def test_providers_list_marks_unsupported(tmp_path, monkeypatch): + # Wide terminal so content assertions are not split by wrapping. + monkeypatch.setenv("COLUMNS", "300") monkeypatch.setenv("OPENSQUILLA_GATEWAY_CONFIG_PATH", str(tmp_path / "c.toml")) result = runner.invoke(app, ["providers", "list"]) assert result.exit_code == 0 @@ -105,6 +107,8 @@ def test_providers_configure_vllm_requires_base_url(tmp_path, monkeypatch): def test_providers_status_probe_column_surfaces_failure_kind(monkeypatch): + # Wide terminal so content assertions are not split by wrapping. + monkeypatch.setenv("COLUMNS", "300") monkeypatch.setattr( "opensquilla.cli.providers_cmd.run_gateway_sync", lambda *_args, **_kwargs: { diff --git a/tests/test_cli/test_search_query_cmd.py b/tests/test_cli/test_search_query_cmd.py index 838d1469b9..546ef31b99 100644 --- a/tests/test_cli/test_search_query_cmd.py +++ b/tests/test_cli/test_search_query_cmd.py @@ -77,6 +77,8 @@ async def fake_run_canonical_web_search( def test_search_query_web_search_text_output_renders_results(monkeypatch): + # Wide terminal so content assertions are not split by wrapping. + monkeypatch.setenv("COLUMNS", "300") async def fake_run_canonical_web_search( options: SearchOptions, **kwargs: object, diff --git a/tests/test_engine/test_local_checkpoint_retention.py b/tests/test_engine/test_local_checkpoint_retention.py index bfdd227b67..a2896dc8c3 100644 --- a/tests/test_engine/test_local_checkpoint_retention.py +++ b/tests/test_engine/test_local_checkpoint_retention.py @@ -7,6 +7,7 @@ import pytest +from opensquilla import token_estimation from opensquilla.engine import Agent, AgentConfig from opensquilla.engine.agent import _CompactionSummaryMessage, _LiveTurnCheckpointMessage from opensquilla.engine.request_window import compact_request_window_tools @@ -24,6 +25,20 @@ ) +class _CharacterBudgetEncoding: + def encode(self, text: str, *, disallowed_special: tuple[object, ...]) -> range: + del disallowed_special + return range((len(text) + 7) // 8) + + +@pytest.fixture(autouse=True) +def _character_pressure_estimator(monkeypatch): + # These retention boundaries exercise character pressure with token headroom. + # Optional tokenizer downloads must not turn them into token-pressure cases. + encoding = _CharacterBudgetEncoding() + monkeypatch.setattr(token_estimation, "_get_encoding", lambda: encoding) + + def tool_round(index): return [ Message( @@ -231,6 +246,21 @@ async def test_user_written_summary_prefix_cannot_claim_checkpoint_priority(): assert '"id": "read-1"' in wire +async def test_conservative_fallback_preserves_checkpoint_and_all_receipts(monkeypatch): + monkeypatch.setattr(token_estimation, "_get_encoding", lambda: None) + agent = make_agent() + old = checkpoint(agent) + source = [Message(role="user", content=CURRENT), *old, *tool_round(1), *tool_round(2)] + + outcome, wire = await recover(agent, source) + + assert any(message is old[0] for message in outcome.messages) + assert wire.count(SUMMARY) == 1 + receipts = agent._live_turn_execution_receipts(outcome.messages[1:]) + assert receipts is not None + assert [receipt["tool_call_id"] for receipt in receipts] == ["read-0", "read-1", "read-2"] + + def test_checkpoint_provenance_survives_copy_and_tool_pruning_without_wire_fields(): agent = make_agent() checkpoint_message = checkpoint(agent)[0]