ruki pipes

.doc/doki/doc/ideas/plugins.md

@@ -3,6 +3,7 @@
 - [Assign to me](#assign-to-me--plugin-action)
 - [Add tag to task](#add-tag-to-task--plugin-action)
 - [Custom status + reject action](#custom-status--reject-action)
+- [Implement with Claude](#implement-with-claude--pipe-action)
 - [Search all tikis](#search-all-tikis--single-lane-plugin)
 - [Quick assign](#quick-assign--lane-based-assignment)
 - [Stale task detection](#stale-task-detection--time-trigger--plugin)
@@ -58,6 +59,19 @@ statuses:
       action: update where id = id() set status="rejected"
 ```
 
+## Implement with Claude Code — pipe action
+
+Shortcut key that pipes the selected task's title and description to Claude Code for implementation.
+
+```yaml
+actions:
+  - key: "i"
+    label: "Implement"
+    action: >
+      select title, description where id = id()
+      | run("claude -p 'Implement this: $1. Details: $2'")
+```
+
 ## Search all tikis — single-lane plugin
 
 A plugin with one unfiltered lane shows every task. Press `/` to search across all of them.

.doc/doki/doc/ruki/examples.md

@@ -136,6 +136,13 @@ Use `call(...)` in a value:
 create title=call("echo hi")
 ```
 
+Pipe select results to a shell command:
+
+```sql
+select id where id = id() | run("echo $1 | pbcopy")
+select id, title where status = "done" | run("myscript $1 $2")
+```
+
 ## Before triggers
 
 Block completion when dependencies remain open:

.doc/doki/doc/ruki/operators-and-builtins.md

@@ -246,10 +246,32 @@ after update where new.status = "in progress" run("echo hello")
 
 ## Shell-related forms
 
-`ruki` includes two shell-related forms:
+`ruki` includes three shell-related forms:
 
-- `call(...)` as a string-returning expression
-- `run(...)` as an `after`-trigger action
+**`call(...)`** — a string-returning expression
 
-- `call(...)` returns a string
-- `run(...)` is used as the top-level action of an `after` trigger
+- Returns a string result from a shell command
+- Can be used in any expression context
+
+**`run(...)` in triggers** — an `after`-trigger action
+
+- Used as the top-level action of an `after` trigger
+- Command string may reference `old.` and `new.` fields
+- Example: `after update where new.status = "in progress" run("echo hello")`
+
+**`| run(...)` on select** — a pipe suffix
+
+- Executes a command for each row returned by `select`
+- Uses positional arguments `$1`, `$2`, etc. for field substitution
+- Command must be a string literal or expression, but **field references are not allowed in the command** itself
+- `|` is a statement suffix, not an operator
+- Example: `select id, title where status = "done" | run("myscript $1 $2")`
+
+Pipe `| run(...)` is distinct from trigger `run()`:
+
+| Form | Context | Field access | Substitution |
+|---|---|---|---|
+| trigger `run()` | after-trigger action | `old.`, `new.` allowed | expression evaluation |
+| pipe `| run()` | select suffix | field refs disallowed in command | positional args `$1`, `$2` |
+
+See [Semantics](semantics.md#pipe-actions-on-select) for pipe evaluation model.

.doc/doki/doc/ruki/quick-start.md

@@ -63,6 +63,14 @@ delete where id = "TIKI-ABC123"
 delete where status = "cancelled" and "old" in tags
 ```
 
+`select` may pipe results to a shell command:
+
+```sql
+select id, title where status = "done" | run("myscript $1 $2")
+```
+
+Each row executes the command with field values as positional arguments (`$1`, `$2`).
+
 ## Conditions and expressions
 
 Conditions support:

.doc/doki/doc/ruki/semantics.md

@@ -167,3 +167,39 @@ Binary `+` and `-` are semantic rather than purely numeric:
 
 For the detailed type rules and built-ins, see [Types And Values](types-and-values.md) and [Operators And Built-ins](operators-and-builtins.md).
 
+## Pipe actions on select
+
+`select` statements may include an optional pipe suffix:
+
+```text
+select <fields> where <condition> | run(<command>)
+```
+
+Evaluation model:
+
+- The `select` runs first, producing zero or more rows.
+- For each row, the `run()` command is executed with positional arguments (`$1`, `$2`, etc.) substituted from the selected fields in left-to-right order.
+- Each command execution has a **30-second timeout**.
+- Command failures are **non-fatal** — remaining rows still execute.
+- Stdout and stderr are **fire-and-forget** (not captured or returned).
+
+Rules:
+
+- Explicit field names are required — `select *` and bare `select` are rejected when used with a pipe.
+- The command expression must be a string literal or string-typed expression, but **field references are not allowed** in the command string itself.
+- Positional arguments `$1`, `$2`, etc. are substituted by the runtime before each command execution.
+
+Example:
+
+```sql
+select id, title where status = "done" | run("myscript $1 $2")
+```
+
+For a task with `id = "TIKI-ABC123"` and `title = "Fix bug"`, the command becomes:
+
+```bash
+myscript "TIKI-ABC123" "Fix bug"
+```
+
+Pipe `| run(...)` on select is distinct from trigger `run()` actions. See [Triggers](triggers.md) for the difference.
+

.doc/doki/doc/ruki/syntax.md

@@ -48,8 +48,9 @@ The following EBNF-style summary shows the grammar:
 ```text
 statement        = selectStmt | createStmt | updateStmt | deleteStmt ;
 
-selectStmt       = "select" [ fieldList | "*" ] [ "where" condition ] [ orderBy ] ;
+selectStmt       = "select" [ fieldList | "*" ] [ "where" condition ] [ orderBy ] [ pipeAction ] ;
 fieldList        = identifier { "," identifier } ;
+pipeAction       = "|" runAction ;
 createStmt       = "create" assignment { assignment } ;
 
 orderBy          = "order" "by" sortField { "," sortField } ;

.doc/doki/doc/ruki/triggers.md

@@ -248,6 +248,8 @@ The maximum cascade depth is **8**. Termination is graceful — a warning is log
 
 ## The run() action
 
+**Note:** This section describes trigger `run()` actions. For the pipe syntax `| run(...)` on select statements, see [Semantics](semantics.md#pipe-actions-on-select) and [Operators And Built-ins](operators-and-builtins.md#shell-related-forms).
+
 When an after-trigger uses `run(...)`, the command expression is evaluated to a string, then executed:
 
 - The command runs via `sh -c <command-string>`.

.doc/doki/doc/ruki/validation-and-errors.md

@@ -216,6 +216,49 @@ Order by inside a subquery:
 select where count(select where status = "done" order by priority) >= 1
 ```
 
+## Pipe validation errors
+
+Pipe actions on `select` have several restrictions:
+
+`select *` with pipe:
+
+```sql
+select * where status = "done" | run("echo $1")
+```
+
+Bare `select` with pipe:
+
+```sql
+select | run("echo $1")
+```
+
+Both are rejected because explicit field names are required when using a pipe.
+
+Non-string command:
+
+```sql
+select id where status = "done" | run(42)
+```
+
+Field references in command:
+
+```sql
+select id, title where status = "done" | run(title + " " + id)
+select id, title where status = "done" | run("echo " + title)
+```
+
+Field references are not allowed in the pipe command expression itself. Use positional arguments (`$1`, `$2`) instead.
+
+Pipe on non-select statements:
+
+```sql
+update where status = "done" set priority=1 | run("echo done")
+create title="x" | run("echo created")
+delete where id = "TIKI-ABC123" | run("echo deleted")
+```
+
+Pipe suffix is only valid on `select` statements.
+
 ## Built-in and subquery errors
 
 Unknown function:

controller/plugin.go

@@ -159,7 +159,7 @@ func (pc *PluginController) handlePluginAction(r rune) bool {
 	input := ruki.ExecutionInput{}
 	taskID := pc.getSelectedTaskID(pc.GetFilteredTasksForLane)
 
-	if pa.Action.IsUpdate() || pa.Action.IsDelete() {
+	if pa.Action.IsUpdate() || pa.Action.IsDelete() || pa.Action.IsPipe() {
 		if taskID == "" {
 			return false
 		}
@@ -212,6 +212,16 @@ func (pc *PluginController) handlePluginAction(r rune) bool {
 				return false
 			}
 		}
+	case result.Pipe != nil:
+		for _, row := range result.Pipe.Rows {
+			if err := service.ExecutePipeCommand(ctx, result.Pipe.Command, row); err != nil {
+				slog.Error("pipe command failed", "command", result.Pipe.Command, "args", row, "key", string(r), "error", err)
+				if pc.statusline != nil {
+					pc.statusline.SetMessage(err.Error(), model.MessageLevelError, true)
+				}
+				// non-fatal: continue remaining rows
+			}
+		}
 	}
 
 	slog.Info("plugin action applied", "task_id", taskID, "key", string(r), "label", pa.Label, "plugin", pc.pluginDef.Name)

plugin/parser.go

@@ -210,8 +210,8 @@ func parsePluginActions(configs []PluginActionConfig, parser *ruki.Parser) ([]Pl
 		if err != nil {
 			return nil, fmt.Errorf("parsing action %d (key %q): %w", i, cfg.Key, err)
 		}
-		if actionStmt.IsSelect() {
-			return nil, fmt.Errorf("action %d (key %q) must be UPDATE, CREATE, or DELETE — not SELECT", i, cfg.Key)
+		if actionStmt.IsSelect() && !actionStmt.IsPipe() {
+			return nil, fmt.Errorf("action %d (key %q) must be UPDATE, CREATE, DELETE, or a piped SELECT — not plain SELECT", i, cfg.Key)
 		}
 
 		actions = append(actions, PluginAction{

plugin/parser_test.go

@@ -504,8 +504,25 @@ func TestParsePluginActions_SelectRejectedAsAction(t *testing.T) {
 	if err == nil {
 		t.Fatal("expected error for SELECT as plugin action")
 	}
-	if !strings.Contains(err.Error(), "not SELECT") {
-		t.Errorf("expected 'not SELECT' error, got: %v", err)
+	if !strings.Contains(err.Error(), "not plain SELECT") {
+		t.Errorf("expected 'not plain SELECT' error, got: %v", err)
+	}
+}
+
+func TestParsePluginActions_PipeAcceptedAsAction(t *testing.T) {
+	parser := testParser()
+	configs := []PluginActionConfig{
+		{Key: "c", Label: "Copy ID", Action: `select id where id = id() | run("echo $1")`},
+	}
+	actions, err := parsePluginActions(configs, parser)
+	if err != nil {
+		t.Fatalf("expected pipe action to be accepted, got error: %v", err)
+	}
+	if len(actions) != 1 {
+		t.Fatalf("expected 1 action, got %d", len(actions))
+	}
+	if !actions[0].Action.IsPipe() {
+		t.Error("expected IsPipe() = true for pipe action")
 	}
 }
 

ruki/ast.go

@@ -13,11 +13,12 @@ type Statement struct {
 	Delete *DeleteStmt
 }
 
-// SelectStmt represents "select [fields] [where <condition>] [order by <field> [asc|desc], ...]".
+// SelectStmt represents "select [fields] [where <condition>] [order by <field> [asc|desc], ...] [| run(...)]".
 type SelectStmt struct {
 	Fields  []string        // nil = all ("select" or "select *"); non-nil = specific fields
 	Where   Condition       // nil = select all
 	OrderBy []OrderByClause // nil = unordered
+	Pipe    *RunAction      // optional pipe suffix: "| run(...)"
 }
 
 // CreateStmt represents "create <field>=<value>...".

ruki/executor.go

@@ -38,6 +38,14 @@ type Result struct {
 	Update *UpdateResult
 	Create *CreateResult
 	Delete *DeleteResult
+	Pipe   *PipeResult
+}
+
+// PipeResult holds the shell command and per-row positional args from a piped select.
+// The ruki executor builds this; the service layer performs the actual shell execution.
+type PipeResult struct {
+	Command string
+	Rows    [][]string
 }
 
 // UpdateResult holds the cloned, mutated tasks produced by an UPDATE statement.
@@ -138,6 +146,10 @@ func (e *Executor) executeSelect(sel *SelectStmt, tasks []*task.Task) (*Result,
 		e.sortTasks(filtered, sel.OrderBy)
 	}
 
+	if sel.Pipe != nil {
+		return e.buildPipeResult(sel.Pipe, sel.Fields, filtered, tasks)
+	}
+
 	return &Result{
 		Select: &TaskProjection{
 			Tasks:  filtered,
@@ -146,6 +158,42 @@ func (e *Executor) executeSelect(sel *SelectStmt, tasks []*task.Task) (*Result,
 	}, nil
 }
 
+func (e *Executor) buildPipeResult(pipe *RunAction, fields []string, matched []*task.Task, allTasks []*task.Task) (*Result, error) {
+	// evaluate command once with a nil-sentinel task — validation ensures no field refs
+	cmdVal, err := e.evalExpr(pipe.Command, nil, allTasks)
+	if err != nil {
+		return nil, fmt.Errorf("pipe command: %w", err)
+	}
+	cmdStr, ok := cmdVal.(string)
+	if !ok {
+		return nil, fmt.Errorf("pipe command must evaluate to string, got %T", cmdVal)
+	}
+
+	rows := make([][]string, len(matched))
+	for i, t := range matched {
+		row := make([]string, len(fields))
+		for j, f := range fields {
+			row[j] = pipeArgString(extractField(t, f))
+		}
+		rows[i] = row
+	}
+
+	return &Result{Pipe: &PipeResult{Command: cmdStr, Rows: rows}}, nil
+}
+
+// pipeArgString space-joins list fields (tags, dependsOn) instead of using Go's
+// default fmt.Sprint which produces "[a b c]" with brackets.
+func pipeArgString(val interface{}) string {
+	if list, ok := val.([]interface{}); ok {
+		parts := make([]string, len(list))
+		for i, elem := range list {
+			parts[i] = normalizeToString(elem)
+		}
+		return strings.Join(parts, " ")
+	}
+	return normalizeToString(val)
+}
+
 func (e *Executor) executeUpdate(upd *UpdateStmt, tasks []*task.Task) (*Result, error) {
 	matched, err := e.filterTasks(upd.Where, tasks)
 	if err != nil {

ruki/grammar.go

@@ -23,6 +23,7 @@ type selectGrammar struct {
 	Fields  *fieldNamesGrammar `parser:"           | @@ )?"`
 	Where   *orCond            `parser:"( 'where' @@ )?"`
 	OrderBy *orderByGrammar    `parser:"@@?"`
+	Pipe    *runGrammar        `parser:"( Pipe @@ )?"`
 }
 
 // --- order by grammar ---

ruki/lexer.go

@@ -25,6 +25,7 @@ var rukiLexer = lexer.MustSimple([]lexer.SimpleRule{
 	{Name: "LBracket", Pattern: `\[`},
 	{Name: "RBracket", Pattern: `\]`},
 	{Name: "Comma", Pattern: `,`},
+	{Name: "Pipe", Pattern: `\|`},
 	{Name: "Keyword", Pattern: keywordPattern()},
 	{Name: "Ident", Pattern: `[a-zA-Z_][a-zA-Z0-9_]*`},
 })

ruki/lower.go

@@ -17,6 +17,13 @@ func lowerStatement(g *statementGrammar) (*Statement, error) {
 		if err != nil {
 			return nil, err
 		}
+		if g.Select.Pipe != nil {
+			cmd, err := lowerExpr(&g.Select.Pipe.Command)
+			if err != nil {
+				return nil, err
+			}
+			s.Pipe = &RunAction{Command: cmd}
+		}
 		return &Statement{Select: s}, nil
 	case g.Create != nil:
 		s, err := lowerCreate(g.Create)

ruki/pipe_test.go

@@ -0,0 +1,369 @@
+package ruki
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/boolean-maybe/tiki/task"
+)
+
+// --- lexer ---
+
+func TestTokenizePipe(t *testing.T) {
+	pipeType := rukiLexer.Symbols()["Pipe"]
+
+	t.Run("bare pipe", func(t *testing.T) {
+		tokens := tokenize(t, "|")
+		if len(tokens) != 1 {
+			t.Fatalf("expected 1 token, got %d: %v", len(tokens), tokens)
+		}
+		if tokens[0].Type != pipeType {
+			t.Errorf("expected Pipe token, got type %d", tokens[0].Type)
+		}
+	})
+
+	t.Run("pipe in context", func(t *testing.T) {
+		tokens := tokenize(t, `select id | run("echo $1")`)
+		found := false
+		for _, tok := range tokens {
+			if tok.Type == pipeType {
+				found = true
+				break
+			}
+		}
+		if !found {
+			t.Error("expected to find Pipe token in tokenized output")
+		}
+	})
+}
+
+// --- parser ---
+
+func TestParsePipeSelect(t *testing.T) {
+	p := newTestParser()
+
+	tests := []struct {
+		name  string
+		input string
+	}{
+		{"basic pipe", `select id where id = id() | run("echo $1")`},
+		{"multi-field pipe", `select id, title where status = "done" | run("myscript $1 $2")`},
+		{"pipe without where", `select id | run("echo $1")`},
+		{"pipe with order by", `select id, priority where status = "ready" order by priority | run("echo $1 $2")`},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			stmt, err := p.ParseAndValidateStatement(tt.input, ExecutorRuntimePlugin)
+			if err != nil {
+				t.Fatalf("parse error: %v", err)
+			}
+			if !stmt.IsSelect() {
+				t.Fatal("expected IsSelect() true")
+			}
+			if !stmt.IsPipe() {
+				t.Fatal("expected IsPipe() true")
+			}
+		})
+	}
+}
+
+func TestParsePipeDoesNotBreakPlainSelect(t *testing.T) {
+	p := newTestParser()
+
+	tests := []struct {
+		name  string
+		input string
+	}{
+		{"bare select", "select"},
+		{"select star", "select *"},
+		{"select with where", `select where status = "done"`},
+		{"select with fields", `select id, title where status = "done"`},
+		{"select with order by", `select order by priority`},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			stmt, err := p.ParseStatement(tt.input)
+			if err != nil {
+				t.Fatalf("parse error: %v", err)
+			}
+			if stmt.Select == nil {
+				t.Fatal("expected Select")
+			}
+			if stmt.Select.Pipe != nil {
+				t.Fatal("expected no Pipe on plain select")
+			}
+		})
+	}
+}
+
+// --- lowering ---
+
+func TestLowerPipeProducesSelectAndPipe(t *testing.T) {
+	p := newTestParser()
+	stmt, err := p.ParseStatement(`select id where id = id() | run("echo $1")`)
+	if err != nil {
+		t.Fatalf("parse error: %v", err)
+	}
+	if stmt.Select == nil {
+		t.Fatal("expected Select non-nil")
+	}
+	if stmt.Select.Pipe == nil {
+		t.Fatal("expected Select.Pipe non-nil")
+	}
+	if stmt.Select.Pipe.Command == nil {
+		t.Fatal("expected Select.Pipe.Command non-nil")
+	}
+}
+
+// --- validation ---
+
+func TestPipeRejectSelectStar(t *testing.T) {
+	p := newTestParser()
+	_, err := p.ParseStatement(`select * | run("echo")`)
+	if err == nil {
+		t.Fatal("expected error for select * with pipe")
+	}
+	if !strings.Contains(err.Error(), "explicit field names") {
+		t.Errorf("expected 'explicit field names' error, got: %v", err)
+	}
+}
+
+func TestPipeRejectBareSelect(t *testing.T) {
+	p := newTestParser()
+	_, err := p.ParseStatement(`select | run("echo")`)
+	if err == nil {
+		t.Fatal("expected error for bare select with pipe")
+	}
+	if !strings.Contains(err.Error(), "explicit field names") {
+		t.Errorf("expected 'explicit field names' error, got: %v", err)
+	}
+}
+
+func TestPipeRejectNonStringCommand(t *testing.T) {
+	p := newTestParser()
+	_, err := p.ParseStatement(`select id | run(42)`)
+	if err == nil {
+		t.Fatal("expected error for non-string pipe command")
+	}
+	if !strings.Contains(err.Error(), "pipe command must be string") {
+		t.Errorf("expected 'pipe command must be string' error, got: %v", err)
+	}
+}
+
+func TestPipeRejectFieldRefInCommand(t *testing.T) {
+	p := newTestParser()
+
+	tests := []struct {
+		name  string
+		input string
+	}{
+		{"bare field ref", `select id | run(title)`},
+		{"qualified ref", `select id | run(old.title)`},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := p.ParseStatement(tt.input)
+			if err == nil {
+				t.Fatal("expected error for field ref in pipe command")
+			}
+			// either the field-ref check catches it, or type inference
+			// rejects it earlier (e.g. "old. qualifier not valid")
+		})
+	}
+}
+
+func TestExprContainsFieldRef(t *testing.T) {
+	tests := []struct {
+		name     string
+		expr     Expr
+		expected bool
+	}{
+		{"string literal", &StringLiteral{Value: "echo"}, false},
+		{"int literal", &IntLiteral{Value: 42}, false},
+		{"bare field ref", &FieldRef{Name: "title"}, true},
+		{"qualified ref", &QualifiedRef{Qualifier: "old", Name: "title"}, true},
+		{"field in binary expr", &BinaryExpr{
+			Op:    "+",
+			Left:  &StringLiteral{Value: "echo "},
+			Right: &FieldRef{Name: "title"},
+		}, true},
+		{"field in function arg", &FunctionCall{
+			Name: "concat",
+			Args: []Expr{&FieldRef{Name: "title"}},
+		}, true},
+		{"clean function", &FunctionCall{
+			Name: "id",
+			Args: nil,
+		}, false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := exprContainsFieldRef(tt.expr)
+			if got != tt.expected {
+				t.Errorf("exprContainsFieldRef() = %v, want %v", got, tt.expected)
+			}
+		})
+	}
+}
+
+// --- semantic validation ---
+
+func TestIsPipeMethod(t *testing.T) {
+	p := newTestParser()
+
+	pipeStmt, err := p.ParseAndValidateStatement(`select id where id = id() | run("echo $1")`, ExecutorRuntimePlugin)
+	if err != nil {
+		t.Fatalf("parse error: %v", err)
+	}
+	if !pipeStmt.IsPipe() {
+		t.Error("expected IsPipe() = true for pipe statement")
+	}
+
+	plainStmt, err := p.ParseAndValidateStatement(`select where status = "done"`, ExecutorRuntimeCLI)
+	if err != nil {
+		t.Fatalf("parse error: %v", err)
+	}
+	if plainStmt.IsPipe() {
+		t.Error("expected IsPipe() = false for plain select")
+	}
+
+	updateStmt, err := p.ParseAndValidateStatement(`update where id = id() set status = "done"`, ExecutorRuntimePlugin)
+	if err != nil {
+		t.Fatalf("parse error: %v", err)
+	}
+	if updateStmt.IsPipe() {
+		t.Error("expected IsPipe() = false for update statement")
+	}
+}
+
+func TestPipeIDDetectedInWhereClause(t *testing.T) {
+	p := newTestParser()
+	stmt, err := p.ParseAndValidateStatement(`select id where id = id() | run("echo $1")`, ExecutorRuntimePlugin)
+	if err != nil {
+		t.Fatalf("parse error: %v", err)
+	}
+	if !stmt.UsesIDBuiltin() {
+		t.Error("expected UsesIDBuiltin() = true for pipe with id() in where")
+	}
+}
+
+// --- executor ---
+
+func TestExecutePipeReturnsResult(t *testing.T) {
+	e := NewExecutor(testSchema{}, nil, ExecutorRuntime{Mode: ExecutorRuntimePlugin})
+	p := newTestParser()
+	tasks := makeTasks()
+
+	stmt, err := p.ParseStatement(`select id, title where status = "done" | run("echo $1 $2")`)
+	if err != nil {
+		t.Fatalf("parse: %v", err)
+	}
+
+	result, err := e.Execute(stmt, tasks, ExecutionInput{SelectedTaskID: "TIKI-000003"})
+	if err != nil {
+		t.Fatalf("execute: %v", err)
+	}
+	if result.Pipe == nil {
+		t.Fatal("expected Pipe result")
+	}
+	if result.Select != nil {
+		t.Fatal("expected Select to be nil when pipe is present")
+	}
+	if result.Pipe.Command != "echo $1 $2" {
+		t.Errorf("command = %q, want %q", result.Pipe.Command, "echo $1 $2")
+	}
+	if len(result.Pipe.Rows) != 1 {
+		t.Fatalf("expected 1 row (status=done), got %d", len(result.Pipe.Rows))
+	}
+	row := result.Pipe.Rows[0]
+	if len(row) != 2 {
+		t.Fatalf("expected 2 fields, got %d", len(row))
+	}
+	if row[0] != "TIKI-000003" {
+		t.Errorf("row[0] (id) = %q, want %q", row[0], "TIKI-000003")
+	}
+	if row[1] != "Write docs" {
+		t.Errorf("row[1] (title) = %q, want %q", row[1], "Write docs")
+	}
+}
+
+func TestExecuteSelectStillWorksWithoutPipe(t *testing.T) {
+	e := newTestExecutor()
+	p := newTestParser()
+	tasks := makeTasks()
+
+	stmt, err := p.ParseStatement(`select where status = "done"`)
+	if err != nil {
+		t.Fatalf("parse: %v", err)
+	}
+
+	result, err := e.Execute(stmt, tasks)
+	if err != nil {
+		t.Fatalf("execute: %v", err)
+	}
+	if result.Select == nil {
+		t.Fatal("expected Select result for plain select")
+	}
+	if result.Pipe != nil {
+		t.Fatal("expected no Pipe result for plain select")
+	}
+}
+
+func TestExecutePipeListFieldSpaceJoined(t *testing.T) {
+	e := NewExecutor(testSchema{}, nil, ExecutorRuntime{Mode: ExecutorRuntimePlugin})
+	p := newTestParser()
+	tasks := []*task.Task{
+		{ID: "TIKI-000001", Title: "Test", Status: "ready", Type: "story",
+			Priority: 1, Tags: []string{"a", "b", "c"}},
+	}
+
+	stmt, err := p.ParseStatement(`select id, tags where id = id() | run("echo $1 $2")`)
+	if err != nil {
+		t.Fatalf("parse: %v", err)
+	}
+
+	result, err := e.Execute(stmt, tasks, ExecutionInput{SelectedTaskID: "TIKI-000001"})
+	if err != nil {
+		t.Fatalf("execute: %v", err)
+	}
+	if result.Pipe == nil {
+		t.Fatal("expected Pipe result")
+	}
+	if len(result.Pipe.Rows) != 1 {
+		t.Fatalf("expected 1 row, got %d", len(result.Pipe.Rows))
+	}
+	row := result.Pipe.Rows[0]
+	if row[1] != "a b c" {
+		t.Errorf("tags field = %q, want %q", row[1], "a b c")
+	}
+}
+
+// --- pipeArgString ---
+
+func TestPipeArgString(t *testing.T) {
+	tests := []struct {
+		name string
+		val  interface{}
+		want string
+	}{
+		{"string", "hello", "hello"},
+		{"int", 3, "3"},
+		{"string slice", []interface{}{"a", "b"}, "a b"},
+		{"empty slice", []interface{}{}, ""},
+		{"nil", nil, "<nil>"},
+		{"status", task.Status("done"), "done"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := pipeArgString(tt.val)
+			if got != tt.want {
+				t.Errorf("pipeArgString(%v) = %q, want %q", tt.val, got, tt.want)
+			}
+		})
+	}
+}

ruki/semantic_validate.go

@@ -46,6 +46,9 @@ func (v *ValidatedStatement) IsCreate() bool {
 func (v *ValidatedStatement) IsDelete() bool {
 	return v != nil && v.statement != nil && v.statement.Delete != nil
 }
+func (v *ValidatedStatement) IsPipe() bool {
+	return v != nil && v.statement != nil && v.statement.Select != nil && v.statement.Select.Pipe != nil
+}
 
 func (v *ValidatedStatement) mustBeSealed() error {
 	if v == nil || v.seal != validatedSeal || v.statement == nil {
@@ -384,7 +387,18 @@ func scanSelectSemantics(sel *SelectStmt) (usesID bool, hasCall bool, err error)
 	if sel == nil {
 		return false, false, nil
 	}
-	return scanConditionSemantics(sel.Where)
+	u, c, err := scanConditionSemantics(sel.Where)
+	if err != nil {
+		return false, false, err
+	}
+	if sel.Pipe != nil {
+		u2, c2, err := scanExprSemantics(sel.Pipe.Command)
+		if err != nil {
+			return false, false, err
+		}
+		u, c = u || u2, c || c2
+	}
+	return u, c, nil
 }
 
 func scanTriggerSemantics(trig *Trigger) (usesID bool, hasCall bool, err error) {
@@ -566,11 +580,15 @@ func cloneSelect(sel *SelectStmt) *SelectStmt {
 	if sel.OrderBy != nil {
 		orderBy = append([]OrderByClause(nil), sel.OrderBy...)
 	}
-	return &SelectStmt{
+	out := &SelectStmt{
 		Fields:  fields,
 		Where:   cloneCondition(sel.Where),
 		OrderBy: orderBy,
 	}
+	if sel.Pipe != nil {
+		out.Pipe = &RunAction{Command: cloneExpr(sel.Pipe.Command)}
+	}
+	return out
 }
 
 func cloneCreate(cr *CreateStmt) *CreateStmt {

ruki/validate.go

@@ -67,7 +67,25 @@ func (p *Parser) validateStatement(s *Statement) error {
 				return err
 			}
 		}
-		return p.validateOrderBy(s.Select.OrderBy)
+		if err := p.validateOrderBy(s.Select.OrderBy); err != nil {
+			return err
+		}
+		if s.Select.Pipe != nil {
+			if len(s.Select.Fields) == 0 {
+				return fmt.Errorf("pipe requires explicit field names in select (not select * or bare select)")
+			}
+			typ, err := p.inferExprType(s.Select.Pipe.Command)
+			if err != nil {
+				return fmt.Errorf("pipe command: %w", err)
+			}
+			if typ != ValueString {
+				return fmt.Errorf("pipe command must be string, got %s", typeName(typ))
+			}
+			if exprContainsFieldRef(s.Select.Pipe.Command) {
+				return fmt.Errorf("pipe command must not contain field references — use $1, $2 for positional args")
+			}
+		}
+		return nil
 	default:
 		return fmt.Errorf("empty statement")
 	}
@@ -881,3 +899,33 @@ func typeName(t ValueType) string {
 		return "unknown"
 	}
 }
+
+// exprContainsFieldRef returns true if the expression tree contains any
+// *FieldRef or *QualifiedRef node. Used to reject field references in
+// pipe commands, where positional args ($1, $2) should be used instead.
+func exprContainsFieldRef(expr Expr) bool {
+	switch e := expr.(type) {
+	case *FieldRef:
+		return true
+	case *QualifiedRef:
+		return true
+	case *BinaryExpr:
+		return exprContainsFieldRef(e.Left) || exprContainsFieldRef(e.Right)
+	case *FunctionCall:
+		for _, arg := range e.Args {
+			if exprContainsFieldRef(arg) {
+				return true
+			}
+		}
+		return false
+	case *ListLiteral:
+		for _, elem := range e.Elements {
+			if exprContainsFieldRef(elem) {
+				return true
+			}
+		}
+		return false
+	default:
+		return false
+	}
+}

service/shell.go

@@ -0,0 +1,40 @@
+package service
+
+import (
+	"context"
+	"log/slog"
+	"os/exec"
+	"time"
+)
+
+// shellCommandTimeout is the timeout for shell commands executed by pipes and triggers.
+const shellCommandTimeout = 30 * time.Second
+
+// RunShellCommand executes a shell command via "sh -c" with optional positional args.
+// When args are provided, "_" occupies $0 and args land in $1, $2, etc.
+// This is standard POSIX shell behavior for "sh -c <script> <$0> <$1> ...".
+func RunShellCommand(ctx context.Context, cmdStr string, args ...string) ([]byte, error) {
+	runCtx, cancel := context.WithTimeout(ctx, shellCommandTimeout)
+	defer cancel()
+	argv := []string{"-c", cmdStr}
+	if len(args) > 0 {
+		argv = append(argv, "_")     // $0 placeholder
+		argv = append(argv, args...) // $1, $2, ...
+	}
+	cmd := exec.CommandContext(runCtx, "sh", argv...) //nolint:gosec // cmdStr is a user-configured action, intentionally dynamic
+	setProcessGroup(cmd)
+	cmd.WaitDelay = 3 * time.Second
+	return cmd.CombinedOutput()
+}
+
+// ExecutePipeCommand runs a pipe run() command with positional args for one row.
+// Errors are logged and returned; callers decide whether to continue with remaining rows.
+func ExecutePipeCommand(ctx context.Context, cmdStr string, args []string) error {
+	output, err := RunShellCommand(ctx, cmdStr, args...)
+	if err != nil {
+		slog.Error("pipe run() command failed", "command", cmdStr, "args", args, "output", string(output), "error", err)
+		return err
+	}
+	slog.Debug("pipe run() command succeeded", "command", cmdStr, "args", args)
+	return nil
+}

service/shell_test.go

@@ -0,0 +1,62 @@
+package service
+
+import (
+	"context"
+	"strings"
+	"testing"
+)
+
+func TestRunShellCommand_NoArgs(t *testing.T) {
+	output, err := RunShellCommand(context.Background(), "echo hello")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if got := strings.TrimSpace(string(output)); got != "hello" {
+		t.Errorf("output = %q, want %q", got, "hello")
+	}
+}
+
+func TestRunShellCommand_PositionalArgs(t *testing.T) {
+	output, err := RunShellCommand(context.Background(), "echo $1 $2", "world", "foo")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if got := strings.TrimSpace(string(output)); got != "world foo" {
+		t.Errorf("output = %q, want %q", got, "world foo")
+	}
+}
+
+func TestRunShellCommand_PositionalArgsWithSpaces(t *testing.T) {
+	output, err := RunShellCommand(context.Background(), "echo \"$1\"", "hello world")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if got := strings.TrimSpace(string(output)); got != "hello world" {
+		t.Errorf("output = %q, want %q", got, "hello world")
+	}
+}
+
+func TestRunShellCommand_DollarZeroIsPlaceholder(t *testing.T) {
+	// $0 should be "_" (the placeholder), not the first real arg
+	output, err := RunShellCommand(context.Background(), "echo $0", "real-arg")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if got := strings.TrimSpace(string(output)); got != "_" {
+		t.Errorf("$0 = %q, want %q", got, "_")
+	}
+}
+
+func TestExecutePipeCommand_Success(t *testing.T) {
+	err := ExecutePipeCommand(context.Background(), "echo $1", []string{"hello"})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestExecutePipeCommand_Failure(t *testing.T) {
+	err := ExecutePipeCommand(context.Background(), "exit 1", nil)
+	if err == nil {
+		t.Fatal("expected error for failing command")
+	}
+}

service/trigger_engine.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
-	"os/exec"
 	"time"
 
 	"github.com/boolean-maybe/tiki/config"
@@ -18,9 +17,6 @@ import (
 // Root mutation is depth 0; up to 8 cascades are allowed.
 const maxTriggerDepth = 8
 
-// runCommandTimeout is the timeout for run() commands executed by triggers.
-const runCommandTimeout = 30 * time.Second
-
 // triggerEntry holds a parsed trigger and its description for logging.
 type triggerEntry struct {
 	description string
@@ -219,13 +215,7 @@ func (te *TriggerEngine) execRun(ctx context.Context, entry triggerEntry, tc *ru
 		return fmt.Errorf("trigger %q run evaluation failed: %w", entry.description, err)
 	}
 
-	runCtx, cancel := context.WithTimeout(ctx, runCommandTimeout)
-	defer cancel()
-
-	cmd := exec.CommandContext(runCtx, "sh", "-c", cmdStr) //nolint:gosec // cmdStr is a user-configured trigger action, intentionally dynamic
-	setProcessGroup(cmd)
-	cmd.WaitDelay = 3 * time.Second
-	output, err := cmd.CombinedOutput()
+	output, err := RunShellCommand(ctx, cmdStr)
 	if err != nil {
 		slog.Error("trigger run() command failed",
 			"trigger", entry.description,