stasis/application/runtime/
grapheme_textops_job_handler.rs

1use std::collections::HashMap;
2use std::sync::Arc;
3
4use async_trait::async_trait;
5use serde::Deserialize;
6use serde_json::json;
7
8use crate::application::runtime::grapheme_job_handler::GraphemeJobHandler;
9use crate::application::runtime::in_memory_runtime::{JobExecutionOutcome, JobHandler};
10use crate::domain::errors::Result;
11use crate::domain::runtime::job::Job;
12use crate::ports::outbound::runtime::workflow_engine::WorkflowEngine;
13
14const MAX_TEXT_LEN: usize = 4096;
15const DEFAULT_MAX_ITEMS: usize = 3;
16const MAX_ITEMS_LIMIT: usize = 10;
17
18#[derive(Clone, Deserialize)]
19#[serde(rename_all = "snake_case")]
20enum TextOpsMode {
21    Summarize,
22    ExtractKeywords,
23}
24
25#[derive(Deserialize)]
26struct TextOpsPayload {
27    mode: TextOpsMode,
28    text: String,
29    max_items: Option<usize>,
30}
31
32pub struct GraphemeTextOpsJobHandler {
33    delegate: GraphemeJobHandler,
34}
35
36impl GraphemeTextOpsJobHandler {
37    pub fn new(engine: Arc<dyn WorkflowEngine>) -> Self {
38        Self {
39            delegate: GraphemeJobHandler::new(engine),
40        }
41    }
42
43    fn build_failure(message: String) -> JobExecutionOutcome {
44        let diagnostics = json!({
45            "provider": "grapheme-sdk",
46            "status": "failure",
47            "guardrail_code": "POLICY_VIOLATION",
48            "policy_reason": &message,
49        })
50        .to_string();
51
52        JobExecutionOutcome::FatalFailure {
53            message,
54            execution_id: None,
55            diagnostics: Some(diagnostics),
56        }
57    }
58
59    fn parse_payload(raw: &str) -> std::result::Result<TextOpsPayload, String> {
60        let payload: TextOpsPayload = serde_json::from_str(raw)
61            .map_err(|err| format!("policy violation: invalid textops payload json: {err}"))?;
62
63        if payload.text.trim().is_empty() {
64            return Err("policy violation: textops payload.text must be non-empty".to_string());
65        }
66
67        if payload.text.len() > MAX_TEXT_LEN {
68            return Err(format!(
69                "policy violation: textops payload.text exceeds max length {}",
70                MAX_TEXT_LEN
71            ));
72        }
73
74        let max_items = payload.max_items.unwrap_or(DEFAULT_MAX_ITEMS);
75        if !(1..=MAX_ITEMS_LIMIT).contains(&max_items) {
76            return Err(format!(
77                "policy violation: textops payload.max_items must be between 1 and {}",
78                MAX_ITEMS_LIMIT
79            ));
80        }
81
82        Ok(TextOpsPayload {
83            mode: payload.mode,
84            text: payload.text,
85            max_items: Some(max_items),
86        })
87    }
88
89    fn summarize(text: &str, max_items: usize) -> String {
90        let mut sentences = Vec::new();
91        let mut current = String::new();
92
93        for ch in text.chars() {
94            current.push(ch);
95            if matches!(ch, '.' | '!' | '?') {
96                let sentence = current.trim();
97                if !sentence.is_empty() {
98                    sentences.push(sentence.to_string());
99                }
100                current.clear();
101            }
102        }
103
104        if sentences.is_empty() {
105            let fallback = text
106                .split_whitespace()
107                .take(24)
108                .collect::<Vec<_>>()
109                .join(" ");
110            return fallback;
111        }
112
113        sentences
114            .into_iter()
115            .take(max_items)
116            .collect::<Vec<_>>()
117            .join(" ")
118    }
119
120    fn extract_keywords(text: &str, max_items: usize) -> String {
121        let stop_words = [
122            "the", "and", "for", "with", "that", "this", "from", "into", "have", "are", "was",
123            "were", "you", "your", "our", "not", "but", "can", "will", "all",
124        ];
125
126        let mut counts: HashMap<String, usize> = HashMap::new();
127        for token in text
128            .split(|c: char| !c.is_alphanumeric())
129            .map(str::to_lowercase)
130            .filter(|word| word.len() >= 4)
131        {
132            if stop_words.contains(&token.as_str()) {
133                continue;
134            }
135            *counts.entry(token).or_insert(0) += 1;
136        }
137
138        let mut ranked = counts.into_iter().collect::<Vec<_>>();
139        ranked.sort_by(|(a_word, a_count), (b_word, b_count)| {
140            b_count.cmp(a_count).then_with(|| a_word.cmp(b_word))
141        });
142
143        ranked
144            .into_iter()
145            .take(max_items)
146            .map(|(word, _)| word)
147            .collect::<Vec<_>>()
148            .join(", ")
149    }
150
151    fn transform(mode: TextOpsMode, text: &str, max_items: usize) -> String {
152        match mode {
153            TextOpsMode::Summarize => Self::summarize(text, max_items),
154            TextOpsMode::ExtractKeywords => Self::extract_keywords(text, max_items),
155        }
156    }
157
158    fn build_inline_source(message: &str) -> String {
159        let cleaned = message
160            .replace('"', "'")
161            .replace(['\n', '\r'], " ");
162
163        format!(
164            "import core from \"grapheme/core\"\n\nquery TextOps {{\n  core.echo(message: \"{}\") {{\n    state {{ current }}\n  }}\n}}\n",
165            cleaned
166        )
167    }
168}
169
170#[async_trait]
171impl JobHandler for GraphemeTextOpsJobHandler {
172    fn job_type(&self) -> &'static str {
173        "workflow.grapheme.textops"
174    }
175
176    async fn execute(&self, job: &Job) -> Result<JobExecutionOutcome> {
177        let payload = match Self::parse_payload(&job.payload_ref) {
178            Ok(payload) => payload,
179            Err(message) => return Ok(Self::build_failure(message)),
180        };
181
182        let transformed = Self::transform(
183            payload.mode,
184            &payload.text,
185            payload.max_items.unwrap_or(DEFAULT_MAX_ITEMS),
186        );
187        if transformed.trim().is_empty() {
188            return Ok(Self::build_failure(
189                "policy violation: textops transform produced empty output".to_string(),
190            ));
191        }
192
193        let source = Self::build_inline_source(&transformed);
194        let synthetic_job = Job {
195            payload_ref: format!("grapheme:inline:{}", source),
196            ..job.clone()
197        };
198
199        self.delegate.execute(&synthetic_job).await
200    }
201}
stasis/application/runtime/grapheme_textops_job_handler.rs

stasis/application/runtime/
grapheme_textops_job_handler.rs
