Define source-scoped summary completion barriers

[zzylol]

Why

Cross-source maintenance needs an explicit proof that one ordered producer partition has delivered every window through an event-time boundary. The worker's diagnostic maximum timestamp cannot provide that proof, and silently accepting a barrier in a sink that does not implement coordination can publish incomplete state.

What

• Define strict, language-neutral SDS completion and watermark contracts.
• Scope source identity by producer, partition, and producer epoch.
• Carry the existing SummaryInstanceId for the concrete SDS instance and keep time/group data as SummaryInstanceCoordinates.
• Require sinks to opt into watermark support; the default returns Unsupported.

How

Completion records bind catalog generation, source incarnation, concrete instance ID, half-open time range, named group coordinates, and retry-stable lineage. Validation rejects empty sources and lineage, zero epochs and sequences, invalid ranges, and invalid catalog identities. Wire structs deny unknown fields so incompatible producers fail closed.

Before

Maintenance code could infer completion from emitted payloads or unrelated worker timestamps. A sink could also discard a watermark through the trait default while reporting success.

After

A coordinator can distinguish producer restarts, reject malformed completion claims, and negotiate barrier support explicitly before using source watermarks.

Verification

• cargo test -p asap_types sds::tests — 17 passed
• git diff --check
• The focused data-plane test compiled through the data-plane crate but local linking failed because the workspace filesystem reached 100% and lld exited with SIGBUS. The worktree target was cleaned; CI provides the clean-environment result.

Limitations

This PR defines the shared boundary. The next stacked change persists staged inputs, barriers, and atomic publication records for restart-safe coordination.

[zzylol]

Correctness findings before the coordination journal builds on this contract:

1. OutputSink::advance_summary_watermark defaults to Ok(()) while discarding the barrier. A producer can therefore treat a completeness claim as accepted even when the sink cannot persist or forward it. The default should return an explicit unsupported error, or support must be negotiated as a capability.
2. Monotonic sequence is scoped only by (CatalogGeneration, producer_id, partition_id), but the contract does not define producer_id as a restart incarnation. A restarted producer can reuse sequence numbers under the same catalog generation. Add a producer epoch/incarnation or make it an explicit invariant in a validated typed identity.
3. Add validation/tests for nonempty source components, a nonempty half-open window, retry-stable nonempty lineage, and malformed serde input. The next journal must reject regressions and equal-sequence/different-payload conflicts rather than relying on these structs alone.
4. SummaryWindowIdentity repeats the coordinate fields already present in SummaryInstance. Please make the relationship explicit: either rename it to SummaryInstanceCoordinates and derive/use the stable SummaryInstanceId, or carry that instance ID in completion records. Otherwise storage inventory and completion idempotency can create two identities for the same state.

The source-partition event-time watermark model is otherwise the right boundary, provided downstream closes only windows whose end is covered by the barrier.