[konlue]

Describe what this PR does / why we need it

Fixes #276: the standalone ReAct agent registered a model stream subscription and immediately requested Long.MAX_VALUE from it. Cancellation only reached a subscription that was already parked in the slot, so a subscription that arrived after the run was cancelled still started the upstream model call — a stopped run kept spending tokens and holding a connection. The same window exists when the execution control is stopped before the stream is created (deadline/ownership loss): onLoss fires while the slot is still empty, and the late onSubscribe used to request anyway.

Does this pull request fix one issue?

Fixes #276

Describe how you did it

• onSubscribe now checks cancelled.get() || future.isCancelled() after winning the slot CAS; when the run is over it returns the slot and cancels the subscription without requesting. The future.isCancelled() disjunct covers the control-stopped path, where the run-level flag is not set.
• cancel() and the onLoss callback now take the subscription with getAndSet(null), so ownership transfer is atomic: the subscription is cancelled exactly once and a later slot cleanup cannot erase it.
• No sentinel subscription: the monotone cancelled flag answers "may this run start?", the slot reference answers "who cancels?", and the cleanup keeps its existing shape.
• The remaining interleaving (guard read wins, cancellation lands before the cancel write) may still issue one request; this is allowed by the reactive-streams contract (RS 3.6 is the publisher's responsibility) and is not locked away.

How was it tested / verified?

New offline ReActStreamSubscriptionCancellationTest with a deferred publisher and a counting subscription:

1. Cancel before onSubscribe → requests=0, cancels=1 (the issue's repro);
2. Normal streaming completion → requests=1, cancels=0, one Finished, onComplete;
3. Two streaming turns → one request per turn, no cancels (guards the slot cleanup);
4. Cancel during the second turn → only the active subscription is cancelled;
5. context.control().cancel() before onSubscribe → no request, one cancel.

Red before the fix: Tests run: 5, Failures: 3. Green after: 5/5; argi-agent-core 103/103; checkstyle passes; downstream argi-agent-framework regression shows no new failures.

Additional notes

• No public API, event, configuration or persistence format changes.
• Scope is the standalone ReAct model stream only; tool-operation cancellation and the A2A transport cancellation ([BUG][P1] A2A 流取消应中止阻塞的 HTTP 读取并释放连接与工作线程 #277) are untouched.

[yuluo-yx]

LGTM