diff --git a/apps/admin_console/database/repositories/step_repository.py b/apps/admin_console/database/repositories/step_repository.py index c1a0cd9..3d18126 100644 --- a/apps/admin_console/database/repositories/step_repository.py +++ b/apps/admin_console/database/repositories/step_repository.py @@ -390,20 +390,20 @@ class StepRepository: steps.append(step_dict) - # Fetch step-less tool traces (e.g. planner tools) + # Fetch step-less tool traces (e.g. planner tools or turns without recorded step) cursor.execute( "SELECT t1.trace_id, t1.parent_trace_id, t1.step_id, t1.type," " t1.name, t1.timestamp, t1.duration, t1.status, t1.payload," " t2.name as agent_name FROM traces t1 LEFT JOIN traces t2 ON" " t1.parent_trace_id = t2.trace_id WHERE t1.session_id = ? AND" - " (t1.step_id IS NULL OR t1.step_id = '') AND (t1.type = 'tool' OR" + " (t1.step_id IS NULL OR t1.step_id = '' OR t1.step_id NOT IN (SELECT step_id FROM steps WHERE session_id = ?)) AND (t1.type = 'tool' OR" " (t1.type = 'llm_call' AND t1.status IN ('failed', 'retrying')) OR (t1.type = 'log'" " AND t1.payload LIKE '%LLM Error:%Pausing execution%' AND NOT EXISTS" " (SELECT 1 FROM traces t3 WHERE t3.session_id = t1.session_id AND" " t3.type = 'llm_call' AND t3.status = 'failed' AND" " ABS(t3.timestamp - t1.timestamp) < 2))) ORDER BY" " t1.timestamp ASC", - (session_id,), + (session_id, session_id), ) stepless_rows = cursor.fetchall() if stepless_rows: diff --git a/apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.ts b/apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.ts index a9a6f64..472bd4e 100644 --- a/apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.ts +++ b/apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.ts @@ -130,13 +130,20 @@ export class FloatingVideoPlayerComponent implements OnDestroy { { allowSignalWrites: true } ); - // Reset step playback state when session changes + // Reset step playback state only when session actually changes + let lastTrackedSessionId: string | null = null; effect( () => { - const _ = this.stepFrames(); - this.activeStepIndex.set(0); - this.viewMode.set('pre'); - this.pauseStepPlay(); + const curSessionId = this.agentService.currentSessionId(); + const total = this.totalStepFrames(); + if (curSessionId !== lastTrackedSessionId) { + lastTrackedSessionId = curSessionId; + this.activeStepIndex.set(0); + this.viewMode.set('pre'); + this.pauseStepPlay(); + } else if (total > 0 && this.activeStepIndex() >= total) { + this.activeStepIndex.set(total - 1); + } }, { allowSignalWrites: true } ); @@ -171,6 +178,12 @@ export class FloatingVideoPlayerComponent implements OnDestroy { this.seek(target, true); } }); + + effect(() => { + const request = this.agentService.stepSeekRequest(); + if (!request) return; + this.seekStepFrame(request.index); + }); } public onLiveStreamError(): void { diff --git a/apps/showcase_ui/src/app/core/models/stream.model.ts b/apps/showcase_ui/src/app/core/models/stream.model.ts index d11e79c..d66042f 100644 --- a/apps/showcase_ui/src/app/core/models/stream.model.ts +++ b/apps/showcase_ui/src/app/core/models/stream.model.ts @@ -101,6 +101,7 @@ export interface CheckerResult { export interface StepReplayFrame { index: number; stepNumber: number; + rawStepNumber?: number; stepId?: string; title: string; imageUrl: string; @@ -115,5 +116,6 @@ export interface StepReplayFrame { isPost?: boolean; timestamp?: number; phaseId?: string; + summary?: string; } diff --git a/apps/showcase_ui/src/app/services/agent.service.ts b/apps/showcase_ui/src/app/services/agent.service.ts index 629dbeb..1768484 100644 --- a/apps/showcase_ui/src/app/services/agent.service.ts +++ b/apps/showcase_ui/src/app/services/agent.service.ts @@ -232,9 +232,11 @@ export class AgentService { public recordingPlaybackMessage = signal(''); public shouldAutoplayVideo = signal(false); public videoSeekRequest = signal<{ seconds: number; requestId: number } | null>(null); + public stepSeekRequest = signal<{ index: number; requestId: number } | null>(null); public playerMode = signal<'video' | 'steps'>('video'); private activeVideoSessionId: string | null = null; private videoSeekRequestId = 0; + private stepSeekRequestId = 0; private videoRequestGeneration = 0; private videoRetryTimer: ReturnType | null = null; private videoWaitStartedAt = 0; @@ -258,6 +260,11 @@ export class AgentService { this.playerMode.update((mode) => (mode === 'video' ? 'steps' : 'video')); } + public requestStepSeek(index: number): void { + this.stepSeekRequestId++; + this.stepSeekRequest.set({ index, requestId: this.stepSeekRequestId }); + } + /** * Clear user-pinned selection so subsequent runs automatically follow active runner */ @@ -1407,7 +1414,8 @@ export class AgentService { sessionId?: string, videoUrl?: string, title?: string, - seekSeconds?: number + seekSeconds?: number, + stepIndex?: number ): void { const targetSessionId = sessionId || this.currentSessionId(); const session = this.sessions().find(s => s.session_id === targetSessionId); @@ -1424,6 +1432,10 @@ export class AgentService { this.recordingPlaybackMessage.set(''); this.activeVideoSegments.set([]); if (Number.isFinite(seekSeconds)) this.requestVideoSeek(Number(seekSeconds)); + if (Number.isFinite(stepIndex)) { + this.playerMode?.set('steps'); + this.requestStepSeek(Number(stepIndex)); + } if (!targetSessionId) { this.activeVideoUrl.set(null); this.isVideoLoading.set(false); diff --git a/apps/showcase_ui/src/app/utils/action-formatter.util.spec.ts b/apps/showcase_ui/src/app/utils/action-formatter.util.spec.ts index bca2865..553e288 100644 --- a/apps/showcase_ui/src/app/utils/action-formatter.util.spec.ts +++ b/apps/showcase_ui/src/app/utils/action-formatter.util.spec.ts @@ -274,4 +274,132 @@ describe('extractStepReplayFrames', () => { expect(frames[0].stepNumber).toBe(1); expect(frames[1].stepNumber).toBe(2); }); + + it('should start from Step 1 even when step_updated events lack step_number (preventing the Step 4 bug)', () => { + const logs = [ + { + type: 'step_recorded', + data: { + step_id: 'step1-uuid', + step_number: 1, + pre_image_name: 'img1', + post_image_name: 'img2', + action_taken: { action: 'manage_app', app_name: 'Maps' }, + timestamp: 100 + } + }, + { + type: 'step_recorded', + data: { + step_id: 'step2-uuid', + step_number: 2, + pre_image_name: 'img2', + post_image_name: 'img3', + action_taken: { action: 'input_text', text: 'Coffee' }, + timestamp: 200 + } + }, + { + type: 'step_recorded', + data: { + step_id: 'step3-uuid', + step_number: 3, + pre_image_name: 'img3', + post_image_name: 'img4', + action_taken: { action: 'press_key', key: 'ENTER' }, + timestamp: 300 + } + }, + // VisualStepSummarizer sends step_updated with only step_id and summary (no step_number) + { + type: 'step_updated', + data: { + step_id: 'step1-uuid', + summary: 'In Step 1, I launched Google Maps' + } + }, + { + type: 'step_updated', + data: { + step_id: 'step2-uuid', + summary: 'In Step 2, I typed Coffee' + } + }, + { + type: 'step_updated', + data: { + step_id: 'step3-uuid', + summary: 'In Step 3, I pressed ENTER' + } + } + ]; + + const frames = extractStepReplayFrames(logs); + expect(frames.length).toBe(3); + + // Frame 1 MUST start at Step 1, NOT Step 4! + expect(frames[0].stepNumber).toBe(1); + expect(frames[0].index).toBe(0); + expect(frames[0].title).toContain('Step 1'); + expect(frames[0].summary).toBe('In Step 1, I launched Google Maps'); + expect(frames[0].preImageUrl).toBe('/images/img1'); + + // Frame 2 MUST be Step 2 + expect(frames[1].stepNumber).toBe(2); + expect(frames[1].index).toBe(1); + expect(frames[1].title).toContain('Step 2'); + expect(frames[1].summary).toBe('In Step 2, I typed Coffee'); + + // Frame 3 MUST be Step 3 + expect(frames[2].stepNumber).toBe(3); + expect(frames[2].index).toBe(2); + expect(frames[2].title).toContain('Step 3'); + expect(frames[2].summary).toBe('In Step 3, I pressed ENTER'); + }); + + it('should filter non-visual steps (like pre-planning step 0) without shifting 1-based replay numbering', () => { + const logs = [ + // Virtual step 0 (pre-planning) with no screenshots + { + step_id: 'pre-planning', + step_number: 0, + summary: 'Planning the mission', + timestamp: 50, + pre_image_name: null, + post_image_name: null, + generic_tools: [{ name: 'save_note', type: 'tool' }] + }, + // Real step 1 with screenshots + { + step_id: 'real-step-1', + step_number: 1, + summary: 'Tapping search bar', + action_taken: { action: 'click', coordinates: [100, 200] }, + pre_image_name: 'shot1', + timestamp: 100 + }, + // Real step 2 with screenshots + { + step_id: 'real-step-2', + step_number: 2, + summary: 'Entering search keywords', + action_taken: { action: 'input_text', text: 'latte' }, + pre_image_name: 'shot2', + timestamp: 200 + } + ]; + + const frames = extractStepReplayFrames(logs); + expect(frames.length).toBe(2); + + // Frame 0 must be labeled Step 1 (not Step 2) + expect(frames[0].stepNumber).toBe(1); + expect(frames[0].rawStepNumber).toBe(1); + expect(frames[0].stepId).toBe('real-step-1'); + + // Frame 1 must be labeled Step 2 + expect(frames[1].stepNumber).toBe(2); + expect(frames[1].rawStepNumber).toBe(2); + expect(frames[1].stepId).toBe('real-step-2'); + }); }); diff --git a/apps/showcase_ui/src/app/utils/action-formatter.util.ts b/apps/showcase_ui/src/app/utils/action-formatter.util.ts index 4361e41..1966e45 100644 --- a/apps/showcase_ui/src/app/utils/action-formatter.util.ts +++ b/apps/showcase_ui/src/app/utils/action-formatter.util.ts @@ -798,73 +798,137 @@ export function extractStepReplayFrames(logsOrSteps: any[]): StepReplayFrame[] { } } - // Deduplicate and merge by step_number (fallback to step_id if step_number missing) - const stepMap = new Map(); - for (const step of rawSteps) { - const key = (step.step_number !== undefined && step.step_number !== null) - ? `step_${step.step_number}` - : (step.step_id || Math.random()); - if (key !== undefined) { - const existing = stepMap.get(key); - if (existing) { - // When merging duplicate records with the same step_number: - // Prefer real actions over report_task_status - const existingAct = existing.action_taken; - const newAct = step.action_taken; - const preferExistingAction = existingAct && isAndroidAction(existingAct) && isReportStatusAction(newAct); - const preferNewAction = newAct && isAndroidAction(newAct) && isReportStatusAction(existingAct); + // Unified dual-indexed step store: correlates by step_id and step_number + const stepsList: any[] = []; + const idToIndex = new Map(); + const numToIndex = new Map(); - stepMap.set(key, { - ...existing, - ...step, - step_number: step.step_number !== undefined ? step.step_number : existing.step_number, - pre_image_name: (preferExistingAction ? existing.pre_image_name : (step.pre_image_name || existing.pre_image_name)), - post_image_name: (preferExistingAction ? (existing.post_image_name || step.post_image_name) : (step.post_image_name || existing.post_image_name)), - action_taken: preferExistingAction ? existing.action_taken : (preferNewAction ? newAct : (step.action_taken || existing.action_taken)), - last_execution_result: step.last_execution_result || existing.last_execution_result, - summary: (preferExistingAction ? (existing.summary || step.summary) : (step.summary || existing.summary)), - generic_tools: step.generic_tools || existing.generic_tools - }); - } else { - stepMap.set(key, { ...step }); + for (const step of rawSteps) { + const stepId = (step.step_id !== undefined && step.step_id !== null && String(step.step_id).trim() !== '') + ? String(step.step_id).trim() + : null; + const stepNum = (step.step_number !== undefined && step.step_number !== null) + ? Number(step.step_number) + : null; + + let targetIndex = -1; + if (stepId && idToIndex.has(stepId)) { + targetIndex = idToIndex.get(stepId)!; + } else if (stepNum !== null && numToIndex.has(stepNum)) { + targetIndex = numToIndex.get(stepNum)!; + } + + if (targetIndex >= 0) { + const existing = stepsList[targetIndex]; + + // When merging duplicate records (e.g. status reports vs real physical actions): + const existingAct = existing.action_taken; + const newAct = step.action_taken; + const preferExistingAction = existingAct && isAndroidAction(existingAct) && isReportStatusAction(newAct); + const preferNewAction = newAct && isAndroidAction(newAct) && isReportStatusAction(existingAct); + + const mergedStepNum = (stepNum !== null) ? stepNum : existing.step_number; + const mergedStepId = existing.step_id || step.step_id; + + const merged = { + ...existing, + ...step, + step_id: mergedStepId, + step_number: mergedStepNum, + pre_image_name: preferExistingAction + ? existing.pre_image_name + : (step.pre_image_name || existing.pre_image_name), + post_image_name: preferExistingAction + ? (existing.post_image_name || step.post_image_name) + : (step.post_image_name || existing.post_image_name), + pre_screenshot: preferExistingAction + ? existing.pre_screenshot + : (step.pre_screenshot || existing.pre_screenshot), + post_screenshot: preferExistingAction + ? (existing.post_screenshot || step.post_screenshot) + : (step.post_screenshot || existing.post_screenshot), + action_taken: preferExistingAction + ? existing.action_taken + : (preferNewAction ? newAct : (step.action_taken || existing.action_taken)), + last_execution_result: step.last_execution_result || existing.last_execution_result, + summary: preferExistingAction + ? (existing.summary || step.summary) + : (step.summary || existing.summary), + operator_raw_thinking: step.operator_raw_thinking || existing.operator_raw_thinking, + operator_native_thinking: step.operator_native_thinking || existing.operator_native_thinking, + generic_tools: step.generic_tools || existing.generic_tools, + timestamp: existing.timestamp || step.timestamp + }; + + stepsList[targetIndex] = merged; + if (mergedStepId) idToIndex.set(String(mergedStepId), targetIndex); + if (mergedStepNum !== undefined && mergedStepNum !== null) { + numToIndex.set(Number(mergedStepNum), targetIndex); } + } else { + const newIndex = stepsList.length; + stepsList.push({ ...step }); + if (stepId) idToIndex.set(stepId, newIndex); + if (stepNum !== null) numToIndex.set(stepNum, newIndex); } } - // Sort strictly by step_number ascending - const steps = Array.from(stepMap.values()).sort((a, b) => { - const numA = Number(a.step_number ?? 0); - const numB = Number(b.step_number ?? 0); + // Pre-resolve candidate images and actions for every step + interface VisualCandidate { + stepData: any; + act: any; + preUrl: string | null; + postUrl: string | null; + primaryImg: string; + } + + const visualCandidates: VisualCandidate[] = []; + + for (const stepData of stepsList) { + const act = stepData.action_taken; + const preUrl = getStepPreImageUrl(stepData, act); + const postUrl = getStepPostImageUrl(stepData, act); + const primaryImg = preUrl || postUrl; + + // Only steps with a resolvable screen image qualify as replay frames + if (primaryImg) { + visualCandidates.push({ + stepData, + act, + preUrl, + postUrl, + primaryImg + }); + } + } + + // Sort strictly by original step_number ascending, then by timestamp + visualCandidates.sort((a, b) => { + const numA = Number(a.stepData.step_number ?? 99999); + const numB = Number(b.stepData.step_number ?? 99999); if (numA !== numB) return numA - numB; - return Number(a.timestamp ?? 0) - Number(b.timestamp ?? 0); + return Number(a.stepData.timestamp ?? 0) - Number(b.stepData.timestamp ?? 0); }); + // Construct guaranteed sequential 1-indexed frames const frames: StepReplayFrame[] = []; - for (let i = 0; i < steps.length; i++) { - const stepData = steps[i]; - // Guarantee 1-indexed sequential step numbering so frames never duplicate numbers - const stepNum = i + 1; - const act = stepData.action_taken; + for (let i = 0; i < visualCandidates.length; i++) { + const { stepData, act, preUrl, postUrl, primaryImg } = visualCandidates[i]; + const stepNum = i + 1; // 1-indexed sequential frame number (1, 2, 3...) const title = act ? (getActionTitle(act) || 'Action') : `Step ${stepNum}`; const coords = act ? getActionCoords(act) : ''; const targetText = act ? getActionTargetText(act) : ''; const actionDesc = coords ? `${title} (${coords})` : (targetText ? `${title} (${targetText})` : title); - - const preUrl = getStepPreImageUrl(stepData, act); - const postUrl = getStepPostImageUrl(stepData, act); - - // Each step gets exactly one primary frame (Pre-action screenshot where the action is taken) - const primaryImg = preUrl || postUrl; - if (!primaryImg) { - continue; - } - const failed = isActionFailed(act, stepData); + const rawStepNum = (stepData.step_number !== undefined && stepData.step_number !== null) + ? Number(stepData.step_number) + : stepNum; frames.push({ - index: frames.length, + index: i, stepNumber: stepNum, + rawStepNumber: rawStepNum, stepId: String(stepData.step_id || `step-${stepNum}`), title: `Step ${stepNum}: ${title}`, actionText: actionDesc, @@ -877,6 +941,7 @@ export function extractStepReplayFrames(logsOrSteps: any[]): StepReplayFrame[] { postImageUrl: postUrl, isPost: false, timestamp: stepData.timestamp, + summary: stepData.summary || '', status: failed ? 'failed' : 'success' }); } diff --git a/artemis/data_engine/engine.py b/artemis/data_engine/engine.py index 97e02ff..93883ca 100644 --- a/artemis/data_engine/engine.py +++ b/artemis/data_engine/engine.py @@ -79,6 +79,7 @@ class DataEngine: self._accumulated_logs = {} self._bg_task_to_trace_id = {} self._trace_name_cache = {} + self._step_number_cache: dict[str, int] = {} # Background tasks are written directly to SQLite storage self.ipc_socket = None @@ -284,8 +285,13 @@ class DataEngine: except Exception: existing_steps = [] if existing_steps: + self._step_number_cache.clear() + for s in existing_steps: + if s.step_id and s.step_number is not None: + self._step_number_cache[str(s.step_id)] = s.step_number self.current_step_number = max([s.step_number for s in existing_steps], default=0) else: + self._step_number_cache.clear() self.current_step_number = 0 logger.info(f"Session started: {session_id} (current step counter: {self.current_step_number})") self._publish("session_started", session.model_dump()) @@ -505,6 +511,7 @@ class DataEngine: step_id = self.current_step_id or uuid4() self.last_recorded_step_id = step_id + self._step_number_cache[str(step_id)] = step_number # Reset current_step_id so subsequent steps will allocate a new one self.current_step_id = None @@ -890,11 +897,15 @@ class DataEngine: self._run_in_background(_update_and_write) + step_num = self.get_step_number(step_id) update_payload = { "step_id": str(step_id), "action_taken": action_taken, "generic_tools": self._get_generic_tools_for_sse(step_id), } + if step_num is not None: + update_payload["step_number"] = step_num + tokens = self._get_step_token_usage_for_sse(step_id) if tokens: update_payload["token_usage"] = tokens @@ -902,6 +913,27 @@ class DataEngine: self._publish("step_updated", update_payload) + def get_step_number(self, step_id: UUID | int | str | None) -> int | None: + """Resolve the 1-based sequential step number for a step ID.""" + if step_id is None: + return None + if isinstance(step_id, int): + return step_id + sid = str(step_id) + cached = self._step_number_cache.get(sid) + if cached is not None: + return cached + if self.storage and hasattr(self.storage, "get_step"): + try: + target_uuid = UUID(sid) if isinstance(step_id, str) else step_id + step_record = self.storage.get_step(target_uuid) + if step_record and getattr(step_record, "step_number", None) is not None: + self._step_number_cache[sid] = int(step_record.step_number) + return int(step_record.step_number) + except Exception: + pass + return None + def update_step_summary(self, step_id: UUID | int | str, summary: str): """Update the step with a concise summary in background.""" target_uuid = None @@ -925,26 +957,28 @@ class DataEngine: return self._run_in_background(self.storage.update_step_summary, target_uuid, summary) - self._publish( - "step_updated", - { - "step_id": str(target_uuid), - "summary": summary, - "generic_tools": self._get_generic_tools_for_sse(target_uuid), - }, - ) + step_num = self.get_step_number(target_uuid if not isinstance(step_id, int) else step_id) + payload = { + "step_id": str(target_uuid), + "summary": summary, + "generic_tools": self._get_generic_tools_for_sse(target_uuid), + } + if step_num is not None: + payload["step_number"] = step_num + self._publish("step_updated", payload) def update_step_thinking(self, step_id: UUID, operator_raw_thinking: str): """Update step's thinking in SQLite in background.""" self._run_in_background(self.storage.update_step_thinking, step_id, operator_raw_thinking) - self._publish( - "step_updated", - { - "step_id": str(step_id), - "operator_raw_thinking": operator_raw_thinking, - "generic_tools": self._get_generic_tools_for_sse(step_id), - }, - ) + step_num = self.get_step_number(step_id) + payload = { + "step_id": str(step_id), + "operator_raw_thinking": operator_raw_thinking, + "generic_tools": self._get_generic_tools_for_sse(step_id), + } + if step_num is not None: + payload["step_number"] = step_num + self._publish("step_updated", payload) def update_step_native_thinking(self, step_id: UUID, operator_native_thinking: str): """Update step's native thinking in SQLite in background.""" @@ -953,14 +987,15 @@ class DataEngine: step_id, operator_native_thinking, ) - self._publish( - "step_updated", - { - "step_id": str(step_id), - "operator_native_thinking": operator_native_thinking, - "generic_tools": self._get_generic_tools_for_sse(step_id), - }, - ) + step_num = self.get_step_number(step_id) + payload = { + "step_id": str(step_id), + "operator_native_thinking": operator_native_thinking, + "generic_tools": self._get_generic_tools_for_sse(step_id), + } + if step_num is not None: + payload["step_number"] = step_num + self._publish("step_updated", payload) def update_step_execution_result( self, @@ -975,15 +1010,16 @@ class DataEngine: last_execution_result, post_image_name, ) - self._publish( - "step_updated", - { - "step_id": str(step_id), - "last_execution_result": last_execution_result, - "post_image_name": post_image_name, - "generic_tools": self._get_generic_tools_for_sse(step_id), - }, - ) + step_num = self.get_step_number(step_id) + payload = { + "step_id": str(step_id), + "last_execution_result": last_execution_result, + "post_image_name": post_image_name, + "generic_tools": self._get_generic_tools_for_sse(step_id), + } + if step_num is not None: + payload["step_number"] = step_num + self._publish("step_updated", payload) def get_relative_time(self, timestamp: float) -> str: """Compute relative time since session start.""" diff --git a/tests/unit/data_engine/test_engine.py b/tests/unit/data_engine/test_engine.py index 141926c..d65b944 100644 --- a/tests/unit/data_engine/test_engine.py +++ b/tests/unit/data_engine/test_engine.py @@ -326,3 +326,40 @@ def test_resumed_session_monotonic_step_numbering(tmp_path): assert len(steps) == 3 assert [s.step_number for s in steps] == [1, 2, 3] + +def test_update_step_summary_includes_step_number_in_sse(tmp_path): + mock_ctx = MagicMock(spec=ArtemisContext) + mock_execution_setup = MagicMock() + mock_execution_setup.traces_path = str(tmp_path) + mock_ctx.execution_setup = mock_execution_setup + mock_ctx.device = None + + published_events = [] + + def mock_subscriber(event_type, data): + published_events.append((event_type, data)) + + engine = DataEngine(mock_ctx) + engine.subscribe(mock_subscriber) + sid = engine.start_session("SSE step_number verification") + + step_id = engine.record_step(summary="Initial step 1") + for t in list(engine._pending_threads): + t.join() + + assert engine.get_step_number(step_id) == 1 + + # Call update_step_summary + engine.update_step_summary(step_id, "Updated concise summary for Step 1") + for t in list(engine._pending_threads): + t.join() + + # Find the published step_updated event + step_updated_events = [e for e in published_events if e[0] == "step_updated"] + assert len(step_updated_events) >= 1 + last_update = step_updated_events[-1][1] + assert last_update["step_id"] == str(step_id) + assert last_update["summary"] == "Updated concise summary for Step 1" + assert last_update["step_number"] == 1 + +