feat(runtime): publish model retry lifecycle (#80)

Author: yanpgwang

docs/architecture/domain-model.md

@@ -54,12 +54,14 @@ The public session status is a projection:
 
 - `idle` — waiting for input or a required client action;
 - `running` — accepted work is executing or queued;
-- `rescheduling` — reserved for automatic retry behavior;
+- `rescheduling` — an active model request is waiting for its next bounded
+  provider retry;
 - `terminated` — terminal failure or completion.
 
-Temporal may retry failed Activities internally without changing the public
-status. The public `rescheduling` transition is not implemented; an exhausted
-or terminal turn failure projects directly to `terminated`.
+Temporal may retry infrastructure-failed Activities internally without changing
+the public status. Retryable provider responses instead use the public
+`running -> rescheduling -> running` lifecycle. Exhausting that bounded budget
+returns the Session to `idle`; permanent failures project to `terminated`.
 
 ## Event
 

docs/architecture/session-lifecycle.md

@@ -15,6 +15,9 @@ stateDiagram-v2
   [*] --> idle
   idle --> running: claimable input admitted
   running --> idle: turn completed or requires action
+  running --> rescheduling: retryable model failure
+  rescheduling --> running: retry delay elapsed
+  rescheduling --> idle: interrupted
   running --> terminated: terminal failure
   idle --> terminated: deleted or terminal operation
   terminated --> [*]
@@ -23,8 +26,8 @@ stateDiagram-v2
 - `idle` means the Session is waiting for ordinary input or a required client
   action.
 - `running` means accepted, claimable work is queued or executing.
-- `rescheduling` is reserved for a future public retry policy. Temporal
-  Activity retries remain private and do not use it.
+- `rescheduling` means an immutable model request is waiting for its next
+  bounded provider retry. Later input remains queued behind the active turn.
 - `terminated` is final.
 
 The status is a projection of committed events. It is not derived from worker
@@ -129,16 +132,21 @@ not supported.
 
 ## Failure and delivery semantics
 
-- Permanent model failures commit `session.error` and
-  `session.status_terminated`.
-- Retryable model and infrastructure failures remain Activity failures so
-  Temporal can apply its retry policy.
+- Permanent model failures commit `session.error` with `retry_status: terminal`
+  and `session.status_terminated`.
+- Retryable provider responses publish `session.error` with
+  `retry_status: retrying`, atomically move through
+  `session.status_rescheduled`, and publish `session.status_running` before the
+  next attempt. Exhaustion returns the Session to idle with
+  `stop_reason: retries_exhausted` and flushes later queued messages.
+- Infrastructure failures remain Activity failures so Temporal can recover
+  without spending the public provider retry budget.
 - Authoritative output is visible only after the PostgreSQL completion
   transaction commits.
 - NATS previews and wakeups are best-effort. Clients recover complete events
   from the PostgreSQL-backed event cursor.
-- Workflow and Activity retries are private implementation details and never
-  create duplicate public events.
+- Activity retries for infrastructure recovery remain private. Workflow-owned
+  provider retries use deterministic event IDs and never create duplicates.
 
 ## Session deletion
 

docs/compatibility.md

@@ -46,6 +46,7 @@ compatibility.
 | Tool confirmations | Supported | Interceptable `always_ask` built-ins and MCP tools park durably. Allow executes the original server-owned call through the tool journal; deny returns an error tool result with the optional denial message. Provider-native Web Search/Fetch cannot be intercepted and reject `always_ask`. |
 | User interrupt | Limited | An untargeted interrupt durably cancels an active model, outcome grader, or tool Activity across API and worker processes. PostgreSQL defines finish-vs-interrupt ordering, closes model spans, emits one idle `end_turn`, and fences uncertain started tool steps as ambiguous. Targeted multi-agent interrupts are not supported. |
 | Terminal execution errors | Supported | A deterministic turn failure emits `session.error` with the documented `unknown_error` variant and `retry_status: terminal`, followed by `session.status_terminated`. MCP setup failures retain their more specific documented event variant. Older persisted histories may contain the legacy `api_error` spelling. |
+| Model retry lifecycle | Supported | Retryable provider responses emit typed `model_overloaded_error`, `model_rate_limited_error`, or `model_request_failed_error` events. Each bounded retry publishes `retrying`, `session.status_rescheduled`, then `session.status_running`; exhaustion emits `exhausted`, returns idle with `retries_exhausted`, and flushes later queued messages. Provider `Retry-After` is honored up to the server cap. Infrastructure recovery remains private to Temporal. |
 | MCP execution | Limited | Remote tools are discovered over unauthenticated Streamable HTTP, pinned per Session, permission-checked, journaled, and executed with large or binary results materialized in the Session sandbox. Calls publish `agent.mcp_tool_use` with the required `mcp_server_name` and the bare server-side tool name, answered by `agent.mcp_tool_result` carrying `mcp_tool_use_id`; an `always_ask` MCP call still parks for a `user.tool_confirmation` referencing `tool_use_id`. Authentication, private-network access, deprecated-SSE fallback, resources, and prompts are not supported. |
 | Files, skills, memory, and vaults | Not supported | These product surfaces are outside the core harness scope. File-backed outcome rubrics and file-sourced message content are rejected explicitly; inline or URL image/document content still participates in the provider transcript and context projection. |
 | Multi-agent orchestration | Not supported | `multiagent` configuration can be stored, but rosters, threads, delegation, and orchestration are not executed. |

internal/domain/session.go

@@ -17,7 +17,7 @@ const (
 var allowed = map[Status]map[Status]bool{
 	StatusIdle:         {StatusRunning: true, StatusTerminated: true},
 	StatusRunning:      {StatusIdle: true, StatusRescheduling: true, StatusTerminated: true},
-	StatusRescheduling: {StatusRunning: true, StatusTerminated: true},
+	StatusRescheduling: {StatusIdle: true, StatusRunning: true, StatusTerminated: true},
 	StatusTerminated:   {},
 }
 

internal/domain/session_test.go

@@ -9,6 +9,7 @@ func TestStatusTransitions(t *testing.T) {
 	ok := [][2]Status{
 		{StatusIdle, StatusRunning}, {StatusRunning, StatusIdle},
 		{StatusRunning, StatusRescheduling}, {StatusRescheduling, StatusRunning},
+		{StatusRescheduling, StatusIdle},
 		{StatusIdle, StatusTerminated}, {StatusRunning, StatusTerminated},
 	}
 	for _, p := range ok {

internal/httpapi/openapi.yaml

@@ -2195,7 +2195,16 @@ components:
         type:
           type: string
           description: api_error may appear only in histories written before the terminal Session error contract.
-          enum: [unknown_error, mcp_connection_failed_error, api_error]
+          enum:
+            - unknown_error
+            - model_overloaded_error
+            - model_rate_limited_error
+            - model_request_failed_error
+            - mcp_connection_failed_error
+            - mcp_authentication_failed_error
+            - billing_error
+            - credential_host_unreachable_error
+            - api_error
         message: {type: string}
         retry_status:
           type: object

internal/httpapi/sdk_test.go

@@ -88,6 +88,66 @@ func TestSDK_TerminalSessionErrorEvent(t *testing.T) {
 	}
 }
 
+func TestSDK_ModelRetryLifecycleEvents(t *testing.T) {
+	client, ts, sessions := sdkClientServerAndSessions(t)
+	ctx := context.Background()
+
+	agent := mustAgent(t, client, "opus", "sys")
+	environmentID := mustEnv(t, ts.URL)
+	session, err := client.Beta.Sessions.New(ctx, anthropic.BetaSessionNewParams{
+		Agent:         anthropic.BetaSessionNewParamsAgentUnion{OfString: anthropic.String(agent.ID)},
+		EnvironmentID: environmentID,
+	})
+	if err != nil {
+		t.Fatalf("create session: %v", err)
+	}
+
+	sessions.mu.Lock()
+	sessions.appendEventLocked(session.ID, domain.EventDraft{
+		Type: domain.EvSessionError,
+		Payload: map[string]any{"error": map[string]any{
+			"type": "model_rate_limited_error", "message": "slow down",
+			"retry_status": map[string]any{"type": "retrying"},
+		}},
+	})
+	sessions.appendEventLocked(session.ID, domain.EventDraft{
+		Type: domain.EvSessionStatusRescheduling, Payload: map[string]any{},
+	})
+	sessions.appendEventLocked(session.ID, domain.EventDraft{
+		Type: domain.EvSessionStatusRunning, Payload: map[string]any{},
+	})
+	sessions.mu.Unlock()
+
+	page, err := client.Beta.Sessions.Events.List(
+		ctx,
+		session.ID,
+		anthropic.BetaSessionEventListParams{Types: []string{
+			domain.EvSessionError,
+			domain.EvSessionStatusRescheduling,
+			domain.EvSessionStatusRunning,
+		}},
+	)
+	if err != nil {
+		t.Fatalf("list retry lifecycle: %v", err)
+	}
+	if len(page.Data) != 3 {
+		t.Fatalf("retry lifecycle event count = %d, want 3", len(page.Data))
+	}
+	retry := page.Data[0].AsSessionError().Error.AsModelRateLimitedError()
+	if retry.Type != anthropic.BetaManagedAgentsModelRateLimitedErrorTypeModelRateLimitedError ||
+		retry.Message != "slow down" || retry.RetryStatus.AsRetrying().Type != "retrying" {
+		t.Fatalf("retry event = %s", page.Data[0].RawJSON())
+	}
+	if page.Data[1].AsSessionStatusRescheduled().Type !=
+		anthropic.BetaManagedAgentsSessionStatusRescheduledEventTypeSessionStatusRescheduled {
+		t.Fatalf("rescheduled event = %s", page.Data[1].RawJSON())
+	}
+	if page.Data[2].AsSessionStatusRunning().Type !=
+		anthropic.BetaManagedAgentsSessionStatusRunningEventTypeSessionStatusRunning {
+		t.Fatalf("running event = %s", page.Data[2].RawJSON())
+	}
+}
+
 func TestSDK_AgentLifecycle(t *testing.T) {
 	client, _ := sdkClientAndServer(t)
 	ctx := context.Background()

internal/pg/completion_test.go

@@ -171,3 +171,220 @@ func TestAppendWorkflowEvents_IdempotentAndIncludedInCompletionReplay(t *testing
 		t.Fatal("mismatched idempotency payload unexpectedly succeeded")
 	}
 }
+
+func TestWorkflowRetryTransitionsAreAtomicIdempotentAndKeepQueuedInputBehindTurn(t *testing.T) {
+	store := testStore(t)
+	ctx := context.Background()
+	session := newSession("sess_retry_transitions")
+	if _, err := store.CreateSession(ctx, session, nil); err != nil {
+		t.Fatalf("create: %v", err)
+	}
+	admission, err := store.AdmitEvents(ctx, session.ID, []domain.EventDraft{{
+		Type: domain.EvUserMessage, Payload: map[string]any{"content": "first"},
+	}})
+	if err != nil {
+		t.Fatalf("admit trigger: %v", err)
+	}
+	trigger := admission.Events[0]
+	errorPayload := map[string]any{
+		"type":    "model_overloaded_error",
+		"message": "provider overloaded",
+		"retry_status": map[string]any{
+			"type": "retrying",
+		},
+	}
+	for attempt := 0; attempt < 2; attempt++ {
+		if err := store.RecordWorkflowRetry(
+			ctx,
+			session.ID,
+			trigger.ID,
+			"sevt_retry_error",
+			"sevt_retry_rescheduled",
+			errorPayload,
+		); err != nil {
+			t.Fatalf("record retry %d: %v", attempt+1, err)
+		}
+	}
+	projected, err := store.GetSession(ctx, session.ID)
+	if err != nil {
+		t.Fatalf("get rescheduling session: %v", err)
+	}
+	if projected.Status != domain.StatusRescheduling {
+		t.Fatalf("status = %s, want rescheduling", projected.Status)
+	}
+
+	queued, err := store.AdmitEvents(ctx, session.ID, []domain.EventDraft{{
+		Type: domain.EvUserMessage, Payload: map[string]any{"content": "queued"},
+	}})
+	if err != nil {
+		t.Fatalf("queue later message: %v", err)
+	}
+	if queued.Session.Status != domain.StatusRescheduling {
+		t.Fatalf("queued admission status = %s, want rescheduling", queued.Session.Status)
+	}
+	if len(queued.Events) != 1 || queued.Events[0].Type != domain.EvUserMessage {
+		t.Fatalf("queued admission events = %v", eventTypes(queued.Events))
+	}
+
+	for attempt := 0; attempt < 2; attempt++ {
+		if err := store.ResumeWorkflowRetry(
+			ctx, session.ID, trigger.ID, "sevt_retry_running",
+		); err != nil {
+			t.Fatalf("resume retry %d: %v", attempt+1, err)
+		}
+	}
+	projected, err = store.GetSession(ctx, session.ID)
+	if err != nil {
+		t.Fatalf("get resumed session: %v", err)
+	}
+	if projected.Status != domain.StatusRunning {
+		t.Fatalf("status = %s, want running", projected.Status)
+	}
+	events, err := store.EventsAfter(ctx, session.ID, 0, 100)
+	if err != nil {
+		t.Fatalf("list events: %v", err)
+	}
+	if got := eventTypes(events); !slices.Equal(got, []string{
+		domain.EvUserMessage,
+		domain.EvSessionStatusRunning,
+		domain.EvSessionError,
+		domain.EvSessionStatusRescheduling,
+		domain.EvUserMessage,
+		domain.EvSessionStatusRunning,
+	}) {
+		t.Fatalf("event order = %v", got)
+	}
+}
+
+func TestRetriesExhaustedReturnsIdleAndFlushesQueuedMessages(t *testing.T) {
+	store := testStore(t)
+	ctx := context.Background()
+	session := newSession("sess_retry_flush")
+	if _, err := store.CreateSession(ctx, session, nil); err != nil {
+		t.Fatalf("create: %v", err)
+	}
+	first, err := store.AdmitEvents(ctx, session.ID, []domain.EventDraft{{
+		Type: domain.EvUserMessage, Payload: map[string]any{"content": "first"},
+	}})
+	if err != nil {
+		t.Fatalf("admit first: %v", err)
+	}
+	queued, err := store.AdmitEvents(ctx, session.ID, []domain.EventDraft{
+		{Type: domain.EvUserMessage, Payload: map[string]any{"content": "second"}},
+		{Type: domain.EvSystemMessage, Payload: map[string]any{"content": "context"}},
+	})
+	if err != nil {
+		t.Fatalf("admit queued pair: %v", err)
+	}
+	completion, err := store.CompleteWorkflowTurn(
+		ctx,
+		session.ID,
+		first.Events[0].ID,
+		[]domain.EventDraft{
+			{Type: domain.EvSessionError, Payload: map[string]any{"error": map[string]any{
+				"type":    "model_request_failed_error",
+				"message": "retry budget exhausted",
+				"retry_status": map[string]any{
+					"type": "exhausted",
+				},
+			}}},
+			{Type: domain.EvSessionStatusIdle, Payload: map[string]any{
+				"stop_reason": map[string]any{"type": "retries_exhausted"},
+			}},
+		},
+		domain.StatusIdle,
+		"",
+		"",
+		nil,
+		nil,
+		nil,
+	)
+	if err != nil {
+		t.Fatalf("complete exhausted turn: %v", err)
+	}
+	if completion.Session.Status != domain.StatusIdle {
+		t.Fatalf("status = %s, want idle", completion.Session.Status)
+	}
+	queuedMessage, err := store.GetEvent(ctx, session.ID, queued.Events[0].ID)
+	if err != nil {
+		t.Fatalf("get queued message: %v", err)
+	}
+	if queuedMessage.ProcessedAt == nil {
+		t.Fatal("queued message was not flushed")
+	}
+	queuedSystem, err := store.GetEvent(ctx, session.ID, queued.Events[1].ID)
+	if err != nil {
+		t.Fatalf("get queued system message: %v", err)
+	}
+	if queuedSystem.ProcessedAt == nil {
+		t.Fatal("queued companion system message was not flushed")
+	}
+}
+
+func TestInterruptCanReturnAReschedulingSessionToIdle(t *testing.T) {
+	store := testStore(t)
+	ctx := context.Background()
+	session := newSession("sess_retry_interrupt")
+	if _, err := store.CreateSession(ctx, session, nil); err != nil {
+		t.Fatalf("create: %v", err)
+	}
+	admission, err := store.AdmitEvents(ctx, session.ID, []domain.EventDraft{{
+		Type: domain.EvUserMessage, Payload: map[string]any{"content": "retry me"},
+	}})
+	if err != nil {
+		t.Fatalf("admit trigger: %v", err)
+	}
+	trigger := admission.Events[0]
+	if err := store.RecordWorkflowRetry(
+		ctx,
+		session.ID,
+		trigger.ID,
+		"sevt_interrupt_retry_error",
+		"sevt_interrupt_rescheduled",
+		map[string]any{
+			"type":    "model_request_failed_error",
+			"message": "temporary failure",
+			"retry_status": map[string]any{
+				"type": "retrying",
+			},
+		},
+	); err != nil {
+		t.Fatalf("record retry: %v", err)
+	}
+	interrupt, err := store.AdmitEvents(ctx, session.ID, []domain.EventDraft{{
+		Type: domain.EvUserInterrupt, Payload: map[string]any{},
+	}})
+	if err != nil {
+		t.Fatalf("admit interrupt: %v", err)
+	}
+	completion, err := store.CompleteWorkflowTurn(
+		ctx,
+		session.ID,
+		trigger.ID,
+		[]domain.EventDraft{{
+			Type: domain.EvSessionStatusIdle,
+			Payload: map[string]any{
+				"stop_reason": map[string]any{"type": "end_turn"},
+			},
+		}},
+		domain.StatusIdle,
+		"",
+		"",
+		nil,
+		nil,
+		nil,
+	)
+	if err != nil {
+		t.Fatalf("complete interrupted retry: %v", err)
+	}
+	if completion.Session.Status != domain.StatusIdle {
+		t.Fatalf("status = %s, want idle", completion.Session.Status)
+	}
+	storedInterrupt, err := store.GetEvent(ctx, session.ID, interrupt.Events[0].ID)
+	if err != nil {
+		t.Fatalf("get interrupt: %v", err)
+	}
+	if storedInterrupt.ProcessedAt == nil {
+		t.Fatal("interrupt was not acknowledged")
+	}
+}