552 lines
20 KiB
Go
552 lines
20 KiB
Go
package openai
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net/http"
|
|
"strings"
|
|
)
|
|
|
|
type hotPathEndpointError struct {
|
|
Status int
|
|
Type string
|
|
Message string
|
|
Disposition hotPathTerminalDisposition
|
|
}
|
|
|
|
type hotPathTerminalIntent struct {
|
|
Output normalizedStageOutput
|
|
Error *hotPathEndpointError
|
|
Disposition hotPathTerminalDisposition
|
|
// CleanupCommitted reports whether this terminal intent was produced by a
|
|
// committed workspace cleanup result (consumeCleanupLocked). It is false for
|
|
// terminals retained for TTL without a cleanup commit, so the cleanup
|
|
// observation stays exactly-once with its single winning owner.
|
|
CleanupCommitted bool
|
|
}
|
|
|
|
type hotPathCleanupTurn struct {
|
|
RequestID string
|
|
Output normalizedStageOutput
|
|
}
|
|
|
|
func (i hotPathTerminalIntent) clone() hotPathTerminalIntent {
|
|
out := hotPathTerminalIntent{Output: cloneNormalizedStageOutput(i.Output), Disposition: i.Disposition, CleanupCommitted: i.CleanupCommitted}
|
|
if i.Error != nil {
|
|
endpointErr := *i.Error
|
|
out.Error = &endpointErr
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (i hotPathTerminalIntent) normalized(outer *hotPathOuterTurn) hotPathTerminalIntent {
|
|
out := i.clone()
|
|
if disposition, ok := outer.terminalDisposition(); ok {
|
|
out.Disposition = disposition
|
|
}
|
|
if !out.Disposition.valid() && out.Error != nil && out.Error.Disposition.valid() {
|
|
out.Disposition = out.Error.Disposition
|
|
}
|
|
if !out.Disposition.valid() {
|
|
kind := hotPathDispositionSuccess
|
|
cause := out.Output.TerminalReason
|
|
if out.Error != nil {
|
|
kind = hotPathDispositionProviderError
|
|
cause = out.Error.Message
|
|
if out.Error.Status >= http.StatusBadRequest && out.Error.Status < http.StatusInternalServerError ||
|
|
strings.Contains(strings.ToLower(out.Error.Type), "invalid") {
|
|
kind = hotPathDispositionValidationError
|
|
}
|
|
}
|
|
out.Disposition = hotPathTerminalDisposition{Kind: kind, Cause: cause, Source: "cleanup_handoff"}
|
|
}
|
|
if out.Error != nil {
|
|
out.Error.Disposition = out.Disposition
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (i hotPathTerminalIntent) terminalClass() string {
|
|
if i.Error != nil {
|
|
return "primary_error"
|
|
}
|
|
return "success"
|
|
}
|
|
|
|
func (s *hotPathLightStore) beginCleanup(
|
|
ctx context.Context,
|
|
requestID, ownerEdgeID string,
|
|
intent hotPathTerminalIntent,
|
|
coordinator *logicalRequestCoordinator,
|
|
) (normalizedStageOutput, error) {
|
|
return s.beginCleanupWithOuter(ctx, requestID, ownerEdgeID, intent, nil, coordinator)
|
|
}
|
|
|
|
func (s *hotPathLightStore) beginCleanupWithOuter(
|
|
ctx context.Context,
|
|
requestID, ownerEdgeID string,
|
|
intent hotPathTerminalIntent,
|
|
outer *hotPathOuterTurn,
|
|
coordinator *logicalRequestCoordinator,
|
|
) (normalizedStageOutput, error) {
|
|
if s == nil || coordinator == nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("light cleanup is unavailable")
|
|
}
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
record := s.records[requestID]
|
|
if record == nil || record.ownerEdgeID != ownerEdgeID || !record.running || record.pending != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("review completion cannot enter cleanup")
|
|
}
|
|
if record.phase != hotPathPhaseReviewResolution && record.phase != hotPathPhaseReviewRepair {
|
|
return normalizedStageOutput{}, fmt.Errorf("review completion is not resolution or repair")
|
|
}
|
|
return s.beginCleanupLocked(ctx, record, record.reviewStageID, intent, outer, coordinator)
|
|
}
|
|
|
|
func (s *hotPathLightStore) beginPrimaryErrorCleanup(
|
|
ctx context.Context,
|
|
requestID, ownerEdgeID string,
|
|
primary hotPathEndpointError,
|
|
outer *hotPathOuterTurn,
|
|
coordinator *logicalRequestCoordinator,
|
|
) (normalizedStageOutput, error) {
|
|
if s == nil || coordinator == nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("light cleanup is unavailable")
|
|
}
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
record := s.records[requestID]
|
|
if record == nil || record.ownerEdgeID != ownerEdgeID || record.cleanupTransitions != 0 || record.terminalIntent != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("primary-error cleanup is unavailable")
|
|
}
|
|
fromStageID, err := record.primaryErrorCleanupSource()
|
|
if err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
intent := hotPathTerminalIntent{Error: &primary, Disposition: primary.Disposition}
|
|
return s.beginCleanupLocked(ctx, record, fromStageID, intent, outer, coordinator)
|
|
}
|
|
|
|
func (r *hotPathLightRecord) primaryErrorCleanupSource() (string, error) {
|
|
if r == nil || r.running || r.pending != nil || r.cleanupTransitions != 0 || r.terminalIntent != nil {
|
|
return "", fmt.Errorf("primary-error cleanup source is unavailable")
|
|
}
|
|
if r.selectorCommit.StageID != r.selectorStageID || strings.TrimSpace(r.selectorCommit.ResponseID) == "" {
|
|
return "", fmt.Errorf("primary-error cleanup selector correlation is unavailable")
|
|
}
|
|
|
|
switch r.phase {
|
|
case hotPathPhaseAwaitArtifacts:
|
|
if r.localStageID != "" || r.reviewStageID != "" {
|
|
return "", fmt.Errorf("primary-error cleanup artifact source is mismatched")
|
|
}
|
|
return "", nil
|
|
case hotPathPhaseLocalActive:
|
|
if !r.artifactReady || !validLogicalRequestID(r.localStageID) || r.reviewStageID != "" {
|
|
return "", fmt.Errorf("primary-error cleanup local source is mismatched")
|
|
}
|
|
return r.localStageID, nil
|
|
case hotPathPhaseReviewActive, hotPathPhaseReviewAwaitRead, hotPathPhaseReviewResolution, hotPathPhaseReviewRepair:
|
|
if !r.artifactReady || !validLogicalRequestID(r.localStageID) || !validLogicalRequestID(r.reviewStageID) ||
|
|
r.localCommit.StageID != r.localStageID || strings.TrimSpace(r.localCommit.ResponseID) == "" {
|
|
return "", fmt.Errorf("primary-error cleanup review source is mismatched")
|
|
}
|
|
return r.reviewStageID, nil
|
|
default:
|
|
return "", fmt.Errorf("phase %q cannot enter primary-error cleanup", r.phase)
|
|
}
|
|
}
|
|
|
|
func (s *hotPathLightStore) beginCleanupLocked(
|
|
ctx context.Context,
|
|
record *hotPathLightRecord,
|
|
fromStageID string,
|
|
intent hotPathTerminalIntent,
|
|
outer *hotPathOuterTurn,
|
|
coordinator *logicalRequestCoordinator,
|
|
) (normalizedStageOutput, error) {
|
|
if err := ctx.Err(); err != nil {
|
|
if outer != nil {
|
|
outer.cancelActiveStage(hotPathDispositionCallerCancel, "cleanup_context", err)
|
|
}
|
|
record.terminalDisposition = ptrHotPathDisposition(hotPathTerminalDisposition{
|
|
Kind: hotPathDispositionCallerCancel, Cause: err.Error(), Source: "cleanup_context",
|
|
})
|
|
record.running = false
|
|
_ = coordinator.disconnect(record.requestID, record.ownerEdgeID, "cancelled")
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
if record.cleanupTransitions != 0 || record.terminalIntent != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("cleanup pending was already committed")
|
|
}
|
|
intent = intent.normalized(outer)
|
|
|
|
cleanupStageID, err := coordinator.newStageID()
|
|
if err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
providerCallID, err := coordinator.newCallID()
|
|
if err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
paths := newReservedPaths(record.requestID)
|
|
deleteBinding := record.binding.operation(opKindDelete)
|
|
if deleteBinding == nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("cleanup delete binding is unavailable")
|
|
}
|
|
deleteArgs := make(map[string]any)
|
|
setMappedArgument(deleteArgs, deleteBinding.pathField, paths.JobDir)
|
|
providerCall := normalizedToolCall{
|
|
ID: providerCallID, ProviderCallID: providerCallID, Name: deleteBinding.toolName,
|
|
Arguments: deleteArgs, Path: paths.JobDir,
|
|
}
|
|
mapped, payload, err := mapArtifactCall(record.binding, providerCall, opKindDelete, paths.JobDir, coordinator)
|
|
if err != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("map cleanup delete: %w", err)
|
|
}
|
|
responseID := strings.TrimSpace(intent.Output.ResponseID)
|
|
if responseID == "" {
|
|
responseID = strings.TrimSpace(record.selectorCommit.ResponseID)
|
|
}
|
|
if responseID == "" {
|
|
return normalizedStageOutput{}, fmt.Errorf("cleanup response identity is unavailable")
|
|
}
|
|
cleanupOutput := normalizedStageOutput{
|
|
ResponseID: responseID, Created: intent.Output.Created, CallerStageOnly: true,
|
|
ToolCalls: []normalizedToolCall{mapped}, TerminalReason: "tool_calls",
|
|
}
|
|
if record.protocol == "anthropic" {
|
|
cleanupOutput.TerminalReason = "tool_use"
|
|
}
|
|
if outer != nil {
|
|
if err := runHotPathCollectedStage(ctx, outer, cleanupStageID, cleanupOutput); err != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("collect cleanup outer turn: %w", err)
|
|
}
|
|
visible := hotPathCompatibilityOutput(outer, cleanupOutput, record.protocol)
|
|
if len(visible.ToolCalls) == 0 && outer.outputBudget().Exhausted {
|
|
outer.commitLengthTerminal()
|
|
return hotPathCompatibilityOutput(outer, cleanupOutput, record.protocol), nil
|
|
}
|
|
if err := outer.projectToolIdentities(cleanupOutput.ToolCalls); err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
cleanupOutput = hotPathCompatibilityOutput(outer, cleanupOutput, record.protocol)
|
|
// Cleanup is an internal continuation frontier. Preserve the accumulated
|
|
// outer turn for the terminal response, but expose only the cleanup tool on
|
|
// this intermediate caller turn.
|
|
cleanupOutput.Content = ""
|
|
cleanupOutput.Reasoning = ""
|
|
cleanupOutput.Deltas = nil
|
|
}
|
|
issuedHash, err := directIssuedCallHash(record.protocol, cleanupOutput)
|
|
if err != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("fingerprint cleanup call: %w", err)
|
|
}
|
|
if _, err := coordinator.startCleanup(record.requestID, record.ownerEdgeID, fromStageID, cleanupStageID, intent.terminalClass()); err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
if _, err := coordinator.awaitToolResults(record.requestID, record.ownerEdgeID, cleanupStageID, []logicalRequestExpectedTool{{
|
|
PublicCallID: mapped.ID, ProviderCallID: mapped.ProviderCallID,
|
|
}}, issuedHash); err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
|
|
stored := intent.clone()
|
|
record.terminalIntent = &stored
|
|
record.terminalDisposition = ptrHotPathDisposition(stored.Disposition)
|
|
record.pendingKind = hotPathPendingCleanup
|
|
record.pending = map[string]hotPathPendingCall{
|
|
mapped.ID: {publicCallID: mapped.ID, providerCallID: mapped.ProviderCallID, payload: payload},
|
|
}
|
|
record.pendingHash = issuedHash
|
|
record.pendingOutput = cloneNormalizedStageOutput(cleanupOutput)
|
|
record.phase = hotPathPhaseCleanupPending
|
|
record.cleanupStageID = cleanupStageID
|
|
record.cleanupTransitions++
|
|
record.running = false
|
|
return cleanupOutput, nil
|
|
}
|
|
|
|
func (s *hotPathLightStore) consumeCleanupLocked(
|
|
record *hotPathLightRecord,
|
|
lineage logicalRequestContinuationLineage,
|
|
results []workspaceResult,
|
|
coordinator *logicalRequestCoordinator,
|
|
) (logicalRequestSnapshot, hotPathLightDisposition, bool, error) {
|
|
if record.terminalIntent == nil || len(record.pending) != 1 || len(results) != 1 {
|
|
return logicalRequestSnapshot{}, hotPathLightDisposition{}, true, fmt.Errorf("cleanup result set mismatch")
|
|
}
|
|
result := results[0]
|
|
pending, ok := record.pending[result.callID]
|
|
if !ok || pending.payload == nil {
|
|
return logicalRequestSnapshot{}, hotPathLightDisposition{}, true, fmt.Errorf("cleanup result id is not pending")
|
|
}
|
|
if reason := matchResultCorrelation(record.binding, pending.payload, result); reason != "" {
|
|
return logicalRequestSnapshot{}, hotPathLightDisposition{}, true, fmt.Errorf("cleanup receipt rejected: %s", reason)
|
|
}
|
|
|
|
intent := record.terminalIntent.clone()
|
|
intent.CleanupCommitted = true
|
|
receipt := matchResultReceipt(record.binding, pending.payload, result)
|
|
if !receipt.matched && intent.Error == nil {
|
|
intent.Error = standardCleanupEndpointError(record.protocol)
|
|
intent.Output = normalizedStageOutput{}
|
|
intent.Disposition = intent.Error.Disposition
|
|
}
|
|
snap, err := coordinator.commitCleanupByLineage(record.ownerEdgeID, record.principalRef, lineage)
|
|
if err != nil {
|
|
return logicalRequestSnapshot{}, hotPathLightDisposition{}, true, err
|
|
}
|
|
requestID := record.requestID
|
|
stageID := snap.ActiveStageID
|
|
delete(s.records, requestID)
|
|
return snap, hotPathLightDisposition{
|
|
RequestID: requestID, StageID: stageID, Phase: hotPathPhaseCleanupPending, Terminal: &intent,
|
|
}, true, nil
|
|
}
|
|
|
|
func standardCleanupEndpointError(protocol string) *hotPathEndpointError {
|
|
if protocol == "anthropic" {
|
|
return &hotPathEndpointError{
|
|
Status: http.StatusBadGateway, Type: "api_error", Message: "workspace cleanup failed",
|
|
Disposition: hotPathTerminalDisposition{
|
|
Kind: hotPathDispositionProviderError, Cause: "workspace cleanup failed", Source: "cleanup_receipt",
|
|
},
|
|
}
|
|
}
|
|
return &hotPathEndpointError{
|
|
Status: http.StatusBadGateway, Type: "run_error", Message: "workspace cleanup failed",
|
|
Disposition: hotPathTerminalDisposition{
|
|
Kind: hotPathDispositionProviderError, Cause: "workspace cleanup failed", Source: "cleanup_receipt",
|
|
},
|
|
}
|
|
}
|
|
|
|
// commitCleanupByLineage admits the exact cleanup continuation and removes the
|
|
// coordinator record in the same critical section. This is the terminal owner
|
|
// shared by cleanup-result and TTL races.
|
|
func (c *logicalRequestCoordinator) commitCleanupByLineage(
|
|
ownerEdgeID, principalRef string,
|
|
lineage logicalRequestContinuationLineage,
|
|
) (logicalRequestSnapshot, error) {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
var target *logicalRequestRecord
|
|
for _, record := range c.requests {
|
|
if record.state == logicalRequestStateCleanup && record.cleanup && record.ownerEdgeID == ownerEdgeID && record.principalRef == principalRef &&
|
|
record.lineage == lineage.Prefix && sameLogicalRequestResultIDs(record.expected, lineage.ResultIDs) {
|
|
target = record
|
|
break
|
|
}
|
|
}
|
|
if target == nil {
|
|
return logicalRequestSnapshot{}, errLogicalRequestNotFound
|
|
}
|
|
if err := validateLogicalRequestContinuationLineage(target.lineage, target.expectedIssuedCallHash, target.expected, lineage); err != nil {
|
|
return logicalRequestSnapshot{}, err
|
|
}
|
|
snapshot := target.snapshot()
|
|
delete(c.requests, target.id)
|
|
return snapshot, nil
|
|
}
|
|
|
|
func (s *Server) writeHotPathTerminal(
|
|
w http.ResponseWriter,
|
|
r *http.Request,
|
|
dispatch routeDispatch,
|
|
protocol string,
|
|
stream bool,
|
|
requestID string,
|
|
intent hotPathTerminalIntent,
|
|
) error {
|
|
outer := hotPathCurrentCallerOuterTurn(r, protocol)
|
|
if outer != nil && intent.Disposition.valid() {
|
|
outer.selectDisposition(intent.Disposition)
|
|
}
|
|
// When this terminal was produced by a committed workspace cleanup result,
|
|
// emit the exactly-once cleanup observation before the terminal so the
|
|
// captured lifecycle reflects cleanup-result → terminal order. The outcome
|
|
// distinguishes a successful primary from a primary-error cleanup;
|
|
// TTL-retained primaries carry CleanupCommitted=false and emit no cleanup.
|
|
if intent.CleanupCommitted {
|
|
cleanupOutcome := hotPathCleanupOutcomeSuccess
|
|
if intent.Error != nil {
|
|
cleanupOutcome = hotPathCleanupOutcomePrimaryError
|
|
}
|
|
s.observeHotPathCleanup(r.Context(), cleanupOutcome, requestID, "")
|
|
}
|
|
|
|
var endpointWriteErr error
|
|
var responseErr error
|
|
if intent.Disposition.Kind == hotPathDispositionCallerCancel {
|
|
responseErr = context.Canceled
|
|
} else if intent.Error != nil {
|
|
if outer != nil {
|
|
outer.commitTerminalError(intent.Error.Type, intent.Error.Type)
|
|
}
|
|
if protocol == "anthropic" {
|
|
if codec := hotPathAnthropicCodecFromRequest(r); codec != nil {
|
|
codec.w = w
|
|
endpointWriteErr = codec.writeDisposition(
|
|
intent.Disposition, intent.Error.Status, intent.Error.Type, intent.Error.Message,
|
|
)
|
|
} else {
|
|
policy := anthropicHotPathPolicy(intent.Disposition)
|
|
if !policy.silent {
|
|
writeAnthropicError(w, policy.status, policy.errorType, intent.Error.Message)
|
|
}
|
|
}
|
|
} else {
|
|
turn := &hotPathTurn{Writer: w, Request: r, OuterTurn: outer}
|
|
if !writeHotPathChatOuterError(
|
|
turn, intent.Error.Status, intent.Error.Type, intent.Error.Message, intent.Disposition,
|
|
) {
|
|
policy := chatHotPathPolicy(intent.Disposition)
|
|
if !policy.silent {
|
|
writeError(w, policy.status, policy.errorType, intent.Error.Message)
|
|
}
|
|
}
|
|
}
|
|
responseErr = fmt.Errorf("%s", intent.Error.Message)
|
|
} else {
|
|
if outer != nil {
|
|
outer.commitTerminalSuccess(intent.Output.TerminalReason)
|
|
}
|
|
endpointWriteErr = s.writeHotPathStageResponse(w, r, dispatch, protocol, stream, requestID, intent.Output)
|
|
responseErr = endpointWriteErr
|
|
}
|
|
|
|
winning := resolveHotPathObservedDisposition(outer, intent.Disposition, endpointWriteErr)
|
|
s.observeHotPathTerminal(r.Context(), hotPathModeLight,
|
|
hotPathTerminalDispositionFromKind(winning.Kind), requestID, winning.StageID, dispatch.Preset.ID)
|
|
return responseErr
|
|
}
|
|
|
|
func resolveHotPathObservedDisposition(outer *hotPathOuterTurn, intended hotPathTerminalDisposition, writeErr error) hotPathTerminalDisposition {
|
|
if writeErr != nil {
|
|
return hotPathTerminalDisposition{
|
|
Kind: hotPathDispositionForError(writeErr), Cause: writeErr.Error(), Source: "endpoint_write",
|
|
}
|
|
}
|
|
if selected, ok := outer.terminalDisposition(); ok && selected.valid() {
|
|
return selected
|
|
}
|
|
if intended.valid() {
|
|
return intended
|
|
}
|
|
return hotPathTerminalDisposition{Kind: hotPathDispositionProviderError, Source: "terminal_observation"}
|
|
}
|
|
|
|
func hotPathLightEndpointError(protocol string, status int, message string) hotPathEndpointError {
|
|
errorType := "run_error"
|
|
if protocol == "anthropic" {
|
|
errorType = "api_error"
|
|
}
|
|
kind := hotPathDispositionProviderError
|
|
if status >= http.StatusBadRequest && status < http.StatusInternalServerError {
|
|
kind = hotPathDispositionValidationError
|
|
errorType = "invalid_request_error"
|
|
}
|
|
return hotPathEndpointError{
|
|
Status: status, Type: errorType, Message: message,
|
|
Disposition: hotPathTerminalDisposition{Kind: kind, Cause: message, Source: "light_flow"},
|
|
}
|
|
}
|
|
|
|
func hotPathLightEndpointErrorForCause(protocol string, status int, stageID string, cause error) hotPathEndpointError {
|
|
message := "hot path stage failed"
|
|
if cause != nil {
|
|
message = cause.Error()
|
|
}
|
|
endpointErr := hotPathLightEndpointError(protocol, status, message)
|
|
if disposition, ok := hotPathDispositionFromError(cause); ok {
|
|
endpointErr.Disposition = disposition
|
|
} else if cause != nil {
|
|
endpointErr.Disposition = hotPathTerminalDisposition{
|
|
Kind: hotPathDispositionForError(cause), Cause: cause.Error(), Source: "stage_dispatch", StageID: stageID,
|
|
}
|
|
}
|
|
return endpointErr
|
|
}
|
|
|
|
func ptrHotPathDisposition(disposition hotPathTerminalDisposition) *hotPathTerminalDisposition {
|
|
if !disposition.valid() {
|
|
return nil
|
|
}
|
|
selected := disposition
|
|
return &selected
|
|
}
|
|
|
|
func (s *Server) retainHotPathPrimaryErrorForTTL(requestID string, primary hotPathEndpointError) *hotPathTerminalIntent {
|
|
ownerEdgeID := s.edgeIDValue()
|
|
if s.lightFlows != nil {
|
|
s.lightFlows.abortWithDisposition(requestID, ownerEdgeID, primary.Disposition)
|
|
}
|
|
_ = s.requestCoordinator.disconnect(requestID, ownerEdgeID, "primary_error")
|
|
return &hotPathTerminalIntent{Error: &primary, Disposition: primary.Disposition}
|
|
}
|
|
|
|
func (s *Server) writeHotPathPrimaryError(
|
|
w http.ResponseWriter,
|
|
r *http.Request,
|
|
dispatch routeDispatch,
|
|
protocol string,
|
|
stream bool,
|
|
requestID string,
|
|
primary hotPathEndpointError,
|
|
) error {
|
|
ownerEdgeID := s.edgeIDValue()
|
|
s.lightFlows.abortDispatch(requestID, ownerEdgeID)
|
|
outer := hotPathCurrentCallerOuterTurn(r, protocol)
|
|
if disposition, ok := outer.terminalDisposition(); ok {
|
|
primary.Disposition = disposition
|
|
} else if primary.Disposition.valid() {
|
|
outer.selectDisposition(primary.Disposition)
|
|
}
|
|
if err := r.Context().Err(); err != nil {
|
|
if outer != nil {
|
|
outer.cancelActiveStage(hotPathDispositionCallerCancel, "caller_context", err)
|
|
}
|
|
s.disconnectHotPathRequestWithDisposition(requestID, ownerEdgeID, hotPathTerminalDisposition{
|
|
Kind: hotPathDispositionCallerCancel, Cause: err.Error(), Source: "caller_context",
|
|
})
|
|
return err
|
|
}
|
|
|
|
cleanup, err := s.lightFlows.beginPrimaryErrorCleanup(
|
|
r.Context(), requestID, ownerEdgeID, primary,
|
|
outer, s.requestCoordinator,
|
|
)
|
|
if err == nil {
|
|
s.observeHotPathCleanupTransition(r.Context(), requestID, dispatch.Preset.ID)
|
|
return s.writeHotPathStageResponse(w, r, dispatch, protocol, stream, requestID, cleanup)
|
|
}
|
|
if contextErr := r.Context().Err(); contextErr != nil {
|
|
if outer != nil {
|
|
outer.cancelActiveStage(hotPathDispositionCallerCancel, "caller_context", contextErr)
|
|
}
|
|
s.disconnectHotPathRequestWithDisposition(requestID, ownerEdgeID, hotPathTerminalDisposition{
|
|
Kind: hotPathDispositionCallerCancel, Cause: contextErr.Error(), Source: "caller_context",
|
|
})
|
|
return contextErr
|
|
}
|
|
intent := s.retainHotPathPrimaryErrorForTTL(requestID, primary)
|
|
return s.writeHotPathTerminal(w, r, dispatch, protocol, stream, requestID, *intent)
|
|
}
|
|
|
|
func (s *Server) disconnectHotPathRequest(requestID, ownerEdgeID string) {
|
|
s.disconnectHotPathRequestWithDisposition(requestID, ownerEdgeID, hotPathTerminalDisposition{
|
|
Kind: hotPathDispositionCallerCancel, Cause: "caller disconnected", Source: "caller_context",
|
|
})
|
|
}
|
|
|
|
func (s *Server) disconnectHotPathRequestWithDisposition(requestID, ownerEdgeID string, disposition hotPathTerminalDisposition) {
|
|
if requestID == "" {
|
|
return
|
|
}
|
|
if s.lightFlows != nil {
|
|
s.lightFlows.abortWithDisposition(requestID, ownerEdgeID, disposition)
|
|
}
|
|
_ = s.requestCoordinator.disconnect(requestID, ownerEdgeID, "cancelled")
|
|
}
|