mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
01e10ad693
* feat(conventions): standard schema models + effective-map merge * feat(orchestrator): park provider on persistent server overload (529/500) A 429 rate limit already parks a provider — queue its spawns, probe until it recovers — but a persistent 529/500/503 overload had no such break: the run died and the orchestrator crash-retried straight back into the overload, burning tokens in a respawn loop. Generalize the park to provider-unavailability. On a non-graceful Anthropic agent exit, match the API's overload markers (overloaded_error / internal_server_error / "API Error: 5xx") against the dead container's own output and park the provider with kind="overloaded"; the existing spawn gate already queues any parked provider, and the probe-resume loop revives the task when it recovers. Grok keeps its exit-75 path; both now route through one _park_provider_unavailable helper. Markers are kept specific so an agent that merely writes about HTTP 500/529 can't trip the break. Fix the recovery probe to require a 2xx: it treated any non-429 as recovered, so a probe that itself got a 529 would have resumed agents straight back into the overload — wrong for the new path and for a 429 that lifts into a 5xx. Gated by ROBOCO_OVERLOAD_BREAK_ENABLED (default on; off => crash-retry). --------- Co-authored-by: Renn F <rennf93@users.noreply.github.com>
100 lines
3.1 KiB
Python
100 lines
3.1 KiB
Python
"""Schema-model + YAML-parse tests for the architectural-conventions standard."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from roboco.foundation.policy.conventions.models import (
|
|
BUILTIN_RULES,
|
|
ConventionsParseError,
|
|
ConventionsStandard,
|
|
CustomRule,
|
|
Module,
|
|
Rule,
|
|
Waiver,
|
|
)
|
|
|
|
_VALID_YAML = """
|
|
version: 1
|
|
languages: [python, typescript]
|
|
modules:
|
|
- path: app/routers
|
|
purpose: HTTP routes
|
|
forbidden: [model, helper]
|
|
- path: app/models
|
|
purpose: Pydantic / ORM models
|
|
rules:
|
|
no_models_in_routers: { level: block }
|
|
no_inline_comments: { level: warn }
|
|
custom:
|
|
- id: no-print
|
|
pattern: '\\bprint\\('
|
|
message: use the logger
|
|
level: warn
|
|
languages: [python]
|
|
waivers:
|
|
- path: app/routers/legacy.py
|
|
rule: no_models_in_routers
|
|
reason: extraction tracked separately
|
|
"""
|
|
|
|
|
|
def test_valid_yaml_parses_to_standard() -> None:
|
|
std = ConventionsStandard.parse_yaml(_VALID_YAML)
|
|
assert std.version == 1
|
|
assert std.languages == ["python", "typescript"]
|
|
assert std.modules[0].path == "app/routers"
|
|
assert std.modules[0].forbidden == ["model", "helper"]
|
|
assert std.rules["no_models_in_routers"].level == "block"
|
|
assert std.rules["no_models_in_routers"].name == "no_models_in_routers"
|
|
assert std.custom[0].id == "no-print"
|
|
assert std.custom[0].languages == ["python"]
|
|
assert std.waivers[0].rule == "no_models_in_routers"
|
|
|
|
|
|
def test_empty_yaml_yields_default_standard() -> None:
|
|
std = ConventionsStandard.parse_yaml("")
|
|
assert std == ConventionsStandard()
|
|
assert std.version == 1
|
|
|
|
|
|
def test_unknown_rule_level_raises_parse_error() -> None:
|
|
with pytest.raises(ConventionsParseError):
|
|
ConventionsStandard.parse_yaml(
|
|
"rules:\n no_models_in_routers: { level: explode }\n"
|
|
)
|
|
|
|
|
|
def test_malformed_yaml_raises_parse_error() -> None:
|
|
with pytest.raises(ConventionsParseError):
|
|
ConventionsStandard.parse_yaml("modules: [unterminated\n")
|
|
|
|
|
|
def test_non_mapping_top_level_raises_parse_error() -> None:
|
|
with pytest.raises(ConventionsParseError):
|
|
ConventionsStandard.parse_yaml("- just\n- a\n- list\n")
|
|
|
|
|
|
def test_unknown_definition_kind_in_forbidden_raises() -> None:
|
|
with pytest.raises(ConventionsParseError):
|
|
ConventionsStandard.parse_yaml(
|
|
"modules:\n - path: x\n purpose: y\n forbidden: [wizard]\n"
|
|
)
|
|
|
|
|
|
def test_builtin_rules_cover_the_org_defaults() -> None:
|
|
assert BUILTIN_RULES["no_models_in_routers"] == "block"
|
|
assert BUILTIN_RULES["no_helpers_in_routers"] == "block"
|
|
assert BUILTIN_RULES["no_lint_suppressions"] == "block"
|
|
assert BUILTIN_RULES["no_inline_comments"] == "warn"
|
|
|
|
|
|
def test_models_construct_directly() -> None:
|
|
mod = Module(path="app/services", purpose="logic", forbidden=["route"])
|
|
assert mod.forbidden == ["route"]
|
|
rule = Rule(name="no_print", level="warn")
|
|
assert rule.level == "warn"
|
|
custom = CustomRule(id="x", pattern="y", message="z", level="block")
|
|
assert custom.languages == []
|
|
waiver = Waiver(path="a.py", rule="no_models_in_routers", reason="r")
|
|
assert waiver.path == "a.py"
|