Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.d/analysis-run-retry-lineage-consumer-parity.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- `tepp_api` adds `lineageweave_analysis_run_retry_lineage_exchange`, Naruon compatibility-listener retry-lineage GET (empty `retries` on accepted creates), and a `tepp-loopback` TCP create-cancel-retry-inspect proof (ADR 0045). Metric-free retry-lineage fields are unchanged from ADR 0035. Not GET status, not lifecycle POST, not an ADR 0014 claim.
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@ All notable changes to TEPP are documented here. The format follows Keep a Chang

## [Unreleased]

- Loopback retry-lineage consumer parity: LineageWeave GET-retries exchange, Naruon compatibility-listener inspect (empty `retries` on accepted creates), and `tepp-loopback` TCP create-cancel-retry-inspect proof (ADR 0045).

- `tepp_api` serves `GET /v1/analysis-runs` on the shared loopback listener (ADR 0031). Operators enumerate accepted, running, cancelled, and terminal runs as metric-free collection rows. Collection bodies refuse RMSE/bias/coverage/SE-gate/scientific-acceptance/`terminal_result` keys. GET-by-id and running/terminal POST remain later GAP-003A slices; this is not an ADR 0014 claim.

- `tepp_api` serves `POST /v1/analysis-runs/{run_id}/cancel` on the shared loopback listener (ADR 0029). Accepted and running runs become metric-free `cancelled` status. Succeeded, failed, and unknown runs cannot be cancelled. Cancel bodies refuse RMSE/bias/coverage/SE-gate/scientific-acceptance keys. GET status and running/terminal POST remain later GAP-003A slices; this is not an ADR 0014 claim.
Expand Down
1 change: 1 addition & 0 deletions DOCUMENTATION.md
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ TEPP's approved PRD v0.4 and implementation plan are the primary product baselin
| Analysis-run retry HTTP doctoring | [`docs/research/analysis-run-retry-http.md`](docs/research/analysis-run-retry-http.md) |
| Analysis-run stored-request HTTP doctoring | [`docs/research/analysis-run-stored-request-http.md`](docs/research/analysis-run-stored-request-http.md) |
| Analysis-run retry-lineage HTTP doctoring | [`docs/research/analysis-run-retry-lineage-http.md`](docs/research/analysis-run-retry-lineage-http.md) |
| Analysis-run retry-lineage consumer-parity doctoring | [`docs/research/analysis-run-retry-lineage-consumer-parity.md`](docs/research/analysis-run-retry-lineage-consumer-parity.md) |
| UML/runtime/scientific flows | [`docs/UML.md`](docs/UML.md) |
| Logical/physical ERD | [`docs/ERD.md`](docs/ERD.md) |
| Security policy | [`SECURITY.md`](SECURITY.md) |
Expand Down
2 changes: 2 additions & 0 deletions crates/tepp_api/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -215,6 +215,8 @@ pub use lineageweave_http::LINEAGEWEAVE_CONSUMER_CODE;
pub use lineageweave_http::NARUON_CONSUMER_CODE;
/// Build a `LineageWeave` analysis-run exchange without provider credentials.
pub use lineageweave_http::lineageweave_analysis_run_exchange;
/// Build a `LineageWeave` retry-lineage GET without provider credentials.
pub use lineageweave_http::lineageweave_analysis_run_retry_lineage_exchange;
/// Build a `LineageWeave` project-history exchange without provider credentials.
pub use lineageweave_http::lineageweave_project_history_exchange;
/// Build a credential-free `LineageWeave` temporal-context exchange.
Expand Down
67 changes: 66 additions & 1 deletion crates/tepp_api/src/lineageweave_http.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
//! Published modular-consumer identity and `LineageWeave` TEPP exchanges.

