[Nexus] Minor fixes - Added destination queue and schedule cancellati… (#9790)

## What changed?
Minor fixes.

## Why?
These were missing or misplaced previously.

## How did you test it?
- [x] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)



<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Changes task routing and cancellation scheduling in the Nexus
operation state machine, which can affect execution flow and delivery to
outbound queues. While localized, incorrect destinations or transition
sequencing could cause stuck or misrouted operations.
> 
> **Overview**
> Ensures Nexus invocation tasks are routed to the correct outbound
queue by setting `TaskAttributes.Destination` to the operation endpoint
when scheduling and when rescheduling after backoff.
> 
> Moves the "cancellation already requested" handling from the workflow
`Started` event handler into `TransitionStarted`, so pending
cancellations are automatically scheduled as soon as an operation token
becomes available.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
4eddec6. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: gow

chasm/lib/nexusoperation/operation_statemachine.go

@@ -16,8 +16,9 @@ var TransitionScheduled = chasm.NewTransition(
 	[]nexusoperationpb.OperationStatus{nexusoperationpb.OPERATION_STATUS_UNSPECIFIED},
 	nexusoperationpb.OPERATION_STATUS_SCHEDULED,
 	func(o *Operation, ctx chasm.MutableContext, event EventScheduled) error {
-		// Emit an invocation task to start the operation
-		ctx.AddTask(o, chasm.TaskAttributes{}, &nexusoperationpb.InvocationTask{
+		// Emit an invocation task to start the operation.
+		// The destination is the endpoint name, which routes the task to the correct outbound queue.
+		ctx.AddTask(o, chasm.TaskAttributes{Destination: o.GetEndpoint()}, &nexusoperationpb.InvocationTask{
 			Attempt: o.Attempt,
 		})
 
@@ -93,7 +94,7 @@ var transitionRescheduled = chasm.NewTransition(
 		o.NextAttemptScheduleTime = nil
 
 		// Emit a new invocation task for the retry attempt
-		ctx.AddTask(o, chasm.TaskAttributes{}, &nexusoperationpb.InvocationTask{
+		ctx.AddTask(o, chasm.TaskAttributes{Destination: o.GetEndpoint()}, &nexusoperationpb.InvocationTask{
 			Attempt: o.Attempt,
 		})
 
@@ -142,6 +143,13 @@ var TransitionStarted = chasm.NewTransition(
 			})
 		}
 
+		// If cancellation was already requested, schedule sending the cancellation request now that we have
+		// an operation token.
+		cancellation, ok := o.Cancellation.TryGet(ctx)
+		if ok && cancellation.StateMachineState() == nexusoperationpb.CANCELLATION_STATUS_UNSPECIFIED {
+			return TransitionCancellationScheduled.Apply(cancellation, ctx, EventCancellationScheduled{})
+		}
+
 		return nil
 	},
 )

chasm/lib/nexusoperation/workflow/events.go

@@ -281,20 +281,9 @@ func (d StartedEventDefinition) Apply(ctx chasm.MutableContext, wf *chasmworkflo
 
 	// TODO: Store event.Links on the Operation for standalone mode, where links won't be available via history.
 
-	if err := nexusoperation.TransitionStarted.Apply(op, ctx, nexusoperation.EventStarted{
+	return nexusoperation.TransitionStarted.Apply(op, ctx, nexusoperation.EventStarted{
 		OperationToken: attrs.GetOperationToken(),
-	}); err != nil {
-		return err
-	}
-
-	// If cancellation was already requested, schedule sending the cancellation request now that we have
-	// an operation token.
-	cancellation, ok := op.Cancellation.TryGet(ctx)
-	if ok && cancellation.StateMachineState() == nexusoperationpb.CANCELLATION_STATUS_UNSPECIFIED {
-		return nexusoperation.TransitionCancellationScheduled.Apply(cancellation, ctx, nexusoperation.EventCancellationScheduled{})
-	}
-
-	return nil
+	})
 }
 
 func (d StartedEventDefinition) CherryPick(ctx chasm.MutableContext, wf *chasmworkflow.Workflow, event *historypb.HistoryEvent, excludeTypes map[enumspb.ResetReapplyExcludeType]struct{}) error {