[Neki] Implement kill for live connections

internal/cmd/branch/connections.go

@@ -11,7 +11,7 @@ func ConnectionsCmd(ch *cmdutil.Helper) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "connections <command>",
 		Short: "Show and kill branch connections",
-		Long: `Show and kill MySQL or Postgres branch connections, and show Neki branch connections.
+		Long: `Show and kill MySQL, Postgres, or Neki branch connections.
 
 MySQL (Vitess):
   show/kill/top with --keyspace and --shard.
@@ -21,7 +21,8 @@ Postgres:
   kill-transaction terminates a transaction_id from show.
 
 Neki:
-  show/top list client sessions across the branch's routers. Read-only.
+  show/top list client sessions across the branch's routers.
+  kill-transaction terminates a transaction_id from show.
 
 Agent workflow:
   1. Run: pscale branch connections show <database> <branch> --format json
@@ -33,7 +34,7 @@ Agent workflow:
 Action semantics:
   kill <database> <branch> <query-id> --query        Cancels the listed query_id.
   kill-transaction <database> <branch> <transaction-id>
-                                                     Postgres only. destructive. Terminates the listed transaction_id if it still matches server state.
+                                                     Postgres and Neki only. destructive. Terminates the listed transaction_id if it still matches server state.
   kill <database> <branch> <connection-id>           destructive. Terminates the listed connection_id.
 
 Use --format json when an agent or script needs to inspect query_id,

internal/cmd/branch/connections/actions.go

@@ -137,7 +137,7 @@ func printActionResult(ch *cmdutil.Helper, result live.ActionResult, engine ps.D
 	if ch.Printer.Format() == printer.JSON {
 		return ch.Printer.PrintResource(toActionResult(result))
 	}
-	if engine == ps.DatabaseEnginePostgres {
+	if engine == ps.DatabaseEnginePostgres || engine == ps.DatabaseEngineNeki {
 		return ch.Printer.PrintResource(toCompactActionResult(result))
 	}
 	return ch.Printer.PrintResource(toActionResult(result))

internal/cmd/branch/connections/engine_flags.go

@@ -9,7 +9,6 @@ import (
 var (
 	errVitessTargetOnly   = errors.New("--keyspace/--shard are only supported for Vitess databases")
 	errPostgresFilterOnly = errors.New("--instance/--role are only supported for Postgres databases")
-	errNekiReadOnly       = errors.New("connections are read-only for Neki databases")
 )
 
 // ValidateEngineFlags rejects flags that only apply to another database engine.
@@ -38,11 +37,3 @@ func validateEngineFlags(engine ps.DatabaseEngine, filter connectionFilter, targ
 	}
 	return nil
 }
-
-// ValidateActionEngine rejects kill commands for engines whose connections are read-only.
-func ValidateActionEngine(engine ps.DatabaseEngine) error {
-	if engine == ps.DatabaseEngineNeki {
-		return errNekiReadOnly
-	}
-	return nil
-}

internal/cmd/branch/connections/engine_flags_test.go

@@ -34,11 +34,3 @@ func TestValidateEngineFlagsForNeki(t *testing.T) {
 		})
 	}
 }
-
-func TestValidateActionEngine(t *testing.T) {
-	c := qt.New(t)
-
-	c.Assert(ValidateActionEngine(ps.DatabaseEngineNeki), qt.ErrorMatches, "connections are read-only for Neki databases")
-	c.Assert(ValidateActionEngine(ps.DatabaseEnginePostgres), qt.IsNil)
-	c.Assert(ValidateActionEngine(ps.DatabaseEngineMySQL), qt.IsNil)
-}

internal/cmd/branch/connections/top.go

@@ -36,8 +36,8 @@ previously captured trace in the TUI — actions are rejected in replay mode.
 
 For Postgres, connections top shows session activity across instances. For
 Vitess, pass --keyspace and --shard or run interactively to select them when
-the server reports available targets. For Neki, connections top shows read-only
-client sessions across the branch's routers.`,
+the server reports available targets. For Neki, connections top shows client
+sessions across the branch's routers.`,
 		Args: func(cmd *cobra.Command, args []string) error {
 			if flags.replay != "" {
 				return nil

internal/cmd/branch/connections/top_test.go

@@ -618,7 +618,7 @@ func TestTopCmdRunEHeadlessVitessWritesCapture(t *testing.T) {
 	})
 }
 
