[e2e] Add a checkpoint-recovery test for restored Python state

[weiqingy]

Linked issue: #836

Purpose of change

No test covers Python agent memory surviving the loss of a TaskManager process. Existing e2e suites run on MiniCluster, which never recreates the embedded Python interpreter, so a value Pemja cannot materialize into a JVM type fails only on a real restore. This test hard-kills a real TaskManager and asserts the tool-call context and a raw bytes value both survived.

Runtime flow

1. The harness configures checkpointing and a restart strategy on a standalone cluster, stages this checkout's artifacts, and submits the job detached.
2. The agent writes the blob to memory, bumps a process-local counter, and requests a chat. The mock model emits one tool call; the tool writes tool-entered and blocks on release.
3. On tool-entered it waits for two further completed checkpoints, records the pid, resource id and restore count, SIGKILLs the TaskManager, starts a replacement, waits for a restore, then writes release.
4. The re-executed tool returns its sentinel. The agent checks the restored blob and transcript against the counters, writes verdict.json, and raises on mismatch.

Key decisions

The source must be unbounded. A bounded one raises end-of-input right after the record, and ActionExecutionOperator.endInput() then holds the mailbox thread while the run is parked, so no checkpoint completes. Measured on the pinned stack: unbounded 16 completions in 15s, bounded 0.

Provenance rests on module-level counters outside Flink state, since anything checkpointed is restored with the payload and cannot tell a restored read from the reader's own write. Pemja's singleton interpreter keeps globals across an in-place task restart, so zero counters prove a different OS process.

counts.restored rather than numRestarts, which counts a restart that restored nothing.

Implementation Description

Behavioral contracts

1. A pass requires a published verdict: "pass"; a run recording no assertion exits non-zero.
2. The interval comes from the job's own coordinator config, not the harness's file.
3. The payload sits in a checkpoint taken while parked: two completions past a park-time baseline.
4. counts.restored rises above its pre-kill baseline.
5. latest.restored is non-null. If 4 and 5 disagree the run fails rather than picking one.
6. The process was replaced: pid and resource id both differ from pre-kill.
7. The values came from another process: both counters are 0 where they are asserted.
8. The blob is bytes or bytearray with the exact content; a list of the same ints fails.
9. The kill/restart window fits the tool's deadline: a pre-flight check plus a per-wait clamp.
10. The TaskManager ran this checkout: version 0.3.dev0, flink_agents.api under its path.
11. Config edits are idempotent and reverted, leaving no duplicate top-level YAML key.
12. The new workflow job never runs on a pull request.

Failure behavior

A missing handshake_dir or verdict_dir raises ValueError. Checkpointing off or a wrong interval records a FAIL and exits non-zero, and an infeasible budget exits 1.

An unreachable, slow, or wrongly shaped REST response never reads as a false condition. Requests are bounded by CURL_MAX_TIME; an empty body, bad JSON, a missing path, or a wrongly typed leaf all fail the read.

One path degrades deliberately. The framework absorbs a tool exception into a ... execute failed. response rather than failing the action, so the tool returns a sentinel the assertion requires. The agent publishes the verdict before raising AssertionError, so a failed run leaves a readable file.

Tests

Contract	Tests
1 verdict only	7 assert_verdict, 3 print_summary
2 checkpointing	4 assert_checkpointing_enabled
3 two checkpoints	none; see negative control
4 restored count	none; see provenance control
5 restored non-null	none; see provenance control
6 process replaced	none
7 counters	none; exercised via 8
8 blob predicate	_blob_matches, 9 cases
9 budget	4 assert_handshake_budget, 3 charged_timeout, 1 handshake_budget_left
10 identity	none
11 config idempotent	6 configure_flink / delete_config_key
12 not on a PR	none
handshake, for 3 and 7	3 await_release

About 45 of the 78 bats tests pin harness plumbing the contracts above deliberately do not cover: REST reads, file waits, cluster and venv setup, and EXIT-trap cleanup.

The 78 bats and 12 Python tests run pre-merge; the cluster test does not, since its job is schedule and workflow_dispatch only. Fork runs are the only evidence, and the controls are what make the pass meaningful.

Happy path: 9/9 assertions, verdict pass, pid 4656 to 6034, counts.completed 1 to 3 while parked, both counters 0, blob back as bytes.

Negative control, release marker pre-created: failed at checkpoint-containment, completed stuck at 1.

Provenance control, kill skipped: failed at restore, counts.restored stayed 0, pid unchanged.

It does not reproduce the SIGSEGV that motivated the issue: the memory value validator now raises TypeError at set() before Pemja is reached, so that control no longer exists. Unit tests pin which types the validator admits; only this one shows an admitted type survives a restore.

Known limitations: timeouts are sized from observed runs, not proven bounds; and 11 pre-existing bats assertions are vacuous under bash 3.2 bodies, since errexit does not apply to [[ ]].

API

No public API change, and no product code touched. Six of the seven files are new; the seventh adds a job to the nightly workflow, leaving its existing job, triggers and paths filter untouched. Nothing changes for a caller who does nothing differently.

Documentation

• doc-needed
• doc-not-needed
• doc-included

Was this patch authored or co-authored using generative AI tooling?

• Yes
• No

Generated-by: Claude Code 2.1.224