[Brook] Wrote Workflow Records through Local Activities.

scripts/_proto/temporal/api/streamservice/v1/stream_service.proto

@@ -127,6 +127,14 @@ message CreateStreamRequest {
     // TTL. The Store applies it on the next access, so no call sees the Stream open after the
     // period. UNIMPLEMENTED (detail stream_unsupported) on a Store that can't.
     google.protobuf.Duration auto_close = 5;
+    // The Workflow run that owns the Stream. Workflow kind, with its run_id and epoch.
+    Producer owner = 6;
+    // Close the Stream with the result or the failure of the owner's run chain when the chain
+    // closes, terminate and timeout included. Needs owner, and ref with the owner fields.
+    bool close_with_run = 7;
+    // The ref of the Stream. A Workflow-owned Stream sets workflow_id and
+    // first_execution_run_id, which name the owner's run chain.
+    StreamRef ref = 8;
 }
 
 enum StreamStatus {
@@ -152,7 +160,8 @@ message AppendRecordsRequest {
     // The batch number in this epoch. With the Producer and epoch it is the dedupe key.
     int64 sequence = 4;
     repeated AppendRecord records = 5;
-    // Workflow only. The failed tasks are those since this run's last flush to this Stream.
+    // Local Activity only. The run's last completed Workflow Task and its failed or timed out
+    // Workflow Tasks, so the batch withdraws the run's orphaned Local Activity Records.
     WorkflowTaskEpoch previous_completed_task = 6;
     repeated int64 failed_tasks = 7;
     // Workflow only. A retry run's first batch: the previous run, whose Records it withdraws.

scripts/_proto/temporal/sdk/streams/v1/extension.proto

@@ -19,10 +19,9 @@ message RedisStreamConfig {
     google.protobuf.Duration retention = 4;
     // How long an open Workflow-owned Stream stays after its last write. Empty is 90 days.
     google.protobuf.Duration safety_ttl = 5;
-    // When above zero, each Workflow flush runs `WAIT durable_writes <durable_timeout>`.
-    int32 durable_writes = 6;
-    // Empty is 1 s.
-    google.protobuf.Duration durable_timeout = 7;
+    // A write is durable as an Activity write is, so a Workflow write has no replica wait.
+    reserved 6, 7;
+    reserved "durable_writes", "durable_timeout";
     // Zero is 256.
     int32 blocking_reads_per_node = 8;
     // Empty is 5 s.
@@ -32,61 +31,6 @@ message RedisStreamConfig {
     string backend = 10;
 }
 
-// Method `BufferWorkflowCalls`, synchronous. Lang calls it at the end of each Activation of a
-// run, with the calls that Workflow code made in that Activation, in order. On replay, lang
-// sends only the creates and the closes. The crate writes the buffer in `before_complete` and
-// sends the closes in `after_complete`.
-message BufferWorkflowCallsRequest {
-    string run_id = 1;
-    // A retry run only: the previous run. The crate puts it on each create and append of the
-    // run, so the first one on each Stream withdraws the previous run's Records.
-    string withdraw_run = 2;
-    repeated WorkflowCall calls = 3;
-    // Set on the run's last call when the server goes on with the Run Chain, which is a failure
-    // that the retry policy retries, or any cron run. The crate then skips the closes that have
-    // `when_run_closes`.
-    bool chain_continues = 4;
-}
-
-message WorkflowCall {
-    oneof call {
-        temporal.api.streamservice.v1.CreateStreamRequest create = 1;
-        // The crate sets the Producer, the sequence and the History facts.
-        temporal.api.streamservice.v1.AppendRecordsRequest append = 2;
-        // The crate sets the Producer and the History facts.
-        temporal.api.streamservice.v1.CloseStreamRequest close = 3;
-    }
-    // For a close only: send it only when the server accepts a completion that closes the run.
-    // `auto_close` sets it.
-    bool when_run_closes = 4;
-}
-
-message BufferWorkflowCallsResponse {
-}
-
-// Method `FlushRunBuffer`, asynchronous. Lang calls it before it runs a Local Activity, also
-// before each retry that Core runs in the same task. The crate writes the run's buffered
-// creates and appends in the epoch of the Workflow Task that dispatched the Local Activity, so
-// the run's Records come before the Local Activity's. Closes still wait for an accepted
-// completion. The task's completion writes only what came later, with the numbering going on,
-// and sends the task's withdrawal facts to each Stream this call wrote. A Local Activity of a
-// task that already completed, or of a task older than an earlier flush, writes nothing. When
-// the call fails, the run's next completion fails with the same error, so the retry withdraws
-// what was written. Lang then fails the Local Activity attempt.
-message FlushRunBufferRequest {
-    string namespace = 1;
-    string run_id = 2;
-    // `ActivityTask.Start.workflow_task` (field 20) of the Local Activity.
-    temporal.api.streamservice.v1.WorkflowTaskEpoch workflow_task = 3;
-    // How long the write may take. Empty is 10 s.
-    google.protobuf.Duration timeout = 4;
-}
-
-message FlushRunBufferResponse {
-    // The Records the call wrote. Zero when the run had nothing buffered.
-    int32 records = 1;
-}
-
 // Method `OpenWriter`, asynchronous. A writer for an Activity or a client Producer, which keeps
 // its epoch and `sequence` order. A writer of an Activity attempt above 1 first sends an empty
 // batch, so the new epoch withdraws the earlier attempt's Records.
@@ -101,6 +45,10 @@ message OpenWriterRequest {
     int32 max_batch_size = 5;
     // How long a write retries a Store failure. Empty is 10 s.
     google.protobuf.Duration max_retry_duration = 6;
+    // Local Activity only. The run's task facts, which each batch sends as AppendRecords does, so
+    // the writes withdraw the run's orphaned Local Activity Records.
+    temporal.api.streamservice.v1.WorkflowTaskEpoch previous_completed_task = 8;
+    repeated int64 failed_tasks = 9;
 }
 
 message OpenWriterResponse {
@@ -144,18 +92,6 @@ message CloseWritersResponse {
     int32 failed = 1;
 }
 
-// Method `DrainCloses`, asynchronous. Waits for the closes that `after_complete` started, for
-// example at Worker shutdown.
-message DrainClosesRequest {
-    // Empty is 10 s.
-    google.protobuf.Duration timeout = 1;
-}
-
-message DrainClosesResponse {
-    // The closes still running when the timeout passed.
-    int32 pending = 1;
-}
-
 // The details of a failed call, in the error's details bytes.
 message StreamErrorDetails {
     // A token for lang, such as `stream_closed`, `stream_fenced` or `stream_format_too_new`.

scripts/_wit/stream-service.wit

@@ -162,6 +162,12 @@ interface stream-service {
     withdraw-run: option<string>,
     /// @nexus.doc "Close with no result after this long with no append. Empty never auto-closes."
     auto-close: option<duration>,
+    /// @nexus.doc "The Workflow run that owns the Stream. The Workflow transport sets it."
+    owner: option<producer>,
+    /// @nexus.doc "Close with the owner run chain's result when it closes. Native only."
+    close-with-run: option<bool>,
+    /// @nexus.doc "The Stream's ref, with the owner fields."
+    %ref: option<stream-ref>,
     /// @nexus.omit
     namespace: placeholder,
   }
@@ -186,9 +192,9 @@ interface stream-service {
     /// @nexus.doc "The batch number in this epoch, 1 or more. It is part of the dedupe key."
     sequence: option<s64>,
     records: list<append-record>,
-    /// @nexus.doc "Workflow only. The previous completed Workflow Task of the run."
+    /// @nexus.doc "Local Activity only. The run's last completed Workflow Task."
     previous-completed-task: option<workflow-task-epoch>,
-    /// @nexus.doc "Workflow only. Failed or timed out tasks since this run's last flush here."
+    /// @nexus.doc "Local Activity only. The run's failed or timed out Workflow Tasks."
     failed-tasks: list<s64>,
     /// @nexus.doc "Retry run only. The previous run, whose Records this withdraws."
     withdraw-run: option<string>,

scripts/_wit/stream-writer.wit

@@ -16,7 +16,9 @@ world writer {
 /// @nexus.experimental
 interface stream-writer {
   use nexus:temporal-types/model@1.0.0.{duration, placeholder};
-  use temporal:nexus/stream-service@1.0.0.{append-record, producer, stream-ref};
+  use temporal:nexus/stream-service@1.0.0.{
+    append-record, producer, stream-ref, workflow-task-epoch,
+  };
 
   /// @nexus.doc "Open a writer for one Producer of one Stream."
   /// @nexus.proto "temporal.sdk.streams.v1.OpenWriterRequest"
@@ -27,6 +29,10 @@ interface stream-writer {
     max-batch-size: option<s32>,
     /// @nexus.doc "How long a write retries a Store failure. Empty is 10 s."
     max-retry-duration: option<duration>,
+    /// @nexus.doc "Local Activity only. The run's last completed Workflow Task."
+    previous-completed-task: option<workflow-task-epoch>,
+    /// @nexus.doc "Local Activity only. The run's failed or timed out Workflow Tasks."
+    failed-tasks: list<s64>,
     /// @nexus.omit
     namespace: placeholder,
   }

temporalio/activity.py

@@ -108,6 +108,12 @@ class WorkflowTaskAttempt:
     started_time: datetime | None
     """When the server started the task. None if the server did not send it."""
 
+    previous_started_event_id: int = 0
+    """ID of the ``WorkflowTaskStarted`` event of the run's last completed task. 0 if none."""
+
+    failed_tasks: Sequence[int] = ()
+    """``WorkflowTaskStarted`` event IDs of the run's failed and timed out tasks, the last 100."""
+
 
 @dataclass(frozen=True)
 class Info: