go-sip/internal/rpc/authorized_execution.go

package rpc

import (
	"context"
	"encoding/hex"
	"log/slog"

	agentv1 "git.ipao.vip/rogee/go-sip/gen/agent/v1"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/proto"
)

const authorizedOriginationVersion = "agent-authorized-origination.v0.1"

// ExecuteAuthorized only runs an explicit mock adapter. D has already made and
// durably claimed the final dial decision. A enforces the D-issued exclusive
// deadline and session identity, but never recalculates business policy.
func (s *Server) ExecuteAuthorized(ctx context.Context, req *agentv1.ExecuteAuthorizedRequest) (*agentv1.ExecuteAuthorizedResponse, error) {
	if req == nil || req.Meta == nil || req.Binding == nil {
		return nil, status.Error(codes.InvalidArgument, "request metadata and execution binding are required")
	}
	if s.mode != "mock" || s.mockAuthorizedOriginate == nil {
		return nil, status.Error(codes.FailedPrecondition, "authorized origination requires an explicit mock adapter")
	}
	if err := s.authorize(ctx, req.Meta); err != nil {
		return nil, err
	}
	if err := requireIdempotency(req.Meta); err != nil {
		return nil, err
	}
	binding := req.Binding
	if req.SchemaVersion != authorizedOriginationVersion || binding.ExecutionId == "" || binding.TenantId == "" || binding.TenantKey == "" ||
		binding.TaskId == "" || binding.TaskItemId == "" || req.SelectedTrunkId == "" || req.CallerId == "" || req.Callee == "" ||
		req.RingTimeoutMs <= 0 || req.MaxCallDurationMs <= 0 || req.DialBeforeUnixMs <= 0 || !validLowerSHA256(req.BoundSnapshotSha256) {
		return nil, status.Error(codes.InvalidArgument, "invalid authorized origination version, binding or dial decision")
	}
	digest := messageDigest(req)
	key := s.operationKey(req.Meta)
	s.mu.Lock()
	if s.executionErr != nil {
		s.mu.Unlock()
		return nil, status.Errorf(codes.Internal, "execution journal unavailable: %v", s.executionErr)
	}
	if previous, exists := s.operations[key]; exists {
		if previous.digest != digest {
			s.mu.Unlock()
			return &agentv1.ExecuteAuthorizedResponse{Receipt: s.receipt(req.Meta, agentv1.ResultCode_RESULT_CODE_CONFLICT, agentv1.FailureCode_FAILURE_CODE_ABORTED, "idempotency key content conflict", false)}, nil
		}
		entry := s.executions[binding.ExecutionId]
		if entry == nil {
			s.mu.Unlock()
			return nil, status.Error(codes.Internal, "persisted operation lacks execution state")
		}
		response := &agentv1.ExecuteAuthorizedResponse{Receipt: proto.Clone(previous.receipt).(*agentv1.OperationReceipt), State: entry.state}
		s.mu.Unlock()
		return response, nil
	}
	if _, exists := s.executions[binding.ExecutionId]; exists {
		s.mu.Unlock()
		return &agentv1.ExecuteAuthorizedResponse{Receipt: s.receipt(req.Meta, agentv1.ResultCode_RESULT_CODE_CONFLICT, agentv1.FailureCode_FAILURE_CODE_ABORTED, "execution already exists", false)}, nil
	}
	if s.now().UnixMilli() >= req.DialBeforeUnixMs {
		s.mu.Unlock()
		return nil, status.Error(codes.FailedPrecondition, "Dispatcher-issued dial deadline expired")
	}
	// Persist an unknown one-shot attempt BEFORE contacting the mock adapter.
	// After a crash or lost response, replay/query cannot cause a second dial.
	s.executions[binding.ExecutionId] = &executionRecord{
		executeDigest: digest, binding: proto.Clone(binding).(*agentv1.ExecutionBinding),
		state: agentv1.ExecutionState_EXECUTION_STATE_UNKNOWN, taskRevision: binding.TaskRevision,
		unknown: true, callState: "mock_originating_unknown",
	}
	pending := s.receipt(req.Meta, agentv1.ResultCode_RESULT_CODE_UNKNOWN, agentv1.FailureCode_FAILURE_CODE_UNAVAILABLE, "mock originate pending; do not retry", false)
	s.operations[key] = operationRecord{digest: digest, receipt: pending}
	if err := s.persistExecutionJournalLocked(); err != nil {
		s.mu.Unlock()
		return nil, err
	}
	s.mu.Unlock()

	// A newly closed deadline between durable claim and adapter invocation must
	// not dial. This is a technical check of D's instruction, not a new policy.
	expired := s.now().UnixMilli() >= req.DialBeforeUnixMs
	var dialErr error
	if !expired {
		dialErr = s.mockAuthorizedOriginate(ctx, proto.Clone(req).(*agentv1.ExecuteAuthorizedRequest))
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	record := s.executions[binding.ExecutionId]
	var result agentv1.ResultCode
	var failure agentv1.FailureCode
	var detail string
	switch {
	case expired:
		record.state, record.unknown, record.callState = agentv1.ExecutionState_EXECUTION_STATE_TERMINAL, false, "mock_deadline_closed_without_dial"
		record.terminalObservedAtUnixMs = s.now().UnixMilli()
		result, failure, detail = agentv1.ResultCode_RESULT_CODE_REJECTED, agentv1.FailureCode_FAILURE_CODE_FAILED_PRECONDITION, "dial deadline expired before mock adapter"
	case dialErr != nil:
		slog.Error("mock originate failed; execution remains unknown and cannot be retried", "execution_id", binding.ExecutionId, "error", dialErr)
		record.state, record.unknown, record.callState = agentv1.ExecutionState_EXECUTION_STATE_UNKNOWN, true, "mock_originating_unknown"
		result, failure, detail = agentv1.ResultCode_RESULT_CODE_UNKNOWN, agentv1.FailureCode_FAILURE_CODE_UNAVAILABLE, "mock originate outcome unknown; reconcile instead of retry"
	default:
		record.state, record.unknown, record.callState = agentv1.ExecutionState_EXECUTION_STATE_TERMINAL, false, "mock_no_answer"
		record.terminalObservedAtUnixMs = s.now().UnixMilli()
		result, failure, detail = agentv1.ResultCode_RESULT_CODE_APPLIED, agentv1.FailureCode_FAILURE_CODE_UNSPECIFIED, ""
	}
	receipt := s.receipt(req.Meta, result, failure, detail, false)
	s.operations[key] = operationRecord{digest: digest, receipt: receipt}
	if err := s.persistExecutionJournalLocked(); err != nil {
		return nil, err
	}
	return &agentv1.ExecuteAuthorizedResponse{Receipt: receipt, State: record.state}, nil
}

func validLowerSHA256(value string) bool {
	if len(value) != 64 {
		return false
	}
	decoded, err := hex.DecodeString(value)
	return err == nil && hex.EncodeToString(decoded) == value
}