diff --git a/internal/app/intentflow/emit.go b/internal/app/intentflow/emit.go
index cda3d0e4..3f6c0e63 100644
--- a/internal/app/intentflow/emit.go
+++ b/internal/app/intentflow/emit.go
@@ -83,10 +83,27 @@ func (p *Plans) emitFact(ctx context.Context, plan *Plan, stepID string, now tim
 		return err
 	}
 	eventID := fmt.Sprintf("intent-plan:%s:%s:%d:%s", plan.PlanID, name, plan.Revision, stepID)
+	aggType := "intent_plan"
+	aggID := plan.PlanID
+	aggVer := plan.Revision
+	if stepID != "" {
+		// Step facts must not collide with plan-revision unique (aggregate_type, id, version).
+		aggType = "intent_plan_step"
+		aggID = stepID
+		aggVer = 1
+		for _, s := range plan.Steps {
+			if s.StepID == stepID {
+				if s.Version > 0 {
+					aggVer = s.Version
+				}
+				break
+			}
+		}
+	}
 	env := events.Envelope{
 		EventID: eventID, EventName: name, SchemaVersion: events.EnvelopeSchema,
 		Environment: plan.Principal.Environment, ProjectID: plan.Principal.ProjectID,
-		AggregateType: "intent_plan", AggregateID: plan.PlanID, AggregateVersion: plan.Revision,
+		AggregateType: aggType, AggregateID: aggID, AggregateVersion: aggVer,
 		ActorRef: fmt.Sprintf("user:%d", plan.Principal.UserID), IdempotencyKey: eventID,
 		PayloadDigest: events.Digest(body), Producer: "app/intentflow",
 		OccurredAt: now, Payload: body,
diff --git a/internal/app/intentflow/pgsink_advance_test.go b/internal/app/intentflow/pgsink_advance_test.go
new file mode 100644
index 00000000..0a7255e4
--- /dev/null
+++ b/internal/app/intentflow/pgsink_advance_test.go
@@ -0,0 +1,74 @@
+package intentflow_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/app/intentflow"
+	"chat-agent-system/internal/app/uow"
+	"chat-agent-system/internal/db"
+	"chat-agent-system/internal/events"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+func TestAdvanceEligible_WithPGSinkWritesReceipt(t *testing.T) {
+	if testing.Short() {
+		t.Skip("short")
+	}
+	dsn := db.SkipIfNoTestDatabaseURL(t)
+	ctx := context.Background()
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(pool.Close)
+	owner := intentflow.NewFakeActionOwner()
+	plans := intentflow.NewPlans(uow.NewPG(pool), intentflow.NewRepository(), intentflow.Options{
+		Owner:  owner,
+		Events: events.PGSink{},
+		Now:    func() time.Time { return time.Date(2026, 9, 16, 12, 0, 0, 0, time.UTC) },
+	})
+	prin := intentflow.Principal{UserID: 1, Environment: "test", ProjectID: 992, SessionID: "sess-adv-sink2"}
+	created, err := plans.CreatePlan(ctx, intentflow.CreateCommand{
+		Principal: prin, ClientTurnID: "adv-sink-2", RequestDigest: "d-sink2", WritePlan: true,
+		ExecutionMode: intentflow.ModeLive,
+		Steps: []intentflow.StepDraft{
+			{ActionKey: "regulation.query", ActionDigest: "q", ExecutionMode: intentflow.ModeLive},
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DELETE FROM intent_plan_decisions WHERE plan_id=$1`, created.PlanID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM intent_plan_steps WHERE plan_id=$1`, created.PlanID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM intent_plans WHERE plan_id=$1`, created.PlanID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM event_deliveries WHERE event_id LIKE $1`, "intent-plan:"+created.PlanID+"%")
+		_, _ = pool.Exec(context.Background(), `DELETE FROM event_envelopes WHERE aggregate_id=$1`, created.PlanID)
+	})
+	confirmed, err := plans.DecidePlan(ctx, intentflow.DecisionCommand{
+		Principal: prin, PlanID: created.PlanID, ExpectedRevision: created.Revision,
+		DecisionNonce: "n-sink2", DecisionDigest: "dd-sink2", Kind: intentflow.DecisionConfirm,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	claimed, err := plans.ClaimStep(ctx, "w-sink", confirmed.PlanID, confirmed.Steps[0].StepID)
+	if err != nil || claimed == nil {
+		t.Fatalf("claim: %v %+v", err, claimed)
+	}
+	t.Logf("claimed fence=%d status=%s", claimed.FencingToken, claimed.Step.Status)
+	view, err := plans.ExecuteClaimed(ctx, claimed)
+	t.Logf("execute view=%+v err=%v", view, err)
+	if err != nil {
+		t.Fatalf("execute: %v", err)
+	}
+	var status, ref string
+	_ = pool.QueryRow(ctx, `SELECT status, result_ref FROM intent_plan_steps WHERE step_id=$1`, confirmed.Steps[0].StepID).Scan(&status, &ref)
+	t.Logf("db status=%s ref=%s", status, ref)
+	if status != "succeeded" {
+		t.Fatalf("want succeeded got %s", status)
+	}
+}
diff --git a/internal/app/intentflow/repository.go b/internal/app/intentflow/repository.go
index a4ca9e85..b83a2d60 100644
--- a/internal/app/intentflow/repository.go
+++ b/internal/app/intentflow/repository.go
@@ -77,13 +77,14 @@ func insertStep(ctx context.Context, tx pgx.Tx, s Step) error {
 	if len(slots) == 0 {
 		slots = []byte("{}")
 	}
+	dependsOn := textArrayOrEmpty(s.DependsOn)
 	_, err := tx.Exec(ctx, `
 INSERT INTO intent_plan_steps (
   step_id, plan_id, ordinal, action_key, slots, depends_on, status, expected_resource_version,
   action_digest, idempotency_key, attempt, result_ref, result_digest, safe_summary,
   lease_owner, lease_until, fencing_token, version, execution_mode, created_at, updated_at
 ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21)`,
-		s.StepID, s.PlanID, s.Ordinal, s.ActionKey, slots, s.DependsOn, string(s.Status),
+		s.StepID, s.PlanID, s.Ordinal, s.ActionKey, slots, dependsOn, string(s.Status),
 		s.ExpectedResourceVersion, s.ActionDigest, s.IdempotencyKey, s.Attempt, s.ResultRef,
 		s.ResultDigest, s.SafeSummary, s.LeaseOwner, s.LeaseUntil, s.FencingToken, s.Version,
 		string(s.ExecutionMode), s.CreatedAt, s.UpdatedAt,
@@ -208,7 +209,7 @@ INSERT INTO intent_plan_decisions (
   exclude_unselected, decision_digest, outcome, created_at
 ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)`,
 		d.DecisionID, d.PlanID, d.Nonce, d.PlanRevision, d.ActorUserID, string(d.Kind),
-		d.StepIDs, d.ExcludeUnselected, d.DecisionDigest, d.Outcome, d.CreatedAt,
+		textArrayOrEmpty(d.StepIDs), d.ExcludeUnselected, d.DecisionDigest, d.Outcome, d.CreatedAt,
 	)
 	if isUniqueViolation(err) {
 		return uow.ErrDigestConflict
@@ -256,7 +257,7 @@ SET status=$1, depends_on=$2, expected_resource_version=$3, action_digest=$4,
     lease_owner=$10, lease_until=$11, fencing_token=$12, version=$13, execution_mode=$14,
     updated_at=$15, slots=$16
 WHERE step_id=$17 AND plan_id=$18`,
-			string(s.Status), s.DependsOn, s.ExpectedResourceVersion, s.ActionDigest,
+			string(s.Status), textArrayOrEmpty(s.DependsOn), s.ExpectedResourceVersion, s.ActionDigest,
 			s.IdempotencyKey, s.Attempt, s.ResultRef, s.ResultDigest, s.SafeSummary,
 			s.LeaseOwner, s.LeaseUntil, s.FencingToken, s.Version, string(s.ExecutionMode),
 			s.UpdatedAt, jsonbOrNil(s.Slots), s.StepID, planID,
@@ -473,6 +474,13 @@ func jsonbOrNil(raw []byte) []byte {
 	return raw
 }
 
+func textArrayOrEmpty(vals []string) []string {
+	if vals == nil {
+		return []string{}
+	}
+	return vals
+}
+
 func statusStrings(in []PlanStatus) []string {
 	out := make([]string, len(in))
 	for i, s := range in {
diff --git a/internal/app/intentflow/turn.go b/internal/app/intentflow/turn.go
index 9963bb58..e050674d 100644
--- a/internal/app/intentflow/turn.go
+++ b/internal/app/intentflow/turn.go
@@ -151,7 +151,11 @@ func (p *Plans) AdvanceEligible(ctx context.Context, workerID, planID string) (*
 		last = got
 	}
 	if last == nil {
-		return p.GetPlan(ctx, prin, planID)
+		view, getErr := p.GetPlan(ctx, prin, planID)
+		if firstErr != nil {
+			return view, firstErr
+		}
+		return view, getErr
 	}
 	return last, firstErr
 }
diff --git a/internal/taskrun/finalize.go b/internal/taskrun/finalize.go
index b88c0766..7403299e 100644
--- a/internal/taskrun/finalize.go
+++ b/internal/taskrun/finalize.go
@@ -448,7 +448,7 @@ func (c *Commands) appendTerminalEvent(ctx context.Context, record *RunRecord, r
 		OccurredAt:       time.Now().UTC(),
 		Payload:          body,
 	}
-	return (events.PGSink{}).Append(ctx, env, []string{"projection", "fav"})
+	return (events.PGSink{}).Append(ctx, env, []string{events.ConsumerNotify})
 }
 
 func (c *Commands) stageControlledDoc(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, status string, man *RunManifest) error {