Fix TimeSensor dag_version churn via TimeOfDayTrigger

[Vamsi-klu]

Problem

TimeSensor(start_from_trigger=True) caused unnecessary DagVersion churn. There are two defects, not one:

1. Shared class mutation (bug regardless of API): start_trigger_args is a class-level StartTriggerArgs. __init__ did self.start_trigger_args.trigger_kwargs = {...}, which mutates the shared class object — every TimeSensor got the last-constructed sensor's moment.
2. Parse-time absolute moment: The mutated trigger_kwargs["moment"] was an absolute datetime from datetime.now() at parse time, so every Dag-processor parse produced a different serialized blob → new DagVersion every parse.

What Changed

Preserves the start_from_trigger API (no major bump) by adding a lazy-resolution trigger:

• TimeOfDayTrigger — stores parse-stable target_time (ISO string) + timezone (IANA name or fixed offset seconds) + end_from_trigger. Resolves the concrete UTC moment when the trigger starts, then delegates to DateTimeTrigger's wait loop.
• TimeSensor.__init__ — keeps start_from_trigger in the signature; builds a per-instance StartTriggerArgs (never mutates the class attribute).
• execute() / poke() — resolve moment once per attempt and cache (no midnight recompute drift).
• DST: spring-forward gap shifts forward; fall-back ambiguous uses fold=0 (first occurrence).
• No Dag context: falls back to UTC without crashing.

Why not remove start_from_trigger?

Removing it would be a functional regression for users who start the sensor directly on the triggerer (worker slot avoided). That would need a major bump on apache-airflow-providers-standard and a changelog breaking-change note. Lazy moment calculation via TimeOfDayTrigger fixes the hash churn while keeping the feature, so no major bump is required.

trigger_kwargs remains accepted-and-ignored (same as main).

Impact

• Fixes dag_version churn from TimeSensor(start_from_trigger=True).
• Preserves start-from-trigger behavior and public API.
• Providers do not use newsfragments; this is a non-breaking bugfix (no Changelog breaking-change entry needed).

Fixes: #69543

Testing

pytest providers/standard/tests/unit/standard/sensors/test_time.py -q
pytest providers/standard/tests/unit/standard/triggers/test_temporal.py -q

Coverage includes: already-passed today, midnight cache, DST spring/fall, no-Dag UTC fallback, TimeSensorAsync, FixedTimezone + named tz, shared-class isolation, serialization stability across mocked now(), end_from_trigger propagation.

[Vamsi-klu]

Reviewers: @ashb (core serialization / sensors) + @potiuk @kaxil for standard provider. This fixes dag_version churn by moving target_datetime computation to execution time.

Fixes #69543

---

Drafted-by: Muse Spark 1.1; reviewed by @Vamsi-klu before posting

[potiuk]

Thanks — the underlying diagnosis is right and worth fixing: baking target_datetime into start_trigger_args at parse time means the serialized Dag hash changes every day, and the hash-stability test you added is a good way to pin that.

But the fix is to remove start-from-trigger support from TimeSensor, and that's a bigger call than the PR framing suggests. Anyone using TimeSensor(start_from_trigger=True) today has their sensor start directly in the Triggerer; after this it warns, is ignored, and the task instead occupies a worker slot to poke or defer. That's a functional regression for those users, not just churn cleanup.

Concretely that needs: a major version bump on apache-airflow-providers-standard (currently 1.16.0) and an entry under the Changelog header in providers/standard/docs/changelog.rst spelling out the required user action. Neither is in the diff. (No newsfragment — providers don't use those.)

The design question I'd like your view on: was removal the only option? Computing moment lazily — so start_trigger_args serializes a stable placeholder and the concrete datetime is resolved when the trigger actually starts — would fix the hash churn while keeping the feature. If that was tried and doesn't work, saying why in the description would help; if it wasn't, it seems worth a look before dropping a documented capability.

One correction to something I'd have flagged: trigger_kwargs is already accepted-and-ignored on main (it's declared in __init__ but never read), so dropping it here changes nothing. Not an issue with this PR.

---

To be clear about where this stands: the points above are mostly design questions rather than defects, and I'd rather you formed your own view on them than took mine as settled. My review here was AI-assisted, and I'd guess parts of this PR were too — that's fine on both sides, but it means neither of us should treat the output as authoritative. Please push back where you think the reasoning is wrong, and say what you think the right trade-off is between fixing the hash churn and keeping the feature.

