fix(amber): pass unstamped boundary states through LoopEnd instead of consuming them

[aglinxinyuan]

What changes were proposed in this PR?

Follow-up to #6661, addressing this review comment: a loop-body operator that emits its own boundary state (produce_state_on_start/finish — a public API on both engine sides) sends it with the "no loop" envelope (counter 0, loop_start_id ""). The LoopEnd matching branch treated every counter-0 frame as the loop's own boundary state:

LoopStart ──(0, LS-id)──▶ stateful body op ──▶ LoopEnd
                             │ produce_state_on_finish
                             └──(0, "")────────▶ LoopEnd   ← arrives AFTER the loop state

Consuming the unstamped state (1) clobbers the captured back-jump id with "" → no loop-back state URI configured for LoopStart '', and (2) hands run_update a State with no table payload → KeyError. Either way the loop breaks. The reviewer reasoned this for a stateful JVM operator; it is language-independent (a Python UDF hits the identical path — no JVM built-in currently overrides produceStateOnFinish, so the Python UDF is also the practical repro).

Fix (consumer-side). A real loop state is always stamped — the matching LoopStart stamps its own id on every iteration's output state — so the LoopEnd runtime now keys on the stamp:

Frame at LoopEnd (counter 0)	before	after
stamped (loop_start_id set)	consume + capture id	unchanged
unstamped ("")	consume → id clobber / KeyError	forward downstream unchanged, skip the operator (default pass-through semantics), captured id untouched

Also adds the comment the review asked for at the two Scala boundary-state emit sites (StartChannelHandler / EndChannelHandler), documenting that their "no loop" envelope defaults are deliberate and how the LoopEnd runtime treats them.

Any related issues, documentation, discussions?

Follow-up to #6661 (review discussion r3648708075). Related engine context: #6660.

How was this PR tested?

• Unit (test_main_loop.py): new test pins that an unstamped counter-0 frame at a LoopEnd is forwarded with its envelope unchanged, the operator is not invoked, and the captured back-jump id is not clobbered (red before the fix); the existing stamped-consume test is unchanged and still passing (95/95 across the loop suites locally).
• E2E (LoopIntegrationSpec, CI-only): new case runs TextInput → LoopStart → stateful Python UDF → LoopEnd where the UDF emits boundary state via produce_state_on_finish — it crashes without this fix and completes exactly 3 iterations with it.
• scalafmtCheckAll + scalafixAll --check + full test-compile + ruff format/check pass locally (Java 17).

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

Generated-by: Claude Code (Fable 5)

[github-actions]

Automated Reviewer Suggestions

Based on the git blame history of the changed files, we recommend the following reviewers:

• Contributors with relevant context: @Yicong-Huang
  You can notify them by mentioning @Yicong-Huang in a comment.

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 78.69%. Comparing base (f521750) to head (852ba0c).

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #6913      +/-   ##
============================================
- Coverage     78.70%   78.69%   -0.01%     
+ Complexity     3740     3738       -2     
============================================
  Files          1161     1161              
  Lines         46084    46088       +4     
  Branches       5110     5110              
============================================
+ Hits          36269    36271       +2     
- Misses         8206     8207       +1     
- Partials       1609     1610       +1     

Flag	Coverage Δ		*Carryforward flag
access-control-service	70.00% <ø> (ø)		Carriedforward from f521750
agent-service	76.76% <ø> (ø)		Carriedforward from f521750
amber	71.51% <ø> (-0.02%)	⬇️
computing-unit-managing-service	20.49% <ø> (ø)		Carriedforward from f521750
config-service	66.66% <ø> (ø)		Carriedforward from f521750
file-service	67.21% <ø> (ø)		Carriedforward from f521750
frontend	82.91% <ø> (ø)		Carriedforward from f521750
notebook-migration-service	78.94% <ø> (ø)		Carriedforward from f521750
pyamber	95.39% <100.00%> (+<0.01%)	⬆️
workflow-compiling-service	55.14% <ø> (ø)		Carriedforward from f521750

*This pull request uses carry forward flags. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

✅ No material benchmark regressions detected

🟢 4 better · 🔴 0 worse · ⚪ 11 noise (<±5%) · 0 without baseline

Compared against main f521750 benchmarked on this same runner, so the delta is largely free of cross-runner hardware noise. The "7d avg" column still reflects the gh-pages dashboard. Treat <±5% as noise unless repeated.

Dashboard · Run

	config	throughput	MB/s	latency	max Δ latest / 7d
🟢	bs=10 sw=10 sl=64	413	0.252	23,576/30,143/30,143 us	🟢 -11.1% / 🔴 +86.3%
🟢	bs=100 sw=10 sl=64	812	0.495	122,024/149,888/149,888 us	🟢 -9.5% / 🔴 +37.1%
⚪	bs=1000 sw=10 sl=64	947	0.578	1,061,713/1,097,831/1,097,831 us	⚪ within ±5% / 🔴 +5.4%

Baseline details

Latest main f521750 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	413 tuples/sec	420 tuples/sec	759.72 tuples/sec	-1.7%	-45.6%
bs=10 sw=10 sl=64	MB/s	0.252 MB/s	0.256 MB/s	0.464 MB/s	-1.6%	-45.7%
bs=10 sw=10 sl=64	p50	23,576 us	22,854 us	12,653 us	+3.2%	+86.3%
bs=10 sw=10 sl=64	p95	30,143 us	33,917 us	16,371 us	-11.1%	+84.1%
bs=10 sw=10 sl=64	p99	30,143 us	33,917 us	19,941 us	-11.1%	+51.2%
bs=100 sw=10 sl=64	throughput	812 tuples/sec	819 tuples/sec	966.78 tuples/sec	-0.9%	-16.0%
bs=100 sw=10 sl=64	MB/s	0.495 MB/s	0.5 MB/s	0.59 MB/s	-1.0%	-16.1%
bs=100 sw=10 sl=64	p50	122,024 us	116,670 us	103,654 us	+4.6%	+17.7%
bs=100 sw=10 sl=64	p95	149,888 us	165,561 us	109,308 us	-9.5%	+37.1%
bs=100 sw=10 sl=64	p99	149,888 us	165,561 us	116,369 us	-9.5%	+28.8%
bs=1000 sw=10 sl=64	throughput	947 tuples/sec	930 tuples/sec	997.95 tuples/sec	+1.8%	-5.1%
bs=1000 sw=10 sl=64	MB/s	0.578 MB/s	0.568 MB/s	0.609 MB/s	+1.8%	-5.1%
bs=1000 sw=10 sl=64	p50	1,061,713 us	1,074,361 us	1,007,348 us	-1.2%	+5.4%
bs=1000 sw=10 sl=64	p95	1,097,831 us	1,143,338 us	1,050,149 us	-4.0%	+4.5%
bs=1000 sw=10 sl=64	p99	1,097,831 us	1,143,338 us	1,079,497 us	-4.0%	+1.7%

Raw CSV

config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,484.32,200,128000,413,0.252,23575.80,30142.89,30142.89
1,100,10,64,20,2463.59,2000,1280000,812,0.495,122024.29,149888.01,149888.01
2,1000,10,64,20,21117.80,20000,12800000,947,0.578,1061712.65,1097831.35,1097831.35