Demonstrate coordinated conversation workflows

docs/consumers/pkg-promptkit.md

@@ -695,6 +695,112 @@ application owns any semantic retry budget. When a
 pre-execution equality check is required, use `PrepareExecution` and compare
 its opaque rendered-prompt hash before invoking `RunPrepared`.
 
+### Capture And Continue A Conversation
+
+Capture records the messages actually executed and the final assistant content.
+An exchange is an immutable in-process capability; `Details()` returns detached,
+encodable history that may contain sensitive content. Ordinary result encoding
+does not include the exchange. Continuations imply capture and may branch or use
+a different engine, profile, output contract, credential, or session:
+
+```go
+initial, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "proposal", CaptureExchange: true})
+if err != nil { return err }
+left, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "refine", ContinueFrom: initial.Exchange})
+if err != nil { return err }
+right, err := anotherEngine.Run(ctx, promptkit.RunRequest{PromptID: "alternative", ContinueFrom: initial.Exchange})
+if err != nil { return err }
+_, _ = left, right // Independent branches; each result has its own exchange.
+```
+
+For a frozen execution, set `CaptureExchange` in the request passed to
+`PrepareExecution`, then obtain the exchange from `RunPrepared`'s result.
+Preparation and validation alone do not produce an exchange. Captured history
+precedes newly rendered definition messages and appended messages. A completed
+structurally invalid result still has an exchange; an operational error returns
+no result. The caller chooses what constitutes acceptance.
+
+```go
+prepared, err := engine.PrepareExecution(ctx, promptkit.RunRequest{PromptID: "proposal", CaptureExchange: true})
+if err != nil { return err }
+result, err := engine.RunPrepared(ctx, prepared)
+if err != nil { return err }
+details := result.Exchange.Details()
+_ = details // Explicitly encode or inspect when appropriate.
+```
+
+### Coordinate Application-Directed Turns
+
+The following sketches use illustrative application functions. The first records
+each newly delivered response before selecting another question:
+
+```go
+outcome, err := engine.Coordinate(ctx, promptkit.CoordinateRequest{
+	Start: initial.Exchange, AdditionalTurns: 3,
+	Next: func(ctx context.Context, turn promptkit.ConversationTurn) (*promptkit.RunRequest, error) {
+		if turn.Result != nil {
+			if !turn.Result.Validation.IsValid { return nil, fmt.Errorf("invalid response") }
+			analysis.Record(turn.Result)
+		}
+		question, ok := analysis.NextQuestion()
+		if !ok { return nil, nil }
+		return &promptkit.RunRequest{PromptID: "research/investigate", Vars: map[string]string{"question": question}}, nil
+	},
+})
+```
+
+An application can instead start with existing findings, assess each new
+candidate, and keep its decisions in a closure:
+
+```go
+outcome, err := engine.Coordinate(ctx, promptkit.CoordinateRequest{
+	Start: proposal.Exchange, AdditionalTurns: 3, Observer: branchGroup.Observe,
+	Next: func(ctx context.Context, turn promptkit.ConversationTurn) (*promptkit.RunRequest, error) {
+		if turn.Result != nil {
+			if !turn.Result.Validation.IsValid { return nil, fmt.Errorf("invalid revision") }
+			candidate, withdrawn, err := decodeRevision(turn.Result.RawOutput)
+			if err != nil { return nil, err }
+			if withdrawn { branch.MarkWithdrawn(); return nil, nil }
+			findings, err := assess(ctx, engine, candidate, validatorGroup.Observe)
+			if err != nil { return nil, err }
+			branch.Update(candidate, findings)
+			if branch.Done() { return nil, nil }
+		}
+		return &promptkit.RunRequest{PromptID: branch.FollowUpPrompt(), Vars: branch.Variables()}, nil
+	},
+})
+```
+
+`assess` is application code and explicitly selects its observer on any nested
+run. A completed turn releases its admission lease before the callback, so that
+assessment may use the same backend. `stopped` and `exhausted` describe control
+flow, not content acceptance. Cancellation before callback delivery can return one
+completed response in `Undelivered`; errors retain their ordinary identities.
+
+### Collect Selected Execution Facts
+
+Groups are independent of continuation and enrollment is explicit:
+
+```go
+var group, validatorGroup promptkit.ExecutionGroup
+request.Observer = promptkit.ComposeObservers(diagnosticObserver, group.Observe)
+prepared, err := engine.PrepareExecution(ctx, request) // no event yet
+if err != nil { return err }
+result, err := engine.RunPrepared(ctx, prepared)
+if err != nil { return err }
+_ = result
+// Set CoordinateRequest.Observer to group.Observe for its continuation runs.
+// A nested validator selects validatorGroup.Observe on its own request.
+summary, validatorSummary := group.Snapshot(), validatorGroup.Snapshot()
+_, _ = summary, validatorSummary
+```
+
+Only the first terminal event per run contributes counts and usage; coverage of
+currently observed members is separate from token-usage completeness. Empty
+groups report complete, known zero totals, while members without terminals do
+not invent totals. Overlapping groups are not generally additive. See the
+[offline example](../../examples/go-library/observe/main.go) for complete setup.
+
 ## Inputs, Profiles, And Overrides
 
 Use `File`, `Inline`, or `InlineWithURI` to construct request inputs. A request