I'd also like other maintainers to weigh in before this is decided. Removing a documented capability from a released provider isn't a call I want to make on my own, and folks closer to the start-from-trigger design will have better context here than I do.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

[Vamsi-klu]

@potiuk Thanks for the careful review — you're right that removing start_from_trigger was the wrong trade-off. I've reworked this to keep the feature.

Two defects (not one)

Worth leading with the shared-class bug, because it is independent of the API discussion:

1. start_trigger_args is a class-level StartTriggerArgs. init did self.start_trigger_args.trigger_kwargs = {...}, which mutates the shared class object. Every TimeSensor ended up with the last-constructed sensor's moment. That is a real bug regardless of whether we keep start_from_trigger.

2. The mutated trigger_kwargs["moment"] was an absolute datetime from datetime.now() at parse time, so every Dag-processor parse produced a different serialized blob and a new DagVersion.

Design: TimeOfDayTrigger (API preserved, no major bump)

I took the lazy-moment path you suggested rather than dropping the feature.

• New TimeOfDayTrigger stores only parse-stable kwargs: target_time (ISO string — datetime.time is not serde-safe), timezone (IANA name or fixed offset seconds), end_from_trigger.
• Concrete moment is resolved when the trigger starts, then the wait loop is delegated to DateTimeTrigger (no duplicated sleep logic).
• TimeSensor.init keeps start_from_trigger in the explicit signature (so inspect/IDE/docs still see it — addresses your kwargs.pop note).
• When start_from_trigger=True it builds a per-instance StartTriggerArgs and never mutates the class attribute (fixes defect 1).
• execute()/poke() resolve once per attempt and cache (no midnight recompute drift). target_datetime is documented as cached-on-first-access for the life of the instance.
• trigger_kwargs stays accepted-and-ignored, matching main (agreed — not a real change).

No major bump: public API is preserved, so this is a minor bugfix. Providers don't use newsfragments; no breaking-change Changelog entry under the Changelog header.

Inline notes

• Signature: start_from_trigger is back in init (not kwargs.pop).
• target_datetime property: docstring now states it is resolved once and cached for the attempt; value semantics are "today at first access", not "recompute every read".
• Dag timezone: uses self._dag (no bare except Exception). Falls back to UTC only when the operator is not yet attached to a Dag (unit tests / construct-before-add_task). When a Dag is present, its timezone is used.

DST

• Spring-forward gap (e.g. 02:30 America/New_York on 2024-03-10): shift forward to next valid local time (03:30).
• Fall-back ambiguous: fold=0 (first occurrence).

Tests (39 passed)

Headline: serialize the same Dag twice with datetime.now() mocked to two different values → LazyDeserializedDAG hash and to_dict() identical.

Also: already-passed today, midnight cache, DST spring/fall, no-Dag UTC fallback, TimeSensorAsync, FixedTimezone + named tz round-trip, two sensors don't share start_trigger_args, end_from_trigger on both start_from_trigger and defer paths, start_from_trigger still in inspect.signature.

Happy to adjust further if other maintainers want a different DST policy or want TimeOfDayTrigger to live elsewhere.

[Vamsi-klu]

You were right that removal wasn't the only option, and I ended up building the lazy version you suggested. start_from_trigger stays, so no major bump: there's a new TimeOfDayTrigger that stores the wall-clock time plus timezone and only resolves the concrete UTC moment when the trigger actually starts, then hands off to DateTimeTrigger's existing wait loop. start_trigger_args is built per instance now, which also kills the shared-class mutation bug. There are tests that serialize the same Dag under two different frozen clocks and expect byte-identical payloads, so the hash churn can't quietly come back.

The earlier inline comments (the kwargs.pop signature, recomputed target_datetime, the getattr UTC fallback) were all about the removal version. That code doesn't exist anymore.

The latest push also fixes the three CI failures that were mine: an RST error in the new trigger docstring that broke the docs build, the pendulum in_timezone typing, and the serialization tests now pick whichever API exists on the running Airflow (2.11 through main) instead of assuming LazyDeserializedDAG.from_dag.

On DST, the semantics I went with: a target time inside the spring-forward gap shifts to the next valid wall time, an ambiguous fall-back time takes the first occurrence, and "already passed today" still fires today rather than rolling to tomorrow. All documented and tested, but if maintainers want different behavior I'm fine changing it. A second pair of eyes on the trigger serialization would be welcome.

---

Drafted-by: Claude Code (Fable 5); reviewed by @Vamsi-klu before posting

[uranusjr]

This and #69925 fix the same thing in different ways. We should first figure out the correct approach in the issue.