tenseleyflow/shithub / 29771c7

Browse files

actions/events: emit workflow lifecycle webhooks (S41h)

Authored by mfwolffe <wolffemf@dukes.jmu.edu>
SHA
29771c735e8a627015d417ffe6298da1ac9412b8
Parents
582d0a5
Tree
57f0be9

12 changed files

StatusFile+-
M docs/public/api/webhooks.md 11 0
M docs/public/user/webhooks.md 14 1
A internal/actions/events/emit.go 143 0
A internal/actions/events/emit_test.go 83 0
M internal/actions/lifecycle/cancel.go 38 1
M internal/actions/queries/workflow_runs.sql 13 0
M internal/actions/sqlc/querier.go 1 0
M internal/actions/sqlc/workflow_runs.sql.go 45 0
M internal/actions/trigger/enqueue.go 54 0
M internal/actions/trigger/enqueue_test.go 24 0
M internal/auth/audit/audit.go 11 0
M internal/web/handlers/api/runners.go 114 5
docs/public/api/webhooks.mdmodified
@@ -45,6 +45,9 @@ The events shippable today, by `X-Shithub-Event` header:
4545
 - `check_run` (actions: `created`, `completed`, `rerequested`)
4646
 - `check_suite` (actions: `requested`, `completed`,
4747
   `rerequested`)
48
+- `workflow_run` (actions: `queued`, `running`, `completed`)
49
+- `workflow_job` (actions: `queued`, `running`, `completed`,
50
+  `cancelled`)
4851
 - `star`
4952
 - `fork`
