refactor: remove activity source tracking (#62)

Author: i-trytoohard

backend/internal/cdc/cdc_test.go

@@ -32,7 +32,7 @@ func seedSession(t *testing.T, s *sqlite.Store) domain.SessionRecord {
 	}
 	r, err := s.CreateSession(ctx, domain.SessionRecord{
 		ProjectID: "mer", Kind: domain.KindWorker,
-		Activity:  domain.ActivitySubstate{State: domain.ActivityActive, LastActivityAt: now, Source: domain.SourceNative},
+		Activity:  domain.Activity{State: domain.ActivityActive, LastActivityAt: now},
 		CreatedAt: now, UpdatedAt: now,
 	})
 	if err != nil {

backend/internal/daemon/wiring_test.go

@@ -42,14 +42,14 @@ func TestWiring_WriteFlowsToBroadcaster(t *testing.T) {
 	}
 	rec, err := store.CreateSession(ctx, domain.SessionRecord{
 		ProjectID: "mer", Kind: domain.KindWorker,
-		Activity: domain.ActivitySubstate{State: domain.ActivityIdle, LastActivityAt: time.Now(), Source: domain.SourceNone},
+		Activity: domain.Activity{State: domain.ActivityIdle, LastActivityAt: time.Now()},
 	})
 	if err != nil {
 		t.Fatal(err)
 	}
 	// A real transition through the engine, which writes the row and fires the
 	// activity_state/is_terminated CDC trigger.
-	if err := lcm.ApplyActivitySignal(ctx, rec.ID, ports.ActivitySignal{Valid: true, State: domain.ActivityActive, Timestamp: time.Now(), Source: domain.SourceNative}); err != nil {
+	if err := lcm.ApplyActivitySignal(ctx, rec.ID, ports.ActivitySignal{Valid: true, State: domain.ActivityActive, Timestamp: time.Now()}); err != nil {
 		t.Fatal(err)
 	}
 

backend/internal/domain/activity.go

@@ -5,59 +5,23 @@ import "time"
 // ActivityState is how busy the agent is, derived from its output/JSONL.
 type ActivityState string
 
-// Activity states. WaitingInput and Blocked are sticky (see IsSticky).
+// Activity states. WaitingInput is sticky (see IsSticky).
 const (
 	ActivityActive       ActivityState = "active"
 	ActivityIdle         ActivityState = "idle"
 	ActivityWaitingInput ActivityState = "waiting_input"
-	ActivityBlocked      ActivityState = "blocked"
 	ActivityExited       ActivityState = "exited"
 )
 
 // IsSticky reports whether an activity state must NOT be aged/demoted by the
 // passage of time (a paused agent is still paused until a new signal says so).
 func (a ActivityState) IsSticky() bool {
-	return a == ActivityWaitingInput || a == ActivityBlocked
+	return a == ActivityWaitingInput
 }
 
-// ActivitySource records where an activity reading came from, so a weaker
-// source can't override a stronger one.
-type ActivitySource string
-
-// Activity signal sources, strongest first.
-const (
-	SourceNative   ActivitySource = "native"
-	SourceTerminal ActivitySource = "terminal"
-	SourceHook     ActivitySource = "hook"
-	SourceRuntime  ActivitySource = "runtime"
-	SourceNone     ActivitySource = "none"
-)
-
-// CanOverride reports whether a reading from source a may replace a current
-// reading from source current. Unknown sources are treated as weakest.
-func (a ActivitySource) CanOverride(current ActivitySource) bool {
-	return activitySourceRank(a) <= activitySourceRank(current)
-}
-
-func activitySourceRank(s ActivitySource) int {
-	switch s {
-	case SourceNative:
-		return 0
-	case SourceTerminal:
-		return 1
-	case SourceHook:
-		return 2
-	case SourceRuntime:
-		return 3
-	default:
-		return 4
-	}
-}
-
-// ActivitySubstate is the persisted activity reading: the state, when it was
-// last observed, and which source reported it.
-type ActivitySubstate struct {
-	State          ActivityState  `json:"state"`
-	LastActivityAt time.Time      `json:"lastActivityAt"`
-	Source         ActivitySource `json:"source"`
+// Activity captures the persisted activity reading: the state and when it was
+// last observed.
+type Activity struct {
+	State          ActivityState `json:"state"`
+	LastActivityAt time.Time     `json:"lastActivityAt"`
 }

backend/internal/domain/session.go

@@ -36,16 +36,16 @@ type SessionMetadata struct {
 // facts: identity, agent harness, activity_state, is_terminated, and operational
 // metadata. The user-facing Status is derived from these facts plus PR facts.
 type SessionRecord struct {
-	ID           SessionID        `json:"id"`
-	ProjectID    ProjectID        `json:"projectId"`
-	IssueID      IssueID          `json:"issueId,omitempty"`
-	Kind         SessionKind      `json:"kind"`
-	Harness      AgentHarness     `json:"harness,omitempty"`
-	Activity     ActivitySubstate `json:"activity"`
-	IsTerminated bool             `json:"isTerminated"`
-	Metadata     SessionMetadata  `json:"-"`
-	CreatedAt    time.Time        `json:"createdAt"`
-	UpdatedAt    time.Time        `json:"updatedAt"`
+	ID           SessionID       `json:"id"`
+	ProjectID    ProjectID       `json:"projectId"`
+	IssueID      IssueID         `json:"issueId,omitempty"`
+	Kind         SessionKind     `json:"kind"`
+	Harness      AgentHarness    `json:"harness,omitempty"`
+	Activity     Activity        `json:"activity"`
+	IsTerminated bool            `json:"isTerminated"`
+	Metadata     SessionMetadata `json:"-"`
+	CreatedAt    time.Time       `json:"createdAt"`
+	UpdatedAt    time.Time       `json:"updatedAt"`
 }
 
 // Session is the read-model returned across the API boundary: a SessionRecord

backend/internal/domain/status.go

@@ -16,7 +16,6 @@ const (
 	StatusMergeable        SessionStatus = "mergeable"
 	StatusMerged           SessionStatus = "merged"
 	StatusNeedsInput       SessionStatus = "needs_input"
-	StatusStuck            SessionStatus = "stuck"
 	StatusIdle             SessionStatus = "idle"
 	StatusTerminated       SessionStatus = "terminated"
 )
@@ -32,11 +31,8 @@ func DeriveStatus(rec SessionRecord, pr *PRFacts) SessionStatus {
 		return StatusTerminated
 	}
 
-	switch rec.Activity.State {
-	case ActivityWaitingInput:
+	if rec.Activity.State == ActivityWaitingInput {
 		return StatusNeedsInput
-	case ActivityBlocked:
-		return StatusStuck
 	}
 
 	if pr != nil {

backend/internal/domain/status_test.go

@@ -3,7 +3,7 @@ package domain
 import "testing"
 
 func rec(activity ActivityState, terminated bool) SessionRecord {
-	return SessionRecord{Activity: ActivitySubstate{State: activity}, IsTerminated: terminated}
+	return SessionRecord{Activity: Activity{State: activity}, IsTerminated: terminated}
 }
 
 func pr(facts PRFacts) *PRFacts { return &facts }
@@ -18,7 +18,6 @@ func TestDeriveStatusFromSessionFactsAndPR(t *testing.T) {
 		{"terminated", rec(ActivityExited, true), nil, StatusTerminated},
 		{"merged-pr", rec(ActivityIdle, true), pr(PRFacts{Merged: true}), StatusMerged},
 		{"needs-input", rec(ActivityWaitingInput, false), pr(PRFacts{CI: CIFailing}), StatusNeedsInput},
-		{"blocked", rec(ActivityBlocked, false), pr(PRFacts{CI: CIFailing}), StatusStuck},
 		{"ci-failed", rec(ActivityIdle, false), pr(PRFacts{CI: CIFailing}), StatusCIFailed},
 		{"draft", rec(ActivityIdle, false), pr(PRFacts{Draft: true}), StatusDraft},
 		{"changes-requested", rec(ActivityIdle, false), pr(PRFacts{Review: ReviewChangesRequest}), StatusChangesRequested},

backend/internal/lifecycle/manager.go

@@ -65,7 +65,7 @@ func (m *Manager) ApplyRuntimeObservation(ctx context.Context, id domain.Session
 		}
 		next := cur
 		next.IsTerminated = true
-		next.Activity = domain.ActivitySubstate{State: domain.ActivityExited, LastActivityAt: timeOr(f.ObservedAt, now), Source: domain.SourceRuntime}
+		next.Activity = domain.Activity{State: domain.ActivityExited, LastActivityAt: timeOr(f.ObservedAt, now)}
 		return next, true
 	})
 }
@@ -79,11 +79,8 @@ func (m *Manager) ApplyActivitySignal(ctx context.Context, id domain.SessionID,
 		if cur.IsTerminated {
 			return cur, false
 		}
-		if !s.Source.CanOverride(cur.Activity.Source) {
-			return cur, false
-		}
 		next := cur
-		act := domain.ActivitySubstate{State: s.State, LastActivityAt: timeOr(s.Timestamp, now), Source: s.Source}
+		act := domain.Activity{State: s.State, LastActivityAt: timeOr(s.Timestamp, now)}
 		if sameActivity(cur.Activity, act) {
 			return cur, false
 		}
@@ -108,7 +105,7 @@ func (m *Manager) MarkSpawned(ctx context.Context, id domain.SessionID, metadata
 	}
 	now := m.clock()
 	rec.IsTerminated = false
-	rec.Activity = domain.ActivitySubstate{State: domain.ActivityIdle, LastActivityAt: now, Source: domain.SourceRuntime}
+	rec.Activity = domain.Activity{State: domain.ActivityIdle, LastActivityAt: now}
 	rec.Metadata = mergeMetadata(rec.Metadata, metadata)
 	rec.UpdatedAt = now
 	return m.store.UpdateSession(ctx, rec)
@@ -121,13 +118,13 @@ func (m *Manager) MarkTerminated(ctx context.Context, id domain.SessionID) error
 			return cur, false
 		}
 		cur.IsTerminated = true
-		cur.Activity = domain.ActivitySubstate{State: domain.ActivityExited, LastActivityAt: now, Source: domain.SourceRuntime}
+		cur.Activity = domain.Activity{State: domain.ActivityExited, LastActivityAt: now}
 		return cur, true
 	})
 }
 
-func sameActivity(a, b domain.ActivitySubstate) bool {
-	return a.State == b.State && a.Source == b.Source && a.LastActivityAt.Equal(b.LastActivityAt)
+func sameActivity(a, b domain.Activity) bool {
+	return a.State == b.State && a.LastActivityAt.Equal(b.LastActivityAt)
 }
 
 func mergeMetadata(base, in domain.SessionMetadata) domain.SessionMetadata {

backend/internal/lifecycle/manager_test.go

@@ -51,7 +51,7 @@ func newManager() (*Manager, *fakeStore, *fakeMessenger) {
 }
 
 func working(id domain.SessionID) domain.SessionRecord {
-	return domain.SessionRecord{ID: id, ProjectID: "mer", Activity: domain.ActivitySubstate{State: domain.ActivityActive, LastActivityAt: time.Now(), Source: domain.SourceNative}}
+	return domain.SessionRecord{ID: id, ProjectID: "mer", Activity: domain.Activity{State: domain.ActivityActive, LastActivityAt: time.Now()}}
 }
 
 func TestRuntimeObservation_InferredDeathSetsTerminated(t *testing.T) {
@@ -92,30 +92,6 @@ func TestActivity_InvalidIsIgnored(t *testing.T) {
 	}
 }
 
-func TestActivity_WeakerSourceDoesNotOverrideStronger(t *testing.T) {
-	m, st, _ := newManager()
-	st.sessions["mer-1"] = working("mer-1")
-	before := st.sessions["mer-1"]
-	if err := m.ApplyActivitySignal(ctx, "mer-1", ports.ActivitySignal{Valid: true, State: domain.ActivityIdle, Source: domain.SourceRuntime}); err != nil {
-		t.Fatal(err)
-	}
-	if st.sessions["mer-1"] != before {
-		t.Fatalf("weaker runtime signal should not override native activity, got %+v", st.sessions["mer-1"])
-	}
-}
-
-func TestActivity_StrongerSourceOverridesWeaker(t *testing.T) {
-	m, st, _ := newManager()
-	st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", Activity: domain.ActivitySubstate{State: domain.ActivityIdle, LastActivityAt: time.Now(), Source: domain.SourceRuntime}}
-	if err := m.ApplyActivitySignal(ctx, "mer-1", ports.ActivitySignal{Valid: true, State: domain.ActivityActive, Source: domain.SourceNative}); err != nil {
-		t.Fatal(err)
-	}
-	got := st.sessions["mer-1"].Activity
-	if got.State != domain.ActivityActive || got.Source != domain.SourceNative {
-		t.Fatalf("stronger native signal should override runtime, got %+v", got)
-	}
-}
-
 func TestMarkTerminated(t *testing.T) {
 	m, st, _ := newManager()
 	st.sessions["mer-1"] = working("mer-1")

backend/internal/lifecycle/reactions.go

@@ -39,7 +39,7 @@ func (m *Manager) ApplyPRObservation(ctx context.Context, id domain.SessionID, o
 	if err != nil || !ok {
 		return err
 	}
-	if rec.IsTerminated || rec.Activity.State == domain.ActivityBlocked || rec.Activity.State == domain.ActivityWaitingInput {
+	if rec.IsTerminated || rec.Activity.State == domain.ActivityWaitingInput {
 		return nil
 	}
 	if o.CI == domain.CIFailing {

backend/internal/lifecycle/runtime.go

@@ -9,7 +9,7 @@ import (
 
 const defaultRecentActivityWindow = 60 * time.Second
 
-func hasRecentActivity(a domain.ActivitySubstate, now time.Time, window time.Duration) bool {
+func hasRecentActivity(a domain.Activity, now time.Time, window time.Duration) bool {
 	switch {
 	case a.State == domain.ActivityExited:
 		return false
@@ -22,7 +22,7 @@ func hasRecentActivity(a domain.ActivitySubstate, now time.Time, window time.Dur
 	}
 }
 
-func runtimeClearlyDead(f ports.RuntimeFacts, activity domain.ActivitySubstate, now time.Time, window time.Duration) bool {
+func runtimeClearlyDead(f ports.RuntimeFacts, activity domain.Activity, now time.Time, window time.Duration) bool {
 	observedAt := timeOr(f.ObservedAt, now)
 	return f.Probe == ports.ProbeDead && !hasRecentActivity(activity, observedAt, window)
 }