docs/roadmap/implementation.md

@@ -585,6 +585,8 @@ retry/repair matrices.
 
 ## Stage 12 — Demonstrate The Consumer Workflows
 
+**Status:** Complete.
+
 **Scope:** practical API use and one maintained offline example.
 
 1. Extend `examples/go-library/observe` and its semantic assertions to demonstrate

examples/go-library/observe/main.go

@@ -75,6 +75,10 @@ type summary struct {
 	AssessmentAttempts         uint64                        `json:"assessment_attempts"`
 	AssessmentTransientRetries uint64                        `json:"assessment_transient_retries"`
 	AssessmentOutcome          string                        `json:"assessment_outcome"`
+	CoordinationOutcome        promptkit.CoordinationOutcome `json:"coordination_outcome"`
+	ExhaustedOutcome           promptkit.CoordinationOutcome `json:"exhausted_outcome"`
+	GroupMembers               int                           `json:"group_members"`
+	GroupAttempts              uint64                        `json:"group_attempts"`
 }
 
 func run() (summary, error) {
@@ -104,8 +108,9 @@ func run() (summary, error) {
 	}
 
 	var proposalLog eventLog
+	var executions promptkit.ExecutionGroup
 	proposal, err := engine.Run(context.Background(), promptkit.RunRequest{
-		PromptID: "example.proposal", Inputs: map[string]promptkit.ArtifactRef{"idea": promptkit.Inline("Reduce queue cleanup toil.")}, Observer: proposalLog.record,
+		PromptID: "example.proposal", Inputs: map[string]promptkit.ArtifactRef{"idea": promptkit.Inline("Reduce queue cleanup toil.")}, CaptureExchange: true, Observer: promptkit.ComposeObservers(proposalLog.record, executions.Observe),
 	})
 	if err != nil {
 		return summary{}, err
@@ -118,18 +123,22 @@ func run() (summary, error) {
 		return summary{}, err
 	}
 	var assessmentLog eventLog
-	assessment, err := engine.Run(context.Background(), promptkit.RunRequest{
+	decision := "rejected"
-		PromptID: "example.assessment", Inputs: map[string]promptkit.ArtifactRef{"proposal": promptkit.Inline(proposal.RawOutput)}, Observer: assessmentLog.record,
+	coordination, err := engine.Coordinate(context.Background(), promptkit.CoordinateRequest{
-	})
+		Start: proposal.Exchange, AdditionalTurns: 1, Observer: executions.Observe,
+		Next: func(ctx context.Context, turn promptkit.ConversationTurn) (*promptkit.RunRequest, error) {
+			if turn.Result == nil {
+				return &promptkit.RunRequest{PromptID: "example.proposal", Inputs: map[string]promptkit.ArtifactRef{"idea": promptkit.Inline("Refine the proposal.")}}, nil
+			}
+			if !turn.Result.Validation.IsValid {
+				return nil, fmt.Errorf("follow-up failed structural validation")
+			}
+			assessment, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "example.assessment", Inputs: map[string]promptkit.ArtifactRef{"proposal": promptkit.Inline(turn.Result.RawOutput)}, Observer: promptkit.ComposeObservers(assessmentLog.record, executions.Observe)})
 			if err != nil {
-		return summary{}, err
+				return nil, err
 			}
 			if !assessment.Validation.IsValid {
-		return summary{}, fmt.Errorf("assessment failed structural validation")
+				return nil, fmt.Errorf("assessment failed structural validation")
-	}
-	assessmentTerminal, err := assessmentLog.terminal()
-	if err != nil {
-		return summary{}, err
 			}
 			var proposalValue struct {
 				Risk string `json:"risk"`
@@ -137,16 +146,32 @@ func run() (summary, error) {
 			var assessmentValue struct {
 				Recommendation string `json:"recommendation"`
 			}
-	if err := json.Unmarshal([]byte(proposal.RawOutput), &proposalValue); err != nil {
+			if err := json.Unmarshal([]byte(turn.Result.RawOutput), &proposalValue); err != nil {
-		return summary{}, err
+				return nil, err
 			}
 			if err := json.Unmarshal([]byte(assessment.RawOutput), &assessmentValue); err != nil {
-		return summary{}, err
+				return nil, err
 			}
-	decision := "rejected"
 			if proposalValue.Risk == "low" && assessmentValue.Recommendation == "accept" {
 				decision = "accepted"
 			}
+			return nil, nil
+		},
+	})
+	if err != nil {
+		return summary{}, err
+	}
+	exhausted, err := engine.Coordinate(context.Background(), promptkit.CoordinateRequest{Start: proposal.Exchange, Next: func(context.Context, promptkit.ConversationTurn) (*promptkit.RunRequest, error) {
+		return &promptkit.RunRequest{PromptID: "example.proposal"}, nil
+	}})
+	if err != nil {
+		return summary{}, err
+	}
+	assessmentTerminal, err := assessmentLog.terminal()
+	if err != nil {
+		return summary{}, err
+	}
+	group := executions.Snapshot()
 	return summary{
 		ProposalAttempts:           proposalTerminal.Attempts,
 		ProposalTransientRetries:   proposalTerminal.TransientRetries,
@@ -154,6 +179,10 @@ func run() (summary, error) {
 		AssessmentAttempts:         assessmentTerminal.Attempts,
 		AssessmentTransientRetries: assessmentTerminal.TransientRetries,
 		AssessmentOutcome:          decision,
+		CoordinationOutcome:        coordination.Outcome,
+		ExhaustedOutcome:           exhausted.Outcome,
+		GroupMembers:               len(group.Members),
+		GroupAttempts:              group.Attempts,
 	}, nil
 }
 

examples/go-library/observe/main_test.go

@@ -18,13 +18,17 @@ func TestObserveExample(t *testing.T) {
 		AssessmentAttempts  *int   `json:"assessment_attempts"`
 		AssessmentRetries   *int   `json:"assessment_transient_retries"`
 		Outcome             string `json:"assessment_outcome"`
+		CoordinationOutcome string `json:"coordination_outcome"`
+		ExhaustedOutcome    string `json:"exhausted_outcome"`
+		GroupMembers        int    `json:"group_members"`
+		GroupAttempts       int    `json:"group_attempts"`
 	}
 	if err := json.Unmarshal(output.Bytes(), &got); err != nil {
 		t.Fatal(err)
 	}
 	if got.ProposalAttempts == nil || *got.ProposalAttempts != 2 || got.ProposalRetries == nil || *got.ProposalRetries != 1 ||
 		got.UsageComplete == nil || *got.UsageComplete || got.AssessmentAttempts == nil || *got.AssessmentAttempts != 1 ||
-		got.AssessmentRetries == nil || *got.AssessmentRetries != 0 || got.Outcome != "accepted" {
+		got.AssessmentRetries == nil || *got.AssessmentRetries != 0 || got.Outcome != "accepted" || got.CoordinationOutcome != "stopped" || got.ExhaustedOutcome != "exhausted" || got.GroupMembers != 3 || got.GroupAttempts != 4 {
 		t.Fatalf("unexpected observation summary: %+v", got)
 	}
 }