Skip to content
Merged
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
16 changes: 8 additions & 8 deletions src/qp_vault/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,11 +32,11 @@

from qp_vault.enums import (
AdversarialStatus,
CISResult,
CISStage,
DataClassification,
EventType,
Lifecycle,
MembraneResult,
MembraneStage,
MemoryLayer,
ResourceStatus,
ResourceType,
Expand All @@ -54,11 +54,11 @@
)
from qp_vault.models import (
Chunk,
CISPipelineStatus,
CISStageRecord,
Collection,
ContentProvenance,
HealthScore,
MembranePipelineStatus,
MembraneStageRecord,
MerkleProof,
Resource,
SearchResult,
Expand Down Expand Up @@ -91,8 +91,8 @@
"MerkleProof",
"VaultEvent",
"ContentProvenance",
"CISStageRecord",
"CISPipelineStatus",
"MembraneStageRecord",
"MembranePipelineStatus",
# Enums
"TrustTier",
"DataClassification",
Expand All @@ -102,8 +102,8 @@
"MemoryLayer",
"EventType",
"AdversarialStatus",
"CISStage",
"CISResult",
"MembraneStage",
"MembraneResult",
"UploadMethod",
# Protocols (for implementors)
"StorageBackend",
Expand Down
8 changes: 0 additions & 8 deletions src/qp_vault/cis/__init__.py

This file was deleted.

10 changes: 5 additions & 5 deletions src/qp_vault/enums.py
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,7 @@ class AdversarialStatus(StrEnum):
"""Flagged by one or more Membrane stages. Multiplier: 0.3x."""


class CISStage(StrEnum):
class MembraneStage(StrEnum):
"""Membrane pipeline stages."""

INGEST = "ingest"
Expand Down Expand Up @@ -119,7 +119,7 @@ class CISStage(StrEnum):
"""Stage 8: Attack Pattern Registry feedback loop."""


class CISResult(StrEnum):
class MembraneResult(StrEnum):
"""Result of a single Membrane stage evaluation."""

PASS = "pass" # nosec B105 — Membrane stage result, not a password
Expand Down Expand Up @@ -202,7 +202,7 @@ class EventType(StrEnum):
SUPERSEDE = "supersede"
VERIFY = "verify"
SEARCH = "search"
CIS_SCAN = "cis_scan"
CIS_RELEASE = "cis_release"
CIS_FLAG = "cis_flag"
MEMBRANE_SCAN = "cis_scan"
MEMBRANE_RELEASE = "cis_release"
MEMBRANE_FLAG = "cis_flag"
ADVERSARIAL_STATUS_CHANGE = "adversarial_status_change"
8 changes: 8 additions & 0 deletions src/qp_vault/membrane/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
# Copyright 2026 Quantum Pipes Technologies, LLC
# SPDX-License-Identifier: Apache-2.0

"""Membrane: multi-stage content screening pipeline."""

from qp_vault.membrane.pipeline import MembranePipeline

__all__ = ["MembranePipeline"]
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,8 @@
import re
from dataclasses import dataclass, field

from qp_vault.enums import CISResult, CISStage
from qp_vault.models import CISStageRecord
from qp_vault.enums import MembraneResult, MembraneStage
from qp_vault.models import MembraneStageRecord

# Default blocklist patterns (prompt injection, jailbreak attempts, data exfiltration)
DEFAULT_BLOCKLIST: list[str] = [
Expand Down Expand Up @@ -47,7 +47,7 @@ class InnateScanConfig:
async def run_innate_scan(
content: str,
config: InnateScanConfig | None = None,
) -> CISStageRecord:
) -> MembraneStageRecord:
"""Run innate scan on content.

Checks content against regex blocklist patterns.
Expand All @@ -57,7 +57,7 @@ async def run_innate_scan(
config: Optional scan configuration.

Returns:
CISStageRecord with PASS, FLAG, or FAIL result.
MembraneStageRecord with PASS, FLAG, or FAIL result.
"""
if config is None:
config = InnateScanConfig()
Expand All @@ -73,15 +73,15 @@ async def run_innate_scan(
continue # Skip malformed patterns

if matches:
return CISStageRecord(
stage=CISStage.INNATE_SCAN,
result=CISResult.FLAG,
return MembraneStageRecord(
stage=MembraneStage.INNATE_SCAN,
result=MembraneResult.FLAG,
matched_patterns=matches[:5],
reasoning=f"Matched {len(matches)} blocklist patterns",
)

return CISStageRecord(
stage=CISStage.INNATE_SCAN,
result=CISResult.PASS, # nosec B105 — CIS stage result, not a password
return MembraneStageRecord(
stage=MembraneStage.INNATE_SCAN,
result=MembraneResult.PASS, # nosec B105 — Membrane stage result, not a password
reasoning=f"Checked {len(config.blocklist_patterns)} patterns, none matched",
)
40 changes: 20 additions & 20 deletions src/qp_vault/cis/pipeline.py → src/qp_vault/membrane/pipeline.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,9 @@
# Copyright 2026 Quantum Pipes Technologies, LLC
# SPDX-License-Identifier: Apache-2.0

"""CIS Pipeline: orchestrates multi-stage content screening.
"""Membrane Pipeline: orchestrates multi-stage content screening.

Runs content through the Content Immune System stages:
Runs content through the Membrane stages:
1. INNATE_SCAN — pattern-based detection (regex, blocklists)
2. RELEASE — risk-proportionate gating decision

Expand All @@ -13,21 +13,21 @@

from __future__ import annotations

from qp_vault.cis.innate_scan import InnateScanConfig, run_innate_scan
from qp_vault.cis.release_gate import evaluate_release
from qp_vault.enums import CISResult, CISStage, ResourceStatus
from qp_vault.models import CISPipelineStatus, CISStageRecord
from qp_vault.enums import MembraneResult, MembraneStage, ResourceStatus
from qp_vault.membrane.innate_scan import InnateScanConfig, run_innate_scan
from qp_vault.membrane.release_gate import evaluate_release
from qp_vault.models import MembranePipelineStatus, MembraneStageRecord


class CISPipeline:
"""Content Immune System pipeline.
class MembranePipeline:
"""Membrane pipeline.

Screens content through multiple stages before allowing indexing.
Content that fails screening is quarantined.

Args:
innate_config: Configuration for the innate scan stage.
enabled: Whether CIS screening is active. Default True.
enabled: Whether Membrane screening is active. Default True.
"""

def __init__(
Expand All @@ -39,29 +39,29 @@ def __init__(
self._innate_config = innate_config
self._enabled = enabled

async def screen(self, content: str) -> CISPipelineStatus:
"""Run content through the CIS pipeline.
async def screen(self, content: str) -> MembranePipelineStatus:
"""Run content through the Membrane pipeline.

Args:
content: Text content to screen.

Returns:
CISPipelineStatus with stage results and overall decision.
MembranePipelineStatus with stage results and overall decision.
"""
if not self._enabled:
return CISPipelineStatus(
return MembranePipelineStatus(
stages=[
CISStageRecord(
stage=CISStage.RELEASE,
result=CISResult.PASS, # nosec B105
MembraneStageRecord(
stage=MembraneStage.RELEASE,
result=MembraneResult.PASS, # nosec B105
reasoning="Released: screening disabled",
),
],
overall_result=CISResult.PASS, # nosec B105
overall_result=MembraneResult.PASS, # nosec B105
recommended_status=ResourceStatus.INDEXED,
)

stages: list[CISStageRecord] = []
stages: list[MembraneStageRecord] = []

# Stage 1: Innate scan
innate_result = await run_innate_scan(content, self._innate_config)
Expand All @@ -73,12 +73,12 @@ async def screen(self, content: str) -> CISPipelineStatus:

# Determine overall result and recommended status
overall = release_result.result
if overall == CISResult.FAIL or overall == CISResult.FLAG:
if overall == MembraneResult.FAIL or overall == MembraneResult.FLAG:
status = ResourceStatus.QUARANTINED
else:
status = ResourceStatus.INDEXED

return CISPipelineStatus(
return MembranePipelineStatus(
stages=stages,
overall_result=overall,
recommended_status=status,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,21 +3,21 @@

"""Release gate: risk-proportionate gating before indexing.

Evaluates CIS stage results and decides whether content should be:
Evaluates Membrane stage results and decides whether content should be:
- RELEASED (indexed normally)
- HELD (quarantined for human review)
- REJECTED (blocked from indexing)
"""

from __future__ import annotations

from qp_vault.enums import CISResult, CISStage
from qp_vault.models import CISStageRecord
from qp_vault.enums import MembraneResult, MembraneStage
from qp_vault.models import MembraneStageRecord


async def evaluate_release(
stage_records: list[CISStageRecord],
) -> CISStageRecord:
stage_records: list[MembraneStageRecord],
) -> MembraneStageRecord:
"""Evaluate whether content should be released for indexing.

Decision logic:
Expand All @@ -26,32 +26,32 @@ async def evaluate_release(
- Otherwise: PASS (release)

Args:
stage_records: Results from previous CIS stages.
stage_records: Results from previous Membrane stages.

Returns:
CISStageRecord for the RELEASE stage.
MembraneStageRecord for the RELEASE stage.
"""
has_fail = any(r.result == CISResult.FAIL for r in stage_records)
has_flag = any(r.result == CISResult.FLAG for r in stage_records)
has_fail = any(r.result == MembraneResult.FAIL for r in stage_records)
has_flag = any(r.result == MembraneResult.FLAG for r in stage_records)

if has_fail:
failed = [r.stage.value for r in stage_records if r.result == CISResult.FAIL]
return CISStageRecord(
stage=CISStage.RELEASE,
result=CISResult.FAIL,
failed = [r.stage.value for r in stage_records if r.result == MembraneResult.FAIL]
return MembraneStageRecord(
stage=MembraneStage.RELEASE,
result=MembraneResult.FAIL,
reasoning=f"Rejected: {', '.join(failed)} failed",
)

if has_flag:
flagged = [r.stage.value for r in stage_records if r.result == CISResult.FLAG]
return CISStageRecord(
stage=CISStage.RELEASE,
result=CISResult.FLAG,
flagged = [r.stage.value for r in stage_records if r.result == MembraneResult.FLAG]
return MembraneStageRecord(
stage=MembraneStage.RELEASE,
result=MembraneResult.FLAG,
reasoning=f"Quarantined: {', '.join(flagged)} flagged",
)

return CISStageRecord(
stage=CISStage.RELEASE,
result=CISResult.PASS, # nosec B105
return MembraneStageRecord(
stage=MembraneStage.RELEASE,
result=MembraneResult.PASS, # nosec B105
reasoning=f"Released: {len(stage_records)} stages passed",
)
22 changes: 11 additions & 11 deletions src/qp_vault/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,11 +9,11 @@

from qp_vault.enums import (
AdversarialStatus,
CISResult,
CISStage,
DataClassification,
EventType,
Lifecycle,
MembraneResult,
MembraneStage,
MemoryLayer,
ResourceStatus,
ResourceType,
Expand Down Expand Up @@ -213,7 +213,7 @@ class ContentProvenance(BaseModel):
created_at: datetime = Field(default_factory=_utcnow)


class CISStageRecord(BaseModel):
class MembraneStageRecord(BaseModel):
"""Result of a single Membrane pipeline stage evaluation.

Every stage (INGEST through REMEMBER) creates one record per document.
Expand All @@ -222,8 +222,8 @@ class CISStageRecord(BaseModel):

id: str = ""
resource_id: str = ""
stage: CISStage = CISStage.INGEST
result: CISResult = CISResult.PASS
stage: MembraneStage = MembraneStage.INGEST
result: MembraneResult = MembraneResult.PASS
risk_score: float = 0.0
reasoning: str = ""
matched_patterns: list[str] = Field(default_factory=list)
Expand All @@ -232,17 +232,17 @@ class CISStageRecord(BaseModel):
created_at: datetime = Field(default_factory=_utcnow)


class CISPipelineStatus(BaseModel):
class MembranePipelineStatus(BaseModel):
"""Aggregate status of the Membrane pipeline for a single document."""

resource_id: str = ""
stages: list[CISStageRecord] = Field(default_factory=list)
stages_completed: list[CISStage] = Field(default_factory=list)
stages_pending: list[CISStage] = Field(default_factory=list)
overall_result: CISResult = CISResult.PASS
stages: list[MembraneStageRecord] = Field(default_factory=list)
stages_completed: list[MembraneStage] = Field(default_factory=list)
stages_pending: list[MembraneStage] = Field(default_factory=list)
overall_result: MembraneResult = MembraneResult.PASS
recommended_status: ResourceStatus = ResourceStatus.INDEXED
aggregate_risk_score: float = 0.0
recommended_action: str = "pending"
stage_results: list[CISStageRecord] = Field(default_factory=list)
stage_results: list[MembraneStageRecord] = Field(default_factory=list)
started_at: datetime | None = None
completed_at: datetime | None = None
Loading
Loading