5053
 - `repository` (actions: `created`, `deleted`, `archived`,
@@ -55,3 +58,11 @@ The events shippable today, by `X-Shithub-Event` header:
5558
 Each event's payload is documented per-type in the webhook detail
5659
 page's "Recent deliveries" inspector — that's currently the
5760
 authoritative reference until per-event documentation lands here.
61
+
62
+### Actions payload safety
63
+
64
+`workflow_run` and `workflow_job` payloads are structural snapshots:
65
+ids, run index, workflow path/name, head SHA/ref, event kind, status,
66
+conclusion, timestamps, job key/name, runner id, needs, timeout, and
67
+cancellation state. They intentionally do **not** include workflow
68
+event payloads, env, permissions, logs, runner JWTs, or secret values.
docs/public/user/webhooks.mdmodified
@@ -21,7 +21,8 @@ Repository → Settings → Webhooks → "Add webhook".
2121
2222
 Each delivery includes:
2323
24
-- `X-Shithub-Event: <event-name>` — e.g., `push`, `pull_request`.
24
+- `X-Shithub-Event: <event-name>` — e.g., `push`, `pull_request`,
25
+  `workflow_run`.
2526
 - `X-Shithub-Delivery: <uuid>` — unique per delivery (idempotent).
2627
 - `X-Shithub-Signature-256: sha256=<hex>` — HMAC-SHA256 of the
2728
   raw body using your configured secret.
@@ -105,6 +106,18 @@ Webhook detail page → "Recent deliveries". Each row shows:
105106
 Stored bodies are capped at 32 KiB (your endpoint can accept
106107
 bigger; we just don't keep more for the inspector).
107108
109
+## Actions events
110
+
111
+Repository webhooks can subscribe to Actions lifecycle events:
112
+
113
+- `workflow_run` actions: `queued`, `running`, `completed`.
114
+- `workflow_job` actions: `queued`, `running`, `completed`,
115
+  `cancelled`.
116
+
117
+Actions payloads only carry structural run/job metadata. shithub does
118
+not include workflow event payloads, env, permissions, logs, runner
119
+tokens, or secrets in webhook bodies.
120
+
108121
 ## SSRF defense
109122
110123
 shithub validates webhook URLs server-side: hostnames are
internal/actions/events/emit.goadded
@@ -0,0 +1,143 @@
1
+// SPDX-License-Identifier: AGPL-3.0-or-later
2
+
3
+// Package events emits webhook-facing Actions lifecycle events.
4
+//
5
+// These payloads are deliberately structural. Do not include workflow
6
+// event_payload, env, permissions, step logs, runner JWTs, or secret material.
7
+package events
8
+
9
+import (
10
+	"context"
11
+
12
+	"github.com/jackc/pgx/v5"
13
+	"github.com/jackc/pgx/v5/pgtype"
14
+
15
+	actionsdb "github.com/tenseleyFlow/shithub/internal/actions/sqlc"
16
+	"github.com/tenseleyFlow/shithub/internal/notif"
17
+)
18
+
19
+const (
20
+	KindWorkflowRun = "workflow_run"
21
+	KindWorkflowJob = "workflow_job"
22
+
23
+	ActionQueued    = "queued"
24
+	ActionRunning   = "running"
25
+	ActionCompleted = "completed"
26
+	ActionCancelled = "cancelled"
27
+)
28
+
29
+// EmitRunTx writes one workflow_run domain event inside the caller's
30
+// transaction. Use it when the run mutation and event row must commit
31
+// atomically.
32
+func EmitRunTx(ctx context.Context, tx pgx.Tx, run actionsdb.WorkflowRun, action string) error {
33
+	return notif.EmitTx(ctx, tx, runEvent(run, action))
34
+}
35
+
36
+// EmitJobTx writes one workflow_job domain event inside the caller's
37
+// transaction. The parent run snapshot is included so webhook subscribers do
38
+// not need a second API call to identify the workflow execution.
39
+func EmitJobTx(ctx context.Context, tx pgx.Tx, run actionsdb.WorkflowRun, job actionsdb.WorkflowJob, action string) error {
40
+	return notif.EmitTx(ctx, tx, jobEvent(run, job, action))
41
+}
42
+
43
+func runEvent(run actionsdb.WorkflowRun, action string) notif.Event {
44
+	return notif.Event{
45
+		ActorUserID: int8Value(run.ActorUserID),
46
+		Kind:        KindWorkflowRun,
47
+		RepoID:      run.RepoID,
48
+		SourceKind:  KindWorkflowRun,
49
+		SourceID:    run.ID,
50
+		Public:      false,
51
+		Extra: map[string]any{
52
+			"action":       action,
53
+			"workflow_run": runPayload(run),
54
+		},
55
+	}
56
+}
57
+
58
+func jobEvent(run actionsdb.WorkflowRun, job actionsdb.WorkflowJob, action string) notif.Event {
59
+	return notif.Event{
60
+		ActorUserID: int8Value(run.ActorUserID),
61
+		Kind:        KindWorkflowJob,
62
+		RepoID:      run.RepoID,
63
+		SourceKind:  KindWorkflowJob,
64
+		SourceID:    job.ID,
65
+		Public:      false,
66
+		Extra: map[string]any{
67
+			"action":       action,
68
+			"workflow_run": runPayload(run),
69
+			"workflow_job": jobPayload(job),
70
+		},
71
+	}
72
+}
73
+
74
+func runPayload(run actionsdb.WorkflowRun) map[string]any {
75
+	return map[string]any{
76
+		"id":               run.ID,
77
+		"repo_id":          run.RepoID,
78
+		"run_index":        run.RunIndex,
79
+		"workflow_file":    run.WorkflowFile,
80
+		"workflow_name":    run.WorkflowName,
81
+		"head_sha":         run.HeadSha,
82
+		"head_ref":         run.HeadRef,
83
+		"event":            string(run.Event),
84
+		"status":           string(run.Status),
85
+		"conclusion":       conclusionValue(run.Conclusion),
86
+		"actor_user_id":    int8Nullable(run.ActorUserID),
87
+		"parent_run_id":    int8Nullable(run.ParentRunID),
88
+		"created_at":       timeValue(run.CreatedAt),
89
+		"updated_at":       timeValue(run.UpdatedAt),
90
+		"started_at":       timeValue(run.StartedAt),
91
+		"completed_at":     timeValue(run.CompletedAt),
92
+		"trigger_event_id": run.TriggerEventID,
93
+	}
94
+}
95
+
96
+func jobPayload(job actionsdb.WorkflowJob) map[string]any {
97
+	return map[string]any{
98
+		"id":               job.ID,
99
+		"run_id":           job.RunID,
100
+		"job_index":        job.JobIndex,
101
+		"job_key":          job.JobKey,
102
+		"job_name":         job.JobName,
103
+		"runs_on":          job.RunsOn,
104
+		"runner_id":        int8Nullable(job.RunnerID),
105
+		"needs_jobs":       job.NeedsJobs,
106
+		"timeout_minutes":  job.TimeoutMinutes,
107
+		"status":           string(job.Status),
108
+		"conclusion":       conclusionValue(job.Conclusion),
109
+		"cancel_requested": job.CancelRequested,
110
+		"created_at":       timeValue(job.CreatedAt),
111
+		"updated_at":       timeValue(job.UpdatedAt),
112
+		"started_at":       timeValue(job.StartedAt),
113
+		"completed_at":     timeValue(job.CompletedAt),
114
+	}
115
+}
116
+
117
+func conclusionValue(v actionsdb.NullCheckConclusion) any {
118
+	if !v.Valid {
119
+		return nil
120
+	}
121
+	return string(v.CheckConclusion)
122
+}
123
+
124
+func int8Value(v pgtype.Int8) int64 {
125
+	if !v.Valid {
126
+		return 0
127
+	}
128
+	return v.Int64
129
+}
130
+
131
+func int8Nullable(v pgtype.Int8) any {
132
+	if !v.Valid {
133
+		return nil
134
+	}
135
+	return v.Int64
136
+}
137
+
138
+func timeValue(v pgtype.Timestamptz) any {
139
+	if !v.Valid {
140
+		return nil
141
+	}
142
+	return v.Time.UTC().Format("2006-01-02T15:04:05.999999999Z07:00")
143
+}
internal/actions/events/emit_test.goadded
@@ -0,0 +1,83 @@
1
+// SPDX-License-Identifier: AGPL-3.0-or-later
2
+
3
+package events
4
+
5
+import (
6
+	"encoding/json"
7
+	"strings"
8
+	"testing"
9
+	"time"
10
+
11
+	"github.com/jackc/pgx/v5/pgtype"
12
+
13
+	actionsdb "github.com/tenseleyFlow/shithub/internal/actions/sqlc"
14
+)
15
+
16
+func TestPayloadsExcludeSensitiveWorkflowState(t *testing.T) {
17
+	now := pgtype.Timestamptz{Time: time.Date(2026, 5, 12, 12, 0, 0, 0, time.UTC), Valid: true}
18
+	run := actionsdb.WorkflowRun{
19
+		ID:             11,
20
+		RepoID:         22,
21
+		RunIndex:       3,
22
+		WorkflowFile:   ".shithub/workflows/ci.yml",
23
+		WorkflowName:   "CI",
24
+		HeadSha:        strings.Repeat("a", 40),
25
+		HeadRef:        "refs/heads/trunk",
26
+		Event:          actionsdb.WorkflowRunEventPush,
27
+		EventPayload:   []byte(`{"secret":"do-not-emit"}`),
28
+		ActorUserID:    pgtype.Int8{Int64: 33, Valid: true},
29
+		Status:         actionsdb.WorkflowRunStatusRunning,
30
+		TriggerEventID: "push:demo",
31
+		CreatedAt:      now,
32
+		UpdatedAt:      now,
33
+		StartedAt:      now,
34
+	}
35
+	job := actionsdb.WorkflowJob{
36
+		ID:             44,
37
+		RunID:          run.ID,
38
+		JobIndex:       0,
39
+		JobKey:         "build",
40
+		JobName:        "Build",
41
+		RunsOn:         "ubuntu-latest",
42
+		NeedsJobs:      []string{},
43
+		TimeoutMinutes: 30,
44
+		Permissions:    []byte(`{"contents":"read"}`),
45
+		JobEnv:         []byte(`{"TOKEN":"do-not-emit"}`),
46
+		Status:         actionsdb.WorkflowJobStatusCompleted,
47
+		Conclusion:     actionsdb.NullCheckConclusion{CheckConclusion: actionsdb.CheckConclusionSuccess, Valid: true},
48
+		CreatedAt:      now,
49
+		UpdatedAt:      now,
50
+		StartedAt:      now,
51
+		CompletedAt:    now,
52
+	}
53
+
54
+	payload, err := json.Marshal(jobEvent(run, job, ActionCompleted).Extra)
55
+	if err != nil {
56
+		t.Fatalf("marshal payload: %v", err)
57
+	}
58
+	got := string(payload)
59
+	for _, forbidden := range []string{
60
+		"event_payload",
61
+		"do-not-emit",
62
+		"permissions",
63
+		"job_env",
64
+		"TOKEN",
65
+		"secret",
66
+	} {
67
+		if strings.Contains(got, forbidden) {
68
+			t.Fatalf("payload leaked %q: %s", forbidden, got)
69
+		}
70
+	}
71
+	for _, required := range []string{
72
+		`"action":"completed"`,
73
+		`"workflow_run"`,
74
+		`"workflow_job"`,
75
+		`"status":"completed"`,
76
+		`"conclusion":"success"`,
77
+		`"trigger_event_id":"push:demo"`,
78
+	} {
79
+		if !strings.Contains(got, required) {
80
+			t.Fatalf("payload missing %q: %s", required, got)
81
+		}
82
+	}
83
+}
internal/actions/lifecycle/cancel.gomodified
@@ -12,6 +12,7 @@ import (
1212
 	"github.com/jackc/pgx/v5"
1313
1414
 	"github.com/tenseleyFlow/shithub/internal/actions/checksync"
15
+	actionsevents "github.com/tenseleyFlow/shithub/internal/actions/events"
1516
 	"github.com/tenseleyFlow/shithub/internal/actions/runstate"
1617
 	actionsdb "github.com/tenseleyFlow/shithub/internal/actions/sqlc"
1718
 	"github.com/tenseleyFlow/shithub/internal/infra/metrics"
@@ -50,7 +51,8 @@ func CancelRun(ctx context.Context, deps Deps, runID int64, reason string) (Canc
5051
 		}
5152
 	}()
5253
53
-	if _, err := q.GetWorkflowRunByID(ctx, tx, runID); err != nil {
54
+	run, err := q.GetWorkflowRunByID(ctx, tx, runID)
55
+	if err != nil {
5456
 		return CancelResult{}, err
5557
 	}
5658
 	changed, err := q.RequestWorkflowRunCancel(ctx, tx, runID)
@@ -73,6 +75,13 @@ func CancelRun(ctx context.Context, deps Deps, runID int64, reason string) (Canc
7375
 		if err != nil {
7476
 			return CancelResult{}, err
7577
 		}
78
+		run, err = q.GetWorkflowRunByID(ctx, tx, runID)
79
+		if err != nil {
80
+			return CancelResult{}, err
81
+		}
82
+		if err := emitCancelEvents(ctx, tx, run, changed, runCompleted); err != nil {
83
+			return CancelResult{}, err
84
+		}
7685
 	}
7786
 	if err := tx.Commit(ctx); err != nil {
7887
 		return CancelResult{}, err
@@ -138,6 +147,13 @@ func CancelJob(ctx context.Context, deps Deps, jobID int64, reason string) (Canc
138147
 		if err != nil {
139148
 			return CancelResult{}, err
140149
 		}
150
+		run, err := q.GetWorkflowRunByID(ctx, tx, runID)
151
+		if err != nil {
152
+			return CancelResult{}, err
153
+		}
154
+		if err := emitCancelEvents(ctx, tx, run, changed, runCompleted); err != nil {
155
+			return CancelResult{}, err
156
+		}
141157
 	}
142158
 	if err := tx.Commit(ctx); err != nil {
143159
 		return CancelResult{}, err
@@ -161,6 +177,27 @@ func recordCancelledJobs(jobs []actionsdb.WorkflowJob, reason string) {
161177
 	metrics.ActionsJobsCancelledTotal.WithLabelValues(cancelReason(reason)).Add(float64(len(jobs)))
162178
 }
163179
180
+func emitCancelEvents(ctx context.Context, tx pgx.Tx, run actionsdb.WorkflowRun, jobs []actionsdb.WorkflowJob, runCompleted bool) error {
181
+	for _, job := range jobs {
182
+		if job.Status != actionsdb.WorkflowJobStatusCancelled {
183
+			continue
184
+		}
185
+		if err := actionsevents.EmitJobTx(ctx, tx, run, job, actionsevents.ActionCancelled); err != nil {
186
+			return err
187
+		}
188
+	}
189
+	if runCompleted {
190
+		action := actionsevents.ActionCompleted
191
+		if run.Status == actionsdb.WorkflowRunStatusCancelled {
192
+			action = actionsevents.ActionCancelled
193
+		}
194
+		if err := actionsevents.EmitRunTx(ctx, tx, run, action); err != nil {
195
+			return err
196
+		}
197
+	}
198
+	return nil
199
+}
200
+
164201
 func cancelReason(reason string) string {
165202
 	switch strings.TrimSpace(reason) {
166203
 	case CancelReasonUser:
internal/actions/queries/workflow_runs.sqlmodified
@@ -109,6 +109,19 @@ SET status = 'running',
109109
     updated_at = now()
110110
 WHERE id = $1 AND status = 'queued';
111111
112
+-- name: StartWorkflowRun :one
113
+UPDATE workflow_runs
114
+SET status = 'running',
115
+    started_at = COALESCE(started_at, now()),
116
+    version = version + 1,
117
+    updated_at = now()
118
+WHERE id = $1 AND status = 'queued'
119
+RETURNING id, repo_id, run_index, workflow_file, workflow_name,
120
+          head_sha, head_ref, event, event_payload,
121
+          actor_user_id, parent_run_id, concurrency_group,
122
+          status, conclusion, pinned, need_approval, approved_by_user_id,
123
+          started_at, completed_at, version, created_at, updated_at, trigger_event_id;
124
+
112125
 -- name: CompleteWorkflowRun :one
113126
 UPDATE workflow_runs
114127
 SET status = 'completed',
internal/actions/sqlc/querier.gomodified
@@ -102,6 +102,7 @@ type Querier interface {
102102
 	RequestWorkflowJobCancel(ctx context.Context, db DBTX, id int64) (WorkflowJob, error)
103103
 	RequestWorkflowRunCancel(ctx context.Context, db DBTX, runID int64) ([]WorkflowJob, error)
104104
 	RevokeAllTokensForRunner(ctx context.Context, db DBTX, runnerID int64) error
105
+	StartWorkflowRun(ctx context.Context, db DBTX, id int64) (WorkflowRun, error)
105106
 	TouchRunnerHeartbeat(ctx context.Context, db DBTX, arg TouchRunnerHeartbeatParams) error
106107
 	UpdateStepLogChunk(ctx context.Context, db DBTX, arg UpdateStepLogChunkParams) error
107108
 	UpdateWorkflowJobStatus(ctx context.Context, db DBTX, arg UpdateWorkflowJobStatusParams) (WorkflowJob, error)
internal/actions/sqlc/workflow_runs.sql.gomodified
@@ -698,3 +698,48 @@ func (q *Queries) NextRunIndexForRepo(ctx context.Context, db DBTX, repoID int64
698698
 	err := row.Scan(&next_index)
699699
 	return next_index, err
700700
 }
701
+
702
+const startWorkflowRun = `-- name: StartWorkflowRun :one
703
+UPDATE workflow_runs
704
+SET status = 'running',
705
+    started_at = COALESCE(started_at, now()),
706
+    version = version + 1,
707
+    updated_at = now()
708
+WHERE id = $1 AND status = 'queued'
709
+RETURNING id, repo_id, run_index, workflow_file, workflow_name,
710
+          head_sha, head_ref, event, event_payload,
711
+          actor_user_id, parent_run_id, concurrency_group,
712
+          status, conclusion, pinned, need_approval, approved_by_user_id,
713
+          started_at, completed_at, version, created_at, updated_at, trigger_event_id
714
+`
715
+
716
+func (q *Queries) StartWorkflowRun(ctx context.Context, db DBTX, id int64) (WorkflowRun, error) {
717
+	row := db.QueryRow(ctx, startWorkflowRun, id)
718
+	var i WorkflowRun
719
+	err := row.Scan(
720
+		&i.ID,
721
+		&i.RepoID,
722
+		&i.RunIndex,
723
+		&i.WorkflowFile,
724
+		&i.WorkflowName,
725
+		&i.HeadSha,
726
+		&i.HeadRef,
727
+		&i.Event,
728
+		&i.EventPayload,
729
+		&i.ActorUserID,
730
+		&i.ParentRunID,
731
+		&i.ConcurrencyGroup,
732
+		&i.Status,
733
+		&i.Conclusion,
734
+		&i.Pinned,
735
+		&i.NeedApproval,
736
+		&i.ApprovedByUserID,
737
+		&i.StartedAt,
738
+		&i.CompletedAt,
739
+		&i.Version,
740
+		&i.CreatedAt,
741
+		&i.UpdatedAt,
742
+		&i.TriggerEventID,
743
+	)
744
+	return i, err
745
+}
internal/actions/trigger/enqueue.gomodified
@@ -15,6 +15,7 @@ import (
1515
1616
 	"github.com/tenseleyFlow/shithub/internal/actions/checksync"
1717
 	"github.com/tenseleyFlow/shithub/internal/actions/concurrency"
18
+	actionsevents "github.com/tenseleyFlow/shithub/internal/actions/events"
1819
 	actionsdb "github.com/tenseleyFlow/shithub/internal/actions/sqlc"
1920
 	"github.com/tenseleyFlow/shithub/internal/actions/workflow"
2021
 	"github.com/tenseleyFlow/shithub/internal/checks"
@@ -184,6 +185,9 @@ func Enqueue(ctx context.Context, deps Deps, p EnqueueParams) (Result, error) {
184185
 		}
185186
 		return Result{}, fmt.Errorf("trigger: insert run: %w", err)
186187
 	}
188
+	if err := actionsevents.EmitRunTx(ctx, tx, run, actionsevents.ActionQueued); err != nil {
189
+		return Result{}, fmt.Errorf("trigger: emit run queued event: %w", err)
190
+	}
187191
188192
 	// Persist child jobs + their steps. Order in Workflow.Jobs is YAML
189193
 	// document order, which we preserve via job_index.
@@ -218,6 +222,9 @@ func Enqueue(ctx context.Context, deps Deps, p EnqueueParams) (Result, error) {
218222
 			return Result{}, fmt.Errorf("trigger: insert job %s: %w", j.Key, err)
219223
 		}
220224
 		jobIDs[i] = job.ID
225
+		if err := actionsevents.EmitJobTx(ctx, tx, run, job, actionsevents.ActionQueued); err != nil {
226
+			return Result{}, fmt.Errorf("trigger: emit job queued event for %s: %w", j.Key, err)
227
+		}
221228
222229
 		for si, s := range j.Steps {
223230
 			stepEnvJSON, err := marshalEnv(s.Env)
@@ -253,6 +260,9 @@ func Enqueue(ctx context.Context, deps Deps, p EnqueueParams) (Result, error) {
253260
 	if err != nil {
254261
 		return Result{}, fmt.Errorf("trigger: enforce concurrency: %w", err)
255262
 	}
263
+	if err := emitConcurrencyCancelEvents(ctx, tx, q, concurrencyResult.CancelledJobs); err != nil {
264
+		return Result{}, err
265
+	}
256266
257267
 	if err := tx.Commit(ctx); err != nil {
258268
 		return Result{}, fmt.Errorf("trigger: commit run tx: %w", err)
@@ -348,6 +358,50 @@ func lookupExistingRun(ctx context.Context, pool *pgxpool.Pool, p EnqueueParams)
348358
 	return rows, nil
349359
 }
350360
361
+func emitConcurrencyCancelEvents(
362
+	ctx context.Context,
363
+	tx pgx.Tx,
364
+	q *actionsdb.Queries,
365
+	jobs []actionsdb.WorkflowJob,
366
+) error {
367
+	if len(jobs) == 0 {
368
+		return nil
369
+	}
370
+	emittedRun := map[int64]struct{}{}
371
+	for _, job := range jobs {
372
+		run, err := q.GetWorkflowRunByID(ctx, tx, job.RunID)
373
+		if err != nil {
374
+			return fmt.Errorf("trigger: load concurrency-cancelled run: %w", err)
375
+		}
376
+		if job.Status == actionsdb.WorkflowJobStatusCancelled {
377
+			if err := actionsevents.EmitJobTx(ctx, tx, run, job, actionsevents.ActionCancelled); err != nil {
378
+				return fmt.Errorf("trigger: emit concurrency job cancelled event: %w", err)
379
+			}
380
+		}
381
+		if _, ok := emittedRun[run.ID]; ok {
382
+			continue
383
+		}
384
+		if workflowRunTerminal(run.Status) {
385
+			if err := actionsevents.EmitRunTx(ctx, tx, run, runTerminalAction(run)); err != nil {
386
+				return fmt.Errorf("trigger: emit concurrency run terminal event: %w", err)
387
+			}
388
+			emittedRun[run.ID] = struct{}{}
389
+		}
390
+	}
391
+	return nil
392
+}
393
+
394
+func workflowRunTerminal(status actionsdb.WorkflowRunStatus) bool {
395
+	return status == actionsdb.WorkflowRunStatusCompleted || status == actionsdb.WorkflowRunStatusCancelled
396
+}
397
+
398
+func runTerminalAction(run actionsdb.WorkflowRun) string {
399
+	if run.Status == actionsdb.WorkflowRunStatusCancelled {
400
+		return actionsevents.ActionCancelled
401
+	}
402
+	return actionsevents.ActionCompleted
403
+}
404
+
351405
 func pgInt8(v int64) pgtype.Int8 {
352406
 	return pgtype.Int8{Int64: v, Valid: v != 0}
353407
 }
internal/actions/trigger/enqueue_test.gomodified
@@ -143,6 +143,10 @@ func TestEnqueue_HappyPath(t *testing.T) {
143143
 	if run.Status != actionsdb.WorkflowRunStatusQueued {
144144
 		t.Errorf("status: got %s want queued", run.Status)
145145
 	}
146
+	assertDomainEventCounts(t, f.pool, f.repoID, map[string]int64{
147
+		"workflow_run": 1,
148
+		"workflow_job": 1,
149
+	})
146150
 }
147151
148152
 func TestEnqueue_ResolvesConcurrencyGroupExpression(t *testing.T) {
@@ -235,6 +239,26 @@ func TestEnqueue_CancelInProgressCancelsOlderQueuedRun(t *testing.T) {
235239
 	if newRun.Status != actionsdb.WorkflowRunStatusQueued {
236240
 		t.Fatalf("new run status: got %s want queued", newRun.Status)
237241
 	}
242
+	assertDomainEventCounts(t, f.pool, f.repoID, map[string]int64{
243
+		"workflow_run": 3,
244
+		"workflow_job": 3,
245
+	})
246
+}
247
+
248
+func assertDomainEventCounts(t *testing.T, db actionsdb.DBTX, repoID int64, want map[string]int64) {
249
+	t.Helper()
250
+	for kind, n := range want {
251
+		var got int64
252
+		if err := db.QueryRow(context.Background(),
253
+			`SELECT count(*) FROM domain_events WHERE repo_id = $1 AND kind = $2`,
254
+			repoID, kind,
255
+		).Scan(&got); err != nil {
256
+			t.Fatalf("count domain events %s: %v", kind, err)
257
+		}
258
+		if got != n {
259
+			t.Fatalf("domain_events[%s] = %d, want %d", kind, got, n)
260
+		}
261
+	}
238262
 }
239263
240264
 func TestClaimQueuedWorkflowJob_BlocksYoungerConcurrencyRun(t *testing.T) {
internal/auth/audit/audit.gomodified
@@ -92,6 +92,17 @@ const (
9292
 	ActionActionsVariableSet     Action = "actions_variable_set"
9393
 	ActionActionsVariableDeleted Action = "actions_variable_deleted"
9494
95
+	// S41h — Actions run/job lifecycle. Metadata must stay structural:
96
+	// run/job ids, status, conclusion, workflow path/name. Never include
97
+	// event payloads, env, logs, permissions, tokens, or secret values.
98
+	ActionWorkflowRunCreated   Action = "workflow_run_created"
99
+	ActionWorkflowRunStarted   Action = "workflow_run_started"
100
+	ActionWorkflowRunCompleted Action = "workflow_run_completed"
101
+	ActionWorkflowJobCreated   Action = "workflow_job_created"
102
+	ActionWorkflowJobStarted   Action = "workflow_job_started"
103
+	ActionWorkflowJobCompleted Action = "workflow_job_completed"
104
+	ActionWorkflowJobCancelled Action = "workflow_job_cancelled"
105
+
95106
 	// S34 — site admin actions. Always recorded with the real admin's
96107
 	// id in actor_id; impersonation flows additionally carry the
97108
 	// impersonated user's id in meta.impersonated_user_id.
internal/web/handlers/api/runners.gomodified
@@ -20,6 +20,7 @@ import (
2020
 	"github.com/jackc/pgx/v5"
2121
 	"github.com/jackc/pgx/v5/pgtype"
2222
23
+	actionsevents "github.com/tenseleyFlow/shithub/internal/actions/events"
2324
 	"github.com/tenseleyFlow/shithub/internal/actions/finalize"
2425
 	actionslifecycle "github.com/tenseleyFlow/shithub/internal/actions/lifecycle"
2526
 	"github.com/tenseleyFlow/shithub/internal/actions/logstream"
@@ -220,7 +221,20 @@ func (h *Handlers) claimRunnerJob(
220221
 		committed = true
221222
 		return actionsdb.ClaimQueuedWorkflowJobRow{}, nil, nil, false, nil
222223
 	}
223
-	if err := q.MarkWorkflowRunRunning(ctx, tx, job.RunID); err != nil {
224
+	run, err := q.StartWorkflowRun(ctx, tx, job.RunID)
225
+	if err == nil {
226
+		if err := actionsevents.EmitRunTx(ctx, tx, run, actionsevents.ActionRunning); err != nil {
227
+			return actionsdb.ClaimQueuedWorkflowJobRow{}, nil, nil, false, err
228
+		}
229
+	} else if errors.Is(err, pgx.ErrNoRows) {
230
+		run, err = q.GetWorkflowRunByID(ctx, tx, job.RunID)
231
+		if err != nil {
232
+			return actionsdb.ClaimQueuedWorkflowJobRow{}, nil, nil, false, err
233
+		}
234
+	} else {
235
+		return actionsdb.ClaimQueuedWorkflowJobRow{}, nil, nil, false, err
236
+	}
237
+	if err := actionsevents.EmitJobTx(ctx, tx, run, claimRowWorkflowJob(job), actionsevents.ActionRunning); err != nil {
224238
 		return actionsdb.ClaimQueuedWorkflowJobRow{}, nil, nil, false, err
225239
 	}
226240
 	steps, err := q.ListRunnerStepsForJob(ctx, tx, job.ID)
@@ -733,15 +747,45 @@ func (h *Handlers) applyJobStatus(
733747
 		return actionsdb.WorkflowJob{}, false, "", err
734748
 	}
735749
 	runConclusion, complete := deriveWorkflowRunConclusion(jobs)
750
+	runAfter, err := q.GetWorkflowRunByID(ctx, tx, updated.RunID)
751
+	if err != nil {
752
+		return actionsdb.WorkflowJob{}, false, "", err
753
+	}
754
+	runBefore := runAfter
755
+	runStarted := false
756
+	runTerminalChanged := false
736757
 	if complete {
737
-		if _, err := q.CompleteWorkflowRun(ctx, tx, actionsdb.CompleteWorkflowRunParams{
758
+		runAfter, err = q.CompleteWorkflowRun(ctx, tx, actionsdb.CompleteWorkflowRunParams{
738759
 			ID:         updated.RunID,
739760
 			Conclusion: runConclusion,
740
-		}); err != nil {
761
+		})
762
+		if err != nil {
763
+			return actionsdb.WorkflowJob{}, false, "", err
764
+		}
765
+		runTerminalChanged = workflowRunLifecycleChanged(runBefore, runAfter)
766
+	} else {
767
+		startedRun, err := q.StartWorkflowRun(ctx, tx, updated.RunID)
768
+		if err == nil {
769
+			runAfter = startedRun
770
+			runStarted = true
771
+		} else if !errors.Is(err, pgx.ErrNoRows) {
772
+			return actionsdb.WorkflowJob{}, false, "", err
773
+		}
774
+	}
775
+	if jobLifecycleChanged(job, updated) {
776
+		if err := actionsevents.EmitJobTx(ctx, tx, runAfter, updated, workflowJobEventAction(updated.Status)); err != nil {
777
+			return actionsdb.WorkflowJob{}, false, "", err
778
+		}
779
+	}
780
+	if runStarted {
781
+		if err := actionsevents.EmitRunTx(ctx, tx, runAfter, actionsevents.ActionRunning); err != nil {
782
+			return actionsdb.WorkflowJob{}, false, "", err
783
+		}
784
+	}
785
+	if complete && runTerminalChanged {
786
+		if err := actionsevents.EmitRunTx(ctx, tx, runAfter, workflowRunEventAction(runAfter.Status)); err != nil {
741787
 			return actionsdb.WorkflowJob{}, false, "", err
742788
 		}
743
-	} else if err := q.MarkWorkflowRunRunning(ctx, tx, updated.RunID); err != nil {
744
-		return actionsdb.WorkflowJob{}, false, "", err
745789
 	}
746790
 	if err := tx.Commit(ctx); err != nil {
747791
 		return actionsdb.WorkflowJob{}, false, "", err
@@ -755,6 +799,71 @@ func (h *Handlers) applyJobStatus(
755799
 	return updated, complete, runConclusion, nil
756800
 }
757801
802
+func claimRowWorkflowJob(row actionsdb.ClaimQueuedWorkflowJobRow) actionsdb.WorkflowJob {
803
+	return actionsdb.WorkflowJob{
804
+		ID:              row.ID,
805
+		RunID:           row.RunID,
806
+		JobIndex:        row.JobIndex,
807
+		JobKey:          row.JobKey,
808
+		JobName:         row.JobName,
809
+		RunsOn:          row.RunsOn,
810
+		RunnerID:        row.RunnerID,
811
+		NeedsJobs:       row.NeedsJobs,
812
+		IfExpr:          row.IfExpr,
813
+		TimeoutMinutes:  row.TimeoutMinutes,
814
+		Permissions:     row.Permissions,
815
+		JobEnv:          row.JobEnv,
816
+		Status:          row.Status,
817
+		Conclusion:      row.Conclusion,
818
+		CancelRequested: row.CancelRequested,
819
+		StartedAt:       row.StartedAt,
820
+		CompletedAt:     row.CompletedAt,
821
+		Version:         row.Version,
822
+		CreatedAt:       row.CreatedAt,
823
+		UpdatedAt:       row.UpdatedAt,
824
+	}
825
+}
826
+
827
+func jobLifecycleChanged(before, after actionsdb.WorkflowJob) bool {
828
+	if before.Status != after.Status {
829
+		return true
830
+	}
831
+	if before.Conclusion.Valid != after.Conclusion.Valid {
832
+		return true
833
+	}
834
+	return before.Conclusion.Valid && before.Conclusion.CheckConclusion != after.Conclusion.CheckConclusion
835
+}
836
+
837
+func workflowRunLifecycleChanged(before, after actionsdb.WorkflowRun) bool {
838
+	if before.Status != after.Status {
839
+		return true
840
+	}
841
+	if before.Conclusion.Valid != after.Conclusion.Valid {
842
+		return true
843
+	}
844
+	return before.Conclusion.Valid && before.Conclusion.CheckConclusion != after.Conclusion.CheckConclusion
845
+}
846
+
847
+func workflowJobEventAction(status actionsdb.WorkflowJobStatus) string {
848
+	switch status {
849
+	case actionsdb.WorkflowJobStatusCancelled:
850
+		return actionsevents.ActionCancelled
851
+	case actionsdb.WorkflowJobStatusCompleted, actionsdb.WorkflowJobStatusSkipped:
852
+		return actionsevents.ActionCompleted
853
+	case actionsdb.WorkflowJobStatusRunning:
854
+		return actionsevents.ActionRunning
855
+	default:
856
+		return actionsevents.ActionQueued
857
+	}
858
+}
859
+
860
+func workflowRunEventAction(status actionsdb.WorkflowRunStatus) string {
861
+	if status == actionsdb.WorkflowRunStatusCancelled {
862
+		return actionsevents.ActionCancelled
863
+	}
864
+	return actionsevents.ActionCompleted
865
+}
866
+
758867
 func deriveWorkflowRunConclusion(jobs []actionsdb.ListJobsForRunRow) (actionsdb.CheckConclusion, bool) {
759868
 	if len(jobs) == 0 {
760869
 		return actionsdb.CheckConclusionFailure, true