Strengthen continuation notification ownership coverage

prepared_execution_contract_test.go

@@ -47,19 +47,32 @@ func TestPreparedExecutionFreezesSourcesAndReturnsIndependentDetails(t *testing.
 	}
 
 	temperature := 0.25
+	initial, err := newSimpleEngine(t, "history", promptkit.WithLLMClient(&fakeLLMClient{response: &promptkit.GenerateResponse{Content: "prior"}})).Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", CaptureExchange: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	initialDetails := initial.Exchange.Details()
 	extraParams := map[string]any{
 		"nested": map[string]any{"source": "original"},
 	}
 	var notified *promptkit.PreparedRun
 	request := promptkit.RunRequest{
-		PromptID: "prepared",
+		PromptID:         "prepared",
-		APIKey:   "synthetic-direct-key",
+		APIKey:           "synthetic-direct-key",
+		ContinueFrom:     initial.Exchange,
+		SessionID:        "continuation-session",
+		AppendedMessages: []promptkit.RenderedMessage{{Role: promptkit.RoleUser, Content: "appended", CacheControl: &promptkit.CacheControl{Type: promptkit.CacheControlEphemeral}}},
 		OnPrepared: func(_ context.Context, details promptkit.PreparedRun) {
 			notified = &details
+			if details.ParentRunID != initial.RunID || details.SessionID != "continuation-session" || details.Messages[0].Content != "history" || details.Messages[2].Content != "Input=original artifact Label=original variable" || details.InputHashes["input"] != "original-input-hash" || details.StructuredOutput.JSONSchema.Schema.(map[string]any)["title"] != "original root" {
+				t.Error("notification did not describe the frozen continuation")
+			}
 			if strings.Contains(fmt.Sprintf("%#v", details), "synthetic-direct-key") {
 				t.Error("notification exposed credential")
 			}
-			details.Messages[0].Content = "mutated notification"
+			details.Messages[0].Content = "mutated history"
+			details.Messages[2].Content = "mutated notification"
+			details.Messages[3].CacheControl.Type = "changed"
 			details.InputHashes["input"] = "mutated notification"
 			details.EffectiveModelParams.ExtraParams["nested"].(map[string]any)["source"] = "mutated notification"
 			details.StructuredOutput.JSONSchema.Schema.(map[string]any)["title"] = "mutated notification"
@@ -99,14 +112,14 @@ func TestPreparedExecutionFreezesSourcesAndReturnsIndependentDetails(t *testing.
 	reader.set("changed artifact", "changed-input-hash")
 
 	first := prepared.Details()
-	first.Messages[0].Content = "changed details"
+	first.Messages[2].Content = "changed details"
 	first.InputHashes["input"] = "changed-details-hash"
 	first.EffectiveModelParams.ExtraParams["nested"].(map[string]any)["source"] = "changed details"
 	first.StructuredOutput.JSONSchema.Schema.(map[string]any)["title"] = "changed details"
 
 	second := prepared.Details()
-	if second.Messages[0].Content != "Input=original artifact Label=original variable" {
+	if second.Messages[2].Content != "Input=original artifact Label=original variable" {
-		t.Fatalf("details message changed: %q", second.Messages[0].Content)
+		t.Fatalf("details message changed: %q", second.Messages[2].Content)
 	}
 	if second.InputHashes["input"] != "original-input-hash" {
 		t.Fatalf("details input hash changed: %q", second.InputHashes["input"])
@@ -131,7 +144,7 @@ func TestPreparedExecutionFreezesSourcesAndReturnsIndependentDetails(t *testing.
 	if reader.callCount() != 1 {
 		t.Fatalf("execution reopened artifact source: calls=%d", reader.callCount())
 	}
-	if notified == nil || prepared.Details().Messages[0].Content != second.Messages[0].Content || prepared.Details().StructuredOutput.JSONSchema.Schema.(map[string]any)["title"] != "original root" {
+	if notified == nil || prepared.Details().Messages[2].Content != second.Messages[2].Content || prepared.Details().Messages[3].CacheControl.Type != promptkit.CacheControlEphemeral || !reflect.DeepEqual(initial.Exchange.Details(), initialDetails) || result.Exchange.Details().Messages[0].Content != "history" || prepared.Details().StructuredOutput.JSONSchema.Schema.(map[string]any)["title"] != "original root" {
 		t.Fatal("notification changed prepared state")
 	}
 
@@ -143,7 +156,7 @@ func TestPreparedExecutionFreezesSourcesAndReturnsIndependentDetails(t *testing.
 	if generated.APIKey != "synthetic-direct-key" {
 		t.Fatal("direct credential did not reach client")
 	}
-	if generated.Prompt.Messages[0].Content != second.Messages[0].Content ||
+	if generated.Prompt.Messages[0].Content != "history" || generated.Prompt.Messages[2].Content != second.Messages[2].Content || generated.Prompt.Messages[3].CacheControl.Type != promptkit.CacheControlEphemeral ||
 		generated.Target.Model != second.EffectiveModelParams.Model ||
 		!reflect.DeepEqual(generated.Target.ExtraParams, second.EffectiveModelParams.ExtraParams) ||
 		!reflect.DeepEqual(generated.StructuredOutput, second.StructuredOutput) {

prepared_notification_contract_test.go

@@ -126,6 +126,9 @@ func TestPreparedNotificationInvalidHandlesAndPreparationOnlyCalls(t *testing.T)
 	if _, err := foreign.RunPrepared(context.Background(), handle); !errors.Is(err, promptkit.ErrInvalidRequest) {
 		t.Fatal(err)
 	}
+	if _, err := engine.RunPrepared(context.Background(), nil); !errors.Is(err, promptkit.ErrInvalidRequest) {
+		t.Fatal(err)
+	}
 	handle.Discard()
 	if _, err := engine.RunPrepared(context.Background(), handle); !errors.Is(err, promptkit.ErrInvalidRequest) {
 		t.Fatal(err)
@@ -136,6 +139,30 @@ func TestPreparedNotificationInvalidHandlesAndPreparationOnlyCalls(t *testing.T)
 	if calls != 0 {
 		t.Fatalf("unexpected notifications=%d", calls)
 	}
+	handle, err = engine.PrepareExecution(context.Background(), req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := engine.RunPrepared(context.Background(), handle); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := engine.RunPrepared(context.Background(), handle); !errors.Is(err, promptkit.ErrInvalidRequest) || calls != 1 {
+		t.Fatalf("reused handle error=%v notifications=%d", err, calls)
+	}
+}
+
+func TestNotificationHoldsAdmission(t *testing.T) {
+	engine := newBackendCapacityEngine(t, observationClientFunc(func(context.Context, promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
+		return &promptkit.GenerateResponse{Content: "ok"}, nil
+	}), 1, capacityInt(0), nil)
+	_, err := engine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", OnPrepared: func(ctx context.Context, _ promptkit.PreparedRun) {
+		if _, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "prompt"}); !errors.Is(err, promptkit.ErrCapacityExceeded) {
+			t.Errorf("nested admission error=%v", err)
+		}
+	}})
+	if err != nil {
+		t.Fatal(err)
+	}
 }
 
 func TestNotificationPanicReleasesCapacityAndConsumesHandle(t *testing.T) {