feat(canvas): show live workflow history and simplify sharing (#3877)

This commit is contained in:
Carlos Florêncio
2026-09-23 10:57:06 +01:00
committed by GitHub
parent a2d4ee8c18
commit 41eeaf0c70
75 changed files with 4910 additions and 590 deletions
@@ -18,6 +18,7 @@ func registerCanvasDistributionRoutes(router *gin.Engine, handler *canvasHTTPHan
return
}
router.POST("/api/v1/canvases/:canvasID/exports", handler.prepareCanvasExport)
router.GET("/api/v1/canvases/:canvasID/export-defaults", handler.getCanvasExportDefaults)
router.GET("/api/v1/canvases/exports/:preparationID", handler.getCanvasExport)
router.GET("/api/v1/canvases/exports/:preparationID/bundle", handler.downloadCanvasBundle)
router.GET("/api/v1/canvases/exports/:preparationID/source", handler.downloadCanvasSource)
@@ -28,6 +29,23 @@ func registerCanvasDistributionRoutes(router *gin.Engine, handler *canvasHTTPHan
router.DELETE("/api/v1/canvases/install-preparations/:preparationID", handler.cancelCanvasInstall)
}
func (h *canvasHTTPHandler) getCanvasExportDefaults(c *gin.Context) {
item, err := h.canvases.Get(c.Request.Context(), c.Param("canvasID"))
if err != nil {
writeCanvasDistributionError(c, err)
return
}
if !h.authorizeWorkspace(c, item.WorkspaceID) {
return
}
defaults, err := h.distribution.GetExportDefaults(c.Request.Context(), canvasRuntimeUser(c), item.WorkspaceID, item.ID)
if err != nil {
writeCanvasDistributionError(c, err)
return
}
writeCanvasJSON(c, http.StatusOK, defaults)
}
type canvasExportRequest struct {
WorkspaceID string `json:"workspace_id"`
ExpectedReleaseID string `json:"expected_release_id"`
@@ -1,13 +1,48 @@
package backendapp
import (
"context"
"errors"
"net/http"
"net/http/httptest"
"testing"
"github.com/gin-gonic/gin"
"github.com/kandev/kandev/internal/auth/authn"
canvasservice "github.com/kandev/kandev/internal/canvas"
)
func TestCanvasRenameRequiresWorkspaceManageScope(t *testing.T) {
taskSvc, _, _ := newRunSubscriptionCheckHarness(t)
workspaceID := seedOwnedWorkspace(t, taskSvc, "canvas-owner")
owner := authn.WithIdentity(context.Background(), authn.Identity{UserID: "canvas-owner", Role: authn.RoleMember})
if _, err := taskSvc.UpsertWorkspaceMember(owner, workspaceID, "canvas-viewer", "viewer"); err != nil {
t.Fatal(err)
}
handler := &canvasHTTPHandler{tasks: taskSvc}
for _, test := range []struct {
name string
user string
allowed bool
status int
}{
{name: "owner", user: "canvas-owner", allowed: true, status: http.StatusOK},
{name: "viewer", user: "canvas-viewer", allowed: false, status: http.StatusForbidden},
} {
t.Run(test.name, func(t *testing.T) {
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
ctx.Request = httptest.NewRequest(http.MethodPatch, "/", nil).WithContext(authn.WithIdentity(context.Background(), authn.Identity{UserID: test.user, Role: authn.RoleMember}))
if allowed := handler.authorizeWorkspaceManage(ctx, workspaceID); allowed != test.allowed {
t.Fatalf("allowed = %v, want %v", allowed, test.allowed)
}
if !test.allowed && recorder.Code != test.status {
t.Fatalf("status = %d, want %d", recorder.Code, test.status)
}
})
}
}
func TestCanvasDistributionErrorStatusUsesInternalErrorForUnexpectedFailures(t *testing.T) {
status, code := canvasDistributionErrorStatus(errors.New("database unavailable"))
if status != http.StatusInternalServerError || code != "internal_error" {
@@ -21,3 +56,14 @@ func TestCanvasDistributionErrorStatusMapsIncompatibleInstall(t *testing.T) {
t.Fatalf("status = %d %q, want 409 incompatible_install", status, code)
}
}
func TestCanvasReleaseResponseIncludesSafeDistributionFields(t *testing.T) {
response := releaseResponseFromMetadata(&canvasservice.ReleaseMetadata{
ID: "release-1", PackageID: "canvas-example", Version: "1.1.0", DisplayName: "Example",
Description: "Example canvas", Author: "Author", License: "MIT", SourceMode: "static",
MinKandevVersion: "0.95.0", RepoURL: "https://example.test/canvas",
})
if response.PackageID != "canvas-example" || response.Version != "1.1.0" || response.DisplayName != "Example" || response.Description != "Example canvas" || response.Author != "Author" || response.License != "MIT" || response.SourceMode != "static" || response.MinKandevVersion != "0.95.0" || response.RepoURL != "https://example.test/canvas" {
t.Fatalf("distribution projection = %+v", response)
}
}
@@ -14,6 +14,7 @@ import (
"github.com/gin-gonic/gin"
"github.com/kandev/kandev/internal/authz"
canvasservice "github.com/kandev/kandev/internal/canvas"
"github.com/kandev/kandev/internal/common/constants"
"github.com/kandev/kandev/internal/orchestrator"
@@ -26,7 +27,7 @@ import (
v1 "github.com/kandev/kandev/pkg/api/v1"
)
// canvasHTTPHandler is the owner-authorized browser API for canvas metadata
// canvasHTTPHandler is the workspace-authorized browser API for canvas metadata
// and host operations. Source transfer and publishing stay on the agent MCP
// path; browsers receive only metadata and short-lived runtime capabilities.
type canvasHTTPHandler struct {
@@ -365,6 +366,7 @@ func registerCanvasRoutes(p routeParams) {
p.router.GET("/api/v1/canvases/workspaces/:workspaceID", h.listWorkspace)
registerCanvasDistributionRoutes(p.router, h)
p.router.GET("/api/v1/canvases/:canvasID", h.get)
p.router.PATCH("/api/v1/canvases/:canvasID", h.rename)
p.router.GET("/api/v1/canvases/:canvasID/releases", h.releases)
p.router.GET("/api/v1/canvases/:canvasID/promotion-preview", h.promotionPreview)
p.router.POST("/api/v1/canvases/:canvasID/promotion", h.promote)
@@ -406,6 +408,15 @@ type canvasHTTPResponse struct {
type canvasReleaseResponse struct {
ID string `json:"id"`
PackageID string `json:"package_id,omitempty"`
Version string `json:"version,omitempty"`
DisplayName string `json:"display_name,omitempty"`
Description string `json:"description,omitempty"`
Author string `json:"author,omitempty"`
License string `json:"license,omitempty"`
SourceMode string `json:"source_mode,omitempty"`
MinKandevVersion string `json:"min_kandev_version,omitempty"`
RepoURL string `json:"repo_url,omitempty"`
PackageDigest string `json:"package_digest,omitempty"`
ValidationStatus string `json:"validation_status"`
ValidationError string `json:"validation_error,omitempty"`
@@ -518,6 +529,30 @@ func (h *canvasHTTPHandler) get(c *gin.Context) {
writeCanvasJSON(c, http.StatusOK, response)
}
func (h *canvasHTTPHandler) rename(c *gin.Context) {
item, err := h.canvases.Get(c.Request.Context(), c.Param("canvasID"))
if err != nil {
h.writeError(c, err)
return
}
if !h.authorizeWorkspaceManage(c, item.WorkspaceID) {
return
}
var request struct {
Title string `json:"title"`
}
if err := c.ShouldBindJSON(&request); err != nil {
writeCanvasError(c, http.StatusBadRequest, "invalid_request", nil)
return
}
updated, err := h.canvases.Rename(c.Request.Context(), item.ID, request.Title)
if err != nil {
h.writeError(c, err)
return
}
writeCanvasJSON(c, http.StatusOK, canvasResponse(*updated))
}
func (h *canvasHTTPHandler) releases(c *gin.Context) {
canvas, err := h.canvases.Get(c.Request.Context(), c.Param("canvasID"))
if err != nil {
@@ -553,16 +588,7 @@ func (h *canvasHTTPHandler) releases(c *gin.Context) {
}
func releaseResponse(release instances.Release, scope string, grants []instances.Grant) canvasReleaseResponse {
permissions := canvasservice.ReleasePermissionSummary(release)
return canvasReleaseResponse{
ID: release.ID, PackageDigest: release.PackageDigest,
ValidationStatus: release.ValidationStatus, ValidationError: release.ValidationError,
Permissions: &permissions, MissingPermissions: canvasservice.MissingPermissionKeys(permissions, scope, grants),
PermissionDigest: canvasservice.PermissionDigest(release), SourceActorKind: release.SourceActorKind,
SourceUserID: release.SourceUserID, SourceTaskID: release.SourceTaskID,
SourceSessionID: release.SourceSessionID, ProtocolVersion: release.ProtocolVersion,
CreatedAt: release.CreatedAt,
}
return *releaseResponseFromMetadata(canvasservice.ReleaseMetadataForHTTP(release, scope, grants))
}
func releaseResponseFromMetadata(value *canvasservice.ReleaseMetadata) *canvasReleaseResponse {
@@ -571,6 +597,9 @@ func releaseResponseFromMetadata(value *canvasservice.ReleaseMetadata) *canvasRe
}
return &canvasReleaseResponse{
ID: value.ID, PackageDigest: value.PackageDigest, ValidationStatus: value.ValidationStatus,
PackageID: value.PackageID, Version: value.Version, DisplayName: value.DisplayName,
Description: value.Description, Author: value.Author, License: value.License,
SourceMode: value.SourceMode, MinKandevVersion: value.MinKandevVersion, RepoURL: value.RepoURL,
ValidationError: value.ValidationError, Permissions: value.Permissions,
MissingPermissions: value.MissingPermissions, PermissionDigest: value.PermissionDigest,
SourceActorKind: value.SourceActorKind, SourceUserID: value.SourceUserID,
@@ -846,6 +875,17 @@ func (h *canvasHTTPHandler) authorizeWorkspace(c *gin.Context, workspaceID strin
return true
}
func (h *canvasHTTPHandler) authorizeWorkspaceManage(c *gin.Context, workspaceID string) bool {
if !h.authorizeWorkspace(c, workspaceID) {
return false
}
if err := h.tasks.AuthorizeWorkspaceScope(c.Request.Context(), workspaceID, authz.ScopeWorkspaceManage); err != nil {
writeCanvasError(c, http.StatusForbidden, "forbidden", nil)
return false
}
return true
}
func canvasRouteParam(c *gin.Context, names ...string) string {
for _, name := range names {
if value := c.Param(name); value != "" {
@@ -7,6 +7,7 @@ import (
"fmt"
"path/filepath"
"strconv"
"strings"
"sync"
"testing"
"time"
@@ -83,6 +84,35 @@ func TestGetExposesPendingFirstRelease(t *testing.T) {
}
}
func TestRenameOnlyChangesTitleAndEmitsUpdate(t *testing.T) {
service, _, _ := newCanvasService(t)
created := createCanvas(t, service, CreateCanvasRequest{WorkspaceID: "workspace-1", TaskID: "task-1", Title: "Before"})
var event LifecycleEvent
service.SetEventPublisher(func(_ context.Context, value LifecycleEvent) { event = value })
updated, err := service.Rename(context.Background(), created.ID, " After ")
if err != nil {
t.Fatal(err)
}
if updated.Title != "After" || updated.ID != created.ID || updated.PluginInstanceID != created.PluginInstanceID || updated.TaskID != created.TaskID {
t.Fatalf("unexpected rename projection: %+v", updated)
}
if event.Type != EventUpdated || event.Title != "After" || event.WorkspaceID != created.WorkspaceID {
t.Fatalf("unexpected update event: %+v", event)
}
if _, err := service.Rename(context.Background(), created.ID, " "); !errors.Is(err, ErrInvalidCanvas) {
t.Fatalf("blank rename = %v", err)
}
if _, err := service.Rename(context.Background(), created.ID, strings.Repeat("x", MaxTitleLength+1)); !errors.Is(err, ErrInvalidCanvas) {
t.Fatalf("long rename = %v", err)
}
if err := service.Remove(context.Background(), created.ID); err != nil {
t.Fatal(err)
}
if _, err := service.Rename(context.Background(), created.ID, "Restored"); !errors.Is(err, ErrCanvasNotFound) {
t.Fatalf("removed rename = %v", err)
}
}
func TestRemoveDeletesCreationAuthority(t *testing.T) {
service, _, _ := newCanvasService(t)
created := createCanvas(t, service, CreateCanvasRequest{
@@ -53,6 +53,68 @@ type ExportMetadata struct {
RepoURL string `json:"repo_url,omitempty"`
}
// MinimumCanvasDistributionVersion is the first tagged Kandev release with
// portable canvas distribution support.
const MinimumCanvasDistributionVersion = "0.95.0"
type ExportDefaults struct {
ExpectedReleaseID string `json:"expected_release_id"`
Metadata ExportMetadata `json:"metadata"`
MissingRequired []string `json:"missing_required"`
}
// GetExportDefaults derives a temporary share draft from the active release.
func (s *DistributionService) GetExportDefaults(ctx context.Context, userID, workspaceID, canvasID string) (ExportDefaults, error) {
if err := s.validateReady(userID, workspaceID, canvasID); err != nil {
return ExportDefaults{}, err
}
if err := s.authorizeWorkspace(ctx, workspaceID); err != nil {
return ExportDefaults{}, err
}
item, err := s.canvases.Get(ctx, canvasID)
if err != nil || !validExportCanvas(item, ExportRequest{WorkspaceID: workspaceID}) {
return ExportDefaults{}, ErrCanvasNotFound
}
_, release, files, err := s.loadExportSnapshot(ctx, ExportRequest{WorkspaceID: workspaceID, CanvasID: canvasID, ExpectedReleaseID: item.ActiveReleaseID})
if err != nil {
return ExportDefaults{}, err
}
current := releaseManifest(release)
metadata := fillExportMetadata(ExportMetadata{}, current)
if current.Distribution == nil && !staticExportAvailable(current, files) {
metadata.SourceMode = ""
}
if metadata.DisplayName == "" {
metadata.DisplayName = item.Title
}
if metadata.MinKandevVersion == "" {
metadata.MinKandevVersion = MinimumCanvasDistributionVersion
}
missing := make([]string, 0, 8)
for _, field := range []struct{ key, value string }{
{"package_id", metadata.PackageID}, {"version", metadata.Version}, {"display_name", metadata.DisplayName},
{"description", metadata.Description}, {"author", metadata.Author}, {"license", metadata.License},
{"source_mode", metadata.SourceMode}, {"min_kandev_version", metadata.MinKandevVersion},
} {
if strings.TrimSpace(field.value) == "" {
missing = append(missing, field.key)
}
}
return ExportDefaults{ExpectedReleaseID: release.ID, Metadata: metadata, MissingRequired: missing}, nil
}
func staticExportAvailable(current manifest.Manifest, files map[string][]byte) bool {
if !current.IsStaticWebAppOnly() {
return false
}
for _, app := range current.UI.WebApps {
if _, ok := files[app.Entry]; !ok {
return false
}
}
return true
}
type ExportRequest struct {
UserID string
WorkspaceID string
@@ -164,10 +226,14 @@ func (s *DistributionService) PrepareExport(ctx context.Context, request ExportR
if err != nil {
return ExportReview{}, err
}
metadata, err := normalizeExportMetadata(request.Metadata, releaseManifest(release))
current := releaseManifest(release)
metadata, err := normalizeExportMetadata(request.Metadata, current)
if err != nil {
return ExportReview{}, err
}
if metadata.SourceMode == manifest.SourceModeStatic && !staticExportAvailable(current, files) {
return ExportReview{}, fmt.Errorf("%w: static export is unavailable", ErrExportInvalid)
}
manifestValue, err := exportManifest(release.ManifestJSON, metadata)
if err != nil {
return ExportReview{}, err
@@ -54,6 +54,50 @@ func TestPrepareExportBuildsOneReleaseBoundBundleAndSource(t *testing.T) {
}
}
func TestPrepareExportRejectsStaticModeWhenEntryFileIsMissing(t *testing.T) {
service, _, _ := newDistributionExportTestService(t)
delete(service.artifacts.(*fakeDistributionArtifactReader).files, "ui/index.html")
defaults, err := service.GetExportDefaults(context.Background(), "user-1", "workspace-1", "canvas-1")
if err != nil || defaults.Metadata.SourceMode != "" || !containsString(defaults.MissingRequired, "source_mode") {
t.Fatalf("missing static entry defaults = %+v, error = %v", defaults, err)
}
_, err = service.PrepareExport(context.Background(), ExportRequest{
UserID: "user-1", WorkspaceID: "workspace-1", CanvasID: "canvas-1", ExpectedReleaseID: "release-1",
Metadata: ExportMetadata{PackageID: "canvas-board", Version: "1.2.3", DisplayName: "Board", Description: "A board", Author: "Kandev", License: "MIT", SourceMode: manifest.SourceModeStatic, MinKandevVersion: "1.0.0"},
})
if !errors.Is(err, ErrExportInvalid) || service.preparationsCount() != 0 {
t.Fatalf("missing static entry prepared: error = %v, preparations = %d", err, service.preparationsCount())
}
}
func TestExportDefaultsUseReleaseAndRequireAuthorLicense(t *testing.T) {
service, releases, canvases := newDistributionExportTestService(t)
var value manifest.Manifest
if err := json.Unmarshal(releases.release.ManifestJSON, &value); err != nil {
t.Fatal(err)
}
value.MinKandevVersion = ""
value.Distribution = nil
releases.release.ManifestJSON, _ = json.Marshal(value)
canvases.canvas.Title = "Mutable canvas name"
defaults, err := service.GetExportDefaults(context.Background(), "user-1", "workspace-1", "canvas-1")
if err != nil {
t.Fatal(err)
}
if defaults.ExpectedReleaseID != "release-1" || defaults.Metadata.PackageID != value.ID || defaults.Metadata.Version != value.Version || defaults.Metadata.DisplayName != value.DisplayName || defaults.Metadata.Author != value.Author || defaults.Metadata.MinKandevVersion != MinimumCanvasDistributionVersion {
t.Fatalf("unexpected defaults: %+v", defaults)
}
if len(defaults.MissingRequired) != 1 || defaults.MissingRequired[0] != "license" {
t.Fatalf("missing fields = %v", defaults.MissingRequired)
}
value.MinKandevVersion = "1.5.0"
releases.release.ManifestJSON, _ = json.Marshal(value)
newDefaults, err := service.GetExportDefaults(context.Background(), "user-1", "workspace-1", "canvas-1")
if err != nil || newDefaults.Metadata.MinKandevVersion != "1.5.0" {
t.Fatalf("higher minimum = %+v, %v", newDefaults, err)
}
}
func TestPrepareExportAllowsTaskCanvasWithPortableWorkspacePlacement(t *testing.T) {
service, releases, canvases := newDistributionExportTestService(t)
var sourceManifest manifest.Manifest
@@ -37,6 +37,11 @@ func releaseMetadata(release plugininstances.Release, scope string, grants []plu
}
}
// ReleaseMetadataForHTTP returns safe manifest fields for host projections.
func ReleaseMetadataForHTTP(release plugininstances.Release, scope string, grants []plugininstances.Grant) *ReleaseMetadata {
return releaseMetadata(release, scope, grants)
}
type manifestSeed struct {
PackageID string
Version string
@@ -455,6 +455,22 @@ func (r *Repository) DeleteTx(ctx context.Context, tx *sqlx.Tx, id string) error
return nil
}
// Rename changes only the mutable canvas title. A removed row cannot be recreated.
func (r *Repository) Rename(ctx context.Context, id, title string, updatedAt time.Time) error {
result, err := r.db.ExecContext(ctx, r.db.Rebind(`UPDATE canvas_lifecycle_metadata SET title = ?, updated_at = ? WHERE id = ?`), title, updatedAt.UTC().Format(time.RFC3339Nano), id)
if err != nil {
return err
}
affected, err := result.RowsAffected()
if err != nil {
return err
}
if affected == 0 {
return ErrCanvasNotFound
}
return nil
}
// Promote records the human-owned provenance and clears the current task
// relationship. The instance scope/grants are changed by the plugin
// instance store immediately before this metadata transaction.
+36
View File
@@ -173,6 +173,40 @@ func (s *Service) GetCanvas(ctx context.Context, id string) (*Canvas, error) {
return s.Get(ctx, id)
}
// Rename updates the instance title without changing its release or runtime.
func (s *Service) Rename(ctx context.Context, id, title string) (*Canvas, error) {
if err := s.ready(); err != nil {
return nil, err
}
title = strings.TrimSpace(title)
if title == "" || len([]rune(title)) > MaxTitleLength {
return nil, ErrInvalidCanvas
}
s.mu.Lock()
_, instance, err := s.load(ctx, id)
if err == nil && instance.Status == StatusRemoved {
err = ErrCanvasNotFound
}
if err == nil {
err = s.repo.Rename(ctx, id, title, s.nowUTC())
}
var updated Canvas
if err == nil {
metadata, current, loadErr := s.load(ctx, id)
err = loadErr
if err == nil {
updated, err = s.buildCanvas(ctx, metadata, current)
}
}
publisher := s.publisher
s.mu.Unlock()
if err != nil {
return nil, err
}
publishEvent(ctx, publisher, lifecycleEvent(EventUpdated, updated))
return &updated, nil
}
// ListTaskCanvases lists task-scoped canvases. Removed canvases are never
// returned; archived canvases are returned only when includeArchived is true.
func (s *Service) ListTaskCanvases(ctx context.Context, taskID string, includeArchived bool) ([]Canvas, error) {
@@ -855,6 +889,8 @@ func isWorkspaceCanvas(metadata CanvasMetadata, instance plugininstances.Instanc
func lifecycleEvent(eventType string, canvas Canvas) LifecycleEvent {
return LifecycleEvent{
Type: eventType,
Title: canvas.Title,
UpdatedAt: canvas.UpdatedAt,
CanvasID: canvas.ID,
PluginInstanceID: canvas.PluginInstanceID,
WorkspaceID: canvas.WorkspaceID,
+12 -9
View File
@@ -207,19 +207,22 @@ type transactionalPluginInstanceStore interface {
// It never carries source files, application state, package bodies, or runtime
// capabilities.
type LifecycleEvent struct {
Type string `json:"type"`
CanvasID string `json:"canvas_id"`
PluginInstanceID string `json:"plugin_instance_id"`
WorkspaceID string `json:"workspace_id"`
TaskID string `json:"task_id,omitempty"`
ScopeKind string `json:"scope_kind"`
Status string `json:"status"`
ActiveReleaseID string `json:"active_release_id,omitempty"`
ActiveReleaseStatus string `json:"active_release_status,omitempty"`
Type string `json:"type"`
Title string `json:"title,omitempty"`
UpdatedAt time.Time `json:"updated_at,omitempty"`
CanvasID string `json:"canvas_id"`
PluginInstanceID string `json:"plugin_instance_id"`
WorkspaceID string `json:"workspace_id"`
TaskID string `json:"task_id,omitempty"`
ScopeKind string `json:"scope_kind"`
Status string `json:"status"`
ActiveReleaseID string `json:"active_release_id,omitempty"`
ActiveReleaseStatus string `json:"active_release_status,omitempty"`
}
const (
EventCreated = "canvas.created"
EventUpdated = "canvas.updated"
EventReleaseActivated = "canvas.release.activated"
EventReleasePermissionRequired = "canvas.release.permission_required"
EventPromoted = "canvas.promoted"
+1
View File
@@ -37,6 +37,7 @@ const (
// source files, application state, or runtime capabilities.
const (
CanvasCreated = "canvas.created"
CanvasUpdated = "canvas.updated"
CanvasReleaseActivated = "canvas.release.activated"
CanvasReleasePermissionRequired = "canvas.release.permission_required"
CanvasPromoted = "canvas.promoted"
@@ -40,6 +40,7 @@ func RegisterCanvasNotifications(ctx context.Context, eventBus bus.EventBus, hub
action string
}{
{events.CanvasCreated, ws.ActionCanvasCreated},
{events.CanvasUpdated, ws.ActionCanvasUpdated},
{events.CanvasReleaseActivated, ws.ActionCanvasReleaseActivated},
{events.CanvasReleasePermissionRequired, ws.ActionCanvasReleasePermissionRequired},
{events.CanvasPromoted, ws.ActionCanvasPromoted},
@@ -59,10 +59,12 @@ workspace.
| --- | --- | --- | --- |
| GET | `./_kandev/v1/data/tasks` | `api_read:tasks` | List tasks |
| GET | `./_kandev/v1/data/tasks/{task_id}` | `api_read:tasks` | Read one task |
| GET | `./_kandev/v1/data/tasks/{task_id}/step-transitions` | `api_read:tasks` | Read recorded task moves |
| PATCH | `./_kandev/v1/data/tasks/{task_id}` | `api_write:tasks` | Update a task |
| POST | `./_kandev/v1/data/tasks/{task_id}/messages` | `api_write:messages` | Send a task message |
| GET | `./_kandev/v1/data/workflows` | `api_read:workflows` | List workflows |
| GET | `./_kandev/v1/data/workflows/{workflow_id}/steps` | `api_read:workflows` | Read workflow steps |
| GET | `./_kandev/v1/data/workflows/{workflow_id}/transition-groups` | `api_read:tasks` and `api_read:workflows`; workspace canvas only | Read recorded workflow route counts |
The task-list query accepts `cursor`, `limit`, `include_archived`,
`workflow_id`, `state`, and `parent_id`. `workflow_id` and `state` can be
@@ -120,6 +122,20 @@ A workflow object contains `id`, `workspace_id`, `name`, `description`,
`is_start_step`, `wip_limit`, `agent_profile_id`, and
`on_enter_action_types`.
The task transition route returns rows newest first, with `id` as a decimal
string, nullable `from_workflow_id`, `from_workflow_step_id`,
`to_workflow_id`, and `to_workflow_step_id`, `trigger`, and RFC3339
`occurred_at`. It omits actor and session identity. The opaque `cursor`
continues by ledger ID; it is bound to the requested task. Empty history
means no retained move was recorded, not that the task never moved.
The workflow transition-group route returns `kind` (`within`, `entry`, or
`exit`), nullable `from_step_id` and `to_step_id`, and `count`. It includes
archived tasks and removed step IDs, but not deleted tasks. Its opaque cursor
orders groups by kind and step IDs. Both routes use the collection envelope
and the 1..200 `limit` rule above. A task canvas cannot request workflow-wide
groups; promotion to workspace scope is a user action, not a fallback.
## Writes and workflow movement
`PATCH ./_kandev/v1/data/tasks/{task_id}` accepts one or more of these JSON
@@ -26,6 +26,10 @@ func int64Ptr(v int64) *int64 { return &v }
// ── fakes for the narrow Host data API interfaces ───────────────────────
type fakeTaskDataSource struct {
transitionRows map[string][]taskmodels.StepTransition
transitionGroups map[string][]taskmodels.TransitionGroup
transitionCalls int
groupCalls int
workspaces []*taskmodels.Workspace
tasksByWorkspace map[string][]*taskmodels.Task
tasksByID map[string]*taskmodels.Task
@@ -66,6 +70,16 @@ type fakeTaskDataSource struct {
dependencyViewsTasks []string
}
func (f *fakeTaskDataSource) ListTaskStepTransitions(_ context.Context, taskID string, _ int, _ string) ([]taskmodels.StepTransition, string, error) {
f.transitionCalls++
return f.transitionRows[taskID], "", nil
}
func (f *fakeTaskDataSource) ListWorkflowTransitionGroups(_ context.Context, workflowID string, _ int, _ string) ([]taskmodels.TransitionGroup, string, error) {
f.groupCalls++
return f.transitionGroups[workflowID], "", nil
}
func (f *fakeTaskDataSource) ListWorkspaces(context.Context) ([]*taskmodels.Workspace, error) {
return f.workspaces, nil
}
@@ -402,6 +416,24 @@ func newTestDataHost(caps manifest.Capabilities) *testDataHost {
return d
}
// @covers AC-PLUGINS-WORKFLOW-HISTORY-001.4 AC-PLUGINS-WORKFLOW-HISTORY-002.1
func TestHostTransitionHistoryUsesExistingReadGrants(t *testing.T) {
d := newTestDataHost(manifest.Capabilities{APIRead: []string{"tasks"}})
d.tasks.tasksByID = map[string]*taskmodels.Task{"task-1": {ID: "task-1", WorkspaceID: "ws-1"}}
d.tasks.transitionRows = map[string][]taskmodels.StepTransition{"task-1": {{ID: 9, Trigger: "test", OccurredAt: time.Now().UTC()}}}
reader, ok := pluginsdk.TransitionHistory(d.host)
if !ok {
t.Fatal("plugin Host transition-history extension is unavailable")
}
items, _, err := reader.ListTask(context.Background(), "task-1", pluginsdk.Page{})
if err != nil || len(items) != 1 || items[0].ID != "9" {
t.Fatalf("task history = %+v, err=%v", items, err)
}
if _, _, err := reader.ListWorkflowGroups(context.Background(), "wf-1", pluginsdk.Page{}); status.Code(err) != codes.PermissionDenied {
t.Fatalf("workflow groups without workflows grant: %v", err)
}
}
// ── capability gating: denied without api_read:<resource> ──────────────
func TestPluginHost_Tasks_DeniedWithoutCapability(t *testing.T) {
@@ -0,0 +1,79 @@
package plugins
import (
"context"
"errors"
"strconv"
"time"
taskservice "github.com/kandev/kandev/internal/task/service"
"github.com/kandev/kandev/pkg/pluginsdk"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
func (h *pluginHost) Transitions() pluginsdk.TransitionHistoryReader {
return hostTransitionHistoryReader{host: h}
}
type hostTransitionHistoryReader struct{ host *pluginHost }
func (r hostTransitionHistoryReader) ListTask(ctx context.Context, taskID string, page pluginsdk.Page) ([]pluginsdk.TaskStepTransition, *pluginsdk.PageInfo, error) {
if !r.host.capabilities.CanRead(resourceTasks) {
return nil, nil, permissionDenied(apiReadCapability(resourceTasks))
}
task, err := r.host.fetchTaskForScopeCheck(ctx, taskID)
if err != nil {
return nil, nil, err
}
if task == nil {
return nil, nil, status.Error(codes.NotFound, "task not found")
}
source, ok := r.host.taskData.(transitionDataSource)
if !ok {
return nil, nil, status.Error(codes.Unimplemented, "transition history unavailable")
}
items, next, err := source.ListTaskStepTransitions(ctx, taskID, normalizePageLimit(page.Limit), page.Cursor)
if err != nil {
return nil, nil, hostTransitionReadError(err)
}
result := make([]pluginsdk.TaskStepTransition, len(items))
for i, item := range items {
result[i] = pluginsdk.TaskStepTransition{
ID: strconv.FormatInt(item.ID, 10), FromWorkflowID: item.FromWorkflowID,
FromWorkflowStepID: item.FromWorkflowStepID, ToWorkflowID: item.ToWorkflowID,
ToWorkflowStepID: item.ToWorkflowStepID, Trigger: item.Trigger,
OccurredAt: item.OccurredAt.UTC().Format(time.RFC3339Nano),
}
}
return result, &pluginsdk.PageInfo{NextCursor: next, HasMore: next != ""}, nil
}
func (r hostTransitionHistoryReader) ListWorkflowGroups(ctx context.Context, workflowID string, page pluginsdk.Page) ([]pluginsdk.WorkflowTransitionGroup, *pluginsdk.PageInfo, error) {
if !r.host.capabilities.CanRead(resourceTasks) {
return nil, nil, permissionDenied(apiReadCapability(resourceTasks))
}
if !r.host.capabilities.CanRead(resourceWorkflows) {
return nil, nil, permissionDenied(apiReadCapability(resourceWorkflows))
}
source, ok := r.host.taskData.(transitionDataSource)
if !ok {
return nil, nil, status.Error(codes.Unimplemented, "transition history unavailable")
}
items, next, err := source.ListWorkflowTransitionGroups(ctx, workflowID, normalizePageLimit(page.Limit), page.Cursor)
if err != nil {
return nil, nil, hostTransitionReadError(err)
}
result := make([]pluginsdk.WorkflowTransitionGroup, len(items))
for i, item := range items {
result[i] = pluginsdk.WorkflowTransitionGroup{Kind: item.Kind, FromStepID: item.FromStepID, ToStepID: item.ToStepID, Count: item.Count}
}
return result, &pluginsdk.PageInfo{NextCursor: next, HasMore: next != ""}, nil
}
func hostTransitionReadError(err error) error {
if errors.Is(err, taskservice.ErrInvalidTransitionCursor) {
return status.Error(codes.InvalidArgument, "invalid transition cursor")
}
return err
}
@@ -219,6 +219,8 @@ func (s *Service) handleWebAppTasks(ctx context.Context, w http.ResponseWriter,
s.listWebAppTasks(ctx, w, r, host, binding)
case len(parts) == 1 && r.Method == http.MethodGet:
s.getWebAppTask(ctx, w, r, host, binding, parts[0])
case len(parts) == 2 && parts[1] == "step-transitions" && r.Method == http.MethodGet:
s.listWebAppTaskTransitions(ctx, w, r, host, binding, parts[0])
case len(parts) == 1 && r.Method == http.MethodPatch:
s.updateWebAppTask(ctx, w, r, host, binding, parts[0])
case len(parts) == 2 && parts[1] == "messages" && r.Method == http.MethodPost:
@@ -234,6 +236,8 @@ func (s *Service) handleWebAppWorkflows(ctx context.Context, w http.ResponseWrit
s.listWebAppWorkflows(ctx, w, r, host, binding)
case len(parts) == 2 && parts[1] == "steps" && r.Method == http.MethodGet:
s.listWebAppWorkflowSteps(ctx, w, r, host, binding, parts[0])
case len(parts) == 2 && parts[1] == "transition-groups" && r.Method == http.MethodGet:
s.listWebAppWorkflowTransitionGroups(ctx, w, r, host, binding, parts[0])
default:
writeWebAppError(w, http.StatusMethodNotAllowed, "method_not_allowed")
}
@@ -0,0 +1,146 @@
package plugins
import (
"context"
"errors"
"net/http"
"strconv"
"time"
"github.com/kandev/kandev/internal/plugins/instances"
"github.com/kandev/kandev/internal/plugins/webapp"
taskmodels "github.com/kandev/kandev/internal/task/models"
taskservice "github.com/kandev/kandev/internal/task/service"
)
type transitionDataSource interface {
ListTaskStepTransitions(context.Context, string, int, string) ([]taskmodels.StepTransition, string, error)
ListWorkflowTransitionGroups(context.Context, string, int, string) ([]taskmodels.TransitionGroup, string, error)
}
type webAppTransition struct {
ID string `json:"id"`
FromWorkflowID *string `json:"from_workflow_id"`
FromWorkflowStepID *string `json:"from_workflow_step_id"`
ToWorkflowID *string `json:"to_workflow_id"`
ToWorkflowStepID *string `json:"to_workflow_step_id"`
Trigger string `json:"trigger"`
OccurredAt string `json:"occurred_at"`
}
type webAppTransitionGroup struct {
Kind string `json:"kind"`
FromStepID *string `json:"from_step_id"`
ToStepID *string `json:"to_step_id"`
Count int64 `json:"count"`
}
func (s *Service) listWebAppTaskTransitions(ctx context.Context, w http.ResponseWriter, r *http.Request, host *pluginHost, binding webapp.CapabilityBinding, taskID string) {
if !validWebAppKey(taskID) {
writeWebAppError(w, http.StatusNotFound, "not_found")
return
}
page, err := webAppRequestPage(r)
if err != nil {
writeWebAppError(w, http.StatusBadRequest, "invalid_request")
return
}
task, err := host.fetchTaskForScopeCheck(ctx, taskID)
if err != nil {
writeWebAppError(w, webAppProtocolStatus(err), webAppErrorCode(err))
return
}
if task == nil {
writeWebAppError(w, http.StatusNotFound, "not_found")
return
}
if !webAppTaskMatches(ctx, host, binding, *task) {
writeWebAppError(w, http.StatusNotFound, "not_found")
return
}
source, ok := s.taskData.(transitionDataSource)
if !ok {
writeWebAppError(w, http.StatusNotImplemented, "not_implemented")
return
}
items, next, err := source.ListTaskStepTransitions(ctx, taskID, int(page.Limit), page.Cursor)
if err != nil {
writeTransitionReadError(w, err)
return
}
result := make([]webAppTransition, len(items))
for i, item := range items {
result[i] = webAppTransition{
ID: strconv.FormatInt(item.ID, 10), FromWorkflowID: item.FromWorkflowID,
FromWorkflowStepID: item.FromWorkflowStepID, ToWorkflowID: item.ToWorkflowID,
ToWorkflowStepID: item.ToWorkflowStepID, Trigger: item.Trigger,
OccurredAt: item.OccurredAt.UTC().Format(time.RFC3339Nano),
}
}
writeWebAppJSON(w, r, http.StatusOK, webAppPage[webAppTransition]{Items: result, PageInfo: webAppPageInfo{NextCursor: next, HasMore: next != ""}})
}
func (s *Service) listWebAppWorkflowTransitionGroups(ctx context.Context, w http.ResponseWriter, r *http.Request, host *pluginHost, binding webapp.CapabilityBinding, workflowID string) {
if !validWebAppKey(workflowID) {
writeWebAppError(w, http.StatusNotFound, "not_found")
return
}
if binding.ScopeKind != instances.ScopeWorkspace || binding.WorkspaceID == "" || !host.capabilities.CanRead(resourceTasks) || !host.capabilities.CanRead(resourceWorkflows) {
writeWebAppError(w, http.StatusForbidden, "plugin_permission_denied")
return
}
page, err := webAppRequestPage(r)
if err != nil {
writeWebAppError(w, http.StatusBadRequest, "invalid_request")
return
}
if host.workflows == nil {
writeWebAppError(w, http.StatusNotImplemented, "not_implemented")
return
}
allowed, err := webAppWorkflowAllowed(ctx, host, binding.WorkspaceID, workflowID)
if err != nil {
writeWebAppError(w, webAppProtocolStatus(err), webAppErrorCode(err))
return
}
if !allowed {
writeWebAppError(w, http.StatusNotFound, "not_found")
return
}
source, ok := s.taskData.(transitionDataSource)
if !ok {
writeWebAppError(w, http.StatusNotImplemented, "not_implemented")
return
}
items, next, err := source.ListWorkflowTransitionGroups(ctx, workflowID, int(page.Limit), page.Cursor)
if err != nil {
writeTransitionReadError(w, err)
return
}
result := make([]webAppTransitionGroup, len(items))
for i, item := range items {
result[i] = webAppTransitionGroup{Kind: item.Kind, FromStepID: item.FromStepID, ToStepID: item.ToStepID, Count: item.Count}
}
writeWebAppJSON(w, r, http.StatusOK, webAppPage[webAppTransitionGroup]{Items: result, PageInfo: webAppPageInfo{NextCursor: next, HasMore: next != ""}})
}
func webAppWorkflowAllowed(ctx context.Context, host *pluginHost, workspaceID, workflowID string) (bool, error) {
workflows, err := host.workflows.ListWorkflows(ctx, workspaceID, false)
if err != nil {
return false, err
}
for _, workflow := range workflows {
if workflow != nil && workflow.ID == workflowID && workflow.WorkspaceID == workspaceID {
return true, nil
}
}
return false, nil
}
func writeTransitionReadError(w http.ResponseWriter, err error) {
if errors.Is(err, taskservice.ErrInvalidTransitionCursor) {
writeWebAppError(w, http.StatusBadRequest, "invalid_request")
return
}
writeWebAppError(w, webAppProtocolStatus(err), webAppErrorCode(err))
}
@@ -0,0 +1,84 @@
package plugins
import (
"net/http"
"net/http/httptest"
"strconv"
"testing"
"time"
"github.com/kandev/kandev/internal/plugins/instances"
"github.com/kandev/kandev/internal/plugins/webapp"
taskmodels "github.com/kandev/kandev/internal/task/models"
"github.com/stretchr/testify/require"
)
// @covers AC-PLUGINS-WORKFLOW-HISTORY-001.4 AC-PLUGINS-WORKFLOW-HISTORY-001.5
func TestWebAppTransitionTrailChecksTaskScopeBeforeReading(t *testing.T) {
task := &taskmodels.Task{ID: "task-1", WorkspaceID: "ws-1"}
source := &fakeTaskDataSource{
tasksByID: map[string]*taskmodels.Task{"task-1": task, "task-2": {ID: "task-2", WorkspaceID: "ws-1"}},
transitionRows: map[string][]taskmodels.StepTransition{"task-1": {{ID: 7, ToWorkflowStepID: stringPointer("review"), Trigger: "task_moved", OccurredAt: time.Now().UTC()}}},
}
svc := &Service{taskData: source}
binding := webapp.CapabilityBinding{ScopeKind: instances.ScopeTask, WorkspaceID: "ws-1", TaskID: "task-1", Permissions: []string{"api_read:tasks"}}
denied := httptest.NewRecorder()
svc.handleWebAppProtocol(denied, httptest.NewRequest(http.MethodGet, "/", nil), "", binding, "v1/data/tasks/task-2/step-transitions")
require.Equal(t, http.StatusNotFound, denied.Code)
require.Zero(t, source.transitionCalls)
allowed := httptest.NewRecorder()
svc.handleWebAppProtocol(allowed, httptest.NewRequest(http.MethodGet, "/", nil), "", binding, "v1/data/tasks/task-1/step-transitions")
require.Equal(t, http.StatusOK, allowed.Code, allowed.Body.String())
require.Equal(t, 1, source.transitionCalls)
require.Contains(t, allowed.Body.String(), `"id":"7"`)
require.NotContains(t, allowed.Body.String(), "actor_id")
require.NotContains(t, allowed.Body.String(), "session_id")
}
// @covers AC-PLUGINS-WORKFLOW-HISTORY-002.1
func TestWebAppTransitionGroupsFindWorkflowBeyondFirstListPage(t *testing.T) {
workflows := make([]*taskmodels.Workflow, 201)
for i := range workflows {
workflows[i] = &taskmodels.Workflow{ID: "wf-" + strconv.Itoa(i), WorkspaceID: "ws-1"}
}
source := &fakeTaskDataSource{transitionGroups: map[string][]taskmodels.TransitionGroup{"wf-200": {{Kind: "entry", Count: 1}}}}
svc := &Service{taskData: source, workflows: &fakeWorkflowLister{workflows: map[string][]*taskmodels.Workflow{"ws-1": workflows}}, workflowSteps: &fakeWorkflowStepLister{}}
binding := webapp.CapabilityBinding{ScopeKind: instances.ScopeWorkspace, WorkspaceID: "ws-1", Permissions: []string{"api_read:tasks", "api_read:workflows"}}
recorder := httptest.NewRecorder()
svc.handleWebAppProtocol(recorder, httptest.NewRequest(http.MethodGet, "/", nil), "", binding, "v1/data/workflows/wf-200/transition-groups")
require.Equal(t, http.StatusOK, recorder.Code, recorder.Body.String())
}
func stringPointer(value string) *string { return &value }
// @covers AC-PLUGINS-WORKFLOW-HISTORY-002.1 AC-PLUGINS-WORKFLOW-HISTORY-002.3
func TestWebAppTransitionGroupsRequireWorkspaceAndBothGrants(t *testing.T) {
source := &fakeTaskDataSource{transitionGroups: map[string][]taskmodels.TransitionGroup{
"wf-1": {{Kind: "within", FromStepID: stringPointer("draft"), ToStepID: stringPointer("review"), Count: 2}},
}}
svc := &Service{
taskData: source,
workflows: &fakeWorkflowLister{workflows: map[string][]*taskmodels.Workflow{"ws-1": {{ID: "wf-1", WorkspaceID: "ws-1"}}}},
workflowSteps: &fakeWorkflowStepLister{},
}
binding := webapp.CapabilityBinding{ScopeKind: instances.ScopeTask, WorkspaceID: "ws-1", TaskID: "task-1", Permissions: []string{"api_read:tasks", "api_read:workflows"}}
request := httptest.NewRequest(http.MethodGet, "/", nil)
call := func(workflowID string) *httptest.ResponseRecorder {
recorder := httptest.NewRecorder()
svc.handleWebAppProtocol(recorder, request, "", binding, "v1/data/workflows/"+workflowID+"/transition-groups")
return recorder
}
require.Equal(t, http.StatusForbidden, call("wf-1").Code)
binding.ScopeKind = instances.ScopeWorkspace
binding.Permissions = []string{"api_read:workflows"}
require.Equal(t, http.StatusForbidden, call("wf-1").Code)
binding.Permissions = []string{"api_read:tasks", "api_read:workflows"}
require.Equal(t, http.StatusNotFound, call("foreign-wf").Code)
require.Zero(t, source.groupCalls)
allowed := call("wf-1")
require.Equal(t, http.StatusOK, allowed.Code, allowed.Body.String())
require.Equal(t, 1, source.groupCalls)
require.Contains(t, allowed.Body.String(), `"count":2`)
}
@@ -0,0 +1,22 @@
package models
import "time"
// StepTransition is the public-safe recorded movement of one task.
type StepTransition struct {
ID int64
FromWorkflowID *string
FromWorkflowStepID *string
ToWorkflowID *string
ToWorkflowStepID *string
Trigger string
OccurredAt time.Time
}
// TransitionGroup is one route through, into, or out of a workflow.
type TransitionGroup struct {
Kind string
FromStepID *string
ToStepID *string
Count int64
}
@@ -1090,8 +1090,14 @@ func (r *Repository) initStepTransitionsSchema() error {
);
CREATE INDEX IF NOT EXISTS idx_task_step_transitions_task
ON task_step_transitions(task_id, occurred_at, id);
CREATE INDEX IF NOT EXISTS idx_task_step_transitions_task_id
ON task_step_transitions(task_id, id);
CREATE INDEX IF NOT EXISTS idx_task_step_transitions_occurred
ON task_step_transitions(occurred_at);
CREATE INDEX IF NOT EXISTS idx_task_step_transitions_from_workflow
ON task_step_transitions(from_workflow_id, task_id);
CREATE INDEX IF NOT EXISTS idx_task_step_transitions_to_workflow
ON task_step_transitions(to_workflow_id, task_id);
`)
if err != nil {
return fmt.Errorf("init step transitions schema: %w", err)
@@ -0,0 +1,102 @@
package sqlite
import (
"context"
"testing"
"time"
"github.com/kandev/kandev/internal/task/models"
)
// @covers AC-PLUGINS-WORKFLOW-HISTORY-001.1
func TestTaskTransitionReadReturnsGenesis(t *testing.T) {
repo := newStepTransitionsTestRepo(t)
createStepTransitionsTestTask(t, repo, "history-task", "workflow-1", "draft")
rows, err := repo.ListTaskStepTransitions(context.Background(), "history-task", 0, 10)
if err != nil {
t.Fatalf("list transition history: %v", err)
}
if len(rows) != 1 || rows[0].FromWorkflowID != nil || rows[0].ToWorkflowStepID == nil || *rows[0].ToWorkflowStepID != "draft" {
t.Fatalf("rows = %+v, want genesis row with a null source", rows)
}
}
// @covers AC-PLUGINS-WORKFLOW-HISTORY-001.1 AC-PLUGINS-WORKFLOW-HISTORY-001.2
func TestTaskTransitionReadPagesByIDAndPreservesRemovedEndpoints(t *testing.T) {
repo := newStepTransitionsTestRepo(t)
createStepTransitionsTestTask(t, repo, "history-pages", "workflow-1", "draft")
now := time.Now().UTC()
for _, step := range []string{"review", "removed-step", "done"} {
if _, err := repo.db.Exec(repo.db.Rebind(`INSERT INTO task_step_transitions
(task_id, from_workflow_id, from_workflow_step_id, to_workflow_id, to_workflow_step_id, trigger, actor_kind, contract_version, occurred_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`), "history-pages", "workflow-1", "draft", "workflow-1", step, "test", "system", 1, now); err != nil {
t.Fatal(err)
}
}
first, err := repo.ListTaskStepTransitions(context.Background(), "history-pages", 0, 2)
if err != nil || len(first) != 2 || first[0].ID <= first[1].ID {
t.Fatalf("first page = %+v, err %v", first, err)
}
second, err := repo.ListTaskStepTransitions(context.Background(), "history-pages", first[1].ID, 2)
if err != nil || len(second) != 2 || second[0].ID >= first[1].ID {
t.Fatalf("second page = %+v, err %v", second, err)
}
if second[0].ToWorkflowStepID == nil || *second[0].ToWorkflowStepID != "review" || second[1].FromWorkflowID != nil {
t.Fatalf("removed/nullable endpoints changed: %+v", second)
}
}
// @covers AC-PLUGINS-WORKFLOW-HISTORY-002.1 AC-PLUGINS-WORKFLOW-HISTORY-002.2
func TestWorkflowTransitionGroupsCountRetainedRoutes(t *testing.T) {
repo := newStepTransitionsTestRepo(t)
ctx := context.Background()
createStepTransitionsTestTask(t, repo, "route-task", "workflow-1", "draft")
if err := repo.ArchiveTask(ctx, "route-task"); err != nil {
t.Fatal(err)
}
for _, move := range [][4]any{
{"workflow-1", "draft", "workflow-1", "review"},
{"workflow-1", "review", "workflow-2", "other"},
{"workflow-2", "other", "workflow-1", "removed-step"},
} {
if _, err := repo.db.Exec(repo.db.Rebind(`INSERT INTO task_step_transitions
(task_id, from_workflow_id, from_workflow_step_id, to_workflow_id, to_workflow_step_id, trigger, actor_kind, contract_version, occurred_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`), "route-task", move[0], move[1], move[2], move[3], "test", "system", 1, time.Now().UTC()); err != nil {
t.Fatal(err)
}
}
if err := repo.CreateTask(ctx, &models.Task{ID: "foreign-route", WorkspaceID: "ws-2", WorkflowID: "workflow-1", WorkflowStepID: "draft", Title: "Foreign", Priority: "medium"}); err != nil {
t.Fatal(err)
}
groups, err := repo.ListWorkflowTransitionGroups(ctx, "ws-1", "workflow-1", "", 10)
if err != nil {
t.Fatal(err)
}
if len(groups) != 4 || groups[0].Kind != "entry" || groups[1].Kind != "entry" || groups[2].Kind != "exit" || groups[3].Kind != "within" {
t.Fatalf("groups = %+v, want two entries, exit, within", groups)
}
if groups[1].ToStepID == nil || *groups[1].ToStepID != "removed-step" {
t.Fatalf("removed step lost: %+v", groups[1])
}
page, err := repo.ListWorkflowTransitionGroups(ctx, "ws-1", "workflow-1", "entry||draft", 2)
if err != nil || len(page) != 2 || page[0].Kind != "entry" || page[1].Kind != "exit" {
t.Fatalf("next route page = %+v, err %v", page, err)
}
}
// @covers AC-PLUGINS-WORKFLOW-HISTORY-002.1
func TestWorkflowTransitionReadIndexesExistAfterSchemaReplay(t *testing.T) {
repo := newStepTransitionsTestRepo(t)
if err := repo.initStepTransitionsSchema(); err != nil {
t.Fatal(err)
}
for _, name := range []string{"idx_task_step_transitions_task_id", "idx_task_step_transitions_from_workflow", "idx_task_step_transitions_to_workflow"} {
var count int
if err := repo.db.Get(&count, `SELECT COUNT(*) FROM sqlite_master WHERE type = 'index' AND name = ?`, name); err != nil {
t.Fatal(err)
}
if count != 1 {
t.Fatalf("index %s missing after schema replay", name)
}
}
}
@@ -10,8 +10,87 @@ import (
"github.com/jmoiron/sqlx"
"github.com/kandev/kandev/internal/db/dialect"
"github.com/kandev/kandev/internal/steptelemetry"
"github.com/kandev/kandev/internal/task/models"
)
// ListTaskStepTransitions reads newest-first ledger rows using immutable IDs.
func (r *Repository) ListTaskStepTransitions(ctx context.Context, taskID string, beforeID int64, limit int) ([]models.StepTransition, error) {
query := `SELECT id, from_workflow_id, from_workflow_step_id, to_workflow_id, to_workflow_step_id, trigger, occurred_at
FROM task_step_transitions WHERE task_id = ?`
args := []any{taskID}
if beforeID > 0 {
query += ` AND id < ?`
args = append(args, beforeID)
}
query += ` ORDER BY id DESC LIMIT ?`
args = append(args, limit)
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(query), args...)
if err != nil {
return nil, fmt.Errorf("list task step transitions: %w", err)
}
defer func() { _ = rows.Close() }()
items := make([]models.StepTransition, 0)
for rows.Next() {
var item models.StepTransition
var fromWorkflow, fromStep, toWorkflow, toStep sql.NullString
if err := rows.Scan(&item.ID, &fromWorkflow, &fromStep, &toWorkflow, &toStep, &item.Trigger, &item.OccurredAt); err != nil {
return nil, fmt.Errorf("scan task step transition: %w", err)
}
item.FromWorkflowID = nullableValue(fromWorkflow)
item.FromWorkflowStepID = nullableValue(fromStep)
item.ToWorkflowID = nullableValue(toWorkflow)
item.ToWorkflowStepID = nullableValue(toStep)
items = append(items, item)
}
return items, rows.Err()
}
func nullableValue(value sql.NullString) *string {
if !value.Valid {
return nil
}
return &value.String
}
// ListWorkflowTransitionGroups counts retained routes involving one workflow.
// Rows remain present when a step is removed because the ledger stores IDs.
func (r *Repository) ListWorkflowTransitionGroups(ctx context.Context, workspaceID, workflowID, afterKey string, limit int) ([]models.TransitionGroup, error) {
const query = `WITH scoped AS (
SELECT CASE WHEN h.from_workflow_id = ? AND h.to_workflow_id = ? THEN 'within'
WHEN h.to_workflow_id = ? THEN 'entry' ELSE 'exit' END AS kind,
CASE WHEN h.from_workflow_id = ? THEN h.from_workflow_step_id END AS from_step_id,
CASE WHEN h.to_workflow_id = ? THEN h.to_workflow_step_id END AS to_step_id
FROM task_step_transitions h JOIN tasks t ON t.id = h.task_id
WHERE t.workspace_id = ? AND (h.from_workflow_id = ? OR h.to_workflow_id = ?)
), grouped AS (
SELECT kind, from_step_id, to_step_id, COUNT(*) AS route_count
FROM scoped GROUP BY kind, from_step_id, to_step_id
), ordered AS (
SELECT kind, from_step_id, to_step_id, route_count,
kind || '|' || COALESCE(from_step_id, '') || '|' || COALESCE(to_step_id, '') AS route_key
FROM grouped
)
SELECT kind, from_step_id, to_step_id, route_count FROM ordered
WHERE route_key > ? ORDER BY route_key LIMIT ?`
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(query), workflowID, workflowID, workflowID,
workflowID, workflowID, workspaceID, workflowID, workflowID, afterKey, limit)
if err != nil {
return nil, fmt.Errorf("list workflow transition groups: %w", err)
}
defer func() { _ = rows.Close() }()
items := make([]models.TransitionGroup, 0)
for rows.Next() {
var item models.TransitionGroup
var fromStep, toStep sql.NullString
if err := rows.Scan(&item.Kind, &fromStep, &toStep, &item.Count); err != nil {
return nil, fmt.Errorf("scan workflow transition group: %w", err)
}
item.FromStepID, item.ToStepID = nullableValue(fromStep), nullableValue(toStep)
items = append(items, item)
}
return items, rows.Err()
}
// stepTransitionTx is satisfied by both *sql.Tx and *sqlx.Tx, the two
// transaction types the mutation paths in this package use.
type stepTransitionTx interface {
@@ -5,6 +5,7 @@ package sqlite
// KANDEV_TEST_POSTGRES_DSN is set; CI runs these in postgres-boot.
import (
"context"
"testing"
"time"
@@ -42,4 +43,19 @@ func TestPostgresStepTransitionsSchemaCreatesTableAndIsReplaySafe(t *testing.T)
if err := repo.initStepTransitionsSchema(); err != nil {
t.Fatalf("replay initStepTransitionsSchema twice: %v", err)
}
for _, name := range []string{"idx_task_step_transitions_task_id", "idx_task_step_transitions_from_workflow", "idx_task_step_transitions_to_workflow"} {
var indexCount int
if err := db.Get(&indexCount, `SELECT COUNT(*) FROM pg_indexes WHERE schemaname = current_schema() AND indexname = $1`, name); err != nil || indexCount != 1 {
t.Fatalf("index %s: count=%d err=%v", name, indexCount, err)
}
}
if _, err := db.Exec(db.Rebind(`INSERT INTO task_step_transitions
(task_id, to_workflow_id, to_workflow_step_id, trigger, actor_kind, contract_version, occurred_at)
VALUES (?, ?, ?, ?, ?, ?, ?)`), "task-pre-ledger-pg", "wf-pg", "draft", "test", "system", 1, now); err != nil {
t.Fatal(err)
}
groups, err := repo.ListWorkflowTransitionGroups(context.Background(), "ws-pre-ledger-pg", "wf-pg", "", 10)
if err != nil || len(groups) != 1 || groups[0].Kind != "entry" {
t.Fatalf("postgres groups = %+v, err=%v", groups, err)
}
}
@@ -0,0 +1,133 @@
package service
import (
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"strings"
"github.com/kandev/kandev/internal/task/models"
)
var ErrInvalidTransitionCursor = errors.New("invalid transition cursor")
type stepTransitionReader interface {
ListTaskStepTransitions(context.Context, string, int64, int) ([]models.StepTransition, error)
ListWorkflowTransitionGroups(context.Context, string, string, string, int) ([]models.TransitionGroup, error)
}
type transitionCursor struct {
Scope string `json:"scope"`
ID int64 `json:"id,omitempty"`
Key string `json:"key,omitempty"`
}
func decodeTransitionCursor(encoded, scope string) (transitionCursor, error) {
if encoded == "" {
return transitionCursor{Scope: scope}, nil
}
if len(encoded) > 1024 {
return transitionCursor{}, ErrInvalidTransitionCursor
}
data, err := base64.RawURLEncoding.DecodeString(encoded)
if err != nil {
return transitionCursor{}, ErrInvalidTransitionCursor
}
var cursor transitionCursor
if json.Unmarshal(data, &cursor) != nil || cursor.Scope != scope {
return transitionCursor{}, ErrInvalidTransitionCursor
}
return cursor, nil
}
func encodeTransitionCursor(cursor transitionCursor) string {
data, _ := json.Marshal(cursor)
return base64.RawURLEncoding.EncodeToString(data)
}
func transitionReadLimit(limit int) (int, error) {
if limit < 0 {
return 0, fmt.Errorf("invalid transition limit")
}
if limit == 0 {
return 50, nil
}
if limit > 200 {
return 200, nil
}
return limit, nil
}
// ListTaskStepTransitions returns one bounded page of retained task movement.
func (s *Service) ListTaskStepTransitions(ctx context.Context, taskID string, limit int, encodedCursor string) ([]models.StepTransition, string, error) {
if err := s.authorizeTaskID(ctx, taskID); err != nil {
return nil, "", err
}
reader, ok := s.tasks.(stepTransitionReader)
if !ok {
return nil, "", fmt.Errorf("task transition reads unavailable")
}
pageSize, err := transitionReadLimit(limit)
if err != nil {
return nil, "", err
}
cursor, err := decodeTransitionCursor(encodedCursor, "task:"+taskID)
if err != nil || (encodedCursor != "" && (cursor.ID <= 0 || cursor.Key != "")) {
return nil, "", ErrInvalidTransitionCursor
}
items, err := reader.ListTaskStepTransitions(ctx, taskID, cursor.ID, pageSize+1)
if err != nil {
return nil, "", err
}
if len(items) <= pageSize {
return items, "", nil
}
items = items[:pageSize]
return items, encodeTransitionCursor(transitionCursor{Scope: "task:" + taskID, ID: items[len(items)-1].ID}), nil
}
// ListWorkflowTransitionGroups returns bounded route counts for a workflow.
func (s *Service) ListWorkflowTransitionGroups(ctx context.Context, workflowID string, limit int, encodedCursor string) ([]models.TransitionGroup, string, error) {
if err := s.authorizeWorkflowID(ctx, workflowID); err != nil {
return nil, "", err
}
workflow, err := s.workflows.GetWorkflow(ctx, workflowID)
if err != nil {
return nil, "", err
}
if workflow == nil {
return nil, "", fmt.Errorf("workflow not found")
}
reader, ok := s.tasks.(stepTransitionReader)
if !ok {
return nil, "", fmt.Errorf("workflow transition reads unavailable")
}
pageSize, err := transitionReadLimit(limit)
if err != nil {
return nil, "", err
}
cursor, err := decodeTransitionCursor(encodedCursor, "workflow:"+workflowID)
if err != nil || (encodedCursor != "" && (cursor.Key == "" || cursor.ID != 0)) {
return nil, "", ErrInvalidTransitionCursor
}
items, err := reader.ListWorkflowTransitionGroups(ctx, workflow.WorkspaceID, workflowID, cursor.Key, pageSize+1)
if err != nil {
return nil, "", err
}
if len(items) <= pageSize {
return items, "", nil
}
items = items[:pageSize]
last := items[len(items)-1]
key := strings.Join([]string{last.Kind, stepValue(last.FromStepID), stepValue(last.ToStepID)}, "|")
return items, encodeTransitionCursor(transitionCursor{Scope: "workflow:" + workflowID, Key: key}), nil
}
func stepValue(step *string) string {
if step == nil {
return ""
}
return *step
}
@@ -0,0 +1,67 @@
package service
import (
"context"
"testing"
"github.com/kandev/kandev/internal/task/models"
)
// @covers AC-PLUGINS-WORKFLOW-HISTORY-001.1 AC-PLUGINS-WORKFLOW-HISTORY-001.3
func TestTransitionReadServicePagesRecordedMoves(t *testing.T) {
svc, _, repo := createTestService(t)
ctx := context.Background()
if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "history-ws", Name: "History"}); err != nil {
t.Fatal(err)
}
if err := repo.CreateTask(ctx, &models.Task{ID: "history-task", WorkspaceID: "history-ws", WorkflowID: "history-wf", WorkflowStepID: "draft", Title: "History", Priority: "medium"}); err != nil {
t.Fatal(err)
}
items, next, err := svc.ListTaskStepTransitions(ctx, "history-task", 1, "")
if err != nil {
t.Fatal(err)
}
if len(items) != 1 || next != "" {
t.Fatalf("items = %+v, next = %q, want one genesis row", items, next)
}
}
// @covers AC-PLUGINS-WORKFLOW-HISTORY-001.1 AC-PLUGINS-WORKFLOW-HISTORY-001.3
func TestTransitionReadServiceRejectsForeignAndMalformedCursors(t *testing.T) {
svc, _, repo := createTestService(t)
ctx := context.Background()
if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "cursor-ws", Name: "Cursor"}); err != nil {
t.Fatal(err)
}
for _, id := range []string{"cursor-a", "cursor-b"} {
if err := repo.CreateTask(ctx, &models.Task{ID: id, WorkspaceID: "cursor-ws", WorkflowID: "wf", WorkflowStepID: "draft", Title: id, Priority: "medium"}); err != nil {
t.Fatal(err)
}
}
foreign := encodeTransitionCursor(transitionCursor{Scope: "task:cursor-a", ID: 1})
for _, cursor := range []string{foreign, "not-base64!"} {
items, next, err := svc.ListTaskStepTransitions(ctx, "cursor-b", 10, cursor)
if err != ErrInvalidTransitionCursor || len(items) != 0 || next != "" {
t.Fatalf("cursor %q: items=%+v next=%q err=%v", cursor, items, next, err)
}
}
}
// @covers AC-PLUGINS-WORKFLOW-HISTORY-002.1
func TestWorkflowTransitionGroupServiceReadsOwnedWorkflow(t *testing.T) {
svc, _, repo := createTestService(t)
ctx := context.Background()
if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "route-ws", Name: "Routes"}); err != nil {
t.Fatal(err)
}
if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "route-wf", WorkspaceID: "route-ws", Name: "Routes"}); err != nil {
t.Fatal(err)
}
if err := repo.CreateTask(ctx, &models.Task{ID: "route-task", WorkspaceID: "route-ws", WorkflowID: "route-wf", WorkflowStepID: "draft", Title: "Task", Priority: "medium"}); err != nil {
t.Fatal(err)
}
groups, next, err := svc.ListWorkflowTransitionGroups(ctx, "route-wf", 10, "")
if err != nil || len(groups) != 1 || groups[0].Kind != "entry" || next != "" {
t.Fatalf("groups=%+v next=%q err=%v", groups, next, err)
}
}
@@ -0,0 +1,143 @@
package pluginsdk
import (
"context"
pluginv1 "github.com/kandev/kandev/proto/kandev/plugin/v1"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
// TaskStepTransition is one retained task movement. Identity is a decimal
// string to preserve the ledger's int64 value in browser and SDK clients.
type TaskStepTransition struct {
ID string
FromWorkflowID *string
FromWorkflowStepID *string
ToWorkflowID *string
ToWorkflowStepID *string
Trigger string
OccurredAt string
}
// WorkflowTransitionGroup counts one route through or across a workflow.
type WorkflowTransitionGroup struct {
Kind string
FromStepID *string
ToStepID *string
Count int64
}
// TransitionHistoryReader is an optional Host extension so existing Host
// implementations remain source compatible.
type TransitionHistoryReader interface {
ListTask(ctx context.Context, taskID string, page Page) ([]TaskStepTransition, *PageInfo, error)
ListWorkflowGroups(ctx context.Context, workflowID string, page Page) ([]WorkflowTransitionGroup, *PageInfo, error)
}
type TransitionHistoryHost interface {
Transitions() TransitionHistoryReader
}
// TransitionHistory returns a Host's optional transition-history reader.
func TransitionHistory(host Host) (TransitionHistoryReader, bool) {
provider, ok := host.(TransitionHistoryHost)
if !ok {
return nil, false
}
return provider.Transitions(), true
}
func (h *grpcHostClient) Transitions() TransitionHistoryReader {
return grpcTransitionHistoryReader{client: h.client}
}
type grpcTransitionHistoryReader struct{ client pluginv1.HostClient }
func (r grpcTransitionHistoryReader) ListTask(ctx context.Context, taskID string, page Page) ([]TaskStepTransition, *PageInfo, error) {
response, err := r.client.ListTaskStepTransitions(ctx, &pluginv1.ListTaskStepTransitionsRequest{TaskId: taskID, Page: page.toProto()})
if err != nil {
return nil, nil, err
}
items := make([]TaskStepTransition, len(response.GetItems()))
for i, item := range response.GetItems() {
items[i] = taskStepTransitionFromProto(item)
}
return items, pageInfoFromProto(response.GetPageInfo()), nil
}
func (r grpcTransitionHistoryReader) ListWorkflowGroups(ctx context.Context, workflowID string, page Page) ([]WorkflowTransitionGroup, *PageInfo, error) {
response, err := r.client.ListWorkflowTransitionGroups(ctx, &pluginv1.ListWorkflowTransitionGroupsRequest{WorkflowId: workflowID, Page: page.toProto()})
if err != nil {
return nil, nil, err
}
items := make([]WorkflowTransitionGroup, len(response.GetItems()))
for i, item := range response.GetItems() {
items[i] = workflowTransitionGroupFromProto(item)
}
return items, pageInfoFromProto(response.GetPageInfo()), nil
}
func (s *grpcHostServer) ListTaskStepTransitions(ctx context.Context, request *pluginv1.ListTaskStepTransitionsRequest) (*pluginv1.ListTaskStepTransitionsResponse, error) {
reader, ok := TransitionHistory(s.impl)
if !ok {
return nil, status.Error(codes.Unimplemented, "transition history unavailable")
}
items, page, err := reader.ListTask(ctx, request.GetTaskId(), pageFromProto(request.GetPage()))
if err != nil {
return nil, err
}
converted := make([]*pluginv1.TaskStepTransition, len(items))
for i, item := range items {
converted[i] = item.toProto()
}
return &pluginv1.ListTaskStepTransitionsResponse{Items: converted, PageInfo: page.toProto()}, nil
}
func (s *grpcHostServer) ListWorkflowTransitionGroups(ctx context.Context, request *pluginv1.ListWorkflowTransitionGroupsRequest) (*pluginv1.ListWorkflowTransitionGroupsResponse, error) {
reader, ok := TransitionHistory(s.impl)
if !ok {
return nil, status.Error(codes.Unimplemented, "transition history unavailable")
}
items, page, err := reader.ListWorkflowGroups(ctx, request.GetWorkflowId(), pageFromProto(request.GetPage()))
if err != nil {
return nil, err
}
converted := make([]*pluginv1.WorkflowTransitionGroup, len(items))
for i, item := range items {
converted[i] = item.toProto()
}
return &pluginv1.ListWorkflowTransitionGroupsResponse{Items: converted, PageInfo: page.toProto()}, nil
}
func (item TaskStepTransition) toProto() *pluginv1.TaskStepTransition {
return &pluginv1.TaskStepTransition{
Id: item.ID, FromWorkflowId: item.FromWorkflowID,
FromWorkflowStepId: item.FromWorkflowStepID, ToWorkflowId: item.ToWorkflowID,
ToWorkflowStepId: item.ToWorkflowStepID, Trigger: item.Trigger,
OccurredAt: item.OccurredAt,
}
}
func taskStepTransitionFromProto(item *pluginv1.TaskStepTransition) TaskStepTransition {
if item == nil {
return TaskStepTransition{}
}
return TaskStepTransition{
ID: item.GetId(), FromWorkflowID: item.FromWorkflowId,
FromWorkflowStepID: item.FromWorkflowStepId, ToWorkflowID: item.ToWorkflowId,
ToWorkflowStepID: item.ToWorkflowStepId, Trigger: item.GetTrigger(),
OccurredAt: item.GetOccurredAt(),
}
}
func (item WorkflowTransitionGroup) toProto() *pluginv1.WorkflowTransitionGroup {
return &pluginv1.WorkflowTransitionGroup{Kind: item.Kind, FromStepId: item.FromStepID, ToStepId: item.ToStepID, Count: item.Count}
}
func workflowTransitionGroupFromProto(item *pluginv1.WorkflowTransitionGroup) WorkflowTransitionGroup {
if item == nil {
return WorkflowTransitionGroup{}
}
return WorkflowTransitionGroup{Kind: item.GetKind(), FromStepID: item.FromStepId, ToStepID: item.ToStepId, Count: item.GetCount()}
}
@@ -0,0 +1,60 @@
package pluginsdk
import (
"context"
"reflect"
"testing"
"github.com/stretchr/testify/require"
)
// @covers AC-PLUGINS-WORKFLOW-HISTORY-001.4 AC-PLUGINS-WORKFLOW-HISTORY-002.1
func TestHostClientExposesOptionalTransitionHistory(t *testing.T) {
if _, ok := reflect.TypeOf(&grpcHostClient{}).MethodByName("Transitions"); !ok {
t.Fatal("Host client has no transition-history extension")
}
}
type recordingTransitionHost struct {
*dataRecordingHost
reader *recordingTransitionReader
}
func (h *recordingTransitionHost) Transitions() TransitionHistoryReader { return h.reader }
type recordingTransitionReader struct {
lastTaskID string
lastPage Page
lastWFID string
}
func (r *recordingTransitionReader) ListTask(_ context.Context, taskID string, page Page) ([]TaskStepTransition, *PageInfo, error) {
r.lastTaskID, r.lastPage = taskID, page
return []TaskStepTransition{{ID: "9223372036854775806", ToWorkflowStepID: strPtr("review"), Trigger: "test", OccurredAt: "2026-09-22T12:00:00Z"}}, &PageInfo{NextCursor: "next", HasMore: true}, nil
}
func (r *recordingTransitionReader) ListWorkflowGroups(_ context.Context, workflowID string, page Page) ([]WorkflowTransitionGroup, *PageInfo, error) {
r.lastWFID, r.lastPage = workflowID, page
return []WorkflowTransitionGroup{{Kind: "entry", ToStepID: strPtr("review"), Count: 3}}, &PageInfo{}, nil
}
// @covers AC-PLUGINS-WORKFLOW-HISTORY-001.1 AC-PLUGINS-WORKFLOW-HISTORY-002.1
func TestTransitionHistoryRoundTripsAcrossHostRPC(t *testing.T) {
impl := &recordingTransitionHost{dataRecordingHost: &dataRecordingHost{}, reader: &recordingTransitionReader{}}
host := dialHostOverBufconn(t, impl)
reader, ok := TransitionHistory(host)
require.True(t, ok)
items, page, err := reader.ListTask(context.Background(), "task-1", Page{Limit: 2, Cursor: "before"})
require.NoError(t, err)
require.Equal(t, "task-1", impl.reader.lastTaskID)
require.Equal(t, Page{Limit: 2, Cursor: "before"}, impl.reader.lastPage)
require.Equal(t, "9223372036854775806", items[0].ID)
require.Nil(t, items[0].FromWorkflowStepID)
require.Equal(t, "review", *items[0].ToWorkflowStepID)
require.Equal(t, &PageInfo{NextCursor: "next", HasMore: true}, page)
groups, _, err := reader.ListWorkflowGroups(context.Background(), "wf-1", Page{Limit: 5})
require.NoError(t, err)
require.Equal(t, "wf-1", impl.reader.lastWFID)
require.Equal(t, int64(3), groups[0].Count)
require.Nil(t, groups[0].FromStepID)
}
+1
View File
@@ -103,6 +103,7 @@ const (
// Plugin-backed canvas lifecycle notifications. These are owner-scoped
// workspace/task events and contain no application content or capabilities.
ActionCanvasCreated = "canvas.created"
ActionCanvasUpdated = "canvas.updated"
ActionCanvasReleaseActivated = "canvas.release.activated"
ActionCanvasReleasePermissionRequired = "canvas.release.permission_required"
ActionCanvasPromoted = "canvas.promoted"
File diff suppressed because it is too large Load Diff
@@ -82,9 +82,11 @@ service Host {
// Reads — capability api_read:<resource>
rpc ListTasks(ListTasksRequest) returns (ListTasksResponse);
rpc GetTask(GetTaskRequest) returns (GetTaskResponse);
rpc ListTaskStepTransitions(ListTaskStepTransitionsRequest) returns (ListTaskStepTransitionsResponse);
rpc ListWorkspaces(ListWorkspacesRequest) returns (ListWorkspacesResponse);
rpc ListWorkflows(ListWorkflowsRequest) returns (ListWorkflowsResponse);
rpc ListWorkflowSteps(ListWorkflowStepsRequest) returns (ListWorkflowStepsResponse);
rpc ListWorkflowTransitionGroups(ListWorkflowTransitionGroupsRequest) returns (ListWorkflowTransitionGroupsResponse);
rpc ListAgentProfiles(ListAgentProfilesRequest) returns (ListAgentProfilesResponse);
rpc ListExecutorProfiles(ListExecutorProfilesRequest) returns (ListExecutorProfilesResponse);
rpc ListRepositories(ListRepositoriesRequest) returns (ListRepositoriesResponse);
@@ -517,6 +519,23 @@ message GetTaskRequest {
message GetTaskResponse {
Task task = 1;
}
message TaskStepTransition {
string id = 1; // decimal ledger identity; stable across timestamp ties
optional string from_workflow_id = 2;
optional string from_workflow_step_id = 3;
optional string to_workflow_id = 4;
optional string to_workflow_step_id = 5;
string trigger = 6;
string occurred_at = 7; // RFC3339
}
message ListTaskStepTransitionsRequest {
string task_id = 1;
Page page = 2;
}
message ListTaskStepTransitionsResponse {
repeated TaskStepTransition items = 1;
PageInfo page_info = 2;
}
// ── Workspace / Workflow / Steps ─────────────────────────────────────────────
message Workspace {
@@ -554,6 +573,20 @@ message ListWorkflowsResponse {
repeated Workflow workflows = 1;
PageInfo page_info = 2;
}
message WorkflowTransitionGroup {
string kind = 1; // within | entry | exit
optional string from_step_id = 2;
optional string to_step_id = 3;
int64 count = 4;
}
message ListWorkflowTransitionGroupsRequest {
string workflow_id = 1;
Page page = 2;
}
message ListWorkflowTransitionGroupsResponse {
repeated WorkflowTransitionGroup items = 1;
PageInfo page_info = 2;
}
message WorkflowStep {
string id = 1;
@@ -503,9 +503,11 @@ const (
Host_GetConfig_FullMethodName = "/kandev.plugin.v1.Host/GetConfig"
Host_ListTasks_FullMethodName = "/kandev.plugin.v1.Host/ListTasks"
Host_GetTask_FullMethodName = "/kandev.plugin.v1.Host/GetTask"
Host_ListTaskStepTransitions_FullMethodName = "/kandev.plugin.v1.Host/ListTaskStepTransitions"
Host_ListWorkspaces_FullMethodName = "/kandev.plugin.v1.Host/ListWorkspaces"
Host_ListWorkflows_FullMethodName = "/kandev.plugin.v1.Host/ListWorkflows"
Host_ListWorkflowSteps_FullMethodName = "/kandev.plugin.v1.Host/ListWorkflowSteps"
Host_ListWorkflowTransitionGroups_FullMethodName = "/kandev.plugin.v1.Host/ListWorkflowTransitionGroups"
Host_ListAgentProfiles_FullMethodName = "/kandev.plugin.v1.Host/ListAgentProfiles"
Host_ListExecutorProfiles_FullMethodName = "/kandev.plugin.v1.Host/ListExecutorProfiles"
Host_ListRepositories_FullMethodName = "/kandev.plugin.v1.Host/ListRepositories"
@@ -581,9 +583,11 @@ type HostClient interface {
// Reads — capability api_read:<resource>
ListTasks(ctx context.Context, in *ListTasksRequest, opts ...grpc.CallOption) (*ListTasksResponse, error)
GetTask(ctx context.Context, in *GetTaskRequest, opts ...grpc.CallOption) (*GetTaskResponse, error)
ListTaskStepTransitions(ctx context.Context, in *ListTaskStepTransitionsRequest, opts ...grpc.CallOption) (*ListTaskStepTransitionsResponse, error)
ListWorkspaces(ctx context.Context, in *ListWorkspacesRequest, opts ...grpc.CallOption) (*ListWorkspacesResponse, error)
ListWorkflows(ctx context.Context, in *ListWorkflowsRequest, opts ...grpc.CallOption) (*ListWorkflowsResponse, error)
ListWorkflowSteps(ctx context.Context, in *ListWorkflowStepsRequest, opts ...grpc.CallOption) (*ListWorkflowStepsResponse, error)
ListWorkflowTransitionGroups(ctx context.Context, in *ListWorkflowTransitionGroupsRequest, opts ...grpc.CallOption) (*ListWorkflowTransitionGroupsResponse, error)
ListAgentProfiles(ctx context.Context, in *ListAgentProfilesRequest, opts ...grpc.CallOption) (*ListAgentProfilesResponse, error)
ListExecutorProfiles(ctx context.Context, in *ListExecutorProfilesRequest, opts ...grpc.CallOption) (*ListExecutorProfilesResponse, error)
ListRepositories(ctx context.Context, in *ListRepositoriesRequest, opts ...grpc.CallOption) (*ListRepositoriesResponse, error)
@@ -788,6 +792,16 @@ func (c *hostClient) GetTask(ctx context.Context, in *GetTaskRequest, opts ...gr
return out, nil
}
func (c *hostClient) ListTaskStepTransitions(ctx context.Context, in *ListTaskStepTransitionsRequest, opts ...grpc.CallOption) (*ListTaskStepTransitionsResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(ListTaskStepTransitionsResponse)
err := c.cc.Invoke(ctx, Host_ListTaskStepTransitions_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *hostClient) ListWorkspaces(ctx context.Context, in *ListWorkspacesRequest, opts ...grpc.CallOption) (*ListWorkspacesResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(ListWorkspacesResponse)
@@ -818,6 +832,16 @@ func (c *hostClient) ListWorkflowSteps(ctx context.Context, in *ListWorkflowStep
return out, nil
}
func (c *hostClient) ListWorkflowTransitionGroups(ctx context.Context, in *ListWorkflowTransitionGroupsRequest, opts ...grpc.CallOption) (*ListWorkflowTransitionGroupsResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(ListWorkflowTransitionGroupsResponse)
err := c.cc.Invoke(ctx, Host_ListWorkflowTransitionGroups_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *hostClient) ListAgentProfiles(ctx context.Context, in *ListAgentProfilesRequest, opts ...grpc.CallOption) (*ListAgentProfilesResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(ListAgentProfilesResponse)
@@ -1089,9 +1113,11 @@ type HostServer interface {
// Reads — capability api_read:<resource>
ListTasks(context.Context, *ListTasksRequest) (*ListTasksResponse, error)
GetTask(context.Context, *GetTaskRequest) (*GetTaskResponse, error)
ListTaskStepTransitions(context.Context, *ListTaskStepTransitionsRequest) (*ListTaskStepTransitionsResponse, error)
ListWorkspaces(context.Context, *ListWorkspacesRequest) (*ListWorkspacesResponse, error)
ListWorkflows(context.Context, *ListWorkflowsRequest) (*ListWorkflowsResponse, error)
ListWorkflowSteps(context.Context, *ListWorkflowStepsRequest) (*ListWorkflowStepsResponse, error)
ListWorkflowTransitionGroups(context.Context, *ListWorkflowTransitionGroupsRequest) (*ListWorkflowTransitionGroupsResponse, error)
ListAgentProfiles(context.Context, *ListAgentProfilesRequest) (*ListAgentProfilesResponse, error)
ListExecutorProfiles(context.Context, *ListExecutorProfilesRequest) (*ListExecutorProfilesResponse, error)
ListRepositories(context.Context, *ListRepositoriesRequest) (*ListRepositoriesResponse, error)
@@ -1212,6 +1238,9 @@ func (UnimplementedHostServer) ListTasks(context.Context, *ListTasksRequest) (*L
func (UnimplementedHostServer) GetTask(context.Context, *GetTaskRequest) (*GetTaskResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method GetTask not implemented")
}
func (UnimplementedHostServer) ListTaskStepTransitions(context.Context, *ListTaskStepTransitionsRequest) (*ListTaskStepTransitionsResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method ListTaskStepTransitions not implemented")
}
func (UnimplementedHostServer) ListWorkspaces(context.Context, *ListWorkspacesRequest) (*ListWorkspacesResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method ListWorkspaces not implemented")
}
@@ -1221,6 +1250,9 @@ func (UnimplementedHostServer) ListWorkflows(context.Context, *ListWorkflowsRequ
func (UnimplementedHostServer) ListWorkflowSteps(context.Context, *ListWorkflowStepsRequest) (*ListWorkflowStepsResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method ListWorkflowSteps not implemented")
}
func (UnimplementedHostServer) ListWorkflowTransitionGroups(context.Context, *ListWorkflowTransitionGroupsRequest) (*ListWorkflowTransitionGroupsResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method ListWorkflowTransitionGroups not implemented")
}
func (UnimplementedHostServer) ListAgentProfiles(context.Context, *ListAgentProfilesRequest) (*ListAgentProfilesResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method ListAgentProfiles not implemented")
}
@@ -1524,6 +1556,24 @@ func _Host_GetTask_Handler(srv interface{}, ctx context.Context, dec func(interf
return interceptor(ctx, in, info, handler)
}
func _Host_ListTaskStepTransitions_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(ListTaskStepTransitionsRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(HostServer).ListTaskStepTransitions(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: Host_ListTaskStepTransitions_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(HostServer).ListTaskStepTransitions(ctx, req.(*ListTaskStepTransitionsRequest))
}
return interceptor(ctx, in, info, handler)
}
func _Host_ListWorkspaces_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(ListWorkspacesRequest)
if err := dec(in); err != nil {
@@ -1578,6 +1628,24 @@ func _Host_ListWorkflowSteps_Handler(srv interface{}, ctx context.Context, dec f
return interceptor(ctx, in, info, handler)
}
func _Host_ListWorkflowTransitionGroups_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(ListWorkflowTransitionGroupsRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(HostServer).ListWorkflowTransitionGroups(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: Host_ListWorkflowTransitionGroups_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(HostServer).ListWorkflowTransitionGroups(ctx, req.(*ListWorkflowTransitionGroupsRequest))
}
return interceptor(ctx, in, info, handler)
}
func _Host_ListAgentProfiles_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(ListAgentProfilesRequest)
if err := dec(in); err != nil {
@@ -2029,6 +2097,10 @@ var Host_ServiceDesc = grpc.ServiceDesc{
MethodName: "GetTask",
Handler: _Host_GetTask_Handler,
},
{
MethodName: "ListTaskStepTransitions",
Handler: _Host_ListTaskStepTransitions_Handler,
},
{
MethodName: "ListWorkspaces",
Handler: _Host_ListWorkspaces_Handler,
@@ -2041,6 +2113,10 @@ var Host_ServiceDesc = grpc.ServiceDesc{
MethodName: "ListWorkflowSteps",
Handler: _Host_ListWorkflowSteps_Handler,
},
{
MethodName: "ListWorkflowTransitionGroups",
Handler: _Host_ListWorkflowTransitionGroups_Handler,
},
{
MethodName: "ListAgentProfiles",
Handler: _Host_ListAgentProfiles_Handler,