import pytest from pentestgpt_agent.agents import ( AgentContractError, parse_execution_delta, recover_execution_delta, ) from pentestgpt_agent.execution import ( ExecutionDelta, ExecutionOutcome, ExecutionValidationError, compile_execution, ) from pentestgpt_agent.memory import ObservationRecord, TaskLease from pentestgpt_agent.trace import EpisodeTrace def test_transport_recovery_skips_a_later_incomplete_done_candidate() -> None: marker = '\n' trace = EpisodeTrace( input={}, events=( { "sequence": 1, "type": "tool_call", "name": "StructuredOutput", "input": { "task_id": "task-1", "outcome": "done", "summary": f"Captured the result.{marker}200 OK", }, }, { "sequence": 2, "type": "tool_call", "name": "StructuredOutput", "input": { "task_id": "task-1", "outcome": "done", "summary": "Tried again but omitted evidence.", "evidence_excerpt": None, }, }, ), output=None, truncated_tail=False, ) recovered = recover_execution_delta(trace) assert recovered.task_id == "task-1" assert recovered.outcome is ExecutionOutcome.DONE assert recovered.evidence_excerpt == "200 OK" def test_transport_recovery_accepts_only_an_exact_executor_contract() -> None: raw = { "task_id": "task-1", "outcome": "done", "summary": "Captured evidence.", "evidence_excerpt": "200 OK", "untrusted_extra": "must not survive schema rejection", } with pytest.raises(AgentContractError, match="unexpected or missing fields"): parse_execution_delta(raw) trace = EpisodeTrace( input={}, events=( { "sequence": 1, "type": "tool_call", "name": "StructuredOutput", "input": { "task_id": "task-1", "outcome": "done", "summary": "Captured evidence.", "evidence_excerpt": "200 OK", }, }, ), output=None, truncated_tail=False, ) recovered = recover_execution_delta(trace) assert recovered == ExecutionDelta( task_id="task-1", outcome=ExecutionOutcome.DONE, summary="Captured evidence.", evidence_excerpt="200 OK", ) trace_with_extra = EpisodeTrace( input={}, events=( { "sequence": 1, "type": "tool_call", "name": "StructuredOutput", "input": raw, }, ), output=None, truncated_tail=False, ) with pytest.raises(AgentContractError, match="no recoverable"): recover_execution_delta(trace_with_extra) def test_executor_cannot_complete_from_an_uncaptured_evidence_claim() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-1", "role": "executor", "state_revision": 1, "task_id": "discover", "attempt_id": "attempt-1", }, events=( { "sequence": 1, "type": "command_run", "command": "curl http://target/", "exit_code": 0, "output": "200 OK", }, ), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.DONE, summary="Found an administrative password.", evidence_excerpt="admin_password=secret", ) lease = TaskLease( run_id="run-1", task_id="discover", attempt_id="attempt-1", revision=1, ) execution = compile_execution(delta, lease, trace) assert execution.outcome is ExecutionOutcome.PROGRESS assert execution.observation == "200 OK" assert "admin_password" not in execution.observation assert execution.evidence_fallback is True def test_executor_can_complete_by_exactly_reusing_same_task_evidence() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-2", "role": "executor", "state_revision": 3, "task_id": "discover", "attempt_id": "attempt-2", }, events=(), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.DONE, summary="The prior bounded attempt already met the task contract.", evidence_excerpt="service returned 200 OK", ) lease = TaskLease( run_id="run-1", task_id="discover", attempt_id="attempt-2", revision=3, ) prior = ObservationRecord( id="attempt-1:observation", task_id="discover", attempt_id="attempt-1", statement="HTTP service returned 200 OK with a stable response.", trace_episode_id="executor-1", evidence_sequences=(1,), created_revision=2, ) execution = compile_execution( delta, lease, trace, prior_observations=(prior,), ) assert execution.outcome is ExecutionOutcome.DONE assert execution.observation is None assert execution.evidence_sequences == () assert execution.reused_observation_id == prior.id def test_executor_cannot_reuse_evidence_from_another_task() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-2", "role": "executor", "state_revision": 3, "task_id": "discover", "attempt_id": "attempt-2", }, events=(), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.DONE, summary="Claimed evidence from another task.", evidence_excerpt="admin password found", ) lease = TaskLease( run_id="run-1", task_id="discover", attempt_id="attempt-2", revision=3, ) unrelated = ObservationRecord( id="attempt-1:observation", task_id="enumerate-mail", attempt_id="attempt-1", statement="admin password found", trace_episode_id="executor-1", evidence_sequences=(1,), created_revision=2, ) with pytest.raises(ExecutionValidationError, match="was not captured"): compile_execution( delta, lease, trace, prior_observations=(unrelated,), ) def test_paraphrased_prior_evidence_degrades_without_creating_an_observation() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-2", "role": "executor", "state_revision": 3, "task_id": "discover", "attempt_id": "attempt-2", }, events=(), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.DONE, summary="Paraphrased the result instead of quoting it.", evidence_excerpt="The HTTP endpoint answered successfully.", ) lease = TaskLease("run-1", "discover", "attempt-2", 3) prior = ObservationRecord( id="attempt-1:observation", task_id="discover", attempt_id="attempt-1", statement="HTTP/1.1 200 OK", trace_episode_id="executor-1", evidence_sequences=(1,), created_revision=2, ) execution = compile_execution( delta, lease, trace, prior_observations=(prior,), ) assert execution.outcome is ExecutionOutcome.PROGRESS assert execution.observation is None assert execution.evidence_sequences == () assert execution.evidence_unresolved is True def test_actionful_attempt_cannot_degrade_an_unresolved_prior_evidence_quote() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-2", "role": "executor", "state_revision": 3, "task_id": "discover", "attempt_id": "attempt-2", }, events=( { "sequence": 1, "type": "tool_call", "name": "Shell", "call_id": "call-1", }, ), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.DONE, summary="Claimed unsupported completion after an external action.", evidence_excerpt="The endpoint answered successfully.", ) lease = TaskLease("run-1", "discover", "attempt-2", 3) prior = ObservationRecord( id="attempt-1:observation", task_id="discover", attempt_id="attempt-1", statement="HTTP/1.1 200 OK", trace_episode_id="executor-1", evidence_sequences=(1,), created_revision=2, ) with pytest.raises(ExecutionValidationError, match="proposed evidence was not captured"): compile_execution( delta, lease, trace, prior_observations=(prior,), ) def test_whitespace_equivalent_text_is_not_an_exact_evidence_quote() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-1", "role": "executor", "state_revision": 1, "task_id": "discover", "attempt_id": "attempt-1", }, events=( { "sequence": 1, "type": "command_run", "exit_code": 0, "output": "endpoint=/check_service\nparameter=service_name", }, ), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.DONE, summary="Captured the endpoint and parameter.", evidence_excerpt="endpoint=/check_service parameter=service_name", ) lease = TaskLease( run_id="run-1", task_id="discover", attempt_id="attempt-1", revision=1, ) execution = compile_execution(delta, lease, trace) assert execution.outcome is ExecutionOutcome.PROGRESS assert execution.observation == "endpoint=/check_service\nparameter=service_name" assert execution.evidence_fallback is True def test_crlf_transport_normalization_resolves_to_the_exact_receipt_slice() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-1", "role": "executor", "state_revision": 1, "task_id": "discover", "attempt_id": "attempt-1", }, events=( { "sequence": 5, "type": "tool_result", "is_error": False, "output": "HTTP/1.1 200 OK\r\nServer: Werkzeug\r\nContent-Type: text/html\r\n\r\nbody", }, ), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.DONE, summary="Captured the response headers.", evidence_excerpt="Server: Werkzeug\nContent-Type: text/html", ) lease = TaskLease( run_id="run-1", task_id="discover", attempt_id="attempt-1", revision=1, ) execution = compile_execution(delta, lease, trace) assert execution.evidence_sequences == (5,) assert execution.observation == "Server: Werkzeug\r\nContent-Type: text/html" def test_exact_ordered_lines_widen_to_one_contiguous_receipt_span() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-1", "role": "executor", "state_revision": 1, "task_id": "discover", "attempt_id": "attempt-1", }, events=( { "sequence": 5, "type": "tool_result", "is_error": False, "output": ( "22/tcp open ssh\n" "| ssh-hostkey: captured detail\n" "111/tcp open rpcbind\n" "| rpcinfo: captured detail\n" "2049/tcp open nfs\n" ), }, ), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.DONE, summary="Captured the material services.", evidence_excerpt="22/tcp open ssh\n111/tcp open rpcbind\n2049/tcp open nfs", ) lease = TaskLease("run-1", "discover", "attempt-1", 1) execution = compile_execution(delta, lease, trace) assert execution.observation == ( "22/tcp open ssh\n" "| ssh-hostkey: captured detail\n" "111/tcp open rpcbind\n" "| rpcinfo: captured detail\n" "2049/tcp open nfs" ) assert execution.evidence_sequences == (5,) assert execution.evidence_span_widened is True def test_line_span_widening_rejects_reordered_receipt_lines() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-1", "role": "executor", "state_revision": 1, "task_id": "discover", "attempt_id": "attempt-1", }, events=( { "sequence": 5, "type": "tool_result", "is_error": False, "output": "22/tcp open ssh\n111/tcp open rpcbind\n2049/tcp open nfs", }, ), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.DONE, summary="Reordered the material services.", evidence_excerpt="2049/tcp open nfs\n22/tcp open ssh", ) execution = compile_execution( delta, TaskLease("run-1", "discover", "attempt-1", 1), trace, ) assert execution.outcome is ExecutionOutcome.PROGRESS assert execution.observation == "22/tcp open ssh\n111/tcp open rpcbind\n2049/tcp open nfs" assert execution.evidence_fallback is True def test_exact_unique_line_envelope_recovers_a_rich_reordered_quote() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-1", "role": "executor", "state_revision": 1, "task_id": "test-mail", "attempt_id": "attempt-1", }, events=( { "sequence": 5, "type": "tool_result", "is_error": False, "output": ( "LOGIN 143 OK\n" "Return-Path: \n" "Received: by target\n" "From: sender@example.test\n" "Subject: Welcome\n" "\n" "Credentials are on the shared drive." ), }, ), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="test-mail", outcome=ExecutionOutcome.DONE, summary="Authenticated and read the message.", evidence_excerpt=( "LOGIN 143 OK\n" "Return-Path: \n" "Subject: Welcome\n" "From: sender@example.test\n" "\n" "Credentials are on the shared drive." ), ) execution = compile_execution( delta, TaskLease("run-1", "test-mail", "attempt-1", 1), trace, ) assert execution.observation == ( "LOGIN 143 OK\n" "Return-Path: \n" "Received: by target\n" "From: sender@example.test\n" "Subject: Welcome\n" "\n" "Credentials are on the shared drive." ) assert execution.evidence_sequences == (5,) assert execution.evidence_span_widened is True def test_canonical_execution_fields_have_conservative_size_limits() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-1", "role": "executor", "state_revision": 1, "task_id": "discover", "attempt_id": "attempt-1", }, events=( { "sequence": 1, "type": "command_run", "exit_code": 0, "output": "A" * 4_001, }, ), output={"success": True}, truncated_tail=False, ) lease = TaskLease( run_id="run-1", task_id="discover", attempt_id="attempt-1", revision=1, ) with pytest.raises(ExecutionValidationError, match="summary exceeds"): compile_execution( ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.DONE, summary="S" * 2_001, evidence_excerpt="A", ), lease, trace, ) oversized = compile_execution( ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.DONE, summary="Captured output.", evidence_excerpt="A" * 4_001, ), lease, trace, ) assert oversized.outcome is ExecutionOutcome.PROGRESS assert oversized.observation == "A" * 4_000 assert oversized.evidence_truncated is True invented = compile_execution( ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.DONE, summary="Invented oversized output.", evidence_excerpt="B" * 4_001, ), lease, trace, ) assert invented.outcome is ExecutionOutcome.PROGRESS assert invented.observation == "A" * 4_000 assert invented.evidence_fallback is True assert invented.evidence_truncated is True def test_evidence_cannot_be_assembled_across_action_receipts() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-1", "role": "executor", "state_revision": 1, "task_id": "discover", "attempt_id": "attempt-1", }, events=( { "sequence": 7, "type": "tool_result", "is_error": False, "output": ( "HTTP/1.0 200 OK\nServer: BaseHTTP/0.6 Python/3.12.11\n" 'Driver application' ), }, { "sequence": 10, "type": "tool_result", "is_error": False, "output": '
', }, ), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.DONE, summary="Mapped the application form.", evidence_excerpt=( "GET / -> HTTP/1.0 200 OK; Server: BaseHTTP/0.6 Python/3.12.11; " 'body: Driver application\n' 'GET /apply -> HTTP/1.0 200 OK; body:
' '
' ), ) lease = TaskLease("run-1", "discover", "attempt-1", 1) execution = compile_execution(delta, lease, trace) assert execution.outcome is ExecutionOutcome.PROGRESS assert execution.observation == '
' assert execution.evidence_sequences == (10,) assert execution.evidence_fallback is True def test_exact_cross_receipt_quote_projects_one_rich_grounded_receipt() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-1", "role": "executor", "state_revision": 1, "task_id": "enumerate-nfs", "attempt_id": "attempt-1", }, events=( { "sequence": 5, "type": "command_run", "exit_code": 0, "output": "Export list for target:\n/srv/nfs/onboarding *\n", }, { "sequence": 21, "type": "command_run", "exit_code": 0, "output": ( "Contents of /srv/nfs/onboarding (read-only mount):\n" "-rw-r--r-- root root 1751 New_Employee_Access.pdf\n" ), }, ), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="enumerate-nfs", outcome=ExecutionOutcome.DONE, summary="Found the onboarding export and its PDF.", evidence_excerpt=( "Export list for target:\n" "/srv/nfs/onboarding *\n" "Contents of /srv/nfs/onboarding (read-only mount):\n" "-rw-r--r-- root root 1751 New_Employee_Access.pdf" ), ) execution = compile_execution( delta, TaskLease("run-1", "enumerate-nfs", "attempt-1", 1), trace, ) assert execution.observation == ( "Contents of /srv/nfs/onboarding (read-only mount):\n" "-rw-r--r-- root root 1751 New_Employee_Access.pdf" ) assert execution.evidence_sequences == (21,) assert execution.evidence_projected is True def test_one_captured_fragment_cannot_launder_an_uncaptured_claim() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-1", "role": "executor", "state_revision": 1, "task_id": "discover", "attempt_id": "attempt-1", }, events=( { "sequence": 7, "type": "tool_result", "is_error": False, "output": "HTTP/1.0 200 OK", }, ), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.DONE, summary="Found an administrative password.", evidence_excerpt="GET / -> HTTP/1.0 200 OK; admin_password=secret", ) lease = TaskLease("run-1", "discover", "attempt-1", 1) execution = compile_execution(delta, lease, trace) assert execution.outcome is ExecutionOutcome.PROGRESS assert execution.observation == "HTTP/1.0 200 OK" assert "admin_password" not in execution.observation assert execution.evidence_fallback is True def test_short_exact_evidence_is_still_grounded() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-1", "role": "executor", "state_revision": 1, "task_id": "test-template", "attempt_id": "attempt-1", }, events=( { "sequence": 4, "type": "command_run", "exit_code": 0, "output": "49", }, ), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="test-template", outcome=ExecutionOutcome.DONE, summary="The expression was evaluated.", evidence_excerpt="49", ) lease = TaskLease("run-1", "test-template", "attempt-1", 1) compiled = compile_execution(delta, lease, trace) assert compiled.evidence_sequences == (4,) assert compiled.observation == "49" def test_nonzero_command_receipt_can_ground_a_completed_negative_test() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-1", "role": "executor", "state_revision": 1, "task_id": "test-ssh-credential", "attempt_id": "attempt-1", }, events=( { "sequence": 4, "type": "command_run", "exit_code": 255, "output": "Permission denied (publickey,password).", }, ), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="test-ssh-credential", outcome=ExecutionOutcome.DONE, summary="The supplied credential was conclusively rejected by SSH.", evidence_excerpt="Permission denied (publickey,password).", ) lease = TaskLease("run-1", "test-ssh-credential", "attempt-1", 1) compiled = compile_execution(delta, lease, trace) assert compiled.outcome is ExecutionOutcome.DONE assert compiled.evidence_sequences == (4,) assert compiled.observation == "Permission denied (publickey,password)." assert compiled.evidence_fallback is False def test_progress_without_evidence_does_not_create_an_observation() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-1", "role": "executor", "state_revision": 1, "task_id": "discover", "attempt_id": "attempt-1", }, events=(), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.PROGRESS, summary="Made partial progress.", evidence_excerpt=None, ) lease = TaskLease("run-1", "discover", "attempt-1", 1) compiled = compile_execution(delta, lease, trace) assert compiled.observation is None assert compiled.evidence_sequences == () def test_evidence_trace_must_belong_to_the_exact_leased_run() -> None: trace = EpisodeTrace( input={ "run_id": "different-run", "episode_id": "executor-1", "role": "executor", "state_revision": 1, "task_id": "discover", "attempt_id": "attempt-1", }, events=(), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.PROGRESS, summary="Made partial progress.", evidence_excerpt=None, ) lease = TaskLease("run-1", "discover", "attempt-1", 1) with pytest.raises(ExecutionValidationError, match="different run"): compile_execution(delta, lease, trace) def test_evidence_trace_must_match_the_persisted_episode_identity() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-wrong", "role": "executor", "state_revision": 1, "task_id": "discover", "attempt_id": "attempt-1", }, events=(), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.PROGRESS, summary="Made partial progress.", evidence_excerpt=None, ) lease = TaskLease( "run-1", "discover", "attempt-1", 1, trace_episode_id="executor-expected", ) with pytest.raises(ExecutionValidationError, match="different episode"): compile_execution(delta, lease, trace) @pytest.mark.parametrize( ("task_id", "attempt_id", "message"), ( ("different-task", "attempt-1", "different task"), ("discover", "different-attempt", "different attempt"), ), ) def test_evidence_trace_must_match_the_exact_task_attempt_identity( task_id: str, attempt_id: str, message: str, ) -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-1", "role": "executor", "state_revision": 1, "task_id": task_id, "attempt_id": attempt_id, }, events=(), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.PROGRESS, summary="Made partial progress.", evidence_excerpt=None, ) lease = TaskLease("run-1", "discover", "attempt-1", 1) with pytest.raises(ExecutionValidationError, match=message): compile_execution(delta, lease, trace) def test_structured_output_receipt_cannot_become_canonical_evidence() -> None: trace = EpisodeTrace( input={ "run_id": "run-1", "episode_id": "executor-1", "role": "executor", "state_revision": 1, "task_id": "discover", "attempt_id": "attempt-1", }, events=( { "sequence": 1, "type": "tool_call", "name": "StructuredOutput", "call_id": "structured-1", }, { "sequence": 2, "type": "tool_result", "call_id": "structured-1", "is_error": False, "output": "Structured output provided successfully", }, ), output={"success": True}, truncated_tail=False, ) delta = ExecutionDelta( task_id="discover", outcome=ExecutionOutcome.DONE, summary="Claimed completion from the response transport.", evidence_excerpt="Structured output provided successfully", ) lease = TaskLease("run-1", "discover", "attempt-1", 1) with pytest.raises(ExecutionValidationError, match="not captured"): compile_execution(delta, lease, trace)