AIManage streaming, client caching, and usage

AIManage streams provider deltas incrementally to the client. The stream is not buffered end-to-end: chunks are yielded as they arrive, and persistence runs alongside as bounded, at-most-once work. The wire format is a typed NDJSON frame protocol with a kind discriminator, and every failure surfaces as a single stable error frame that never leaks raw exception text.

1. NDJSON frame contract

Every frame is a JSON object on its own line (application/x-ndjson; charset=utf-8). The kind field is always present and selects the frame shape; null-valued fields are omitted.

{"kind":"data","text":"first delta"}
{"kind":"data","text":"second delta"}
{"kind":"done"}

{"kind":"data","text":"partial delta"}
{"kind":"error","errorCode":"AI.Conversation.StreamFailed","errorType":"Unexpected","detail":"...","traceId":"...","correlationId":"...","requestId":"..."}

The frame type is NdjsonTextStreamFrame (src/Framework/BitzOrcas.Framework.AspNetCore/Results/NdjsonTextStreamFrame.cs). Three factories produce the only valid shapes:

Serialization is source-generated (NdjsonTextStreamJsonContext) for Native AOT compatibility. Each frame is written and flushed immediately so first-byte latency tracks the provider, not a buffer.

const response = await fetch(
  `/api/ai/chat/conversations/${conversationId}/stream`,
  {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      requestId: "turn-42",
      content: prompt,
      modelId: null,
    }),
    signal: abortController.signal,
  },
);

// A license/authorization denial returns a normal JSON ProblemDetails (403) before
// the stream starts — the body is NOT NDJSON in that case.
if (!response.ok) {
  const problem = await response.json();
  showError(problem.errorCode ?? response.status);
  return;
}

const reader = response.body.pipeThrough(new TextDecoderStream()).getReader();
let pending = "";
for (;;) {
  const { value, done } = await reader.read();
  if (done) break;
  pending += value;
  const lines = pending.split("\n");
  pending = lines.pop() ?? "";
  for (const line of lines) {
    if (!line) continue;
    const frame = JSON.parse(line);
    if (frame.kind === "data") renderIncrementally(frame.text);
    else if (frame.kind === "error") {
      showError(frame.errorCode, frame.errorType);
      return; // terminal — the stream stops here
    }
    // kind === "done" is an empty terminal marker; loop exits naturally.
  }
}

Production clients also need HTTP-status checks (the 403-before-start path returns ProblemDetails, not NDJSON), frame-size limits, schema validation, and recovery for split UTF-8/line boundaries.

2. Incremental streaming and the prefetch boundary

NdjsonTextStreamResult.CreateAsync prefetches the first item before returning the HTTP result. This is deliberate: a license or authorization denial happens before response headers are sent, so it can be projected as a standard 403 ProblemDetails by the exception middleware rather than as an in-stream error frame. Once the first item is in hand, headers are set (no-cache, X-Accel-Buffering: no) and subsequent deltas are yielded and flushed as they arrive. The handler does not collect the full response before yielding — first-byte latency approximates provider first-token latency, not full-generation time.

3. Hardened stream error contract

Stream failures never leak internal detail. The contract is enforced by NdjsonTextStreamResult.WriteFramesAsync and pinned by contract tests:

• A Result<string>.IsFailure item, or an unhandled exception after response start, writes exactly one terminal error frame and stops. done is only written when enumeration completes without failure.
• The error detail is produced only from L.GetString(error.Code) — the localized i18n key. The developer-facing Error.WithDescription(...) text is never serialized. A diagnostic description containing “database shard secret-diagnostic” must not appear in the body.
• An unhandled exception after response start is mapped to the caller-declared stable error (AIManageErrors.ConversationStreamFailed = AI.Conversation.StreamFailed), never the raw exception. The raw exception text never reaches the client.
• At most one terminal frame is written even if DisposeAsync throws afterward.
• Client cancellation writes no error frame and emits no LogLevel.Error.

// AIManageEndpoints.cs — the stream route wraps the mediator stream in the result
// and names the stable error to use if an unhandled exception escapes after start.
return NdjsonTextStreamResult.CreateAsync(
    mediator.CreateStream(new StreamMessage.Command(conversationId, request.RequestId, request.Content, request.ModelId)),
    HttpContext,
    AIManageErrors.ConversationStreamFailed,
    logger,
    streamName: "ai-conversation");

4. License and authorization gate the stream before it starts

The stream pipeline is source-generated in a fixed order: LoggingStreamPipelineBehavior → RuntimeLicenseStreamPipelineBehavior → AuthorizationStreamPipelineBehavior. Both gate behaviors throw before the handler enumerates:

• RuntimeLicenseStreamPipelineBehavior throws RuntimeLicenseExecutionDeniedException on deny/failure.
• AuthorizationStreamPipelineBehavior throws AuthorizationExecutionDeniedException on a denied authorization decision.

Because the first item is prefetched (section 2), these denials occur before response start and are mapped by the exception middleware to a standard 403 ProblemDetails with errorCode (Licensing.Runtime.Denied or Authorization.Denied). The conversation store is never invoked on a denied request.

5. Idempotent requests and the claim state machine

Both SendMessage.Command and StreamMessage.Command require a client-supplied RequestId. The IAIMessageRequestCoordinator atomically claims that key (via a DB unique constraint) before any provider call, so retries and concurrent duplicates resolve to one provider attempt.

public sealed record Command(
    string ConversationId,
    string RequestId,   // required; stable across retries
    string Content,
    string? ModelId) : IStreamCommand<Result<string>>, IAuthorizedRequest;

