Dhyaneesh/pwned

1from __future__ import annotations2 3import asyncio4import json5 6import jsonschema7import pytest8from openenv.core.env_server.serialization import serialize_observation9 10from tests.reference_spec._transport import get_json_from_app, post_json_to_app, run_websocket_script11 12 13EXPECTED_PUBLIC_OBSERVATION_KEYS = {14    "raw_output",15    "notebook",16    "suite_name",17    "task_name",18    "difficulty",19    "objective",20    "procedural_profile",21    "mission_brief",22    "submission_schema",23    "steps_remaining",24    "evidence_log",25    "report_feedback",26}27 28 29class _RecordingRubric:30    def __init__(self) -> None:31        self.reset_calls = 032 33    def reset(self) -> None:34        self.reset_calls += 135 36 37def _mark_transformed(observation):38    observation.raw_output = f"TRANSFORMED::{observation.raw_output}"39    observation.metadata = {40        **observation.metadata,41        "transformed": True,42    }43    return observation44 45 46def test_environment_supports_transform_and_rubric_on_reset_and_step() -> None:47    from Pwned.models import PwnedAction48    from Pwned.server.Pwned_environment import PwnedEnvironment49 50    rubric = _RecordingRubric()51    env = PwnedEnvironment(transform=_mark_transformed, rubric=rubric)52 53    reset_observation = env.reset(seed=11, episode_id="layers-transform", level=2)54    step_observation = env.step(PwnedAction(command="whoami"))55 56    assert rubric.reset_calls == 157    assert reset_observation.raw_output.startswith("TRANSFORMED::")58    assert step_observation.raw_output.startswith("TRANSFORMED::user")59    assert reset_observation.metadata == {}60    assert step_observation.metadata == {}61 62 63def test_reset_is_atomic_when_transform_raises() -> None:64    from Pwned.server.Pwned_environment import PwnedEnvironment65 66    class Boom(RuntimeError):67        pass68 69    def explode(_observation):70        raise Boom("boom")71 72    rubric = _RecordingRubric()73    env = PwnedEnvironment(rubric=rubric)74    original_state = env.state.model_dump(mode="json")75    env._transform = explode76 77    try:78        env.reset(seed=11, episode_id="layers-reset-atomic", level=2)79    except Boom:80        pass81    else:  # pragma: no cover82        raise AssertionError("reset() should propagate transform failures")83 84    assert env.state.model_dump(mode="json") == original_state85    assert rubric.reset_calls == 086 87 88def test_reset_commits_state_before_rubric_reset() -> None:89    from Pwned.server.Pwned_environment import PwnedEnvironment90 91    env = PwnedEnvironment()92    seen: dict[str, object] = {}93 94    def record_reset() -> None:95        seen["level"] = env.state.level96 97    env._reset_rubric = record_reset  # type: ignore[method-assign]98    env.reset(seed=11, episode_id="layers-reset-order", level=2)99 100    assert seen["level"] == 2101 102 103def test_environment_exposes_last_reset_and_last_step_info_for_trusted_inprocess_callers() -> None:104    from Pwned.models import PwnedAction105    from Pwned.server.Pwned_environment import PwnedEnvironment106 107    env = PwnedEnvironment()108    env.reset(seed=11, episode_id="layers-local-info", level=2)109    step_observation = env.step(PwnedAction(command="whoami"))110 111    assert step_observation.done is False112    assert env.last_reset_info["level"] == 2113    assert env.last_reset_info["step_budget"] == 20114    assert env.last_info["step"] == 1115    assert env.last_info["risk_pool"] == env.state.cumulative_risk_pool116    assert env.last_info["termination_cause"] is None117 118 119def test_environment_exposes_typed_local_reset_and_step_helpers() -> None:120    from Pwned.models import PwnedAction121    from Pwned.server.Pwned_environment import PwnedEnvironment122 123    env = PwnedEnvironment()124    reset_observation, reset_info = env.reset_result(125        seed=11,126        episode_id="layers-local-typed-result",127        config={"level": 2, "task_mode": "unguided"},128    )129    step_result = env.step_result(PwnedAction(command="whoami"))130 131    assert reset_observation.notebook.step == 0132    assert reset_info["level"] == 2133    assert reset_info["step_budget"] == 20134    assert step_result.observation.raw_output == "user"135    assert step_result.reward == -0.01136    assert step_result.done is False137    assert step_result.info["termination_cause"] is None138    assert step_result.info["step"] == 1139 140 141def test_reset_exposes_named_mission_brief_and_schema() -> None:142    from Pwned.server.Pwned_environment import PwnedEnvironment143 144    env = PwnedEnvironment()145    obs = env.reset(seed=7, episode_id="alert-triage", task_name="alert_triage")146 147    assert obs.task_name == "alert_triage"148    assert "initially compromised host" in obs.mission_brief.lower()149    assert "cited_evidence_ids" in obs.submission_schema["required"]150    assert obs.steps_remaining == env.state.step_budget151 152 153def test_reset_returns_pentesting_observation_with_objective_and_profile() -> None:154    from Pwned.server.Pwned_environment import PwnedEnvironment155 156    env = PwnedEnvironment()157    obs = env.reset(158        seed=7,159        episode_id="pentesting-reset",160        suite_name="pentesting",161        task_name="pivot_path",162        difficulty="medium",163    )164 165    assert obs.suite_name == "pentesting"166    assert obs.task_name == "pivot_path"167    assert obs.difficulty == "medium"168    assert obs.objective is not None169    assert obs.objective.status == "in_progress"170    assert obs.procedural_profile is not None171    assert obs.procedural_profile.source == "preset"172 173 174def test_reset_derives_pentesting_suite_from_task_name_when_suite_is_omitted() -> None:175    from Pwned.server.Pwned_environment import PwnedEnvironment176 177    env = PwnedEnvironment()178    obs = env.reset(179        seed=7,180        episode_id="pentesting-derived-suite",181        task_name="pivot_path",182        difficulty="easy",183    )184 185    assert obs.suite_name == "pentesting"186    assert obs.task_name == "pivot_path"187    assert obs.difficulty == "easy"188 189 190def test_secops_reset_ignores_pentesting_only_fields() -> None:191    from Pwned.server.Pwned_environment import PwnedEnvironment192 193    env = PwnedEnvironment()194    obs = env.reset(195        seed=7,196        episode_id="secops-ignore-pentesting-fields",197        suite_name="secops",198        task_name="alert_triage",199        difficulty="hard",200        pentesting_overrides={"required_pivot_count": 3},201    )202 203    assert obs.suite_name == "secops"204    assert obs.task_name == "alert_triage"205    assert obs.difficulty is None206    assert obs.procedural_profile is None207    assert obs.objective is None208 209 210def test_reset_canonicalizes_task_name_before_persisting_hidden_state() -> None:211    from Pwned.server.Pwned_environment import PwnedEnvironment212 213    env = PwnedEnvironment()214    obs = env.reset(seed=7, episode_id="alert-triage-canonical", task_name=" alert_triage ")215 216    assert obs.task_name == "alert_triage"217    assert env.state.current_task_name == "alert_triage"218 219 220def test_observation_canonicalizes_preset_hidden_task_name() -> None:221    from Pwned.models import PwnedAction222    from Pwned.server.Pwned_environment import PwnedEnvironment223 224    env = PwnedEnvironment()225    env.reset(seed=7, episode_id="alert-triage-step-canonical", task_name="alert_triage")226    object.__setattr__(env._state, "current_task_name", " alert_triage ")227 228    obs = env.step(PwnedAction(command="whoami"))229 230    assert obs.task_name == "alert_triage"231    assert env.state.current_task_name == "alert_triage"232 233 234def test_reset_honors_nested_config_payload() -> None:235    from Pwned.server.Pwned_environment import PwnedEnvironment236 237    env = PwnedEnvironment()238    observation = env.reset(239        seed=0,240        episode_id="layers-nested-config",241        config={"level": 4, "task_mode": "subtask"},242    )243 244    assert observation.notebook.step == 0245    assert env.state.level == 4246    assert env.state.task_mode == "subtask"247    assert env.last_reset_info["level"] == 4248    assert env.last_reset_info["task_mode"] == "subtask"249    assert env.last_reset_info["step_budget"] == env.state.step_budget250 251 252def test_local_api_wrappers_preserve_standard_dump_kwargs() -> None:253    from Pwned.contracts import HackerNotebook, PwnedObservation, PwnedStepResult254 255    step_result = PwnedStepResult(256        observation=PwnedObservation(257            raw_output="user",258            recent_turns=[],259            notebook=HackerNotebook(),260            phase="pwned",261        ),262        reward=None,263        done=False,264        info={},265    )266 267    payload = step_result.model_dump(exclude_none=True)268    payload_json = step_result.model_dump_json(indent=2, exclude_none=True)269 270    assert "reward" not in payload271    assert '\n  "done": false' in payload_json272 273 274def test_step_raises_after_terminal_episode() -> None:275    from Pwned.models import PwnedAction276    from Pwned.server.Pwned_environment import PwnedEnvironment277 278    env = PwnedEnvironment()279    terminal_observation = env.step(PwnedAction(command="state()"))280 281    assert terminal_observation.done is True282 283    with pytest.raises(RuntimeError, match="Episode is done"):284        env.step(PwnedAction(command="whoami"))285 286 287def test_step_is_atomic_when_transform_raises() -> None:288    from Pwned.models import PwnedAction289    from Pwned.server.Pwned_environment import PwnedEnvironment290 291    class Boom(RuntimeError):292        pass293 294    env = PwnedEnvironment()295    env.reset(seed=11, episode_id="layers-step-atomic", level=2)296    original_state = env.state.model_dump(mode="json")297 298    def explode(_observation):299        raise Boom("boom")300 301    env._transform = explode302 303    with pytest.raises(Boom, match="boom"):304        env.step(PwnedAction(command="whoami"))305 306    assert env.state.model_dump(mode="json") == original_state307 308 309def test_serialize_observation_excludes_internal_metadata_from_public_transport() -> None:310    from Pwned.contracts import HackerNotebook, PwnedObservation311 312    payload = serialize_observation(313        PwnedObservation(314            raw_output="user",315            recent_turns=[],316            notebook=HackerNotebook(),317            task_name="alert_triage",318            mission_brief="Identify the initially compromised host.",319            submission_schema={320                "type": "object",321                "properties": {},322                "required": ["cited_evidence_ids"],323                "additionalProperties": False,324            },325            steps_remaining=15,326            evidence_log=[],327            phase="pwned",328            done=False,329            reward=0.0,330            metadata={"canonical_command": "whoami"},331        )332    )333 334    assert payload["reward"] == 0.0335    assert payload["done"] is False336    assert set(payload["observation"]) == EXPECTED_PUBLIC_OBSERVATION_KEYS337 338 339def test_http_transport_serves_mission_safe_state_and_public_schema() -> None:340    from Pwned.server.app import app341 342    missing_state_status, missing_state_payload = asyncio.run(get_json_from_app(app, "/state"))343    reset_status, reset_payload = asyncio.run(344        post_json_to_app(345            app,346            "/reset",347            {"seed": 3, "episode_id": "layers-http-state", "task_name": "alert_triage"},348        )349    )350    state_status, state_payload = asyncio.run(351        get_json_from_app(app, f"/state?session_id={reset_payload['session_id']}")352    )353    schema_status, schema_payload = asyncio.run(get_json_from_app(app, "/schema"))354 355    assert missing_state_status == 409356    assert missing_state_payload["detail"]["code"] == "SESSION_NOT_FOUND"357    assert reset_status == 200358    assert state_status == 200359    assert state_payload["session_id"] == reset_payload["session_id"]360    assert state_payload["status"] == "active"361    assert set(state_payload["observation"]) == EXPECTED_PUBLIC_OBSERVATION_KEYS362    assert schema_status == 200363    assert "oneOf" in schema_payload["action"]364    assert '"metadata"' not in json.dumps(schema_payload["action"])365    assert set(schema_payload["observation"]["properties"]) == EXPECTED_PUBLIC_OBSERVATION_KEYS366    assert set(schema_payload["state"]["properties"]) == {367        "session_id",368        "status",369        "done",370        "observation",371    }372 373 374def test_http_reset_fallback_returns_public_step_result_payload() -> None:375    from Pwned.server.app import app376 377    status_code, payload = asyncio.run(378        asyncio.wait_for(379            post_json_to_app(380                app,381                "/reset",382                {"seed": 3, "episode_id": "layers-http-reset", "level": 1},383            ),384            timeout=1.0,385        )386    )387 388    assert status_code == 200389    assert set(payload) == {"session_id", "observation", "reward", "done"}390    assert payload["session_id"].startswith("sess_")391    assert set(payload["observation"]) == EXPECTED_PUBLIC_OBSERVATION_KEYS392 393 394def test_http_reset_fallback_accepts_documented_public_config_payload() -> None:395    from Pwned.server.app import app396 397    status_code, payload = asyncio.run(398        asyncio.wait_for(399            post_json_to_app(400                app,401                "/reset",402                {403                    "seed": 3,404                    "episode_id": "layers-http-reset-config",405                    "config": {"task_name": "alert_triage"},406                },407            ),408            timeout=1.0,409        )410    )411 412    assert status_code == 200413    assert set(payload) == {"session_id", "observation", "reward", "done"}414    assert payload["observation"]["notebook"]["step"] == 0415    assert payload["observation"]["task_name"] == "alert_triage"416    assert set(payload["observation"]) == EXPECTED_PUBLIC_OBSERVATION_KEYS417 418 419def test_http_reset_rejects_unknown_task_name() -> None:420    from Pwned.server.app import app421 422    status_code, payload = asyncio.run(423        post_json_to_app(app, "/reset", {"seed": 1, "task_name": "not-a-task"})424    )425 426    assert status_code == 422427    assert payload["detail"]428 429 430def test_http_step_fallback_accepts_bare_pwned_action_payload() -> None:431    from Pwned.server.app import app432 433    reset_status, reset_payload = asyncio.run(434        post_json_to_app(435            app,436            "/reset",437            {"seed": 3, "episode_id": "layers-http-step", "task_name": "alert_triage"},438        )439    )440 441    assert reset_status == 200442 443    status_code, payload = asyncio.run(444        asyncio.wait_for(445            post_json_to_app(446                app,447                "/step",448                {"session_id": reset_payload["session_id"], "command": "whoami"},449            ),450            timeout=1.0,451        )452    )453 454    assert status_code == 200455    assert set(payload) == {"session_id", "observation", "reward", "done"}456    assert payload["session_id"] == reset_payload["session_id"]457    assert payload["observation"]["raw_output"] == "user"458    assert set(payload["observation"]) == EXPECTED_PUBLIC_OBSERVATION_KEYS459 460 461def test_http_step_without_session_id_returns_conflict() -> None:462    from Pwned.server.app import app463 464    status_code, payload = asyncio.run(post_json_to_app(app, "/step", {"command": "whoami"}))465 466    assert status_code == 409467    assert payload["detail"]["code"] == "SESSION_NOT_FOUND"468 469 470def test_client_step_payload_preserves_command_mode_wire_shape() -> None:471    from Pwned.client import PwnedEnv472    from Pwned.models import PwnedAction473 474    client = object.__new__(PwnedEnv)475 476    payload = client._step_payload(PwnedAction(command="whoami"))477 478    assert payload == {"command": "whoami"}479 480 481def test_client_step_payload_serializes_submit_report_mode() -> None:482    from Pwned.client import PwnedEnv483    from Pwned.models import PwnedAction484 485    client = object.__new__(PwnedEnv)486    action = PwnedAction.model_validate(487        {488            "kind": "submit_report",489            "report": {490                "task_name": "alert_triage",491                "verdict": "confirmed",492                "confidence": 0.8,493                "cited_evidence_ids": ["ev-0001"],494                "initial_compromised_host": "10.0.0.10",495                "suspected_entry_vector": "stolen ssh key",496            },497        }498    )499 500    payload = client._step_payload(action)501 502    assert payload == {503        "kind": "submit_report",504        "report": {505            "task_name": "alert_triage",506            "verdict": "confirmed",507            "confidence": 0.8,508            "cited_evidence_ids": ["ev-0001"],509            "initial_compromised_host": "10.0.0.10",510            "suspected_entry_vector": "stolen ssh key",511        },512    }513 514 515def test_submit_report_step_is_explicitly_handled_when_command_is_absent() -> None:516    from Pwned.models import PwnedAction517    from Pwned.server.Pwned_environment import PwnedEnvironment518 519    env = PwnedEnvironment()520    env.reset(seed=7, episode_id="submit-report-placeholder", task_name="alert_triage")521 522    observation = env.step(523        PwnedAction.model_validate(524            {525                "kind": "submit_report",526                "report": {527                    "task_name": "alert_triage",528                    "verdict": "confirmed",529                    "confidence": 0.8,530                    "cited_evidence_ids": ["ev-0001"],531                    "initial_compromised_host": "10.0.0.10",532                    "suspected_entry_vector": "stolen ssh key",533                },534            }535        )536    )537 538    assert observation.done is True539    assert "report" in observation.raw_output.lower()540    assert observation.report_feedback is not None541    assert env.last_info["action_kind"] == "submit_report"542    assert env.last_info["report_status"] == "graded"543 544 545def test_http_step_fallback_accepts_submit_report_payload() -> None:546    from Pwned.server.app import app547 548    reset_status, reset_payload = asyncio.run(549        post_json_to_app(550            app,551            "/reset",552            {"seed": 7, "episode_id": "layers-http-report", "task_name": "alert_triage"},553        )554    )555 556    assert reset_status == 200557 558    status_code, payload = asyncio.run(559        asyncio.wait_for(560            post_json_to_app(561                app,562                "/step",563                {564                    "session_id": reset_payload["session_id"],565                    "kind": "submit_report",566                    "report": {567                        "task_name": "alert_triage",568                        "verdict": "confirmed",569                        "confidence": 0.8,570                        "cited_evidence_ids": ["ev-0001"],571                        "initial_compromised_host": "10.0.0.10",572                        "suspected_entry_vector": "stolen ssh key",573                    },574                },575            ),576            timeout=1.0,577        )578    )579 580    assert status_code == 200581    assert payload["session_id"] == reset_payload["session_id"]582    assert payload["done"] is True583    assert payload["observation"]["report_feedback"] is not None584    assert "report" in payload["observation"]["raw_output"].lower()585 586 587def test_public_schema_payloads_are_valid_json_schema_documents() -> None:588    from Pwned.server.app import app589 590    status_code, payload = asyncio.run(get_json_from_app(app, "/schema"))591 592    assert status_code == 200593    for key in ("action", "observation", "state"):594        jsonschema.validators.validator_for(payload[key]).check_schema(payload[key])595 596 597def test_public_action_schema_matches_the_accepted_two_mode_contract() -> None:598    from Pwned.server.app import app599 600    status_code, payload = asyncio.run(get_json_from_app(app, "/schema"))601 602    assert status_code == 200603    action_schema = payload["action"]604 605    jsonschema.validate({"command": "whoami"}, action_schema)606    jsonschema.validate({"kind": "command", "command": "whoami"}, action_schema)607    jsonschema.validate(608        {609            "kind": "submit_report",610            "report": {611                "task_name": "alert_triage",612                "verdict": "confirmed",613                "confidence": 0.8,614                "cited_evidence_ids": ["ev-0001"],615                "initial_compromised_host": "10.0.0.10",616                "suspected_entry_vector": "stolen ssh key",617            },618        },619        action_schema,620    )621 622 623@pytest.mark.parametrize(624    "invalid_payload",625    [626        {"command": "   "},627        {"kind": "command", "command": "   "},628    ],629)630def test_public_action_schema_rejects_whitespace_only_commands(631    invalid_payload: dict[str, object],632) -> None:633    from Pwned.server.app import app634 635    status_code, payload = asyncio.run(get_json_from_app(app, "/schema"))636 637    assert status_code == 200638    with pytest.raises(jsonschema.ValidationError):639        jsonschema.validate(invalid_payload, payload["action"])640 641 642@pytest.mark.parametrize(643    "invalid_payload",644    [645        {"kind": "command"},646        {"kind": "submit_report"},647        {648            "report": {649                "task_name": "alert_triage",650                "verdict": "confirmed",651                "confidence": 0.8,652                "cited_evidence_ids": ["ev-0001"],653                "initial_compromised_host": "10.0.0.10",654                "suspected_entry_vector": "stolen ssh key",655            }656        },657    ],658)659def test_public_action_schema_rejects_invalid_mode_shapes(660    invalid_payload: dict[str, object],661) -> None:662    from Pwned.server.app import app663 664    status_code, payload = asyncio.run(get_json_from_app(app, "/schema"))665 666    assert status_code == 200667    with pytest.raises(jsonschema.ValidationError):668        jsonschema.validate(invalid_payload, payload["action"])669 670 671def test_websocket_transport_omits_metadata_and_returns_public_state_snapshots() -> None:672    from Pwned.server.app import app673 674    sent_messages = asyncio.run(675        run_websocket_script(676            app,677            [678                {679                    "type": "reset",680                    "data": {"seed": 11, "episode_id": "layers-wire", "level": 2},681                },682                {"type": "step", "data": {"command": "nmap 10.1.0.10"}},683                {"type": "state"},684                {"type": "close"},685            ],686            expected_message_count=4,687        )688    )689 690    reset_message = json.loads(sent_messages[1]["text"])691    step_message = json.loads(sent_messages[2]["text"])692    state_message = json.loads(sent_messages[3]["text"])693 694    assert reset_message["type"] == "observation"695    assert step_message["type"] == "observation"696    assert reset_message["data"]["session_id"].startswith("sess_")697    assert step_message["data"]["session_id"] == reset_message["data"]["session_id"]698    assert "metadata" not in reset_message["data"]["observation"]699    assert "metadata" not in step_message["data"]["observation"]700    assert set(reset_message["data"]["observation"]) == EXPECTED_PUBLIC_OBSERVATION_KEYS701    assert set(step_message["data"]["observation"]) == EXPECTED_PUBLIC_OBSERVATION_KEYS702    assert state_message["type"] == "state"703    assert state_message["data"]["session_id"] == reset_message["data"]["session_id"]704    assert state_message["data"]["status"] == "active"705    assert set(state_message["data"]["observation"]) == EXPECTED_PUBLIC_OBSERVATION_KEYS706 707 708def test_openapi_schema_advertises_public_wire_types() -> None:709    from Pwned.server.app import app710 711    schema = app.openapi()712 713    step_resp = schema["paths"]["/step"]["post"]["responses"]["200"]["content"]["application/json"]["schema"]714    assert step_resp.get("$ref", "").endswith("PwnedPublicStepResult"), (715        f"/step 200 response schema should reference PwnedPublicStepResult, got: {step_resp}"716    )717 718    reset_resp = schema["paths"]["/reset"]["post"]["responses"]["200"]["content"]["application/json"]["schema"]719    assert reset_resp.get("$ref", "").endswith("PwnedPublicStepResult"), (720        f"/reset 200 response schema should reference PwnedPublicStepResult, got: {reset_resp}"721    )722 723    obs_schema = schema["components"]["schemas"]["PwnedPublicObservation"]724    props = obs_schema.get("properties", {})725    assert "done" not in props, f"PwnedPublicObservation must not expose 'done': {props}"726    assert "reward" not in props, f"PwnedPublicObservation must not expose 'reward': {props}"727    assert "metadata" not in props, f"PwnedPublicObservation must not expose 'metadata': {props}"728    assert set(props) == EXPECTED_PUBLIC_OBSERVATION_KEYS, (729        "The canonical public observation schema must expose the benchmark observation shape. "730        f"props={props}"731    )732 733    state_resp = schema["paths"]["/state"]["get"]["responses"]["200"]["content"]["application/json"]["schema"]734    assert state_resp.get("$ref", "").endswith("PwnedPublicState"), (735        f"/state 200 response schema should reference PwnedPublicState, got: {state_resp}"736    )737 738 739def test_openapi_schema_contains_no_dangling_defs_refs() -> None:740    from Pwned.server.app import app741 742    schema = app.openapi()743 744    assert "#/$defs/" not in json.dumps(schema)745 746 747def test_openapi_reset_request_schema_advertises_pwned_runtime_fields() -> None:748    from Pwned.server.app import app749 750    schema = app.openapi()751 752    reset_req = schema["paths"]["/reset"]["post"]["requestBody"]["content"]["application/json"]["schema"]753    assert reset_req.get("$ref", "").endswith("PwnedPublicResetRequest"), (754        f"/reset request schema should reference PwnedPublicResetRequest, got: {reset_req}"755    )756 757    request_schema = schema["components"]["schemas"]["PwnedPublicResetRequest"]758    props = request_schema.get("properties", {})759    assert "seed" in props760    assert "episode_id" in props761    assert "level" in props762    assert "task_name" in props763    assert "task_mode" in props764    assert "config" in props765    assert "resolved_seed" not in props766    assert request_schema.get("additionalProperties") is False767    assert props["config"].get("additionalProperties") is False768    assert "task_name" in props["config"]["properties"]769 770 771def test_openapi_step_request_schema_advertises_session_aware_public_step_payload() -> None:772    from Pwned.server.app import app773 774    schema = app.openapi()775 776    step_req = schema["paths"]["/step"]["post"]["requestBody"]["content"]["application/json"]["schema"]777    assert step_req.get("$ref", "").endswith("PwnedPublicStepRequest"), (778        f"/step request schema should reference PwnedPublicStepRequest, got: {step_req}"779    )780 781    action_schema = schema["components"]["schemas"]["PwnedPublicStepRequest"]782    jsonschema.validate({"session_id": "sess_123", "command": "whoami"}, action_schema)783    jsonschema.validate(784        {"session_id": "sess_123", "kind": "command", "command": "whoami"},785        action_schema,786    )787    with pytest.raises(jsonschema.ValidationError):788        jsonschema.validate({"session_id": "sess_123", "kind": "command"}, action_schema)789    with pytest.raises(jsonschema.ValidationError):790        jsonschema.validate({"session_id": "sess_123", "command": "   "}, action_schema)791    with pytest.raises(jsonschema.ValidationError):792        jsonschema.validate(793            {"session_id": "sess_123", "kind": "command", "command": "   "},794            action_schema,795        )796 797 798def test_openapi_validation_error_schemas_are_not_rewritten_to_public_step_results() -> None:799    from Pwned.server.app import app800 801    schema = app.openapi()802 803    reset_422 = schema["paths"]["/reset"]["post"]["responses"]["422"]["content"]["application/json"]["schema"]804    step_422 = schema["paths"]["/step"]["post"]["responses"]["422"]["content"]["application/json"]["schema"]805 806    assert not reset_422.get("$ref", "").endswith("PwnedPublicStepResult"), (807        f"/reset 422 response schema must not reference PwnedPublicStepResult, got: {reset_422}"808    )809    assert not step_422.get("$ref", "").endswith("PwnedPublicStepResult"), (810        f"/step 422 response schema must not reference PwnedPublicStepResult, got: {step_422}"811    )812 813 814def test_openapi_schema_describes_the_public_schema_surface_explicitly() -> None:815    from Pwned.server.app import app816 817    schema = app.openapi()818 819    schema_resp = schema["paths"]["/schema"]["get"]["responses"]["200"]["content"]["application/json"]["schema"]820    assert schema_resp.get("$ref", "").endswith("PwnedPublicSchemaSurface"), (821        f"/schema 200 response should reference PwnedPublicSchemaSurface, got: {schema_resp}"822    )823    surface_schema = schema["components"]["schemas"]["PwnedPublicSchemaSurface"]824    props = surface_schema.get("properties", {})825    for field_name in ("action", "observation", "state"):826        assert field_name in props, (827            f"PwnedPublicSchemaSurface must expose `{field_name}` in OpenAPI: {surface_schema}"828        )829        field_schema = props[field_name]830        assert "$ref" in field_schema or "properties" in field_schema, (831            "OpenAPI should describe nested /schema fields explicitly instead of "832            f"using opaque generic objects. field={field_name} schema={field_schema}"833        )834 835 836def test_server_app_uses_configurable_concurrent_capacity() -> None:837    import os838    import Pwned.server.app as server_app_module839 840    default_capacity = server_app_module._resolve_max_concurrent_envs()841    original = os.environ.get("PWNED_MAX_SESSIONS")842 843    try:844        os.environ["PWNED_MAX_SESSIONS"] = "3"845        configured_capacity = server_app_module._resolve_max_concurrent_envs()846    finally:847        if original is None:848            os.environ.pop("PWNED_MAX_SESSIONS", None)849        else:850            os.environ["PWNED_MAX_SESSIONS"] = original851 852    assert default_capacity >= 1853    assert configured_capacity == 3854