vik 0.1.0

Vik is an issue-driven coding workflow automation tool.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142

//! Typed orchestrator event channel.
//!
//! One bounded mpsc channel carries every signal: producers are cloned
//! into background tasks, the single consumer lives on the main loop.
//! Event-style coordination (instead of callbacks) keeps every state
//! transition visible at one site — the only place [`super::running::RunningMap`]
//! is mutated.

use std::path::PathBuf;

use tokio::sync::mpsc;

use crate::context::Issue;
use crate::logging::Phase;
use crate::session::{Session, SessionSnapshot};

use super::types::{IssueStage, IssueStageKey};

/// Bounded so a slow main loop applies backpressure to producers. 256 is
/// large enough to swallow a normal intake burst without forcing intake
/// to await mid-cycle.
const EVENT_BUFFER: usize = 256;

#[derive(Clone)]
pub(super) struct EventProducer {
  sender: mpsc::Sender<OrchestratorEvent>,
}

impl EventProducer {
  pub(super) async fn intake_issue(&self, issue: Issue) {
    self.send(OrchestratorEvent::Intake(IntakeEvent::Issue(issue))).await;
  }

  pub(super) async fn intake_failed(&self, error: impl ToString) {
    self
      .send(OrchestratorEvent::Intake(IntakeEvent::Failed(error.to_string())))
      .await;
  }

  pub(super) async fn intake_stopped(&self) {
    self.send(OrchestratorEvent::Intake(IntakeEvent::Stopped)).await;
  }

  pub(super) async fn issue_ready(&self, issue_stages: Vec<IssueStage>, issue_workdir: PathBuf) {
    self
      .send(OrchestratorEvent::Stage(StageEvent::IssueReady {
        issue_stages,
        issue_workdir,
      }))
      .await;
  }

  pub(super) async fn stage_started(&self, issue_stage: IssueStage, session: Session) {
    self
      .send(OrchestratorEvent::Stage(StageEvent::Started {
        issue_stage: Box::new(issue_stage),
        session,
      }))
      .await;
  }

  pub(super) async fn stage_snapshot(&self, key: IssueStageKey, snapshot: SessionSnapshot) {
    self
      .send(OrchestratorEvent::Stage(StageEvent::Snapshot { key, snapshot }))
      .await;
  }

  pub(super) async fn stage_terminal(&self, key: IssueStageKey, snapshot: SessionSnapshot) {
    self
      .send(OrchestratorEvent::Stage(StageEvent::Terminal { key, snapshot }))
      .await;
  }

  /// Pre-session failure path: there is no `Session`, only the reserved
  /// key, so the main loop can release the reservation.
  pub(super) async fn stage_failed(&self, key: IssueStageKey, error: impl ToString) {
    self
      .send(OrchestratorEvent::Stage(StageEvent::Failed {
        key,
        error: error.to_string(),
      }))
      .await;
  }

  async fn send(&self, event: OrchestratorEvent) {
    if self.sender.send(event).await.is_err() {
      tracing::debug!(phase = %Phase::Dispatch, "orchestrator event receiver dropped");
    }
  }
}

pub(super) struct EventConsumer {
  receiver: mpsc::Receiver<OrchestratorEvent>,
}

impl EventConsumer {
  pub(super) async fn recv(&mut self) -> Option<OrchestratorEvent> {
    self.receiver.recv().await
  }
}

pub(super) fn event_channel() -> (EventProducer, EventConsumer) {
  let (sender, receiver) = mpsc::channel(EVENT_BUFFER);
  (EventProducer { sender }, EventConsumer { receiver })
}

pub(super) enum OrchestratorEvent {
  Intake(IntakeEvent),
  Stage(StageEvent),
}

pub(super) enum IntakeEvent {
  Issue(Issue),
  /// Recoverable error during one cycle — the loop keeps going.
  Failed(String),
  /// Natural end (max iterations or cooperative shutdown). Triggers the
  /// main loop's drain check.
  Stopped,
}

pub(super) enum StageEvent {
  IssueReady {
    issue_stages: Vec<IssueStage>,
    issue_workdir: PathBuf,
  },
  Started {
    issue_stage: Box<IssueStage>,
    session: Session,
  },
  Snapshot {
    key: IssueStageKey,
    snapshot: SessionSnapshot,
  },
  Terminal {
    key: IssueStageKey,
    snapshot: SessionSnapshot,
  },
  Failed {
    key: IssueStageKey,
    error: String,
  },
}