The coordinator’s ClaimAsync returns an AIMessageRequestClaim whose State drives the response:

SendMessage.Command also implements INonTransactionalCommand: it skips the generic long transaction, and the coordinator commits the user message + claim (and later the assistant message) in bounded short transactions around the network call. Completed replays the persisted assistant message without re-calling the provider — a retry after a successful turn is free and idempotent.

6. Conversation ownership is enforced

Both handlers inject ICurrentUser and reject any conversation whose TenantId or UserId does not match the caller, before any provider call or message write. The check returns AIManageErrors.ConversationNotOwner (AI.Conversation.NotOwner, Forbidden).

// Resolve identity only from the authenticated caller context.
var caller = currentUser.User;
var userId = currentUser.RequireUserId(AIManageErrors.ConversationUserRequired).GetValueOrThrow();
// Bind both tenant and owner before the provider sees any content.
if (!string.Equals(conversation.TenantId, caller.TenantId, StringComparison.Ordinal)
    || !string.Equals(conversation.UserId, userId, StringComparison.Ordinal))
{
    yield return Result.Failure<string>(AIManageErrors.ConversationNotOwner);
    yield break;
}

Both commands also implement IAuthorizedRequest with ResourceDescriptor(AIManagePermissions.Module, AIManagePermissions.ConversationResource) and AuthorizationAction.Use, so the authorization pipeline evaluates before the handler runs.

7. Cancellation, partial persistence, and the at-most-once invariant

The handler tracks a single AssistantPersistenceAttempt (NotStarted → InProgress → Succeeded/Failed) so that at most one assistant write occurs per stream enumerator, even across cancellation and dispose paths.

• Client cancellation with non-empty partial content: persists the partial text under a bounded 5-second token, then rethrows OperationCanceledException so cancellation still propagates to the client. The partial content is yielded before cancellation in the normal case.
• Provider failure: appends the interruption marker shown below to any delivered partial and persists it, or marks the attempt failed if nothing was delivered, then yields AIManageErrors.ChatRequestFailed.
• Provider-internal cancellation is treated as provider failure, not client cancellation. A token cancelled inside the provider does not take the silent client-cancellation path.
• Ambiguous CompleteAsync failure is never blindly retried. The durable attempt is left in its recorded state rather than guessing.
• The DisposeAsync path independently persists any already-delivered partial content (best-effort), but never produces a second terminal frame.

The persisted marker is a literal compatibility value:

[流式中断]

8. Cross-module seam: IAIConversationPort

Other modules call AIManage through the stable conversation port, not internal commands. AIConversationMediatorPort routes every call through the Mediator, so authorization, runtime license, and logging all apply — there is no bypass path.

public interface IAIConversationPort
{
    // Creation and non-streaming turns return durable summaries through Result.
    Task<Result<ConversationSummary>> CreateConversationAsync(
        AIConversationCreateRequest request, CancellationToken cancellationToken = default);

    Task<Result<MessageDto>> SendMessageAsync(
        AIConversationMessageRequest request, CancellationToken cancellationToken = default);

    // Streaming preserves per-chunk business failures without exposing an internal handler.
    IAsyncEnumerable<Result<string>> StreamMessageAsync(
        AIConversationMessageRequest request, CancellationToken cancellationToken = default);
}

AIConversationCreateRequest(WorkspaceId, Title?, SystemPrompt?) and AIConversationMessageRequest(ConversationId, RequestId, Content, ModelId?) are the request records. Tenant/user identity is not passed by the caller — it is derived from the trusted caller context inside the unified pipeline.

9. ChatClientFactory

The factory keeps up to 32 clients. Its key combines ProviderId, ModelId, and a hash of endpoint/model/key. Hits touch LastAccessed; full caches evict and dispose the oldest. IChatClientBuilder supplies the production Semantic Kernel adapter.

// The cache key covers provider, model, endpoint, and credential identity.
IChatClient client = chatClientFactory.GetOrCreate(provider, modelId);
var response = await client.GetResponseAsync(messages, new ChatOptions
{
    MaxOutputTokens = 4096,
    Temperature = 0.7f
}, cancellationToken);

10. Semantic Kernel boundary

The adapter registers AddOpenAIChatCompletion(modelId, apiKey, endpoint), maps MEAI messages to SK ChatHistory, and maps only MaxTokens/Temperature. It registers no plugins/functions and enables no automatic function calls. Commercial support must be proven per provider with endpoint, auth, model, streaming, usage, and error contract tests.

11. Tokens, cost, and operations governance

Non-streaming stores response total tokens on the assistant message; streaming stores null. DTOs expose no usage split. A durable usage ledger (tenant/workspace/conversation/turn/provider/model, input/output/cached tokens, price version, estimated/settled cost, status) and tenant budget enforcement remain governance targets rather than current behavior.

Operationally, measure first-byte/full duration, chunk gaps, chunks/bytes, cancellation timing, partial persistence, provider errors, cache activity, active streams, tokens, cost, and budget denial. Alert separately for provider latency, proxy buffering, and client disconnects.

12. Troubleshooting stream behavior

13. Verification commands

# Frame contract, error mapping, prefetch, and pipeline ordering.
rg -n "NdjsonTextStreamFrame|CreateAsync\(|ConversationStreamFailed|IAIMessageRequestCoordinator|ConversationNotOwner" \
  src/Platform/AIManage src/Framework/BitzOrcas.Framework.AspNetCore -g '*.cs'

AIManage overview · Conversation security · Runtime licensing