use crate::analysis_run_retry_lineage_http::naruon_analysis_run_retry_lineage_exchange;
use crate::naruon_http::compose_https_target;
use crate::project_history::build_project_history_exchange;
use crate::{
Expand Down Expand Up @@ -38,6 +39,29 @@ pub fn lineageweave_analysis_run_exchange(
Ok(exchange)
}

/// Build a `LineageWeave` → TEPP retry-lineage GET without credentials.
///
/// The function reuses TEPP's existing origin and identity validation, then
/// replaces only the published modular-consumer identity. The response remains
/// a metric-free inspect of direct retry children, not a measurement result.
///
/// # Errors
///
/// Returns the same fail-closed errors as [`naruon_analysis_run_retry_lineage_exchange`].
pub fn lineageweave_analysis_run_retry_lineage_exchange(
origin: &str,
run_id: &str,
) -> Result<NaruonHttpExchange, ApiError> {
let mut exchange = naruon_analysis_run_retry_lineage_exchange(origin, run_id)?;
let consumer_header = exchange
.headers
.iter_mut()
.find(|(name, _)| name.eq_ignore_ascii_case("tepp-consumer"))
.ok_or(ApiError::InvalidWirePayload)?;
LINEAGEWEAVE_CONSUMER_CODE.clone_into(&mut consumer_header.1);
Ok(exchange)
}

/// Build a credential-free `LineageWeave` temporal-context exchange.
///
/// # Errors
Expand Down Expand Up @@ -94,7 +118,7 @@ pub(crate) fn consumer_is_supported(consumer_code: &str) -> bool {
mod tests {
use super::{
LINEAGEWEAVE_CONSUMER_CODE, NARUON_CONSUMER_CODE, consumer_is_supported,
lineageweave_analysis_run_exchange,
lineageweave_analysis_run_exchange, lineageweave_analysis_run_retry_lineage_exchange,
};
use crate::{ANALYSIS_RUN_CONTRACT_VERSION, AnalysisRunRequest, ApiError};

Expand Down Expand Up @@ -132,4 +156,45 @@ mod tests {
Err(ApiError::InvalidWirePayload)
);
}
#[test]
fn lineageweave_retry_lineage_exchange_swaps_only_the_consumer_header() {
let exchange = lineageweave_analysis_run_retry_lineage_exchange(
"https://tepp.example.test",
"tepp-run-1",
)
.expect("exchange");
assert_eq!(exchange.method, "GET");
assert_eq!(
exchange.target_url,
"https://tepp.example.test/v1/analysis-runs/tepp-run-1/retries"
);
assert!(exchange.body.is_empty());
assert!(
exchange
.headers
.contains(&("tepp-consumer".into(), LINEAGEWEAVE_CONSUMER_CODE.into()))
);
assert!(
!exchange
.headers
.contains(&("tepp-consumer".into(), NARUON_CONSUMER_CODE.into()))
);
assert!(exchange.headers.iter().all(|(name, _)| {
!matches!(
name.to_ascii_lowercase().as_str(),
"authorization" | "proxy-authorization" | "cookie" | "x-api-key"
)
}));
assert_eq!(
lineageweave_analysis_run_retry_lineage_exchange(
"http://tepp.example.test",
"tepp-run-1"
),
Err(ApiError::InvalidWirePayload)
);
assert_eq!(
lineageweave_analysis_run_retry_lineage_exchange("https://tepp.example.test", ""),
Err(ApiError::InvalidWirePayload)
);
}
}
156 changes: 150 additions & 6 deletions crates/tepp_api/src/naruon_live.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,10 @@ use std::io::{Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::time::Duration;

use crate::analysis_run_retry_lineage_http::{
AnalysisRunRetryLineage, analysis_run_retry_lineage_path_run_id,
refuse_metrics_on_retry_lineage_payload,
};
use crate::authorization::{
AnalyticalPurpose, ExportAuthorizationRequest, authorize_export, require_export_allowed,
};
Expand All @@ -16,7 +20,7 @@ use crate::live_http::{
use crate::naruon_http::{NARUON_ANALYSIS_RUN_PATH, NARUON_EXPORT_PATH};
use crate::wire::{from_json, to_json};
use crate::{
AnalysisRunAccepted, AnalysisRunRequest, ApiError, ErrorEnvelope,
AnalysisRunAccepted, AnalysisRunRequest, AnalysisRunStatusState, ApiError, ErrorEnvelope,
requests_are_idempotent_matches,
};

Expand Down Expand Up @@ -51,14 +55,15 @@ pub struct NaruonLiveResponse {
/// Production interchange origins remain `https` only. This listener binds
/// loopback TCP so tests and local standalone operation can prove request
/// handling without claiming TLS termination or cross-service table access.
/// This port only accepts versioned naruon POSTs.
/// This port accepts versioned naruon POSTs and Naruon-only retry-lineage GET.
#[derive(Debug)]
pub struct NaruonLiveService {
listener: Option<TcpListener>,
bound_addr: Option<SocketAddr>,
next_run_serial: u64,
next_request_serial: u64,
accepted_runs: HashMap<String, (AnalysisRunRequest, AnalysisRunAccepted)>,
runs_by_id: HashMap<String, String>,
}

impl Default for NaruonLiveService {
Expand All @@ -77,6 +82,7 @@ impl NaruonLiveService {
next_run_serial: 1,
next_request_serial: 1,
accepted_runs: HashMap::new(),
runs_by_id: HashMap::new(),
}
}

Expand Down Expand Up @@ -198,14 +204,24 @@ impl NaruonLiveService {
let mut lines = header_block.split("\r\n");
let request_line = lines.next().unwrap_or("");
let (method, path) = parse_request_line(request_line)?;
let headers = parse_headers(lines)?;
if method == "GET" {
if matches!(
analysis_run_retry_lineage_path_run_id(path),
Ok(_) | Err(ApiError::LimitExceeded)
) {
refuse_live_headers(&headers, self.bound_addr, false)?;
return self.list_analysis_run_retries(path, &headers, body);
}
return Err(ApiError::InvalidWirePayload);
}
if method != "POST" {
return Err(ApiError::InvalidWirePayload);
}
if path != NARUON_ANALYSIS_RUN_PATH && path != NARUON_EXPORT_PATH {
return Err(ApiError::InvalidWirePayload);
}
let headers = parse_headers(lines)?;
refuse_live_headers(&headers, self.bound_addr)?;
refuse_live_headers(&headers, self.bound_addr, true)?;
self.dispatch_path(path, &headers, body)
}

Expand Down Expand Up @@ -246,12 +262,45 @@ impl NaruonLiveService {
let run_id = format!("naruon-run-{}", self.next_run_serial);
self.next_run_serial += 1;
let accepted =
AnalysisRunAccepted::new(run_id, "accepted", request.idempotency_key.clone())?;
AnalysisRunAccepted::new(run_id.clone(), "accepted", request.idempotency_key.clone())?;
let body = accepted.to_json()?;
self.runs_by_id.insert(run_id, replay_key.clone());
self.accepted_runs.insert(replay_key, (request, accepted));
Ok(NaruonLiveResponse::json(202, "Accepted", body))
}

fn list_analysis_run_retries(
&self,
path: &str,
_headers: &HashMap<String, String>,
body: &str,
) -> Result<NaruonLiveResponse, ApiError> {
let run_id = analysis_run_retry_lineage_path_run_id(path)?;
if !body.trim().is_empty() {
return Err(ApiError::InvalidWirePayload);
}
refuse_metrics_on_retry_lineage_payload(body)?;
let replay_key = self
.runs_by_id
.get(&run_id)
.cloned()
.ok_or(ApiError::InvalidWirePayload)?;
let stored_accepted = &self
.accepted_runs
.get(&replay_key)
.ok_or(ApiError::InvalidWirePayload)?
.1;
let payload = AnalysisRunRetryLineage::new(
stored_accepted.run_id.clone(),
AnalysisRunStatusState::Accepted,
stored_accepted.idempotency_key.clone(),
Vec::new(),
)?;
let response_body = payload.to_json()?;
refuse_metrics_on_retry_lineage_payload(&response_body)?;
Ok(NaruonLiveResponse::json(200, "OK", response_body))
}

fn authorize_export(
headers: &HashMap<String, String>,
body: &str,
Expand Down Expand Up @@ -326,6 +375,7 @@ fn status_for(error: ApiError) -> (u16, &'static str) {
fn refuse_live_headers(
headers: &HashMap<String, String>,
bound_addr: Option<SocketAddr>,
require_idempotency: bool,
) -> Result<(), ApiError> {
validate_common_headers(headers, bound_addr)?;
if header_value(headers, "tepp-consumer")? != NARUON_CONSUMER_CODE {
Expand All @@ -334,7 +384,9 @@ fn refuse_live_headers(
if header_value(headers, "tepp-contract-version")? != "1" {
return Err(ApiError::InvalidWirePayload);
}
let _idempotency_key = header_value(headers, "idempotency-key")?;
if require_idempotency {
let _idempotency_key = header_value(headers, "idempotency-key")?;
}
Ok(())
}

Expand Down Expand Up @@ -531,4 +583,96 @@ mod tests {
ApiError::InvalidWirePayload
);
}

#[test]
#[allow(clippy::too_many_lines)]
fn naruon_compatibility_listener_lists_empty_retry_lineage() {
use crate::{ANALYSIS_RUN_CONTRACT_VERSION, AnalysisRunRequest, AnalysisRunRetryLineage};

let run = AnalysisRunRequest {
contract_version: ANALYSIS_RUN_CONTRACT_VERSION,
idempotency_key: "naruon-retries-idem".into(),
tenant_workspace_id: "naruon-retries-tenant".into(),
snapshot_id: "naruon-retries-snapshot".into(),
knowledge_cutoff: "2026-08-01T00:00:00Z".into(),
model_contract_version: "tepp-analysis-run-v1".into(),
output_profile: "calibrated_event_measurement".into(),
};
let body = run.to_json().expect("run json");
let create = format!(
"POST /v1/analysis-runs HTTP/1.1\r\nHost: 127.0.0.1\r\ncontent-type: application/json\r\ntepp-consumer: naruon\r\ntepp-contract-version: 1\r\nidempotency-key: naruon-retries-idem\r\ncontent-length: {}\r\n\r\n{body}",
body.len()
);
let mut service = NaruonLiveService::new();
let accepted = service.handle_http_request(&create);
assert_eq!(accepted.status_code, 202);
let run_id = serde_json::from_str::<serde_json::Value>(&accepted.body)
.expect("accepted json")["run_id"]
.as_str()
.expect("run_id")
.to_owned();

let inspect = format!(
"GET /v1/analysis-runs/{run_id}/retries HTTP/1.1\r\nHost: 127.0.0.1\r\ncontent-type: application/json\r\ntepp-consumer: naruon\r\ntepp-contract-version: 1\r\ncontent-length: 0\r\n\r\n"
);
let inspected = service.handle_http_request(&inspect);
assert_eq!(inspected.status_code, 200);
let lineage = AnalysisRunRetryLineage::from_json(&inspected.body).expect("lineage");
assert_eq!(lineage.run_id, run_id);
assert_eq!(lineage.run_state, crate::AnalysisRunStatusState::Accepted);
assert_eq!(lineage.idempotency_key, run.idempotency_key);
assert!(lineage.retries.is_empty());
assert!(
inspected.body.contains("\"retries\":[]") || inspected.body.contains("\"retries\": []")
);
assert!(!inspected.body.contains("rmse"));
assert!(!inspected.body.contains("scientific_acceptance"));
assert!(!inspected.body.contains("tenant_workspace_id"));
assert!(!inspected.body.contains("snapshot_id"));

let replay = service.handle_http_request(&inspect);
assert_eq!(replay.body, inspected.body);

let lineageweave = format!(
"GET /v1/analysis-runs/{run_id}/retries HTTP/1.1\r\nHost: 127.0.0.1\r\ncontent-type: application/json\r\ntepp-consumer: lineageweave\r\ntepp-contract-version: 1\r\ncontent-length: 0\r\n\r\n"
);
assert_eq!(service.handle_http_request(&lineageweave).status_code, 400);
assert_eq!(
service
.handle_http_request(
"GET /v1/analysis-runs/missing/retries HTTP/1.1\r\nHost: 127.0.0.1\r\ncontent-type: application/json\r\ntepp-consumer: naruon\r\ntepp-contract-version: 1\r\ncontent-length: 0\r\n\r\n"
)
.status_code,
400
);
assert_eq!(
service
.handle_http_request(&format!(
"GET /v1/analysis-runs/{run_id}/retries HTTP/1.1\r\nHost: 127.0.0.1\r\ncontent-type: application/json\r\ntepp-consumer: naruon\r\ntepp-contract-version: 1\r\ncontent-length: 2\r\n\r\n{{}}"
))
.status_code,
400
);
assert_eq!(
service
.handle_http_request(&format!(
"POST /v1/analysis-runs/{run_id}/retries HTTP/1.1\r\nHost: 127.0.0.1\r\ncontent-type: application/json\r\ntepp-consumer: naruon\r\ntepp-contract-version: 1\r\nidempotency-key: naruon-retries-idem\r\ncontent-length: 0\r\n\r\n"
))
.status_code,
400
);
let oversized = "a".repeat(129);
assert_eq!(
service
.handle_http_request(&format!(
"GET /v1/analysis-runs/{oversized}/retries HTTP/1.1\r\nHost: 127.0.0.1\r\ncontent-type: application/json\r\ntepp-consumer: naruon\r\ntepp-contract-version: 1\r\ncontent-length: 0\r\n\r\n"
))
.status_code,
413
);
let metrics = format!(
"GET /v1/analysis-runs/{run_id}/retries HTTP/1.1\r\nHost: 127.0.0.1\r\ncontent-type: application/json\r\ntepp-consumer: naruon\r\ntepp-contract-version: 1\r\ncontent-length: 16\r\n\r\n{{\"rmse\":0.1}}"
);
assert_eq!(service.handle_http_request(&metrics).status_code, 400);
}
}
Loading
Loading