feat(embed): CancelActivityForOrg — org-scoped cancel mutator for in-process hosts

Mirrors ActivitiesForOrg's org-scoped read with a matching mutator so a host
binary (cloud's clients/visor fleet) can cancel a queued/running gpu-jobs
activity it just listed, without a second HTTP hop into its own tasks surface.
Tenancy is the caller's to enforce (validated X-Org-Id). Enables the cloud
POST /v1/fleet/jobs/:id/cancel endpoint.

pkg/tasks/cancel_for_org_test.go

@@ -0,0 +1,55 @@
+// Copyright © 2026 Hanzo AI. MIT License.
+
+package tasks
+
+import (
+	"context"
+	"testing"
+)
+
+// TestCancelActivityForOrg proves the org-scoped cancel MUTATOR: it cancels a job
+// in the caller's shard, is rejected once the job is terminal, and never reaches
+// across the org boundary — the same tenancy contract ActivitiesForOrg reads under.
+func TestCancelActivityForOrg(t *testing.T) {
+	emb, err := Embed(context.Background(), EmbedConfig{ZAPPort: 0})
+	if err != nil {
+		t.Fatalf("embed: %v", err)
+	}
+	defer emb.Stop(context.Background())
+
+	const org, ns = "acme", "gpu-jobs"
+	en := emb.engine.WithOrg(org)
+	if err := en.RegisterNamespace(Namespace{
+		NamespaceInfo: NamespaceInfo{Name: ns, State: "NAMESPACE_STATE_REGISTERED"},
+		Config:        NamespaceCfg{WorkflowExecutionRetentionTtl: "720h", APSLimit: 400},
+	}); err != nil {
+		t.Fatalf("register ns: %v", err)
+	}
+	if _, err := en.StartActivity(ns, "job1", "run1", TypeRef{Name: "studio.render"},
+		"gpu:spark", map[string]any{"prompt": "x"}, nil, "", "", "", "", "", ""); err != nil {
+		t.Fatalf("start activity: %v", err)
+	}
+
+	// Cancel via the org-scoped wrapper → the job is CANCELED in this org's shard.
+	if err := emb.CancelActivityForOrg(org, ns, "job1", "run1", "user canceled", "tester"); err != nil {
+		t.Fatalf("CancelActivityForOrg: %v", err)
+	}
+	rows, err := emb.ActivitiesForOrg(org, ns)
+	if err != nil {
+		t.Fatalf("ActivitiesForOrg: %v", err)
+	}
+	if len(rows) != 1 || rows[0].Status != activityStateCanceled {
+		t.Fatalf("status = %+v, want %s", rows, activityStateCanceled)
+	}
+
+	// A terminal activity refuses a second cancel (matches the HTTP path's 409).
+	if err := emb.CancelActivityForOrg(org, ns, "job1", "run1", "again", "tester"); err == nil {
+		t.Fatal("second cancel on a terminal activity should error")
+	}
+
+	// Tenancy: another org's shard has no such activity — cancel is refused, never
+	// reaching across the boundary.
+	if err := emb.CancelActivityForOrg("other-org", ns, "job1", "run1", "x", "y"); err == nil {
+		t.Fatal("cross-org cancel should error (activity not in the other org's shard)")
+	}
+}

pkg/tasks/embed.go

@@ -463,6 +463,19 @@ func (e *Embedded) ActivitiesForOrg(org, ns string) ([]StandaloneActivity, error
 	return rows, err
 }
 
+// CancelActivityForOrg cancels a standalone activity in org's shard — the
+// org-scoped programmatic MUTATOR that mirrors ActivitiesForOrg's read, so an
+// in-process host (cloud's clients/fleet) can cancel a queued/running job it just
+// listed without a second HTTP hop into its own surface. Tenancy is the caller's
+// to enforce: pass the validated X-Org-Id, never client input. Rejected (error)
+// if the activity is missing or already terminal, exactly like the HTTP path.
+func (e *Embedded) CancelActivityForOrg(org, ns, activityID, runID, reason, identity string) error {
+	if e == nil || e.engine == nil {
+		return fmt.Errorf("tasks engine not ready")
+	}
+	return e.engine.WithOrg(org).CancelActivity(ns, activityID, runID, reason, identity)
+}
+
 // EventsHandler returns the SSE realtime stream of engine events.
 func (e *Embedded) EventsHandler() http.Handler { return e.sseHandler() }