Method EnqueueRunFromPeer

internal/network/tasks.go:292–343

EnqueueRunFromPeer enqueues one task run from an authenticated network peer while preserving origin-scoped idempotency inside the task manager.

(
	ctx context.Context,
	ingress TaskIngressContext,
	spec taskpkg.EnqueueRun,
)

// EnqueueRunFromPeer enqueues one task run from an authenticated network peer
// while preserving origin-scoped idempotency inside the task manager.
func (m *Manager) EnqueueRunFromPeer(
	ctx context.Context,
	ingress TaskIngressContext,
	spec taskpkg.EnqueueRun,
) (*taskpkg.Run, error) {
	peerCtx, err := m.resolveTaskPeerContext(ctx, ingress, networkTaskActionEnqueue)
	if err != nil {
		return nil, err
	}
	view, err := m.tasks.GetTask(ctx, strings.TrimSpace(spec.TaskID), peerCtx.actor)
	if err != nil {
		return nil, m.rejectTaskIngress(ctx, peerCtx.ingress, networkTaskActionEnqueue, err, nil)
	}
	if err := enforceBoundTaskChannel(
		view.Task.ID,
		view.Task.NetworkChannel,
		peerCtx.ingress.Channel,
		nil,
	); err != nil {
		return nil, m.rejectTaskIngress(ctx, peerCtx.ingress, networkTaskActionEnqueue, err, map[string]any{
			tasksTaskIDKey:         view.Task.ID,
			tasksNetworkChannelKey: strings.TrimSpace(view.Task.NetworkChannel),
		})
	}
	if err := validateRequestedTaskChannel(peerCtx.ingress.Channel, spec.NetworkChannel); err != nil {
		return nil, m.rejectTaskIngress(ctx, peerCtx.ingress, networkTaskActionEnqueue, err, map[string]any{
			tasksTaskIDKey:         view.Task.ID,
			tasksNetworkChannelKey: strings.TrimSpace(spec.NetworkChannel),
		})
	}
	spec, err = withNetworkRunMetadata(spec, peerCtx.ingress)
	if err != nil {
		return nil, m.rejectTaskIngress(ctx, peerCtx.ingress, networkTaskActionEnqueue, err, map[string]any{
			tasksTaskIDKey: view.Task.ID,
		})
	}

	run, err := m.tasks.EnqueueRun(ctx, spec, peerCtx.actor)
	if err != nil {
		return nil, m.rejectTaskIngress(ctx, peerCtx.ingress, networkTaskActionEnqueue, err, map[string]any{
			tasksTaskIDKey:    view.Task.ID,
			"idempotency_key": strings.TrimSpace(spec.IdempotencyKey),
		})
	}
	m.recordTaskIngress(ctx, peerCtx.ingress, networkTaskActionEnqueue, AuditDirectionReceived, "", map[string]any{
		tasksTaskIDKey:         run.TaskID,
		"run_id":               run.ID,
		"idempotency_key":      strings.TrimSpace(run.IdempotencyKey),
		tasksNetworkChannelKey: strings.TrimSpace(run.NetworkChannel),
	})
	return run, nil
}

func (m *Manager) resolveTaskPeerContext(
	ctx context.Context,