fix(otel): end recording spans on non-terminal

[ayushiahjolia]

Issue #, if available:
#642

Description of changes:

• Workflow span: created+ended once, on the terminal invocation (both plugins).
• Attempt span: created+ended in the same invocation (both plugins).
• Operation span: invocation plugin → one segment per invocation (fresh ID + link across invocations); execution plugin → one span total, created+ended in the invocation where the operation terminates (held as a non-recording placeholder while it spans invocations).

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[github-actions]

Codex AI review · Finding arf_v1_2pna33ayjyhsqx2ehym77e3n7t

[P1] Do not reset terminal-export deduplication after each invocation. ReplayChildren contexts emit on_operation_end(..., is_replayed=True) on every replay. Clearing this set lets each invocation export the same deterministic span ID again, which collectors may overwrite or reject. Skip replayed terminal callbacks or otherwise ensure execution-wide uniqueness, and add a multi-invocation ReplayChildren test asserting one export.

[github-actions]

Codex AI review · Finding arf_v1_dv5gcy7ci2c3j7gxu7vdwsejwy

[P2] Close attempts overwritten by same-key re-entry. This sweep can retrieve only the latest span stored for each key. If an incomplete STEP resumes in-process with the same attempt, _start_span() replaces the first recording span; ending the replacement here leaves the original unreachable and unended. End or retain the existing attempt before replacement, and test that both span objects stop recording after re-entry.

[github-actions]

Codex AI review · Finding arf_v1_ks67rhkdvrki2tkwtewoob5vbj

[P2] Preserve deferred timestamps when a context re-enters. A suspended child context can re-enter in-process with the same operation ID. Replacing its DurableParentSpan discards the first run's earliest start and child end timestamps; if the asynchronous START checkpoint reports a later start, the eventual parent span can begin after an already-exported child. Reuse the existing placeholder and note the new start time, then test that a resumed context encloses children emitted before suspension.

[github-actions]

Codex AI review

Found one high-severity replay/export-identity regression and two medium-severity re-entry lifecycle defects. The added tests do not cover repeated ReplayChildren exports or same-key re-entry state preservation.

Reviewed commit 00efaf2f592cf28b40b0beb7b83746c4fd2dccee. Workflow run

[github-actions]

Claude AI review · Finding arf_v1_fmn2a33o4jdr6kuqfzdrvwg7ok

note_start_time/note_end_time mutate _earliest_start_time/_latest_end_time with an unsynchronized read-compare-write, but the object is shared as the parent placeholder for durable operations such as PARALLEL/MAP containers and open child contexts. context.parallel()/context.map() run branches concurrently on separate OS threads via ConcurrentExecutor's ThreadPoolExecutor (see operation/parallel.py, operation/child.py), and each branch's on_user_function_start/on_user_function_end in execution_plugin.py calls _note_parent_start(info.parent_id, ...)/_note_parent_end(info.parent_id, ...) on the same container DurableParentSpan instance whenever parent_id names the container operation. Two branch threads racing on if self._earliest_start_time is None or timestamp < self._earliest_start_time: self._earliest_start_time = timestamp can lose one update, so the eventually-materialized operation span can fail to enclose one of its concurrent children — exactly the defect normalized_start_time/normalized_end_time are meant to prevent. Every other piece of shared plugin state (_operation_spans, _context_tokens) is guarded by self._lock/self._operation_spans_lock; this new mutable state on DurableParentSpan is not. Add a threading.Lock inside DurableParentSpan and acquire it in note_start_time, note_end_time, normalized_start_time, and normalized_end_time.

[github-actions]

Claude AI review · Finding arf_v1_d3v4nkts2wbw4vfdae6gv4poj6

_end_open_recording_spans() force-closes any recording span still open at invocation end — realistically a STEP attempt whose outcome was INCOMPLETE (e.g. the invocation suspended mid-attempt) — via a bare popped.end() with no status and none of the durable.attempt.*/error attributes a normally-completed attempt gets from on_user_function_end. In a trace viewer this truncated attempt is indistinguishable from a short, uneventful successful attempt, hiding that the invocation actually ended before the attempt finished. Mark it explicitly before ending, e.g. with a durable.span.truncated_at_invocation_boundary attribute.

Suggested change

	def _end_open_recording_spans(self) -> None:
	"""End recording user-function spans left open by a suspended operation.
	Operation placeholders are non-recording and export their span from
	on_operation_end, so they are skipped. Reverse order keeps each child
	contained within its parent; the invocation span is ended by the caller.
	"""
	with self._lock:
	keys = list(reversed(self._operation_spans))
	for key in keys:
	if key == _INVOCATION_KEY:
	continue
	span = self._get_span(key)
	if span is None or not span.is_recording():
	continue
	popped = self._pop_span(key)
	if popped is not None:
	popped.end()
	def _end_open_recording_spans(self) -> None:
	"""End recording user-function spans left open by a suspended operation.
	Operation placeholders are non-recording and export their span from
	on_operation_end, so they are skipped. Reverse order keeps each child
	contained within its parent; the invocation span is ended by the caller.
	"""
	with self._lock:
	keys = list(reversed(self._operation_spans))
	for key in keys:
	if key == _INVOCATION_KEY:
	continue
	span = self._get_span(key)
	if span is None or not span.is_recording():
	continue
	popped = self._pop_span(key)
	if popped is not None:
	popped.set_attribute(
	"durable.span.truncated_at_invocation_boundary", True
	)
	popped.end()

[github-actions]

Claude AI review

Reviewed the OTel plugin rework that turns operation/Workflow spans into deferred, non-recording DurableParentSpan placeholders and materializes the real recording span once, on the invocation where the Workflow/operation actually terminates. The design (dedup via _ended_operation_ids, skipping non-recording placeholders in _end_open_recording_spans, ensure_end_after_start/normalized_*_time clamping) is sound and correctly fixes the previously-reported duplicate-deterministic-span-ID export bug for ExecutionOtelPlugin (operations are no longer exported as real spans while open, so a replayed/suspended operation can no longer be exported twice under the same span ID — that prior finding is resolved by this PR and not re-reported).

Two issues remain:

1. A genuine, PR-introduced thread-safety gap: DurableParentSpan.note_start_time/note_end_time mutate shared state without any lock, but the object is shared as the parent placeholder for containers whose children run on different threads (e.g. context.parallel()/context.map() branches via ConcurrentExecutor's ThreadPoolExecutor). Concurrent branch completions can race and lose an update, producing an operation span that doesn't fully enclose one of its children.
2. A previously-reported, still-present cosmetic/observability gap: _end_open_recording_spans() force-ends a truncated (INCOMPLETE) attempt span with no status or attribute, making it indistinguishable from a normal short successful attempt.

No other correctness, API-compatibility, or determinism regressions were found in the reworked hooks (on_operation_start/end, on_user_function_start/end, Workflow span export) for either plugin, and the new unit/integration tests adequately cover the new placeholder/materialization behavior, dedup, and trace-state propagation. Residual risk: the concurrency race above has no dedicated test (hard to make deterministic), so a regression there would not be caught by CI.

Reviewed commit 00efaf2f592cf28b40b0beb7b83746c4fd2dccee. Workflow run