fix(amazon): include deferred-task trigger logs when reading from Clo…

[bujjibabukatta]

Title

fix(amazon): include deferred-task trigger logs when reading from CloudWatch (#70317)

Body

Closes #70317.

Root cause

CloudWatchRemoteLogIO.stream() (and the legacy CloudwatchTaskHandler._read_remote_logs()) only ever read the task's own CloudWatch stream. A deferred task's triggerer logs are written to a separate stream, .trigger..log (FileTaskHandler.add_triggerer_suffix). While the task is DEFERRED, the UI tails those logs live from the triggerer over HTTP, which is why they're visible during the deferral — but once the task finishes, that live path goes away and nothing ever looked in the separate CloudWatch stream. The local-file backend doesn't have this problem because FileTaskHandler._read_from_local() uses glob(worker_log_path.name + ""), which picks up attempt=1.log and attempt=1.log.trigger..log for free. CloudWatch has no glob equivalent, so the trigger stream has to be discovered explicitly.

Fix
Added AwsLogsHook.describe_log_streams(): a synchronous, paginated wrapper around DescribeLogStreams with a prefix filter (mirrors the existing async variant's ResourceNotFoundException → [] handling for a not-yet-existing log group).
CloudWatchRemoteLogIO.stream() now looks up any .trigger.*.log streams via that prefix scan and reads each one the same way it reads the base stream, merging them into the returned log groups. Skipped while the task is actively DEFERRED, since the live HTTP path already covers that state and the extra API call would just be repeated on every UI poll for a potentially long-running deferral.
Multiple trigger streams (a task deferred, resumed, and deferred again — even by different triggerer processes) are sorted numerically by job id, not lexicographically — job id 7 must sort before job id 200. (Caught this via test — an initial plain-string sort got it backwards.)
Applied the identical fix to the legacy CloudwatchTaskHandler._read_remote_logs() path (still registered in provider.yaml for older-style log-handler configs). Also removed a pre-existing dead call there (self.io.read(...), whose result was immediately discarded and redone manually right below it — wasting a CloudWatch API call on every read).
Trigger-stream discovery failures (e.g. a permissions issue) are reported as an extra message rather than raising, consistent with the existing error handling on the base stream.
Backward compatibility

When no trigger streams exist (the common, non-deferred case), behavior and API call count are unchanged — verified via test that the discovery call only fires when needed and returns [] cleanly for a non-existent/empty-prefix case.

Testing
12 new tests added to test_cloudwatch_task_handler.py, covering: base+trigger merge, correct numeric ordering across multiple trigger streams, the DEFERRED-state skip, graceful handling of a DescribeLogStreams failure, the legacy handler path, correct colon-replacement in the discovery prefix, and hook-level pagination (tested with 54+ streams) and missing-log-group handling.
All 42 tests in the file pass (33 existing + 12 new — some existing tests exercise overlapping paths).
ruff check, ruff format --check, and mypy all pass clean on the three changed files.
Trade-off worth flagging for reviewers

This adds one DescribeLogStreams API call per log read for tasks that are not currently DEFERRED (previously: zero). For a task that was never deferred, this returns an empty list almost instantly, but it is still an extra round-trip. Happy to gate this further (e.g. only when try_number matches a TaskInstanceHistory row known to have gone through DEFERRED) if that trade-off is a concern — went with the simpler, always-check approach first since it's the correct and easy default and mirrors what the issue reporter's own tested workaround did.

AI Disclosure
This contribution used AI assistance.
Model(s) used: Claude

AI was used for:

Helping write this PR description.
Running the test suite locally to make sure everything passes before submitting.
Reviewing the implementation from a performance perspective.
The implementation and code changes were developed and validated by me.

[ferruzzi]

This adds one DescribeLogStreams API call per log read for tasks that are not currently DEFERRED

Not sure if this is an issue worth making changes, but maybe worth keeping an eye on if we go with this plan: DescribeLogStreams is a fairly low-quota API at 25 TPS per account per region. I don't know how likely a large environment is to hit that but it seems like a low quota to hit?

[bujjibabukatta]

Hey @ferruzzi, thanks for digging into this so carefully — went through all four:

The re-deferral thing was a genuine bug, glad you caught it. Fixed it so we only skip the current triggerer's stream while DEFERRED, not the whole lookup — so logs from an earlier deferral don't disappear anymore. Added a test to lock that in.
Renamed _get_trigger_stream_names to get_trigger_stream_names (dropped the underscore), and did the same for the new shared helper.
Cleaned up the parametrize test too — it's just a plain test now, no more unused param.
And on the quota thing — heads up that the re-deferral fix actually makes DescribeLogStreams get called a bit more often now, not less, since it runs during DEFERRED too. Not opposed to gating it down the line (caching, or keying off TaskInstanceHistory) if it ends up mattering in practice, just didn't want to over-engineer it without real numbers.

[ferruzzi]

So many stacked getattrs here makes me think there should be a better way. Can you try this, and see if it works? It may need a little tweaking, but should be cleaner, I think:

if (ti.state == TaskInstanceState.DEFERRED) and (job := getattr(ti, "triggerer_job", None)):
    live_stream = f"{relative_path.replace(':', '_')}.trigger.{job.id}.log"
    trigger_stream_names = [name for name in trigger_stream_names if name != live_stream]

Plus, hardcoding the prefix assembly here makes me think we should likely move that builder into a shared helper at some point, but that can be done in a different PR.