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.

[vincbeck]

This code looks very similar to the block above, any chance to factorize?

[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?

[ferruzzi]

This may be an edge case, but what happens if the task is deferred again? _get_log_retrieval_url builds the served-log URL from ti.triggerer_job.id so the UI is only tailing the logs of the current deferred task. Do we need some way to persist the logs from previous deferrals into the main logs?

[ferruzzi]

Calling a "private" method from another class is a code smell, should io._get_trigger_stream_names be renamed to drop the leading underscore? It also looks like a genuinely useful helper that may be reused elsewhere if it wasn't private.

[ferruzzi]

Parameterizing with only one param case? It also doesn't look like expected_error_code is used int he test unless I'm missing something