-func TestTopCmdRunsReadOnlyNekiTop(t *testing.T) {
+func TestTopCmdRunsNekiTop(t *testing.T) {
 	c := qt.New(t)
 	restoreTTY := setPrinterTTY(t, true)
 	defer restoreTTY()
@@ -654,8 +654,8 @@ func TestTopCmdRunsReadOnlyNekiTop(t *testing.T) {
 	c.Assert(view, qt.Contains, "router-a")
 	c.Assert(view, qt.Contains, "incomplete results: router router-b unreachable")
 	c.Assert(view, qt.Not(qt.Contains), "BLOCK")
-	c.Assert(view, qt.Not(qt.Contains), "cancel query")
-	c.Assert(view, qt.Not(qt.Contains), "force terminate")
+	c.Assert(view, qt.Contains, "cancel query")
+	c.Assert(view, qt.Contains, "force terminate")
 	c.Assert(blockersView, qt.Not(qt.Contains), "[blockers]")
 	c.Assert(cancelConfirming, qt.IsFalse)
 }

internal/cmd/branch/connections_commands.go

@@ -84,9 +84,6 @@ the current query.`,
 			if err != nil {
 				return err
 			}
-			if err := connections.ValidateActionEngine(engine); err != nil {
-				return err
-			}
 			target := connections.ConnectionTarget{Keyspace: flags.keyspace, Shard: flags.shard}
 			if err := connections.ValidateEngineFlags(engine, connections.ConnectionFilter{}, target); err != nil {
 				return err
@@ -108,11 +105,11 @@ the current query.`,
 func ConnectionsKillTransactionCmd(ch *cmdutil.Helper) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "kill-transaction <database> <branch> <transaction-id>",
-		Short: "Kill a Postgres branch transaction",
-		Long: `Kill a Postgres branch transaction.
+		Short: "Kill a Postgres or Neki branch transaction",
+		Long: `Kill a Postgres or Neki branch transaction.
 
 This is destructive. Pass a transaction_id from connections show to terminate
-the matching Postgres connection.`,
+the matching connection.`,
 		Args: cmdutil.RequiredArgs("database", "branch", "transaction-id"),
 		RunE: func(cmd *cobra.Command, args []string) error {
 			if err := connections.ValidateTransactionID(args[2]); err != nil {
@@ -122,11 +119,8 @@ the matching Postgres connection.`,
 			if err != nil {
 				return err
 			}
-			if err := connections.ValidateActionEngine(engine); err != nil {
-				return err
-			}
-			if engine != ps.DatabaseEnginePostgres {
-				return errors.New("connections kill-transaction is only supported for Postgres databases")
+			if engine != ps.DatabaseEnginePostgres && engine != ps.DatabaseEngineNeki {
+				return errors.New("connections kill-transaction is only supported for Postgres and Neki databases")
 			}
 			return connections.RunKillTransactionForEngine(cmd.Context(), ch, args[0], args[1], args[2], engine, connections.ConnectionTarget{})
 		},

internal/cmd/branch/connections_test.go

@@ -43,7 +43,6 @@ func TestConnectionsHelpListsEngines(t *testing.T) {
 	c.Assert(help, qt.Contains, "MySQL (Vitess):")
 	c.Assert(help, qt.Contains, "Postgres:")
 	c.Assert(help, qt.Contains, "Neki:")
-	c.Assert(help, qt.Contains, "Read-only.")
 }
 
 func TestConnectionsShowHelpListsTargetAndFilterFlags(t *testing.T) {
@@ -541,31 +540,34 @@ func TestConnectionsKillTransactionRejectsVitess(t *testing.T) {
 
 	err := cmd.Execute()
 
-	c.Assert(err, qt.ErrorMatches, "connections kill-transaction is only supported for Postgres databases")
+	c.Assert(err, qt.ErrorMatches, "connections kill-transaction is only supported for Postgres and Neki databases")
 }
 
-func TestConnectionsRejectsNekiActionsBeforeAPI(t *testing.T) {
+func TestConnectionsSendsNekiActions(t *testing.T) {
 	tests := []struct {
-		name string
-		args []string
+		name     string
+		args     []string
+		wantPath string
 	}{
-		{name: "kill", args: []string{"kill", "neki", "main", "primary-123-c"}},
-		{name: "kill query", args: []string{"kill", "neki", "main", "primary-123-q", "--query"}},
-		{name: "kill-transaction", args: []string{"kill-transaction", "neki", "main", "primary-123-t"}},
+		{name: "kill", args: []string{"kill", "neki", "main", "primary-123-c"}, wantPath: "/connections/connection/primary-123-c"},
+		{name: "kill query", args: []string{"kill", "neki", "main", "primary-123-q", "--query"}, wantPath: "/connections/query/primary-123-q"},
+		{name: "kill-transaction", args: []string{"kill-transaction", "neki", "main", "primary-123-t"}, wantPath: "/connections/transaction/primary-123-t"},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			c := qt.New(t)
 			server := liveConnectionsBranchServer(t, func(w http.ResponseWriter, r *http.Request) {
-				c.Fatalf("%s should reject Neki before calling %s", tt.name, r.URL.Path)
+				c.Assert(r.Method, qt.Equals, http.MethodDelete)
+				c.Assert(r.URL.Path, qt.Equals, "/v1/organizations/acme/databases/neki/branches/main"+tt.wantPath)
+				_, _ = io.WriteString(w, `{"success":true}`)
 			})
 			cmd := connectionsCmdForTest(connectionsTestHelper("acme", ps.DatabaseEngineNeki, nil, server.URL, printer.JSON, &bytes.Buffer{}))
 			cmd.SetArgs(tt.args)
 
 			err := cmd.Execute()
 
-			c.Assert(err, qt.ErrorMatches, "connections are read-only for Neki databases")
+			c.Assert(err, qt.IsNil)
 		})
 	}
 }
@@ -696,7 +698,7 @@ func TestConnectionsShowNekiHuman(t *testing.T) {
 	c.Assert(got, qt.Contains, "shard_backends:  1\nshard:           -80\nshard_uid:       shard-1\nsidecar:         sidecar-1\nbackend_pid:     4242\n")
 	c.Assert(got, qt.Contains, "backend_query:   SELECT 1\n")
 	c.Assert(got, qt.Contains, "shard_backends:  0\n")
-	c.Assert(got, qt.Not(qt.Contains), "connection_id:")
+	c.Assert(got, qt.Contains, "connection_id:   aws_useast1b_3:router-b-7-1779364000000000\n")
 	c.Assert(strings.Count(got, "database:"), qt.Equals, 2)
 }
 
@@ -958,7 +960,7 @@ func sampleBranchConnectionsListResponse() string {
 func sampleNekiConnectionsListResponse() string {
 	return `{"type":"list","database_kind":"neki","captured_at":"2026-10-16T12:00:05Z","instances":[],"warnings":["router router-b unreachable"],"data":[` +
 		`{"pid":7,"instance":"router-a","router_name":"router-a","router_cell":"us-east-1a","router_uid":"router-a-us-east-1a-1","availability_zone":"us-east-1a","state":"active","usename":"app","datname":"postgres","application_name":"worker","client_addr":null,"duration_ms":5000,"query_text":"SELECT 1","xact_start":"2026-10-16T12:00:00Z","backend_start":"2026-10-16T12:00:00.123456Z","sidecar_backends":[{"shard_uid":"shard-1","shard_name":"-80","sidecar_uid":"sidecar-1","backend_pid":4242,"query_text":"SELECT\n  1"}]},` +
-		`{"pid":7,"instance":"router-b","router_name":"router-b","router_cell":"us-east-1b","router_uid":"router-b-us-east-1b-1","availability_zone":"us-east-1b","state":"idle","usename":"app","datname":"postgres","application_name":"","duration_ms":null,"query_text":"SELECT 2","xact_start":"2026-10-16T12:00:01Z","backend_start":"2026-10-16T11:59:00Z","sidecar_backends":[]}` +
+		`{"pid":7,"instance":"router-b","router_name":"router-b","connection_id":"aws_useast1b_3:router-b-7-1779364000000000","router_cell":"us-east-1b","router_uid":"router-b-us-east-1b-1","availability_zone":"us-east-1b","state":"idle","usename":"app","datname":"postgres","application_name":"","duration_ms":null,"query_text":"SELECT 2","xact_start":"2026-10-16T12:00:01Z","backend_start":"2026-10-16T11:59:00Z","sidecar_backends":[]}` +
 		`]}`
 }
 

internal/connections/connection_list.go

@@ -284,6 +284,9 @@ func (c Connection) NekiHumanFields() [][2]string {
 		{"application", c.ApplicationName},
 		{"client_addr", c.ClientAddr},
 		{"backend_start", formatOptionalTime(c.BackendStart)},
+		{"query_id", DerefString(c.QueryID)},
+		{"transaction_id", DerefString(c.TransactionID)},
+		{"connection_id", DerefString(c.ConnectionID)},
 		{"shard_backends", strconv.Itoa(len(c.SidecarBackends))},
 	}
 	for _, backend := range c.SidecarBackends {

internal/connections/tui/connection_capabilities.go

@@ -77,10 +77,6 @@ func (s ConnectionCapabilities) supports(kind actionKind) bool {
 	return requirement.field != actionTargetUnset
 }
 
-func (s ConnectionCapabilities) hasActions() bool {
-	return s.supports(actionCancelQuery) || s.supports(actionTerminateTxn) || s.supports(actionTerminateConn)
-}
-
 func (s ConnectionCapabilities) requirement(kind actionKind) ActionRequirement {
 	s = s.effective()
 	switch kind {

internal/connections/tui/help.go

@@ -68,22 +68,15 @@ func helpModalLines(state helpState) []string {
 			"",
 		)
 	}
-	switch {
-	case !support.hasActions():
-		lines = append(lines,
-			headerStyle.Render("Actions"),
-			"  Read-only: queries and connections cannot be cancelled or terminated from this view.",
-			"",
-		)
-	case !support.supports(actionTerminateTxn):
+	if !support.supports(actionTerminateTxn) {
 		lines = append(lines,
 			headerStyle.Render("Actions"),
 			"  c  Kill the selected query (KILL QUERY) using the query_id from the selected process.",
 			"  K  Kill the selected connection (KILL) using the connection_id from the selected process.",
 			"  c and K require confirmation. Replay mode blocks backend actions.",
 			"",
 		)
-	default:
+	} else {
 		lines = append(lines,
 			headerStyle.Render("Actions"),
 			"  c  Cancel the selected query (pg_cancel_backend) only if it is the same active query we observed.",

internal/connections/tui/model.go

@@ -86,12 +86,17 @@ var (
 		defaultSort: live.SortByDuration,
 		sortModes:   []live.SortMode{live.SortByDuration},
 	}
-	// NekiConnectionView presents read-only Neki router sessions.
+	// NekiConnectionView presents Neki router sessions.
 	NekiConnectionView = ConnectionViewProfile{
 		displayPreset: connectionDisplayNeki,
-		capabilities:  ConnectionCapabilities{configured: true},
-		defaultSort:   live.SortByTransactionStart,
-		sortModes:     []live.SortMode{live.SortByTransactionStart, live.SortByDuration},
+		capabilities: ConnectionCapabilities{
+			CancelQuery:          ActionTargetQueryID,
+			TerminateTransaction: ActionTargetTransactionID,
+			TerminateConnection:  ActionTargetConnectionID,
+			configured:           true,
+		},
+		defaultSort: live.SortByTransactionStart,
+		sortModes:   []live.SortMode{live.SortByTransactionStart, live.SortByDuration},
 	}
 )
 

internal/connections/tui/model_test.go

@@ -1937,7 +1937,7 @@ func TestModelPreListPausedFooterHidesRefresh(t *testing.T) {
 	c.Assert(view, qt.Not(qt.Contains), "r refresh")
 }
 
-func TestModelNekiConnectionViewIsReadOnly(t *testing.T) {
+func TestModelNekiConnectionView(t *testing.T) {
 	c := qt.New(t)
 	client := &clientStub{}
 	model := nekiModel(client, nil, nekiConnection("router-a", 7, time.Unix(50, 0)))
@@ -1946,9 +1946,9 @@ func TestModelNekiConnectionViewIsReadOnly(t *testing.T) {
 	c.Assert(tableView, qt.Contains, "ROUTER")
 	c.Assert(tableView, qt.Contains, "router-a")
 	c.Assert(tableView, qt.Not(qt.Contains), "BLOCK")
-	c.Assert(tableView, qt.Not(qt.Contains), "cancel query")
-	c.Assert(tableView, qt.Not(qt.Contains), "kill transaction")
-	c.Assert(tableView, qt.Not(qt.Contains), "force terminate")
+	c.Assert(tableView, qt.Contains, "cancel query")
+	c.Assert(tableView, qt.Contains, "kill transaction")
+	c.Assert(tableView, qt.Contains, "force terminate")
 
 	for _, key := range []string{"c", "k", "K"} {
 		updated, _ := model.Update(tea.KeyMsg{Type: tea.KeyRunes, Runes: []rune(key)})
@@ -1961,8 +1961,7 @@ func TestModelNekiConnectionViewIsReadOnly(t *testing.T) {
 
 	updated, _ = model.Update(tea.KeyMsg{Type: tea.KeyRunes, Runes: []rune("?")})
 	helpView := stripANSI(updated.(Model).View())
-	c.Assert(helpView, qt.Contains, "Read-only")
-	c.Assert(helpView, qt.Not(qt.Contains), "pg_cancel_backend")
+	c.Assert(helpView, qt.Contains, "pg_cancel_backend")
 	c.Assert(helpView, qt.Not(qt.Contains), "KILL QUERY")
 
 	updated, _ = model.Update(tea.KeyMsg{Type: tea.KeyEnter})
@@ -1972,7 +1971,7 @@ func TestModelNekiConnectionViewIsReadOnly(t *testing.T) {
 	c.Assert(detailView, qt.Contains, "shard_backends:  1")
 	c.Assert(detailView, qt.Contains, "backend_pid:     4242")
 	c.Assert(detailView, qt.Not(qt.Contains), "left/right tabs")
-	c.Assert(detailView, qt.Not(qt.Contains), "cancel query")
+	c.Assert(detailView, qt.Contains, "cancel query")
 }
 
 func TestModelNekiConnectionViewSortsWithoutBlockers(t *testing.T) {