fix(attribution): address PR #141 review — guardrails, tests, request_id

Add AGENTCTL_REQUIRE_ATTRIBUTION/--require-attribution guardrail and stderr
warning when local defaults apply; env overrides for tenant/thread/actor.
Consolidate request_id generation in StartRun (UUID). Document idempotency
as stored-only metadata. Remove redundant GetRun in executeEngine; add resume
attribution, OTel span attribute, and actor-only filter tests.

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

Author: leo-aa88

CHANGELOG.md

@@ -21,6 +21,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Changed
 
+- **`agentctl logs` table output** (issue #111): the default (non-JSON) run list adds `TENANT`, `THREAD`, and `ACTOR` columns. Scripts that parse fixed column positions should switch to `-o json` or match by header names.
 - **Breaking — tool calls without explicit policy are no longer unrestricted.** Previously, `CheckToolCall` with a nil [spec.PolicySpec] allowed all tools. Now fail-closed safety always applies from the project graph (even when the workflow omits `spec.policy` or the Policy resource is missing).
 - Tools with **no** `spec.safety` block behave as **untrusted with side effects** after normalization → require `--approve` unless an explicit `approvals.requiredFor` rule matches.
 

docs/ATTRIBUTION.md

@@ -11,7 +11,7 @@ Issue [#111](https://github.com/LAA-Software-Engineering/agentic-control-plane/i
 | `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 |
+| `idempotency_key` | Client reference key (stored only; dedupe is not enforced yet) |
 | `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.
@@ -25,7 +25,14 @@ When flags are omitted, `agentctl run` stores:
 - `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`.
+**Do not rely on these defaults in CI or production.** A stderr warning is emitted when defaults apply. For CI/prod, pass real actor ids, set env vars, or enable the guardrail:
+
+```bash
+export AGENTCTL_REQUIRE_ATTRIBUTION=1
+# or: agentctl run ... --require-attribution --tenant-id ... --thread-id ... --actor-id ...
+```
+
+Env overrides when flags are omitted: `AGENTCTL_TENANT_ID`, `AGENTCTL_THREAD_ID`, `AGENTCTL_ACTOR_ID`.
 
 ```bash
 agentctl run workflow/demo \
@@ -58,7 +65,15 @@ agentctl logs --tenant-id acme --thread-id prod-review-42
 
 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).
 
+## request_id
+
+When omitted, [state.RuntimeStore.StartRun] assigns a new UUID via `util.NewRequestID()`. Legacy rows migrated from pre-005 databases may have `request_id == run_id`.
+
+## Idempotency key
+
+`idempotency_key` is persisted and exposed in JSON for future dedupe. There is no unique index or at-most-once enforcement in this release — do not assume idempotent execution from the key alone.
+
 ## 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.
+- `actor_id` is supplied by the caller and is not authenticated in this release. Do not use attribution for access control.

internal/cli/attribution.go

@@ -0,0 +1,77 @@
+package cli
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"strings"
+
+	"github.com/LAA-Software-Engineering/agentic-control-plane/internal/runtime"
+	"github.com/LAA-Software-Engineering/agentic-control-plane/internal/state"
+)
+
+// EnvRequireAttribution, when set to a truthy value, requires explicit tenant/thread/actor ids on run.
+const EnvRequireAttribution = "AGENTCTL_REQUIRE_ATTRIBUTION"
+
+// EnvTenantID overrides --tenant-id when the flag is omitted.
+const EnvTenantID = "AGENTCTL_TENANT_ID"
+
+// EnvThreadID overrides --thread-id when the flag is omitted.
+const EnvThreadID = "AGENTCTL_THREAD_ID"
+
+// EnvActorID overrides --actor-id when the flag is omitted.
+const EnvActorID = "AGENTCTL_ACTOR_ID"
+
+func resolveRunAttributionFlags(tenantID, threadID, actorID, parentRunID, requestID, idempotencyKey, source string, requireFlag bool) runtime.WorkflowRunOptions {
+	return runtime.WorkflowRunOptions{
+		TenantID:           firstNonEmpty(strings.TrimSpace(tenantID), os.Getenv(EnvTenantID)),
+		ThreadID:           firstNonEmpty(strings.TrimSpace(threadID), os.Getenv(EnvThreadID)),
+		ActorID:            firstNonEmpty(strings.TrimSpace(actorID), os.Getenv(EnvActorID)),
+		ParentRunID:        strings.TrimSpace(parentRunID),
+		RequestID:          strings.TrimSpace(requestID),
+		IdempotencyKey:     strings.TrimSpace(idempotencyKey),
+		Source:             strings.TrimSpace(source),
+		RequireAttribution: requireFlag || envTruthy(EnvRequireAttribution),
+	}
+}
+
+func applyRunAttributionOpts(opts *runtime.WorkflowRunOptions, tenantID, threadID, actorID, parentRunID, requestID, idempotencyKey, source string, requireFlag bool) {
+	if opts == nil {
+		return
+	}
+	resolved := resolveRunAttributionFlags(tenantID, threadID, actorID, parentRunID, requestID, idempotencyKey, source, requireFlag)
+	opts.TenantID = resolved.TenantID
+	opts.ThreadID = resolved.ThreadID
+	opts.ActorID = resolved.ActorID
+	opts.ParentRunID = resolved.ParentRunID
+	opts.RequestID = resolved.RequestID
+	opts.IdempotencyKey = resolved.IdempotencyKey
+	opts.Source = resolved.Source
+	opts.RequireAttribution = resolved.RequireAttribution
+}
+
+// warnAttributionDefaults writes a one-line stderr warning when local attribution defaults apply.
+func warnAttributionDefaults(w io.Writer, attr state.RunAttribution) {
+	if w == nil || !state.UsesAttributionDefaults(attr) {
+		return
+	}
+	_, _ = fmt.Fprintf(w, "warning: run attribution using local defaults (tenant-1/thread-1/user-1); set --tenant-id, --thread-id, and --actor-id or AGENTCTL_REQUIRE_ATTRIBUTION=1 in CI/prod\n")
+}
+
+func firstNonEmpty(values ...string) string {
+	for _, v := range values {
+		if strings.TrimSpace(v) != "" {
+			return strings.TrimSpace(v)
+		}
+	}
+	return ""
+}
+
+func envTruthy(name string) bool {
+	switch strings.ToLower(strings.TrimSpace(os.Getenv(name))) {
+	case "1", "true", "yes", "on":
+		return true
+	default:
+		return false
+	}
+}

internal/cli/attribution_test.go

@@ -0,0 +1,122 @@
+package cli
+
+import (
+	"bytes"
+	"io"
+	"strings"
+	"testing"
+
+	"github.com/LAA-Software-Engineering/agentic-control-plane/internal/state"
+)
+
+func TestResolveRunAttributionFlags_envOverrides(t *testing.T) {
+	t.Setenv(EnvTenantID, "env-tenant")
+	t.Setenv(EnvThreadID, "env-thread")
+	t.Setenv(EnvActorID, "env-actor")
+
+	got := resolveRunAttributionFlags("", "", "", "", "", "", "", false)
+	if got.TenantID != "env-tenant" || got.ThreadID != "env-thread" || got.ActorID != "env-actor" {
+		t.Fatalf("env overrides: %+v", got)
+	}
+
+	got = resolveRunAttributionFlags("flag-tenant", "", "", "", "", "", "", false)
+	if got.TenantID != "flag-tenant" {
+		t.Fatalf("flag wins: %+v", got)
+	}
+}
+
+func TestWarnAttributionDefaults(t *testing.T) {
+	var buf bytes.Buffer
+	warnAttributionDefaults(&buf, state.RunAttribution{})
+	if !strings.Contains(buf.String(), "warning:") || !strings.Contains(buf.String(), "tenant-1") {
+		t.Fatalf("warn: %q", buf.String())
+	}
+	buf.Reset()
+	warnAttributionDefaults(&buf, state.RunAttribution{TenantID: "t", ThreadID: "th", ActorID: "a"})
+	if buf.Len() != 0 {
+		t.Fatalf("no warn expected: %q", buf.String())
+	}
+}
+
+func TestRun_requireAttributionRejectsDefaults(t *testing.T) {
+	root := runProjRoot(t)
+	db := t.TempDir() + "/req.db"
+
+	ResetGlobalsForTest()
+	cmd := NewRootCmd()
+	cmd.SetOut(io.Discard)
+	var errBuf bytes.Buffer
+	cmd.SetErr(&errBuf)
+	cmd.SetArgs([]string{
+		"run", "workflow/demo",
+		"--project", root,
+		"--state", db,
+		"--input", "topic=x",
+		"--require-attribution",
+	})
+	err := cmd.Execute()
+	if err == nil {
+		t.Fatal("expected validation error")
+	}
+	if !strings.Contains(err.Error(), "attribution required") {
+		t.Fatalf("err = %v", err)
+	}
+}
+
+func TestRun_requireAttributionViaEnv(t *testing.T) {
+	t.Setenv(EnvRequireAttribution, "1")
+	root := runProjRoot(t)
+	db := t.TempDir() + "/req-env.db"
+
+	ResetGlobalsForTest()
+	cmd := NewRootCmd()
+	cmd.SetOut(io.Discard)
+	cmd.SetErr(io.Discard)
+	cmd.SetArgs([]string{
+		"run", "workflow/demo",
+		"--project", root,
+		"--state", db,
+		"--input", "topic=x",
+	})
+	err := cmd.Execute()
+	if err == nil {
+		t.Fatal("expected validation error")
+	}
+	if !strings.Contains(err.Error(), "attribution required") {
+		t.Fatalf("err = %v", err)
+	}
+}
+
+func TestRun_warnsOnDefaultAttribution(t *testing.T) {
+	root := runProjRoot(t)
+	db := t.TempDir() + "/warn.db"
+
+	ResetGlobalsForTest()
+	cmd := NewRootCmd()
+	cmd.SetOut(io.Discard)
+	var errBuf bytes.Buffer
+	cmd.SetErr(&errBuf)
+	cmd.SetArgs([]string{
+		"run", "workflow/demo",
+		"--project", root,
+		"--state", db,
+		"--input", "topic=warn-test",
+	})
+	if err := cmd.Execute(); err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(errBuf.String(), "warning:") {
+		t.Fatalf("stderr = %q", errBuf.String())
+	}
+}
+
+func TestEnvTruthy(t *testing.T) {
+	t.Setenv("AGENTCTL_TEST_TRUTHY", "yes")
+	if !envTruthy("AGENTCTL_TEST_TRUTHY") {
+		t.Fatal("expected truthy")
+	}
+	t.Setenv("AGENTCTL_TEST_TRUTHY", "0")
+	if envTruthy("AGENTCTL_TEST_TRUTHY") {
+		t.Fatal("expected false")
+	}
+}

internal/cli/logs_attribution_test.go

@@ -103,6 +103,50 @@ func TestLogs_runListShowsAttributionColumns(t *testing.T) {
 	}
 }
 
+func TestLogs_filterByActorOnly(t *testing.T) {
+	ctx := context.Background()
+	db := filepath.Join(t.TempDir(), "logs-actor.db")
+	st, err := sqlite.Open(ctx, db)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = st.Close() })
+
+	start := time.Now().UTC()
+	for _, r := range []state.Run{
+		{RunID: "a", WorkflowName: "wf", Env: "local", Status: "succeeded", StartedAt: start,
+			InputJSON: `{}`, TenantID: "t1", ThreadID: "th1", ActorID: "target", RequestID: "r1", Source: "cli"},
+		{RunID: "b", WorkflowName: "wf", Env: "local", Status: "succeeded", StartedAt: start,
+			InputJSON: `{}`, TenantID: "t1", ThreadID: "th1", ActorID: "other", RequestID: "r2", Source: "cli"},
+	} {
+		if err := st.StartRun(ctx, r); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	root := runProjRoot(t)
+	ResetGlobalsForTest()
+	var out bytes.Buffer
+	cmd := NewRootCmd()
+	cmd.SetOut(&out)
+	cmd.SetErr(&out)
+	cmd.SetArgs([]string{"logs", "--actor-id", "target", "--project", root, "--state", db, "-o", "json"})
+	if err := cmd.Execute(); err != nil {
+		t.Fatal(err)
+	}
+	var payload struct {
+		Runs []struct {
+			RunID string `json:"runId"`
+		} `json:"runs"`
+	}
+	if err := json.Unmarshal(out.Bytes(), &payload); err != nil {
+		t.Fatal(err)
+	}
+	if len(payload.Runs) != 1 || payload.Runs[0].RunID != "a" {
+		t.Fatalf("runs: %+v", payload.Runs)
+	}
+}
+
 func TestLogs_rejectsRunWithTenantFilter(t *testing.T) {
 	ResetGlobalsForTest()
 	cmd := NewRootCmd()

internal/cli/run.go

@@ -37,6 +37,7 @@ func newRunCmd() *cobra.Command {
 	var requestID string
 	var idempotencyKey string
 	var source string
+	var requireAttribution bool
 
 	cmd := &cobra.Command{
 		Use:          "run workflow/<name>",
@@ -54,8 +55,10 @@ When a run pauses for human approval, resume with --decision and related flags,
 --auto-approve / AGENTCTL_AUTO_APPROVE=1 for non-interactive approval.
 
 Attribution flags (--tenant-id, --thread-id, --actor-id) scope runs for multi-tenant logs and
-compliance. When omitted, local defaults apply (tenant-1 / thread-1 / user-1). Never rely on
-defaults in CI or production; pass real actor ids and include tenant/env in thread_id.
+compliance. When omitted, local defaults apply (tenant-1 / thread-1 / user-1) with a stderr
+warning. Never rely on defaults in CI or production; pass real actor ids, set
+AGENTCTL_REQUIRE_ATTRIBUTION=1, or use --require-attribution. Env overrides: AGENTCTL_TENANT_ID,
+AGENTCTL_THREAD_ID, AGENTCTL_ACTOR_ID. The idempotency-key field is stored metadata only (no dedupe yet).
 
 Examples:
   agentctl run workflow/demo --input topic=hello
@@ -92,7 +95,7 @@ Exit codes (section 11.2):
 				}
 			}
 			return runRun(cmd, wfName, resumeRunID, inputFile, inputPairs, approves, autoApprove, decision, decisionEditJSON, decisionSwitchTarget,
-				tenantID, threadID, actorID, parentRunID, requestID, idempotencyKey, source)
+				tenantID, threadID, actorID, parentRunID, requestID, idempotencyKey, source, requireAttribution)
 		},
 	}
 	cmd.Flags().StringVar(&inputFile, "input-file", "", "path to JSON file with workflow input object")
@@ -108,8 +111,9 @@ Exit codes (section 11.2):
 	cmd.Flags().StringVar(&actorID, "actor-id", "", "who triggered this run (default: user-1; use a real principal in CI/prod)")
 	cmd.Flags().StringVar(&parentRunID, "parent-run-id", "", "origin run for sub-runs (not used for --resume of the same run)")
 	cmd.Flags().StringVar(&requestID, "request-id", "", "per-invocation correlation id (generated when omitted)")
-	cmd.Flags().StringVar(&idempotencyKey, "idempotency-key", "", "optional dedupe key for accidental re-triggers")
+	cmd.Flags().StringVar(&idempotencyKey, "idempotency-key", "", "client reference key stored on the run (dedupe not enforced yet)")
 	cmd.Flags().StringVar(&source, "source", "", "run origin label (default: cli)")
+	cmd.Flags().BoolVar(&requireAttribution, "require-attribution", false, "require explicit --tenant-id, --thread-id, and --actor-id (or set AGENTCTL_REQUIRE_ATTRIBUTION=1)")
 	return cmd
 }
 
@@ -172,6 +176,9 @@ func classifyRunError(err error) int {
 	if _, ok := policy.AsDenied(err); ok {
 		return ExitPolicyDenied
 	}
+	if errors.Is(err, state.ErrAttributionRequired) {
+		return ExitValidationError
+	}
 	msg := err.Error()
 	switch {
 	case strings.Contains(msg, "validate project"):
@@ -198,7 +205,7 @@ func classifyRunError(err error) int {
 }
 
 func runRun(cmd *cobra.Command, wfName, resumeRunID, inputFile string, inputPairs, approves []string, autoApprove bool, decision, decisionEditJSON, decisionSwitchTarget string,
-	tenantID, threadID, actorID, parentRunID, requestID, idempotencyKey, source string) error {
+	tenantID, threadID, actorID, parentRunID, requestID, idempotencyKey, source string, requireAttribution bool) error {
 	ctx := context.Background()
 	g := Globals()
 
@@ -248,13 +255,10 @@ func runRun(cmd *cobra.Command, wfName, resumeRunID, inputFile string, inputPair
 			RunID:           resumeID,
 		}
 		if resumeID == "" {
-			opts.TenantID = strings.TrimSpace(tenantID)
-			opts.ThreadID = strings.TrimSpace(threadID)
-			opts.ActorID = strings.TrimSpace(actorID)
-			opts.ParentRunID = strings.TrimSpace(parentRunID)
-			opts.RequestID = strings.TrimSpace(requestID)
-			opts.IdempotencyKey = strings.TrimSpace(idempotencyKey)
-			opts.Source = strings.TrimSpace(source)
+			applyRunAttributionOpts(&opts, tenantID, threadID, actorID, parentRunID, requestID, idempotencyKey, source, requireAttribution)
+			warnAttributionDefaults(cmd.ErrOrStderr(), state.RunAttribution{
+				TenantID: opts.TenantID, ThreadID: opts.ThreadID, ActorID: opts.ActorID,
+			})
 		}
 		if err := applyHitlRunOptions(&opts, resumeID != "", autoApprove, decision, decisionEditJSON, decisionSwitchTarget); err != nil {
 			return NewExitError(ExitValidationError, err)