dk_protocol/
verify.rs

1use tokio::sync::mpsc;
2use tonic::Status;
3use uuid::Uuid;
4
5use dk_runner::executor::process::ProcessExecutor;
6use dk_runner::Runner;
7
8use crate::server::ProtocolServer;
9use crate::{VerifyRequest, VerifyStepResult};
10use crate::proto::dkod::v1::Finding as ProtoFinding;
11use crate::proto::dkod::v1::Suggestion as ProtoSuggestion;
12
13pub async fn handle_verify(
14    server: &ProtocolServer,
15    req: VerifyRequest,
16    tx: mpsc::Sender<Result<VerifyStepResult, Status>>,
17) {
18    // Validate session
19    let session = match server.validate_session(&req.session_id) {
20        Ok(s) => s,
21        Err(e) => {
22            let _ = tx.send(Err(e)).await;
23            return;
24        }
25    };
26    if let Err(e) = crate::require_live_session::require_live_session(server, &req.session_id).await {
27        let _ = tx.send(Err(e)).await;
28        return;
29    }
30
31    let changeset_id = match req.changeset_id.parse::<Uuid>() {
32        Ok(id) => id,
33        Err(_) => {
34            let _ = tx
35                .send(Err(Status::invalid_argument("invalid changeset_id")))
36                .await;
37            return;
38        }
39    };
40
41    // Verify changeset exists and update status to verifying
42    let engine = server.engine();
43    {
44        if let Err(e) = engine.changeset_store().get(changeset_id).await {
45            let _ = tx.send(Err(Status::not_found(e.to_string()))).await;
46            return;
47        }
48        let _ = engine
49            .changeset_store()
50            .update_status(changeset_id, "verifying")
51            .await;
52    }
53
54    // Resolve repo_id for enriched events
55    let repo_id_str = match engine.get_repo(&session.codebase).await {
56        Ok((rid, _)) => rid.to_string(),
57        Err(_) => String::new(),
58    };
59
60    // Publish verify_started event
61    server.event_bus().publish(crate::WatchEvent {
62        event_type: "changeset.verify_started".to_string(),
63        changeset_id: changeset_id.to_string(),
64        agent_id: session.agent_id.clone(),
65        affected_symbols: vec![],
66        details: String::new(),
67        session_id: req.session_id.clone(),
68        affected_files: vec![],
69        symbol_changes: vec![],
70        repo_id: repo_id_str.clone(),
71        event_id: Uuid::new_v4().to_string(),
72    });
73
74    // Create runner with process executor
75    let runner = Runner::new(
76        server.engine.clone(),
77        Box::new(ProcessExecutor::new()),
78    );
79
80    // Bridge dk-runner StepResults to gRPC VerifyStepResults
81    let (runner_tx, mut runner_rx) = tokio::sync::mpsc::channel(32);
82
83    let codebase = session.codebase.clone();
84    let grpc_tx = tx.clone();
85
86    // Spawn runner in background
87    let runner_handle = tokio::spawn(async move {
88        runner.verify(changeset_id, &codebase, runner_tx).await
89    });
90
91    // Forward results from runner to gRPC stream
92    let mut step_counter = 0i32;
93    while let Some(result) = runner_rx.recv().await {
94        step_counter += 1;
95
96        let step_status_str = result.status.as_str().to_string();
97        let step_name_str = result.step_name.clone();
98
99        let findings: Vec<ProtoFinding> = result.findings.iter().map(|f| {
100            ProtoFinding {
101                severity: f.severity.as_str().to_string(),
102                check_name: f.check_name.clone(),
103                message: f.message.clone(),
104                file_path: f.file_path.clone(),
105                line: f.line,
106                symbol: f.symbol.clone(),
107            }
108        }).collect();
109
110        let suggestions: Vec<ProtoSuggestion> = result.suggestions.iter().map(|s| {
111            ProtoSuggestion {
112                finding_index: s.finding_index as u32,
113                description: s.description.clone(),
114                file_path: s.file_path.clone(),
115                replacement: s.replacement.clone(),
116            }
117        }).collect();
118
119        let _ = grpc_tx
120            .send(Ok(VerifyStepResult {
121                step_order: step_counter,
122                step_name: result.step_name,
123                status: result.status.as_str().to_string(),
124                output: result.output,
125                required: result.required,
126                findings,
127                suggestions,
128            }))
129            .await;
130
131        // Publish verify_step event for each step
132        server.event_bus().publish(crate::WatchEvent {
133            event_type: "changeset.verify_step".to_string(),
134            changeset_id: changeset_id.to_string(),
135            agent_id: session.agent_id.clone(),
136            affected_symbols: vec![],
137            details: format!("{}:{}", step_name_str, step_status_str),
138            session_id: req.session_id.clone(),
139            affected_files: vec![],
140            symbol_changes: vec![],
141            repo_id: repo_id_str.clone(),
142            event_id: Uuid::new_v4().to_string(),
143        });
144    }
145
146    // Get final result and update changeset status
147    let final_status = match runner_handle.await {
148        Ok(Ok(passed)) => if passed { "approved" } else { "rejected" },
149        Ok(Err(e)) => {
150            tracing::error!("runner error: {e}");
151            "rejected"
152        }
153        Err(e) => {
154            tracing::error!("runner task panicked: {e}");
155            "rejected"
156        }
157    };
158
159    let _ = engine
160        .changeset_store()
161        .update_status(changeset_id, final_status)
162        .await;
163
164    // Publish verified event with final status
165    server.event_bus().publish(crate::WatchEvent {
166        event_type: "changeset.verified".to_string(),
167        changeset_id: changeset_id.to_string(),
168        agent_id: session.agent_id.clone(),
169        affected_symbols: vec![],
170        details: final_status.to_string(),
171        session_id: req.session_id.clone(),
172        affected_files: vec![],
173        symbol_changes: vec![],
174        repo_id: repo_id_str.clone(),
175        event_id: Uuid::new_v4().to_string(),
176    });
177}
178
179// ── Event type constants ────────────────────────────────────────────
180// Extracted from the handler above so they can be tested and referenced
181// by other modules without string duplication.
182
183/// Event published when verification begins.
184pub const EVENT_VERIFY_STARTED: &str = "changeset.verify_started";
185/// Event published after each verification step completes.
186pub const EVENT_VERIFY_STEP: &str = "changeset.verify_step";
187/// Event published when the entire verification pipeline finishes.
188pub const EVENT_VERIFIED: &str = "changeset.verified";
189
190#[cfg(test)]
191mod tests {
192    use super::*;
193
194    #[test]
195    fn verify_started_event_type() {
196        assert_eq!(EVENT_VERIFY_STARTED, "changeset.verify_started");
197    }
198
199    #[test]
200    fn verify_step_event_type() {
201        assert_eq!(EVENT_VERIFY_STEP, "changeset.verify_step");
202    }
203
204    #[test]
205    fn verified_event_type() {
206        assert_eq!(EVENT_VERIFIED, "changeset.verified");
207    }
208
209    #[test]
210    fn verify_event_types_use_dot_separator() {
211        for event in [EVENT_VERIFY_STARTED, EVENT_VERIFY_STEP, EVENT_VERIFIED] {
212            assert!(
213                event.contains('.'),
214                "event type '{}' should use dot separator",
215                event
216            );
217            assert!(
218                event.starts_with("changeset."),
219                "event type '{}' should start with 'changeset.'",
220                event
221            );
222        }
223    }
224
225    #[test]
226    fn verify_event_types_are_distinct() {
227        let events = [EVENT_VERIFY_STARTED, EVENT_VERIFY_STEP, EVENT_VERIFIED];
228        for i in 0..events.len() {
229            for j in (i + 1)..events.len() {
230                assert_ne!(
231                    events[i], events[j],
232                    "event types should be distinct: '{}' vs '{}'",
233                    events[i], events[j]
234                );
235            }
236        }
237    }
238}
dk_protocol/verify.rs

dk_protocol/
verify.rs
