feat(cli,inspect,runtime): wire run attribution flags and filters

Add agentctl run attribution flags with local defaults and prod guardrails,
logs/inspector filtering by tenant/thread/actor, OTel gen_ai scope attributes,
and resume behavior that preserves persisted thread identity.

Closes #111.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: leo-aa88

CHANGELOG.md

@@ -8,6 +8,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Added
 
+- **Run attribution** (issue #111): `tenant_id`, `thread_id`, `actor_id`, `parent_run_id`, `request_id`, `idempotency_key`, and `source` on `runs`; trace events carry matching tenant/thread/actor for filterable logs and inspector queries. `agentctl run` accepts `--tenant-id`, `--thread-id`, `--actor-id` (local defaults `tenant-1` / `thread-1` / `user-1`); `agentctl logs` and `GET /api/runs` filter by the same dimensions. `--resume` reuses persisted `run_id` and `thread_id`. OTel spans emit `gen_ai.tenant.id`, `gen_ai.thread.id`, `gen_ai.actor.id`, and `gen_ai.request.id`. See [`docs/ATTRIBUTION.md`](docs/ATTRIBUTION.md).
 - **Trace payload redaction** (issue #110): trace events are sanitized, key-redacted, and size-capped before SQLite storage. Defaults mask common secret key names; override via `Project.spec.traces.redactKeys`, `maxPayloadBytes`, and `spec.traces.redaction` (`maxDepth`, `maxBytes` for binary previews, `maxStringChars`). HITL edit `argsDiff` is redacted before persistence. Local runs use [trace.NewRecorderForGraph] from project spec.
 - **Optional OpenTelemetry trace export** (issue #108): `Project.spec.telemetry` (`enabled`, `serviceName`, `endpoint` with `env:` tokens, `consoleExport`) emits WayFind-aligned `gen_ai.*` spans (`agent.run`, `model.chat`, `tool.exec`, `approval`) alongside SQLite traces. Disabled by default; init failures log a warning and never fail runs. See [`docs/OTEL.md`](docs/OTEL.md) for a Jaeger quick start.
 - **`agentctl inspect --web`** — read-only local inspector (default `http://127.0.0.1:8787`) over SQLite state: runs, trace timeline, run steps, applied deployment resources, and checkpoints ([#109](https://github.com/LAA-Software-Engineering/agentic-control-plane/issues/109)).

docs/ATTRIBUTION.md

@@ -0,0 +1,64 @@
+# Run attribution (tenant, thread, actor)
+
+Issue [#111](https://github.com/LAA-Software-Engineering/agentic-control-plane/issues/111) adds lightweight tenancy and attribution to `runs` and `trace_events`.
+
+## Fields
+
+| Field | Purpose |
+| --- | --- |
+| `tenant_id` | Outermost multi-tenant scope |
+| `thread_id` | Session continuity across runs and `--resume` |
+| `actor_id` | Who triggered the run (caller-asserted for now) |
+| `parent_run_id` | Lineage for sub-runs (not set on resume of the same run) |
+| `request_id` | Per-invocation correlation id (distinct from `run_id`) |
+| `idempotency_key` | Optional dedupe key for accidental re-triggers |
+| `source` | Origin label (`cli`, `actions`, `api`, …) |
+
+Trace events duplicate `tenant_id`, `thread_id`, and `actor_id` from the parent run so `logs` and the inspector can filter without joins.
+
+## CLI defaults (local only)
+
+When flags are omitted, `agentctl run` stores:
+
+- `tenant_id`: `tenant-1`
+- `thread_id`: `thread-1`
+- `actor_id`: `user-1`
+- `source`: `cli`
+
+**Do not rely on these defaults in CI or production.** Pass real actor ids (for example the CI principal) and include tenant/environment context in `thread_id`.
+
+```bash
+agentctl run workflow/demo \
+  --tenant-id acme \
+  --thread-id prod-review-42 \
+  --actor-id github-actions@acme
+```
+
+Filter history:
+
+```bash
+agentctl logs --tenant-id acme --thread-id prod-review-42
+```
+
+## Resume
+
+`agentctl run --resume <run-id>` reuses the original `run_id` and `thread_id` from the persisted run row. Attribution flags on resume are ignored so thread timelines stay coherent. `--parent-run-id` is for genuine sub-runs, not resumes.
+
+## Inspector API
+
+`GET /api/runs` accepts optional query parameters:
+
+- `tenant_id`
+- `thread_id`
+- `actor_id`
+- `workflow`
+- `limit`
+
+## OpenTelemetry
+
+When telemetry is enabled, spans emit `gen_ai.tenant.id`, `gen_ai.thread.id`, `gen_ai.actor.id`, `gen_ai.run.id`, and `gen_ai.request.id` alongside existing gen_ai attributes. See [OTEL.md](./OTEL.md).
+
+## Production guidance
+
+- SQLite attribution is advisory; DB-level tenant isolation belongs to a future remote/Postgres store.
+- `actor_id` is supplied by the caller and is not authenticated in this release.

internal/cli/logs.go

@@ -20,6 +20,9 @@ import (
 func newLogsCmd() *cobra.Command {
 	var runID string
 	var workflow string
+	var tenantID string
+	var threadID string
+	var actorID string
 
 	cmd := &cobra.Command{
 		Use:          "logs",
@@ -28,36 +31,47 @@ func newLogsCmd() *cobra.Command {
 		Long: `Inspect execution history stored in the SQLite state database.
 
 Without filters, lists recent runs (newest first). Use --run to print trace events for one run
-(ordered by seq), or --workflow to print events for recent runs of a workflow name.
+(ordered by seq), --workflow to print events for recent runs of a workflow name, or
+--tenant-id / --thread-id / --actor-id to filter the run list (combinable).
 
 Examples:
   agentctl logs
   agentctl logs --run <run-id>
   agentctl logs --workflow pr-review
+  agentctl logs --tenant-id acme --thread-id prod-session-1
 
 Exit codes (section 11.2):
   0 — success
   1 — generic failure (e.g. cannot open SQLite)
   2 — validation failure (unknown run id, invalid flags)`,
 		RunE: func(cmd *cobra.Command, args []string) error {
 			_ = args
-			return runLogs(cmd, runID, workflow)
+			return runLogs(cmd, runID, workflow, tenantID, threadID, actorID)
 		},
 	}
 	cmd.Flags().StringVar(&runID, "run", "", "show trace events for this run id")
 	cmd.Flags().StringVar(&workflow, "workflow", "", "show trace events for recent runs of this workflow")
+	cmd.Flags().StringVar(&tenantID, "tenant-id", "", "filter runs by tenant id")
+	cmd.Flags().StringVar(&threadID, "thread-id", "", "filter runs by thread id")
+	cmd.Flags().StringVar(&actorID, "actor-id", "", "filter runs by actor id")
 	return cmd
 }
 
-func runLogs(cmd *cobra.Command, runID, workflow string) error {
+func runLogs(cmd *cobra.Command, runID, workflow, tenantID, threadID, actorID string) error {
 	ctx := context.Background()
 	g := Globals()
 
 	runID = strings.TrimSpace(runID)
 	workflow = strings.TrimSpace(workflow)
+	tenantID = strings.TrimSpace(tenantID)
+	threadID = strings.TrimSpace(threadID)
+	actorID = strings.TrimSpace(actorID)
 	if runID != "" && workflow != "" {
 		return NewExitErrorf(ExitValidationError, "logs: use only one of --run or --workflow")
 	}
+	if runID != "" && (tenantID != "" || threadID != "" || actorID != "") {
+		return NewExitErrorf(ExitValidationError, "logs: --run cannot be combined with tenant/thread/actor filters")
+	}
 
 	graph, root, err := prepareProjectGraph(g.ProjectRoot, g)
 	if err != nil {
@@ -82,11 +96,18 @@ func runLogs(cmd *cobra.Command, runID, workflow string) error {
 		}
 	}
 
+	filter := state.RunListFilter{
+		TenantID:     tenantID,
+		ThreadID:     threadID,
+		ActorID:      actorID,
+		WorkflowName: workflow,
+	}
+
 	switch {
 	case runID != "":
 		return writeLogsForRun(cmd, ctx, st, dsn, runID, g)
-	case workflow != "":
-		return writeLogsForWorkflow(cmd, ctx, st, dsn, workflow, g)
+	case workflow != "" || tenantID != "" || threadID != "" || actorID != "":
+		return writeLogsFiltered(cmd, ctx, st, dsn, filter, g)
 	default:
 		return writeLogsRunList(cmd, ctx, st, dsn, g)
 	}
@@ -106,11 +127,13 @@ func writeLogsForRun(cmd *cobra.Command, ctx context.Context, st *sqlite.Store,
 	return writeLogsEventsOutput(cmd, dsn, runID, "", events, g)
 }
 
-func writeLogsForWorkflow(cmd *cobra.Command, ctx context.Context, st *sqlite.Store, dsn, workflow string, g *Global) error {
-	runs, err := st.ListRunsByWorkflow(ctx, workflow, state.DefaultRunListLimit)
+func writeLogsFiltered(cmd *cobra.Command, ctx context.Context, st *sqlite.Store, dsn string, filter state.RunListFilter, g *Global) error {
+	filter.Limit = state.DefaultRunListLimit
+	runs, err := st.ListRunsFiltered(ctx, filter)
 	if err != nil {
 		return fmt.Errorf("logs: list runs: %w", err)
 	}
+	workflow := strings.TrimSpace(filter.WorkflowName)
 
 	if g.Output != render.FormatTable {
 		type runEntry struct {
@@ -145,7 +168,9 @@ func writeLogsForWorkflow(cmd *cobra.Command, ctx context.Context, st *sqlite.St
 	}
 
 	var b strings.Builder
-	fmt.Fprintf(&b, "State: %s\nWorkflow filter: %s\n\n", dsn, workflow)
+	fmt.Fprintf(&b, "State: %s\n", dsn)
+	writeLogsFilterHeader(&b, filter)
+	b.WriteString("\n")
 	if len(runs) == 0 {
 		fmt.Fprintln(&b, "No runs found.")
 		_, err := fmt.Fprint(cmd.OutOrStdout(), b.String())
@@ -166,6 +191,21 @@ func writeLogsForWorkflow(cmd *cobra.Command, ctx context.Context, st *sqlite.St
 	return err
 }
 
+func writeLogsFilterHeader(b *strings.Builder, filter state.RunListFilter) {
+	if w := strings.TrimSpace(filter.WorkflowName); w != "" {
+		fmt.Fprintf(b, "Workflow filter: %s\n", w)
+	}
+	if t := strings.TrimSpace(filter.TenantID); t != "" {
+		fmt.Fprintf(b, "Tenant filter: %s\n", t)
+	}
+	if th := strings.TrimSpace(filter.ThreadID); th != "" {
+		fmt.Fprintf(b, "Thread filter: %s\n", th)
+	}
+	if a := strings.TrimSpace(filter.ActorID); a != "" {
+		fmt.Fprintf(b, "Actor filter: %s\n", a)
+	}
+}
+
 func writeLogsRunList(cmd *cobra.Command, ctx context.Context, st *sqlite.Store, dsn string, g *Global) error {
 	runs, err := st.ListRecentRuns(ctx, state.DefaultRunListLimit)
 	if err != nil {
@@ -186,10 +226,11 @@ func writeLogsRunList(cmd *cobra.Command, ctx context.Context, st *sqlite.Store,
 			return err
 		}
 		w := tabwriter.NewWriter(&b, 0, 0, 2, ' ', 0)
-		fmt.Fprintln(w, "RUN ID\tWORKFLOW\tENV\tSTATUS\tSTARTED")
+		fmt.Fprintln(w, "RUN ID\tWORKFLOW\tENV\tSTATUS\tTENANT\tTHREAD\tACTOR\tSTARTED")
 		for _, r := range runs {
-			fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n",
-				r.RunID, r.WorkflowName, r.Env, r.Status, r.StartedAt.UTC().Format(time.RFC3339),
+			fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n",
+				r.RunID, r.WorkflowName, r.Env, r.Status, r.TenantID, r.ThreadID, r.ActorID,
+				r.StartedAt.UTC().Format(time.RFC3339),
 			)
 		}
 		if err := w.Flush(); err != nil {

internal/cli/logs_attribution_test.go

@@ -0,0 +1,119 @@
+package cli
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"io"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/LAA-Software-Engineering/agentic-control-plane/internal/state"
+	"github.com/LAA-Software-Engineering/agentic-control-plane/internal/state/sqlite"
+	"github.com/LAA-Software-Engineering/agentic-control-plane/internal/statejson"
+)
+
+func TestLogs_filterByTenantAndThread(t *testing.T) {
+	ctx := context.Background()
+	db := filepath.Join(t.TempDir(), "logs-filter.db")
+	st, err := sqlite.Open(ctx, db)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = st.Close() })
+
+	start := time.Date(2026, 6, 4, 10, 0, 0, 0, time.UTC)
+	for _, r := range []state.Run{
+		{RunID: "match", WorkflowName: "wf", Env: "local", Status: "succeeded", StartedAt: start,
+			InputJSON: `{}`, TenantID: "acme", ThreadID: "session-1", ActorID: "u1", RequestID: "r1", Source: "cli"},
+		{RunID: "other-thread", WorkflowName: "wf", Env: "local", Status: "succeeded", StartedAt: start.Add(time.Minute),
+			InputJSON: `{}`, TenantID: "acme", ThreadID: "session-2", ActorID: "u1", RequestID: "r2", Source: "cli"},
+	} {
+		if err := st.StartRun(ctx, r); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := st.AppendTraceEvent(ctx, r.RunID, start, "run.started", "", `{}`); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	ResetGlobalsForTest()
+	var out bytes.Buffer
+	cmd := NewRootCmd()
+	cmd.SetOut(&out)
+	cmd.SetErr(&out)
+	root := runProjRoot(t)
+	cmd.SetArgs([]string{"logs", "--tenant-id", "acme", "--thread-id", "session-1", "--project", root, "--state", db, "-o", "json"})
+	if err := cmd.Execute(); err != nil {
+		t.Fatal(err)
+	}
+
+	var payload struct {
+		Runs []struct {
+			RunID  string                       `json:"runId"`
+			Events []statejson.TraceEventRecord `json:"events"`
+		} `json:"runs"`
+	}
+	if err := json.Unmarshal(out.Bytes(), &payload); err != nil {
+		t.Fatalf("json: %v\n%s", err, out.String())
+	}
+	if len(payload.Runs) != 1 || payload.Runs[0].RunID != "match" {
+		t.Fatalf("runs: %+v", payload.Runs)
+	}
+	if len(payload.Runs[0].Events) != 1 || payload.Runs[0].Events[0].TenantID != "acme" || payload.Runs[0].Events[0].ThreadID != "session-1" {
+		t.Fatalf("events: %+v", payload.Runs[0].Events)
+	}
+}
+
+func TestLogs_runListShowsAttributionColumns(t *testing.T) {
+	ctx := context.Background()
+	db := filepath.Join(t.TempDir(), "logs-table.db")
+	st, err := sqlite.Open(ctx, db)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = st.Close() })
+
+	start := time.Now().UTC()
+	if err := st.StartRun(ctx, state.Run{
+		RunID: "r1", WorkflowName: "wf", Env: "local", Status: "running", StartedAt: start,
+		InputJSON: `{}`, TenantID: "t", ThreadID: "th", ActorID: "a", RequestID: "req", Source: "cli",
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	ResetGlobalsForTest()
+	var out bytes.Buffer
+	cmd := NewRootCmd()
+	cmd.SetOut(&out)
+	cmd.SetErr(io.Discard)
+	root := runProjRoot(t)
+	cmd.SetArgs([]string{"logs", "--project", root, "--state", db})
+	if err := cmd.Execute(); err != nil {
+		t.Fatal(err)
+	}
+	text := out.String()
+	if !strings.Contains(text, "TENANT") || !strings.Contains(text, "THREAD") || !strings.Contains(text, "ACTOR") {
+		t.Fatalf("table headers:\n%s", text)
+	}
+	if !strings.Contains(text, "r1") || !strings.Contains(text, " t ") || !strings.Contains(text, " th ") || !strings.Contains(text, " a ") {
+		t.Fatalf("table values:\n%s", text)
+	}
+}
+
+func TestLogs_rejectsRunWithTenantFilter(t *testing.T) {
+	ResetGlobalsForTest()
+	cmd := NewRootCmd()
+	cmd.SetOut(io.Discard)
+	cmd.SetErr(io.Discard)
+	cmd.SetArgs([]string{"logs", "--run", "r1", "--tenant-id", "acme"})
+	err := cmd.Execute()
+	if err == nil {
+		t.Fatal("expected validation error")
+	}
+	if !strings.Contains(err.Error(), "cannot be combined") {
+		t.Fatalf("err = %v", err)
+	}
+}