From 8ffec646f2be24e217fb6e582bf585dec915bc71 Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Wed, 29 Apr 2026 20:12:55 -0600 Subject: [PATCH] feat(activity_sm): K=2 hysteresis on facet-gone and type-change ends + change reason in events Apply K=2 hysteresis to facet-gone and type-change activity ends while leaving idle and gap endings immediate. Motivated by the 4/16 backfill investigation: about 50-60% of ends came from ended_facet_gone and about 30% from ended_type_change, mostly single-segment wobbles. Persist pending counters through the existing awareness snapshot and add change to activity.detected/activity.persisted JSONL events for end-reason observability. --- tests/test_activity_state_machine.py | 39 ++- .../test_activity_state_machine_durability.py | 11 +- .../test_activity_state_machine_hysteresis.py | 266 ++++++++++++++++++ tests/test_pipeline_smoke.py | 12 +- tests/test_think_activity.py | 87 +++++- think/activity_state_machine.py | 119 +++++--- think/thinking.py | 22 +- 7 files changed, 485 insertions(+), 71 deletions(-) create mode 100644 tests/test_activity_state_machine_hysteresis.py diff --git a/tests/test_activity_state_machine.py b/tests/test_activity_state_machine.py index 38e790879..958ef6355 100644 --- a/tests/test_activity_state_machine.py +++ b/tests/test_activity_state_machine.py @@ -63,8 +63,11 @@ class TestContentTypeChange: sm = ActivityStateMachine() sm.update(_sense(content_type="coding"), "090000_300", "20260304") - changes = sm.update(_sense(content_type="meeting"), "090500_300", "20260304") + pending = sm.update(_sense(content_type="meeting"), "090500_300", "20260304") + changes = sm.update(_sense(content_type="meeting"), "091000_300", "20260304") + assert all(c["state"] != "ended" for c in pending) + assert pending[0]["_change"] == "type_change_pending" assert len(changes) == 2 ended = [c for c in changes if c["state"] == "ended"] started = [c for c in changes if c["state"] == "active"] @@ -152,8 +155,10 @@ class TestFacetDisappearing: ] sm.update(_sense(facets=two_facets), "090000_300", "20260304") one_facet = [{"facet": "work", "activity": "coding", "level": "high"}] - changes = sm.update(_sense(facets=one_facet), "090500_300", "20260304") + pending = sm.update(_sense(facets=one_facet), "090500_300", "20260304") + changes = sm.update(_sense(facets=one_facet), "091000_300", "20260304") + assert pending[0]["_change"] == "facet_gone_pending" ended = [c for c in changes if c["_change"] == "ended_facet_gone"] assert len(ended) == 1 assert ended[0]["facet"] == "personal" @@ -190,6 +195,7 @@ class TestGetCompletedActivities: sm = ActivityStateMachine() sm.update(_sense(content_type="coding"), "090000_300", "20260304") sm.update(_sense(content_type="meeting"), "090500_300", "20260304") + sm.update(_sense(content_type="meeting"), "091000_300", "20260304") completed = sm.get_completed_activities() assert len(completed) == 1 @@ -197,6 +203,7 @@ class TestGetCompletedActivities: assert "id" in rec assert "activity" in rec and rec["activity"] == "coding" assert "segments" in rec and isinstance(rec["segments"], list) + assert rec["segments"] == ["090000_300", "090500_300"] assert "level_avg" in rec and isinstance(rec["level_avg"], float) assert "description" in rec assert "active_entities" in rec @@ -214,11 +221,17 @@ class TestSegmentAccumulation: sm.update(_sense(content_type="coding"), "091000_300", "20260304") # End by type change sm.update(_sense(content_type="meeting"), "091500_300", "20260304") + sm.update(_sense(content_type="meeting"), "092000_300", "20260304") completed = sm.get_completed_activities() assert len(completed) == 1 rec = completed[0] - assert rec["segments"] == ["090000_300", "090500_300", "091000_300"] + assert rec["segments"] == [ + "090000_300", + "090500_300", + "091000_300", + "091500_300", + ] def test_ten_segments_produces_ten_keys(self): from think.activity_state_machine import ActivityStateMachine @@ -312,6 +325,7 @@ class TestSegmentAccumulationEdgeCases: sm.update(_sense(facets=two), "091000_300", "20260304") # personal disappears sm.update(_sense(facets=one), "091500_300", "20260304") + sm.update(_sense(facets=one), "092000_300", "20260304") completed = sm.get_completed_activities() assert len(completed) == 1 @@ -321,6 +335,7 @@ class TestSegmentAccumulationEdgeCases: "090000_300", "090500_300", "091000_300", + "091500_300", ] def test_gap_ending_preserves_accumulated_segments(self): @@ -347,6 +362,7 @@ class TestSegmentAccumulationEdgeCases: sm.update(_sense(content_type="coding"), "091000_300", "20260304") sm.update(_sense(content_type="coding"), "091500_300", "20260304") sm.update(_sense(content_type="meeting"), "092000_300", "20260304") + sm.update(_sense(content_type="meeting"), "092500_300", "20260304") completed = sm.get_completed_activities() assert len(completed) == 1 @@ -355,6 +371,7 @@ class TestSegmentAccumulationEdgeCases: "090500_300", "091000_300", "091500_300", + "092000_300", ] def test_single_segment_activity_has_one_segment(self): @@ -363,7 +380,7 @@ class TestSegmentAccumulationEdgeCases: sm = ActivityStateMachine() sm.update(_sense(content_type="coding"), "090000_300", "20260304") - sm.update(_sense(content_type="meeting"), "090500_300", "20260304") + sm.update(_sense(density="idle"), "090500_300", "20260304") completed = sm.get_completed_activities() assert len(completed) == 1 @@ -378,10 +395,11 @@ class TestSegmentAccumulationEdgeCases: # Same segment key again (shouldn't happen in practice, but defensive) sm.update(_sense(content_type="coding"), "090000_300", "20260304") sm.update(_sense(content_type="meeting"), "090500_300", "20260304") + sm.update(_sense(content_type="meeting"), "091000_300", "20260304") completed = sm.get_completed_activities() assert len(completed) == 1 - assert completed[0]["segments"] == ["090000_300"] + assert completed[0]["segments"] == ["090000_300", "090500_300"] def test_multi_facet_simultaneous_ending_all_have_segments(self): """Multiple facets ending simultaneously each have their own segments.""" @@ -421,6 +439,7 @@ class TestCompletedRecordFields: sm = ActivityStateMachine() sm.update(_sense(content_type="coding"), "090000_300", "20260304") sm.update(_sense(content_type="meeting"), "090500_300", "20260304") + sm.update(_sense(content_type="meeting"), "091000_300", "20260304") after = int(time.time() * 1000) rec = sm.get_completed_activities()[0] @@ -436,6 +455,7 @@ class TestCompletedRecordFields: sm = ActivityStateMachine() sm.update(_sense(content_type="coding"), "090000_300", "20260304") sm.update(_sense(content_type="meeting"), "090500_300", "20260304") + sm.update(_sense(content_type="meeting"), "091000_300", "20260304") rec = sm.get_completed_activities()[0] # Simulate routes.py cutoff calculation @@ -459,6 +479,7 @@ class TestCompletedRecordFields: "20260304", ) sm.update(_sense(content_type="meeting"), "090500_300", "20260304") + sm.update(_sense(content_type="meeting"), "091000_300", "20260304") rec = sm.get_completed_activities()[0] assert isinstance(rec["level_avg"], float) @@ -471,6 +492,7 @@ class TestCompletedRecordFields: sm = ActivityStateMachine() sm.update(_sense(content_type="coding"), "090000_300", "20260304") sm.update(_sense(content_type="meeting"), "090500_300", "20260304") + sm.update(_sense(content_type="meeting"), "091000_300", "20260304") rec = sm.get_completed_activities()[0] required = { @@ -497,7 +519,7 @@ class TestCompletedRecordFields: sm.update( _sense(content_type="coding", entities=entities), "090000_300", "20260304" ) - sm.update(_sense(content_type="meeting"), "090500_300", "20260304") + sm.update(_sense(density="idle"), "090500_300", "20260304") rec = sm.get_completed_activities()[0] assert rec["active_entities"] == ["Alice", "VSCode"] @@ -513,8 +535,10 @@ class TestCumulativeCompletedList: # Activity 1: coding sm.update(_sense(content_type="coding"), "090000_300", "20260304") sm.update(_sense(content_type="meeting"), "090500_300", "20260304") + sm.update(_sense(content_type="meeting"), "091000_300", "20260304") # Activity 2: meeting - sm.update(_sense(content_type="coding"), "091000_300", "20260304") + sm.update(_sense(content_type="coding"), "091500_300", "20260304") + sm.update(_sense(content_type="coding"), "092000_300", "20260304") completed = sm.get_completed_activities() assert len(completed) == 2 @@ -528,6 +552,7 @@ class TestCumulativeCompletedList: sm = ActivityStateMachine() sm.update(_sense(content_type="coding"), "090000_300", "20260304") sm.update(_sense(content_type="meeting"), "090500_300", "20260304") + sm.update(_sense(content_type="meeting"), "091000_300", "20260304") list1 = sm.get_completed_activities() list2 = sm.get_completed_activities() diff --git a/tests/test_activity_state_machine_durability.py b/tests/test_activity_state_machine_durability.py index 8fa20346d..32ca0ceda 100644 --- a/tests/test_activity_state_machine_durability.py +++ b/tests/test_activity_state_machine_durability.py @@ -224,7 +224,7 @@ def test_run_segment_sense_emits_activity_events(tmp_path: Path, monkeypatch): think, "_drain_priority_batch", lambda *args, **kwargs: (1, 0, []) ) monkeypatch.setattr( - think, "_jsonl_log", lambda event, **fields: events.append(event) + think, "_jsonl_log", lambda event, **fields: events.append((event, fields)) ) monkeypatch.setattr(think, "run_activity_prompts", lambda **kwargs: True) monkeypatch.setattr(think, "_callosum", None) @@ -239,5 +239,10 @@ def test_run_segment_sense_emits_activity_events(tmp_path: Path, monkeypatch): ) assert (success, failed, failed_names) == (1, 0, []) - assert "activity.detected" in events - assert "activity.persisted" in events + event_names = [event for event, _fields in events] + assert "activity.detected" in event_names + assert "activity.persisted" in event_names + detected = [fields for event, fields in events if event == "activity.detected"] + persisted = [fields for event, fields in events if event == "activity.persisted"] + assert detected[0]["change"] == "ended_idle" + assert persisted[0]["change"] == "ended_idle" diff --git a/tests/test_activity_state_machine_hysteresis.py b/tests/test_activity_state_machine_hysteresis.py new file mode 100644 index 000000000..9ddc63a73 --- /dev/null +++ b/tests/test_activity_state_machine_hysteresis.py @@ -0,0 +1,266 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Hysteresis tests for the deterministic activity state machine.""" + +import json +from pathlib import Path + +from think.activity_state_machine import ActivityStateMachine +from think.thinking import _write_json_atomic + +DAY = "20260304" + + +def _sense( + content_type: str = "coding", + density: str = "active", + facets: list[dict] | None = None, + summary: str = "Working on code.", + entities: list[dict] | None = None, +) -> dict: + if facets is None: + facets = [{"facet": "work", "activity": content_type, "level": "high"}] + return { + "density": density, + "content_type": content_type, + "activity_summary": summary, + "entities": entities or [], + "facets": facets, + "meeting_detected": content_type == "meeting", + "speakers": [], + "recommend": {}, + } + + +def _persist_snapshot(journal_root: Path, state_machine: ActivityStateMachine) -> None: + snapshot = { + "last_segment_key": state_machine.last_segment_key, + "last_segment_day": state_machine.last_segment_day, + "active": { + facet: {k: v for k, v in entry.items() if k != "_change"} + for facet, entry in state_machine.state.items() + }, + } + _write_json_atomic(journal_root / "awareness" / "activity_state.json", snapshot) + + +def test_single_segment_facet_wobble_does_not_end(): + sm = ActivityStateMachine() + sm.update(_sense(), "090000_300", DAY) + + missing = sm.update(_sense(facets=[]), "090500_300", DAY) + assert not any(change.get("_change") == "ended_facet_gone" for change in missing) + assert missing[0]["_change"] == "facet_gone_pending" + assert sm.state["work"]["segments"] == ["090000_300", "090500_300"] + + returned = sm.update(_sense(), "091000_300", DAY) + assert not any(change.get("_change") == "ended_facet_gone" for change in returned) + assert sm.state["work"]["state"] == "active" + assert sm.state["work"]["segments"] == [ + "090000_300", + "090500_300", + "091000_300", + ] + assert sm.state["work"]["_pending_facet_misses"] == 0 + + +def test_two_segment_facet_gone_ends_at_k(): + sm = ActivityStateMachine() + sm.update(_sense(), "090000_300", DAY) + sm.update(_sense(facets=[]), "090500_300", DAY) + + changes = sm.update(_sense(facets=[]), "091000_300", DAY) + ended = [ + change for change in changes if change.get("_change") == "ended_facet_gone" + ] + + assert len(ended) == 1 + assert ended[0]["state"] == "ended" + assert sm.state == {} + completed = sm.get_completed_activities() + assert completed[0]["segments"] == ["090000_300", "090500_300"] + + +def test_single_segment_type_wobble_does_not_end(): + sm = ActivityStateMachine() + sm.update(_sense(summary="Writing code."), "090000_300", DAY) + + wobble = sm.update( + _sense(content_type="meeting", summary="Stand-up."), "090500_300", DAY + ) + assert not any(change.get("_change") == "ended_type_change" for change in wobble) + assert wobble[0]["_change"] == "type_change_pending" + assert sm.state["work"]["description"] == "Writing code." + assert sm.state["work"]["segments"] == ["090000_300", "090500_300"] + + returned = sm.update(_sense(summary="Still coding."), "091000_300", DAY) + assert not any(change.get("_change") == "ended_type_change" for change in returned) + assert sm.get_completed_activities() == [] + assert sm.state["work"]["activity"] == "coding" + assert sm.state["work"]["description"] == "Still coding." + assert sm.state["work"]["_pending_type"] is None + assert sm.state["work"]["_pending_type_count"] == 0 + assert sm.state["work"]["segments"] == [ + "090000_300", + "090500_300", + "091000_300", + ] + + +def test_two_segment_type_change_ends_at_k(): + sm = ActivityStateMachine() + sm.update(_sense(content_type="coding"), "090000_300", DAY) + sm.update(_sense(content_type="meeting"), "090500_300", DAY) + + changes = sm.update(_sense(content_type="meeting"), "091000_300", DAY) + ended = [ + change for change in changes if change.get("_change") == "ended_type_change" + ] + active = [change for change in changes if change.get("_change") == "new"] + + assert len(ended) == 1 + assert ended[0]["activity"] == "coding" + assert len(active) == 1 + assert active[0]["activity"] == "meeting" + assert active[0]["since"] == "091000_300" + assert sm.state["work"]["segments"] == ["091000_300"] + completed = sm.get_completed_activities() + assert completed[0]["segments"] == ["090000_300", "090500_300"] + + +def test_idle_still_ends_immediately(): + sm = ActivityStateMachine() + sm.update(_sense(content_type="coding"), "090000_300", DAY) + sm.update(_sense(content_type="meeting"), "090500_300", DAY) + + changes = sm.update(_sense(density="idle"), "091000_300", DAY) + ended = [change for change in changes if change.get("_change") == "ended_idle"] + + assert len(ended) == 1 + assert sm.state == {} + assert sm.get_completed_activities()[0]["segments"] == [ + "090000_300", + "090500_300", + ] + + +def test_gap_still_ends_immediately(): + sm = ActivityStateMachine() + sm.update(_sense(content_type="coding"), "090000_300", DAY) + sm.update(_sense(content_type="meeting"), "090500_300", DAY) + + changes = sm.update(_sense(content_type="coding"), "100600_300", DAY) + ended_gap = [change for change in changes if change.get("_change") == "ended_gap"] + new = [change for change in changes if change.get("_change") == "new"] + + assert len(ended_gap) == 1 + assert len(new) == 1 + assert sm.get_completed_activities()[0]["segments"] == [ + "090000_300", + "090500_300", + ] + + +def test_pending_counters_reset_on_continuing_segment(): + sm = ActivityStateMachine() + sm.update(_sense(content_type="coding"), "090000_300", DAY) + sm.update(_sense(content_type="meeting"), "090500_300", DAY) + + sm.update(_sense(content_type="coding"), "091000_300", DAY) + entry = sm.state["work"] + + assert entry["_pending_facet_misses"] == 0 + assert entry["_pending_type"] is None + assert entry["_pending_type_count"] == 0 + + +def test_pending_counters_round_trip_via_awareness_json(tmp_path: Path, monkeypatch): + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + + sm1 = ActivityStateMachine(journal_root=tmp_path) + sm1.update(_sense(), "090000_300", DAY) + sm1.update(_sense(facets=[]), "090500_300", DAY) + _persist_snapshot(tmp_path, sm1) + + state_path = tmp_path / "awareness" / "activity_state.json" + data = json.loads(state_path.read_text(encoding="utf-8")) + assert data["active"]["work"]["_pending_facet_misses"] == 1 + + sm2 = ActivityStateMachine(journal_root=tmp_path) + changes = sm2.update(_sense(facets=[]), "091000_300", DAY) + ended = [ + change for change in changes if change.get("_change") == "ended_facet_gone" + ] + + assert len(ended) == 1 + assert sm2.state == {} + assert sm2.get_completed_activities()[0]["segments"] == [ + "090000_300", + "090500_300", + ] + + +def test_pending_type_counters_round_trip_via_awareness_json( + tmp_path: Path, monkeypatch +): + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + + sm1 = ActivityStateMachine(journal_root=tmp_path) + sm1.update(_sense(content_type="coding"), "090000_300", DAY) + sm1.update(_sense(content_type="meeting"), "090500_300", DAY) + _persist_snapshot(tmp_path, sm1) + + state_path = tmp_path / "awareness" / "activity_state.json" + data = json.loads(state_path.read_text(encoding="utf-8")) + assert data["active"]["work"]["_pending_type"] == "meeting" + assert data["active"]["work"]["_pending_type_count"] == 1 + + sm2 = ActivityStateMachine(journal_root=tmp_path) + changes = sm2.update(_sense(content_type="meeting"), "091000_300", DAY) + ended = [ + change for change in changes if change.get("_change") == "ended_type_change" + ] + + assert len(ended) == 1 + assert ended[0]["state"] == "ended" + + +def test_facet_recovery_resets_pending_counter(): + sm = ActivityStateMachine() + sm.update(_sense(), "090000_300", DAY) + sm.update(_sense(facets=[]), "090500_300", DAY) + sm.update(_sense(), "091000_300", DAY) + + changes = sm.update(_sense(facets=[]), "091500_300", DAY) + + assert not any(change.get("_change") == "ended_facet_gone" for change in changes) + assert sm.state["work"]["_pending_facet_misses"] == 1 + assert sm.state["work"]["segments"] == [ + "090000_300", + "090500_300", + "091000_300", + "091500_300", + ] + + +def test_alternating_type_wobble_does_not_accumulate(): + sm = ActivityStateMachine() + sm.update(_sense(content_type="coding"), "090000_300", DAY) + sm.update(_sense(content_type="meeting"), "090500_300", DAY) + sm.update(_sense(content_type="coding"), "091000_300", DAY) + sm.update(_sense(content_type="meeting"), "091500_300", DAY) + changes = sm.update(_sense(content_type="coding"), "092000_300", DAY) + + assert not any(change.get("_change") == "ended_type_change" for change in changes) + assert sm.get_completed_activities() == [] + assert sm.state["work"]["activity"] == "coding" + assert sm.state["work"]["_pending_type"] is None + assert sm.state["work"]["_pending_type_count"] == 0 + assert sm.state["work"]["segments"] == [ + "090000_300", + "090500_300", + "091000_300", + "091500_300", + "092000_300", + ] diff --git a/tests/test_pipeline_smoke.py b/tests/test_pipeline_smoke.py index 0634f3486..f33a546db 100644 --- a/tests/test_pipeline_smoke.py +++ b/tests/test_pipeline_smoke.py @@ -288,19 +288,19 @@ class TestPipelineSmokeTest: coding_rec = records[0] assert coding_rec["activity"] == "coding" - assert coding_rec["segments"] == ["090000_300", "090500_300"] + assert coding_rec["segments"] == ["090000_300", "090500_300", "091000_300"] assert coding_rec["id"] == make_activity_id("coding", "090000_300") assert isinstance(coding_rec["created_at"], int) assert isinstance(coding_rec["level_avg"], float) - assert coding_rec["level_avg"] == 1.0 + assert coding_rec["level_avg"] == 0.5 assert isinstance(coding_rec["active_entities"], list) assert isinstance(coding_rec["description"], str) assert len(coding_rec["description"]) > 0 meeting_rec = records[1] assert meeting_rec["activity"] == "meeting" - assert meeting_rec["segments"] == ["091000_300", "091500_300"] - assert meeting_rec["id"] == make_activity_id("meeting", "091000_300") + assert meeting_rec["segments"] == ["091500_300"] + assert meeting_rec["id"] == make_activity_id("meeting", "091500_300") assert isinstance(meeting_rec["created_at"], int) assert isinstance(meeting_rec["level_avg"], float) assert meeting_rec["level_avg"] == 0.5 @@ -309,7 +309,7 @@ class TestPipelineSmokeTest: assert len(meeting_rec["description"]) > 0 # Activity agents fire for BOTH endings: - # 1. Coding ended by type change (segment 3, active path) + # 1. Coding ended by type change (segment 4, active path) # 2. Meeting ended by idle (segment 5, idle path) assert len(activity_calls) == 2 @@ -321,7 +321,7 @@ class TestPipelineSmokeTest: assert activity_calls[1]["facet"] == "work" assert activity_calls[1]["activity_id"] == make_activity_id( - "meeting", "091000_300" + "meeting", "091500_300" ) assert activity_calls[1]["day"] == DAY diff --git a/tests/test_think_activity.py b/tests/test_think_activity.py index 23acf82cb..f952c8585 100644 --- a/tests/test_think_activity.py +++ b/tests/test_think_activity.py @@ -694,7 +694,7 @@ class TestActivityPersistence: "090000_300", "20260304", ) - changes = sm.update( + sm.update( { "density": "active", "content_type": "meeting", @@ -710,6 +710,22 @@ class TestActivityPersistence: "090500_300", "20260304", ) + changes = sm.update( + { + "density": "active", + "content_type": "meeting", + "activity_summary": "Stand-up", + "entities": [], + "facets": [ + {"facet": "work", "activity": "meeting", "level": "medium"} + ], + "meeting_detected": True, + "speakers": [], + "recommend": {}, + }, + "091500_300", + "20260304", + ) # Find the ended change ended = [c for c in changes if c.get("state") == "ended"] @@ -768,8 +784,9 @@ class TestActivityPersistenceRoundTrip: sm.update(self._sense(content_type="coding"), "090500_300", "20260304") sm.update(self._sense(content_type="coding"), "091000_300", "20260304") # End via type change + sm.update(self._sense(content_type="meeting"), "091500_300", "20260304") changes = sm.update( - self._sense(content_type="meeting"), "091500_300", "20260304" + self._sense(content_type="meeting"), "092000_300", "20260304" ) ended = [c for c in changes if c.get("state") == "ended"] @@ -787,7 +804,12 @@ class TestActivityPersistenceRoundTrip: records = load_activity_records("work", "20260304") assert len(records) == 1 r = records[0] - assert r["segments"] == ["090000_300", "090500_300", "091000_300"] + assert r["segments"] == [ + "090000_300", + "090500_300", + "091000_300", + "091500_300", + ] assert r["activity"] == "coding" assert isinstance(r["created_at"], int) assert r["created_at"] > 0 @@ -827,6 +849,7 @@ class TestActivityPersistenceRoundTrip: sm = ActivityStateMachine() sm.update(self._sense(content_type="coding"), "090000_300", "20260304") sm.update(self._sense(content_type="meeting"), "090500_300", "20260304") + sm.update(self._sense(content_type="meeting"), "091000_300", "20260304") rec = sm.get_completed_activities()[0] @@ -851,8 +874,9 @@ class TestActivityPersistenceRoundTrip: sm = ActivityStateMachine() # Activity 1 ends sm.update(self._sense(content_type="coding"), "090000_300", "20260304") + sm.update(self._sense(content_type="meeting"), "090500_300", "20260304") changes1 = sm.update( - self._sense(content_type="meeting"), "090500_300", "20260304" + self._sense(content_type="meeting"), "091000_300", "20260304" ) facet_by_id = { c["id"]: c["facet"] for c in changes1 if c.get("state") == "ended" @@ -863,7 +887,7 @@ class TestActivityPersistenceRoundTrip: # Activity 2 continues (no ending) changes2 = sm.update( - self._sense(content_type="meeting"), "091000_300", "20260304" + self._sense(content_type="meeting"), "091500_300", "20260304" ) # No ended changes in this update facet_by_id2 = { @@ -901,6 +925,7 @@ class TestActivityPersistenceRoundTrip: "20260304", ) sm.update(self._sense(content_type="meeting"), "090500_300", "20260304") + sm.update(self._sense(content_type="meeting"), "091000_300", "20260304") rec = sm.get_completed_activities()[0] append_activity_record("work", "20260304", rec) @@ -1008,6 +1033,22 @@ class TestCreatedAtRoutesCompat: "090500_300", "20260304", ) + sm.update( + { + "density": "active", + "content_type": "meeting", + "activity_summary": "standup", + "entities": [], + "facets": [ + {"facet": "work", "activity": "meeting", "level": "medium"} + ], + "meeting_detected": True, + "speakers": [], + "recommend": {}, + }, + "091000_300", + "20260304", + ) rec = sm.get_completed_activities()[0] append_activity_record("work", "20260304", rec) @@ -1055,7 +1096,7 @@ class TestCreatedAtRoutesCompat: "090000_300", "20260304", ) - changes1 = sm.update( + sm.update( { "density": "active", "content_type": "meeting", @@ -1071,6 +1112,22 @@ class TestCreatedAtRoutesCompat: "090500_300", "20260304", ) + changes1 = sm.update( + { + "density": "active", + "content_type": "meeting", + "activity_summary": "second", + "entities": [], + "facets": [ + {"facet": "work", "activity": "meeting", "level": "medium"} + ], + "meeting_detected": True, + "speakers": [], + "recommend": {}, + }, + "091000_300", + "20260304", + ) # Persist first completed facet_by_id = { c["id"]: c["facet"] for c in changes1 if c.get("state") == "ended" @@ -1082,7 +1139,7 @@ class TestCreatedAtRoutesCompat: # Small delay so created_at differs time.sleep(0.01) - changes2 = sm.update( + sm.update( { "density": "active", "content_type": "coding", @@ -1098,6 +1155,22 @@ class TestCreatedAtRoutesCompat: "091000_300", "20260304", ) + changes2 = sm.update( + { + "density": "active", + "content_type": "coding", + "activity_summary": "third", + "entities": [], + "facets": [ + {"facet": "work", "activity": "coding", "level": "high"} + ], + "meeting_detected": False, + "speakers": [], + "recommend": {}, + }, + "092000_300", + "20260304", + ) facet_by_id2 = { c["id"]: c["facet"] for c in changes2 if c.get("state") == "ended" } diff --git a/think/activity_state_machine.py b/think/activity_state_machine.py index 0610ae2dc..75bcdeaa5 100644 --- a/think/activity_state_machine.py +++ b/think/activity_state_machine.py @@ -13,6 +13,7 @@ from think.utils import segment_parse # 10 min; R&D default. Existing LLM hook uses 3600. GAP_THRESHOLD_SECONDS = 600 +END_HYSTERESIS_SEGMENTS = 2 class ActivityStateMachine: @@ -163,20 +164,33 @@ class ActivityStateMachine: facet_map[facet["facet"]] = facet current_facets = set(facet_map.keys()) + # Hysteresis invariant: pending segments bridge into the prior activity's segments[]; + # the segment that commits the K-th condition does NOT. for facet in sorted(set(self.state.keys()) - current_facets): - prior = self.state.pop(facet) - entry = { - "id": prior["id"], - "activity": prior["activity"], - "state": "ended", - "since": prior["since"], - "description": prior["description"], - "_change": "ended_facet_gone", - "facet": facet, - "segment": segment_key, - } - changes.append(entry) - self._completed.append(self._make_completed_record(prior)) + prior = self.state[facet] + misses = prior.get("_pending_facet_misses", 0) + 1 + if misses >= END_HYSTERESIS_SEGMENTS: + prior = self.state.pop(facet) + entry = { + "id": prior["id"], + "activity": prior["activity"], + "state": "ended", + "since": prior["since"], + "description": prior["description"], + "_change": "ended_facet_gone", + "facet": facet, + "segment": segment_key, + } + changes.append(entry) + self._completed.append(self._make_completed_record(prior)) + else: + prior["_pending_facet_misses"] = misses + prior["_change"] = "facet_gone_pending" + prior["segment"] = segment_key + prior.setdefault("segments", [prior["since"]]) + if segment_key not in prior["segments"]: + prior["segments"].append(segment_key) + changes.append(dict(prior)) for facet in sorted(current_facets): facet_data = facet_map.get(facet, {}) @@ -186,39 +200,62 @@ class ActivityStateMachine: if facet in self.state: prior = self.state[facet] + prior["_pending_facet_misses"] = 0 if prior["activity"] != content_type: - ended = { - "id": prior["id"], - "activity": prior["activity"], - "state": "ended", - "since": prior["since"], - "description": prior["description"], - "_change": "ended_type_change", - "facet": facet, - "segment": segment_key, - } - changes.append(ended) - self._completed.append(self._make_completed_record(prior)) - - new_entry = { - "id": make_activity_id(content_type, segment_key), - "activity": content_type, - "state": "active", - "since": segment_key, - "description": activity_summary, - "level": level, - "active_entities": entity_names, - "_change": "new", - "facet": facet, - "segment": segment_key, - "segments": [segment_key], - } - self.state[facet] = new_entry - changes.append(dict(new_entry)) + pending_type = prior.get("_pending_type") + if pending_type == content_type: + pending_count = prior.get("_pending_type_count", 0) + 1 + else: + pending_type = content_type + pending_count = 1 + + if pending_count >= END_HYSTERESIS_SEGMENTS: + ended = { + "id": prior["id"], + "activity": prior["activity"], + "state": "ended", + "since": prior["since"], + "description": prior["description"], + "_change": "ended_type_change", + "facet": facet, + "segment": segment_key, + } + changes.append(ended) + self._completed.append(self._make_completed_record(prior)) + + new_entry = { + "id": make_activity_id(content_type, segment_key), + "activity": content_type, + "state": "active", + "since": segment_key, + "description": activity_summary, + "level": level, + "active_entities": entity_names, + "_change": "new", + "facet": facet, + "segment": segment_key, + "segments": [segment_key], + } + self.state[facet] = new_entry + changes.append(dict(new_entry)) + else: + prior["_pending_type"] = pending_type + prior["_pending_type_count"] = pending_count + prior["_change"] = "type_change_pending" + prior["segment"] = segment_key + prior["level"] = level + prior["active_entities"] = entity_names + prior.setdefault("segments", [prior["since"]]) + if segment_key not in prior["segments"]: + prior["segments"].append(segment_key) + changes.append(dict(prior)) else: prior["description"] = activity_summary prior["level"] = level prior["active_entities"] = entity_names + prior["_pending_facet_misses"] = 0 + prior["_pending_type"] = None + prior["_pending_type_count"] = 0 prior["_change"] = "continuing" prior["segment"] = segment_key prior.setdefault("segments", [prior["since"]]) diff --git a/think/thinking.py b/think/thinking.py index fceb31c79..99aa42e8b 100644 --- a/think/thinking.py +++ b/think/thinking.py @@ -684,13 +684,15 @@ def run_segment_sense( routing_day = state_machine.last_segment_day or day idle_changes = state_machine.update(sense_json, segment, day) # Persist completed activity records from idle transitions - ended_pairs = [ - (c["id"], c["facet"]) for c in idle_changes if c.get("state") == "ended" + ended_triples = [ + (c["id"], c["facet"], c.get("_change")) + for c in idle_changes + if c.get("state") == "ended" ] completed_lookup = {} for rec in state_machine.get_completed_activities(): completed_lookup.setdefault(rec["id"], rec) - for activity_id, facet in ended_pairs: + for activity_id, facet, change in ended_triples: _jsonl_log( "activity.detected", mode=target_schedule, @@ -699,6 +701,7 @@ def run_segment_sense( activity=str(activity_id), facet=str(facet), state="ended", + change=change, ) rec = completed_lookup.get(activity_id) if rec: @@ -710,9 +713,10 @@ def run_segment_sense( segment=segment, activity=str(activity_id), facet=str(facet), + change=change, ) # Run activity agents for completed activities - for activity_id, facet in ended_pairs: + for activity_id, facet, _change in ended_triples: logging.info( "Activity completed (idle): %s facet=%s, running activity agents", activity_id, @@ -940,13 +944,15 @@ def run_segment_sense( routing_day = state_machine.last_segment_day or day changes = state_machine.update(sense_json, segment, day) # Persist completed activity records before running activity agents - ended_pairs = [ - (c["id"], c["facet"]) for c in changes if c.get("state") == "ended" + ended_triples = [ + (c["id"], c["facet"], c.get("_change")) + for c in changes + if c.get("state") == "ended" ] completed_lookup = {} for rec in state_machine.get_completed_activities(): completed_lookup.setdefault(rec["id"], rec) - for activity_id, facet in ended_pairs: + for activity_id, facet, change in ended_triples: _jsonl_log( "activity.detected", mode=target_schedule, @@ -955,6 +961,7 @@ def run_segment_sense( activity=str(activity_id), facet=str(facet), state="ended", + change=change, ) rec = completed_lookup.get(activity_id) if rec: @@ -966,6 +973,7 @@ def run_segment_sense( segment=segment, activity=str(activity_id), facet=str(facet), + change=change, ) if state_machine.journal_root is not None: try: -- 2.51.2