add relationship fields

README.md

@@ -162,6 +162,13 @@ contain them. This lets late-arriving or less frequent fields such as custom
 `lcs_*` application fields stay visible long enough to be reviewed and appended
 to an existing profile.
 
+Shared-field discovery is also used as the base for cross-source correlation.
+SignalScope groups exact aliases and broader semantic families such as source IP,
+user, host, ID, status/result, action, type/category, domain, URL, and custom
+namespaces such as `lcs_*`. These shared groups are the foundation for a global
+correlation profile and future flow graphs that show how users, hosts, IPs,
+applications, IDs, statuses, and destinations relate across streams.
+
 Enabled streams are normalized through the same event model. Stream profiles
 define the entity, timestamp, categorical, and numeric fields used for baselines.
 The dashboard and Ollama then correlate behavior across sources, for example a
@@ -173,6 +180,29 @@ A stream profile can track multiple entities from the same event, such as
 each selected entity value, which makes cross-source investigation work even when
 one source is user-centric and another is IP- or host-centric.
 
+Profiles can also track field relationships as behavior patterns. This is useful
+when the suspicious signal is not a single new value, but a new combination such
+as a known user logging in successfully from a source IP that has never been seen
+for that user before. Add `relationship_fields` to a stream profile, for example:
+
+```json
+[
+  {
+    "stream_id": "windows-security",
+    "entity_field": "username",
+    "categorical_fields": ["action", "eventid"],
+    "relationship_fields": [
+      {"left": "username", "right": "srcip", "name": "user source IP"},
+      {"left": "username", "right": "hostname", "name": "user host"}
+    ]
+  }
+]
+```
+
+After the baseline has learned those relationships, a new `username -> srcip` or
+`username -> hostname` pair is reported as `new_relationship` with sample events.
+The dashboard profile editor exposes this as `Behavior relationships (JSON)`.
+
 SignalScope keeps a common alias map for fields such as source IP, destination
 IP, ports, action, severity, service/protocol, DNS query, URL, message, and event
 type. This lets Related Activity and correlations work with firewall/proxy/DNS

src/fgai/baseline.py

@@ -93,6 +93,10 @@ def _sample_event(event: LogEvent, value: str = "") -> dict[str, object]:
     }
 
 
+def _relationship_key(left: str, right: str) -> str:
+    return f"relationship:{left.lower()}->{right.lower()}"
+
+
 class BaselineStore:
     """Persistent five-minute behavior baseline, implemented with stdlib SQLite."""
 
@@ -230,6 +234,13 @@ class BaselineStore:
                     for detector in event_detector_categories(event):
                         detector_pending[(stream_id, entity, detector, bucket)] += 1
                         detector_temporal_pending[(stream_id, entity, detector, moment.weekday(), moment.hour, bucket)] += 1
+                    for relation in getattr(profile, "relationship_fields", ()):
+                        left = str(getattr(relation, "left", "")).lower()
+                        right = str(getattr(relation, "right", "")).lower()
+                        left_value = event.fields.get(left)
+                        right_value = event.fields.get(right)
+                        if left_value and right_value:
+                            pending_values[(stream_id, left_value, _relationship_key(left, right), right_value)] += 1
                     for field in fields:
                         key = (stream_id, entity, str(field).lower(), bucket)
                         value = _number(event.fields.get(key[2])) if key[2] in numeric else 0
@@ -386,6 +397,63 @@ class BaselineStore:
                 samples = [_sample_event(event, detector) for event in matching[:5]]
                 if score >= MIN_REPORTED_DEVIATION_SCORE:
                     output[entity].append({"detector": f"{detector}_burst", "field": detector, "stream_id": stream, "score": score, "base_score": base_score, "weight": weight, "confidence": confidence, "baseline_samples": len(rows), "baseline_age_days": round(baseline_age_days, 2), "baseline_scope": baseline_scope, "reason": f"{detector.replace('_', ' ')} burst above its {baseline_scope} baseline (z={z_score:.1f}, minimum={minimum})", "current": current_value, "baseline": round(mean(history), 2), "sample_values": [detector], "sample_events": samples})
+            # Detect custom behavior relationships, e.g. username -> srcip or host -> process.name.
+            relationship_counts: Counter[tuple[str, str, str, str, str]] = Counter()
+            relationship_samples: dict[tuple[str, str, str, str, str], list[LogEvent]] = defaultdict(list)
+            for event in events:
+                stream = event.fields.get("fgai_stream_id", "")
+                profile = profiles.get(stream)
+                if not profile:
+                    continue
+                for relation in getattr(profile, "relationship_fields", ()):
+                    left = str(getattr(relation, "left", "")).lower()
+                    right = str(getattr(relation, "right", "")).lower()
+                    left_value = event.fields.get(left)
+                    right_value = event.fields.get(right)
+                    if not left_value or not right_value:
+                        continue
+                    key = (stream, left_value, left, right, right_value)
+                    relationship_counts[key] += 1
+                    if len(relationship_samples[key]) < 5:
+                        relationship_samples[key].append(event)
+            relationship_limit: Counter[tuple[str, str]] = Counter()
+            for (stream, left_value, left, right, right_value), count in relationship_counts.items():
+                field = _relationship_key(left, right)
+                known = connection.execute("select seen_count from profile_values where stream_id=? and entity=? and field=? and value=?", (stream, left_value, field, right_value)).fetchone()
+                known_total = connection.execute("select coalesce(sum(seen_count), 0) from profile_values where stream_id=? and entity=? and field=?", (stream, left_value, field)).fetchone()[0]
+                if known is not None or int(known_total or 0) < 12:
+                    continue
+                if relationship_limit[(stream, left_value)] >= MAX_RARE_VALUES_PER_ENTITY:
+                    continue
+                oldest = connection.execute("select min(bucket_start) from profile_buckets where stream_id=? and entity=?", (stream, left_value)).fetchone()[0]
+                if oldest is None:
+                    oldest = connection.execute("select min(last_seen) from profile_values where stream_id=? and entity=? and field=?", (stream, left_value, field)).fetchone()[0]
+                baseline_age_days = _age_days(oldest, _event_epoch(relationship_samples[(stream, left_value, left, right, right_value)][0], int(time.time())))
+                if baseline_age_days < min_training_days:
+                    continue
+                profile = profiles.get(stream)
+                base_score = 28
+                score, weight = _weighted_score(base_score, profile, field, "new_relationship")
+                if score < MIN_REPORTED_DEVIATION_SCORE:
+                    continue
+                samples = relationship_samples[(stream, left_value, left, right, right_value)]
+                output[left_value].append({
+                    "detector": "new_relationship",
+                    "field": field,
+                    "stream_id": stream,
+                    "score": score,
+                    "base_score": base_score,
+                    "weight": weight,
+                    "confidence": "medium",
+                    "baseline_samples": int(known_total),
+                    "baseline_age_days": round(baseline_age_days, 2),
+                    "baseline_scope": "known field relationships",
+                    "reason": f"new {right} value for {left}={left_value}",
+                    "value": right_value,
+                    "sample_values": [f"{left}={left_value}", f"{right}={right_value}"],
+                    "sample_events": [_sample_event(item, f"{left}={left_value} {right}={right_value}") for item in samples],
+                })
+                relationship_limit[(stream, left_value)] += 1
             # Detect selected categorical values that have not appeared for this entity in prior data.
             rare_counts: Counter[tuple[str, str]] = Counter()
             for event in events:

src/fgai/cli.py

@@ -257,13 +257,15 @@ def _stream_name(config: dict[str, object], stream_id: str) -> str:
 def _profile_fields(profile: object | None) -> tuple[str, ...]:
     if not profile:
         return ()
-        return tuple(
+    return tuple(
-            field for field in (
+        field for field in (
-                str(getattr(profile, "entity_field", "")),
+            str(getattr(profile, "entity_field", "")),
-                *tuple(str(item) for item in getattr(profile, "entity_fields", ())),
+            *tuple(str(item) for item in getattr(profile, "entity_fields", ())),
-                str(getattr(profile, "timestamp_field", "")),
+            str(getattr(profile, "timestamp_field", "")),
             *tuple(str(item) for item in getattr(profile, "categorical_fields", ())),
             *tuple(str(item) for item in getattr(profile, "numeric_fields", ())),
+            *tuple(str(getattr(relation, "left", "")) for relation in getattr(profile, "relationship_fields", ())),
+            *tuple(str(getattr(relation, "right", "")) for relation in getattr(profile, "relationship_fields", ())),
         )
         if field
     )

src/fgai/history.py

@@ -75,6 +75,42 @@ class FieldDiscoveryStore:
                     primary key (stream_id, field)
                 )"""
             )
+            connection.execute(
+                """create table if not exists field_catalog (
+                    stream_id text not null, stream_name text not null, field text not null,
+                    field_type text not null, properties text not null, first_seen integer not null, last_seen integer not null,
+                    primary key (stream_id, field)
+                )"""
+            )
+
+    def ingest_catalog(self, stream_id: str, stream_name: str, fields: list[dict[str, object]], *, retention_days: int = 30) -> int:
+        now = int(time.time())
+        rows = []
+        for item in fields:
+            if not isinstance(item, dict):
+                continue
+            field = str(item.get("name") or item.get("field") or "").strip()
+            if not field or field in IGNORED_DISCOVERY_FIELDS:
+                continue
+            type_info = item.get("type", {})
+            field_type = str(type_info.get("type", "") if isinstance(type_info, dict) else type_info)
+            properties = type_info.get("properties", []) if isinstance(type_info, dict) else []
+            if not isinstance(properties, list):
+                properties = []
+            rows.append((stream_id, stream_name, field, field_type, json.dumps([str(prop) for prop in properties]), now, now))
+        with sqlite3.connect(self.path) as connection:
+            for row in rows:
+                connection.execute(
+                    """insert into field_catalog values (?, ?, ?, ?, ?, ?, ?)
+                       on conflict(stream_id, field) do update set
+                       stream_name=excluded.stream_name,
+                       field_type=excluded.field_type,
+                       properties=excluded.properties,
+                       last_seen=excluded.last_seen""",
+                    row,
+                )
+            connection.execute("delete from field_catalog where last_seen < ?", (now - retention_days * 86400,))
+        return len(rows)
 
     def ingest(self, events: list[object], *, max_values: int = 20, retention_days: int = 30) -> int:
         now = int(time.time())
@@ -130,6 +166,7 @@ class FieldDiscoveryStore:
         events: list[LogEvent] = []
         with sqlite3.connect(self.path) as connection:
             rows = connection.execute("select stream_id, stream_name, field, seen_count, numeric_count, sample_values from field_discovery").fetchall()
+            catalog_rows = connection.execute("select stream_id, stream_name, field, field_type, properties from field_catalog").fetchall()
         for stream_id, stream_name, field, seen_count, numeric_count, sample_values in rows:
             try:
                 values = [str(item) for item in json.loads(str(sample_values))]
@@ -142,4 +179,20 @@ class FieldDiscoveryStore:
                 value = values[index % len(values)]
                 fields = {"fgai_stream_id": str(stream_id), "fgai_stream": str(stream_name), str(field): value}
                 events.append(LogEvent(raw=json.dumps(fields, sort_keys=True), fields=fields))
+        existing = {(event.fields.get("fgai_stream_id", ""), field) for event in events for field in event.fields if field not in {"fgai_stream_id", "fgai_stream"}}
+        for stream_id, stream_name, field, field_type, properties in catalog_rows:
+            if (str(stream_id), str(field)) in existing:
+                continue
+            try:
+                props = set(json.loads(str(properties)))
+            except (json.JSONDecodeError, TypeError):
+                props = set()
+            if "numeric" in props or str(field_type).lower() in {"long", "double", "integer", "int", "float"}:
+                value = "1"
+            elif "enumerable" in props:
+                value = "observed"
+            else:
+                value = "catalog"
+            fields = {"fgai_stream_id": str(stream_id), "fgai_stream": str(stream_name), str(field): value}
+            events.append(LogEvent(raw=json.dumps(fields, sort_keys=True), fields=fields))
         return events

src/fgai/llm.py

@@ -109,22 +109,28 @@ def ollama_dashboard_assessment(analysis: dict[str, object], model: str | None =
     )
 
 
+def _compact_profile_suggestion(item: dict[str, object]) -> dict[str, object]:
+    discovery = item.get("discovery", {}) if isinstance(item.get("discovery"), dict) else {}
+    selected = discovery.get("selected_fields", {}) if isinstance(discovery.get("selected_fields"), dict) else {}
+    return {
+        "stream_id": item.get("stream_id"),
+        "stream_name": item.get("stream_name"),
+        "events": item.get("events"),
+        "detected_log_type": item.get("detected_log_type", ""),
+        "common_fields": item.get("common_fields", [])[:25],
+        "shared_fields": item.get("shared_fields", [])[:20],
+        "top_fields": discovery.get("top_fields", [])[:25],
+        "selected_fields": selected,
+        "rejected_fields": discovery.get("rejected_fields", [])[:10],
+        "heuristic_profile": item.get("profile", {}),
+    }
+
+
 def ollama_profile_advice(suggestions: list[dict[str, object]], model: str | None = None, timeout: int | None = None) -> list[dict[str, object]]:
     host = os.getenv("OLLAMA_HOST", "http://127.0.0.1:11434").rstrip("/")
     selected_model = model or os.getenv("FGAI_PROFILE_ADVISOR_MODEL", "qwen3:8b")
     selected_timeout = timeout or int(os.getenv("FGAI_PROFILE_ADVISOR_TIMEOUT", "120"))
-    compact = [
+    compact = [_compact_profile_suggestion(item) for item in suggestions[:10]]
-        {
-            "stream_id": item.get("stream_id"),
-            "stream_name": item.get("stream_name"),
-            "events": item.get("events"),
-            "common_fields": item.get("common_fields", [])[:20],
-            "shared_fields": item.get("shared_fields", [])[:12],
-            "detected_log_type": item.get("detected_log_type", ""),
-            "heuristic_profile": item.get("profile", {}),
-        }
-        for item in suggestions[:10]
-    ]
     body = json.dumps(
         {
             "model": selected_model,
@@ -136,9 +142,18 @@ def ollama_profile_advice(suggestions: list[dict[str, object]], model: str | Non
                 "Return only valid JSON with this schema: "
                 "{\"profiles\":[{\"stream_id\":\"...\",\"entity_fields\":[\"...\"],\"timestamp_field\":\"...\","
                 "\"categorical_fields\":[\"...\"],\"numeric_fields\":[\"...\"],\"detectors\":{\"auth_failure\":{\"enabled\":true,\"minimum\":5,\"z_threshold\":3}},"
+                "\"relationship_fields\":[{\"left\":\"username\",\"right\":\"srcip\",\"name\":\"user source IP\"}],"
+                "\"field_roles\":{\"field\":\"entity|identity|asset|session|status|action|resource|metric|context\"},"
+                "\"correlation_roles\":{\"entity\":[\"...\"],\"identity\":[\"...\"],\"asset\":[\"...\"],\"session\":[\"...\"],\"activity\":[\"...\"],\"resource\":[\"...\"],\"status\":[\"...\"]},"
                 "\"reason\":\"short reason\"}]}. "
-                "Use only field names present in common_fields, shared_fields, or heuristic_profile. Prefer fields that appear in shared_fields when they are useful categorical or numeric baseline fields. Do not include raw message/full_message fields. "
+                "Use only field names present in common_fields, shared_fields aliases, top_fields, selected_fields, or heuristic_profile. "
-                "Allowed detectors are auth_failure, dns_query, deny_action. Prefer canonical fields such as username, hostname, eventid, srcip, dstip when present. "
+                "Do not invent fields. Do not include raw message/full_message/raw/answer fields. "
+                "Treat shared_fields as deterministic evidence that a field or semantic group appears across streams. "
+                "For custom namespaces such as lcs_* infer roles from the real field names and observed stats, for example tenant/customer/request/session/correlation IDs. "
+                "Prefer profile fields that connect streams: source or client IPs, usernames/accounts, hostnames, event IDs, session/correlation/request IDs, status/result/action/type, service/app, URL/domain/resource, and stable numeric metrics. "
+                "Use categorical_fields for low-cardinality behavior dimensions, numeric_fields for counters/durations/sizes, and entity_fields for fields that identify a user, host, IP, account, device, or application actor. "
+                "Use relationship_fields for behavior pairs where a left identity should learn normal right-side values, such as username->srcip, username->hostname, host->process.name, srcip->dstip, or account->application. "
+                "Allowed detectors are auth_failure, dns_query, deny_action. Prefer canonical fields such as username, hostname, eventid, srcip, dstip when present, but keep useful stream-specific fields too. "
                 f"\n\nObserved streams:\n{json.dumps(compact, sort_keys=True)}"
             ),
         }

src/fgai/monitor.py

@@ -54,6 +54,21 @@ def _range_seconds(value: object) -> int:
         return 300
 
 
+def _graylog_fields_from_result(result: dict[str, object]) -> list[dict[str, object]]:
+    content = result.get("result", {}).get("content", []) if isinstance(result.get("result"), dict) else []
+    text = next((item.get("text", "") for item in content if isinstance(item, dict)), "")
+    if not text:
+        return []
+    try:
+        payload = json.loads(text)
+    except json.JSONDecodeError:
+        return []
+    fields = payload.get("fields", payload) if isinstance(payload, dict) else payload
+    if isinstance(fields, dict) and isinstance(fields.get("fields"), list):
+        fields = fields["fields"]
+    return [item for item in fields if isinstance(item, dict)] if isinstance(fields, list) else []
+
+
 def _stream_coverage(runtime_values: dict[str, object], stream_profiles: dict[str, object], stream_status: dict[str, object], profile_readiness: list[dict[str, object]], stream_titles: dict[str, str]) -> list[dict[str, object]]:
     configured = [
         item for item in runtime_values.get("graylog_streams", [])
@@ -85,7 +100,7 @@ def _stream_coverage(runtime_values: dict[str, object], stream_profiles: dict[st
             "profile": _profile_name(stream_id, stream_titles, profile) if profile else "",
             "profile_ready": bool(profile),
             "entity_field": ", ".join(getattr(profile, "entity_fields", ()) or (str(getattr(profile, "entity_field", "")),)) if profile else "",
-            "tracked_fields": len(getattr(profile, "categorical_fields", ())) + len(getattr(profile, "numeric_fields", ())) if profile else 0,
+            "tracked_fields": len(getattr(profile, "categorical_fields", ())) + len(getattr(profile, "numeric_fields", ())) + len(getattr(profile, "relationship_fields", ())) if profile else 0,
             "ready_fields": ready_fields,
             "total_fields": total_fields,
             "readiness": f"{ready_fields}/{total_fields}" if total_fields else "0/0",
@@ -155,6 +170,8 @@ def build_status(
                 aggregate_events_total = 0
                 client = GraylogMcpClient(url, token, verify_tls=verify_tls)
                 probe_status = client.probe()
+                discovery_store = FieldDiscoveryStore(history_path) if history_path else None
+                catalog_fields_total = 0
                 for stream_config in stream_configs:
                     stream_id = str(stream_config["id"])
                     profile = stream_profiles.get(stream_id)
@@ -164,8 +181,15 @@ def build_status(
                         str(getattr(profile, "timestamp_field", "")),
                         *tuple(str(field) for field in getattr(profile, "categorical_fields", ())),
                         *tuple(str(field) for field in getattr(profile, "numeric_fields", ())),
+                        *tuple(str(getattr(relation, "left", "")) for relation in getattr(profile, "relationship_fields", ())),
+                        *tuple(str(getattr(relation, "right", "")) for relation in getattr(profile, "relationship_fields", ())),
                     ) if profile else ()
                     stream_name = str(stream_config.get("title", "") or stream_titles.get(stream_id) or stream_id)
+                    if discovery_store:
+                        try:
+                            catalog_fields_total += discovery_store.ingest_catalog(stream_id, stream_name, _graylog_fields_from_result(client.call_tool("list_fields", {"streams": [stream_id]})))
+                        except RuntimeError:
+                            pass
                     aggregate_status: dict[str, object] = {}
                     if use_aggregate:
                         aggregate_status = GraylogAggregateSource(client, stream_id, str(runtime_values.get("graylog_query", "*"))).fetch_count(range_seconds=range_seconds, probe_status=probe_status)
@@ -199,6 +223,7 @@ def build_status(
                     "sample_limited_streams": len(sample_limited_streams),
                     "truncated_streams": len(truncated_streams),
                     "aggregate_error_streams": len(aggregate_errors),
+                    "catalog_fields": catalog_fields_total,
                     "coverage_status": "partial" if partial_streams else "truncated" if truncated_streams else "complete_window",
                     "coverage_warning": " ".join(warnings),
                 }
@@ -338,7 +363,7 @@ def build_status(
         "baseline": {"enabled": bool(baseline), "sources_ready": len(profiles), "training_days": baseline_training_days, "new_events_recorded": baseline_events, "profile_fields_recorded": profile_baseline_fields, "discovery_fields_recorded": discovered_profile_fields, "discovery_cache_events": len(discovery_cache_events), "maintenance": baseline_maintenance, "size_bytes": baseline_maintenance.get("size_bytes", 0) if isinstance(baseline_maintenance, dict) else 0},
         "capabilities": {"threat_intel": threat_intel_status, "graylog_mcp": mcp_status, "profile_advisor": profile_advisor_status},
         "configuration": runtime_config,
-        "stream_profiles": [{"stream_id": item.stream_id, "name": _profile_name(item.stream_id, stream_titles, item), "stream_name": _stream_name(item.stream_id, stream_titles, item), "entity_field": item.entity_field, "entity_fields": list(item.entity_fields), "timestamp_field": item.timestamp_field, "categorical_fields": list(item.categorical_fields), "numeric_fields": list(item.numeric_fields), "detectors": item.detectors, "field_weights": item.field_weights} for item in stream_profiles.values()],
+        "stream_profiles": [{"stream_id": item.stream_id, "name": _profile_name(item.stream_id, stream_titles, item), "stream_name": _stream_name(item.stream_id, stream_titles, item), "entity_field": item.entity_field, "entity_fields": list(item.entity_fields), "timestamp_field": item.timestamp_field, "categorical_fields": list(item.categorical_fields), "numeric_fields": list(item.numeric_fields), "detectors": item.detectors, "field_weights": item.field_weights, "relationship_fields": [{"left": relation.left, "right": relation.right, "name": relation.name} for relation in item.relationship_fields]} for item in stream_profiles.values()],
         "stream_coverage": stream_coverage,
         "profile_suggestions": profile_suggestions,
         "profile_readiness": profile_readiness,

src/fgai/profile_suggestions.py

@@ -219,7 +219,46 @@ def _looks_like_time_field(field: str) -> bool:
 
 
 def _semantic_field_group(field: str) -> str:
-    return ALIAS_CANONICAL_BY_FIELD.get(field.lower(), field.lower())
+    return _semantic_field_groups(field)[0]
+
+
+def _semantic_field_groups(field: str) -> list[str]:
+    normalized = field.lower()
+    if normalized in ALIAS_CANONICAL_BY_FIELD:
+        return [ALIAS_CANONICAL_BY_FIELD[normalized]]
+    groups = []
+    if normalized.startswith("lcs_"):
+        groups.append("lcs_*")
+    compact = normalized.replace(".", "_").replace("-", "_")
+    parts = [part for part in compact.split("_") if part]
+    if not parts:
+        return [normalized]
+    if parts[-1] in {"ip", "addr", "address"}:
+        groups.append("*_ip")
+    if parts[-1] in {"user", "username", "account", "principal", "actor", "name"} and any(part in {"user", "account", "principal", "actor"} for part in parts):
+        groups.append("*_user")
+    if parts[-1] in {"host", "hostname", "computer", "node", "name"} and any(part in {"host", "hostname", "computer", "node"} for part in parts):
+        groups.append("*_host")
+    if parts[-1] in {"id", "uuid", "guid"}:
+        groups.append("*_id")
+    if parts[-1] in {"name", "label", "title"}:
+        groups.append("*_name")
+    if parts[-1] in {"status", "state", "result", "outcome", "disposition"}:
+        groups.append("*_status")
+    if parts[-1] in {"action", "operation", "event"}:
+        groups.append("*_action")
+    if parts[-1] in {"type", "category", "class", "kind"}:
+        groups.append("*_type")
+    if parts[-1] in {"code", "reason"}:
+        groups.append("*_code")
+    if parts[-1] in {"port"}:
+        groups.append("*_port")
+    if parts[-1] in {"domain", "fqdn"}:
+        groups.append("*_domain")
+    if parts[-1] in {"url", "uri", "path"}:
+        groups.append("*_url")
+    groups.append(normalized)
+    return list(dict.fromkeys(groups))
 
 
 def _shared_field_kind(field: str, *, cardinality: int, total: int, numeric_ratio: float) -> str:
@@ -413,7 +452,9 @@ def _shared_profile_fields(
             "candidate": candidate,
             "semantic_group": _semantic_field_group(field),
         })
-        if kind == "numeric":
+        if kind == "entity":
+            pass
+        elif kind == "numeric":
             numeric.append(field)
         elif kind == "categorical":
             categorical.append(field)
@@ -443,6 +484,14 @@ def _shared_profile_fields(
         else:
             stats_by_key[group].setdefault("aliases", aliases[:10])
             stats_by_key[group]["streams"] = max(int(stats_by_key[group].get("streams", 0) or 0), stream_count)
+        if kind == "numeric":
+            for field in present_aliases:
+                if field not in numeric:
+                    numeric.append(field)
+        elif kind == "categorical":
+            for field in present_aliases:
+                if field not in categorical:
+                    categorical.append(field)
     stats = sorted(
         stats_by_key.values(),
         key=lambda item: (
@@ -458,6 +507,21 @@ def _shared_profile_fields(
 def _allowed_fields(suggestion: dict[str, object]) -> set[str]:
     profile = suggestion.get("profile", {}) if isinstance(suggestion.get("profile"), dict) else {}
     fields = {str(item.get("field", "")) for item in suggestion.get("common_fields", []) if isinstance(item, dict)}
+    discovery = suggestion.get("discovery", {}) if isinstance(suggestion.get("discovery"), dict) else {}
+    for item in discovery.get("top_fields", []) if isinstance(discovery.get("top_fields"), list) else []:
+        if isinstance(item, dict):
+            fields.add(str(item.get("field", "")))
+    for item in suggestion.get("shared_fields", []) if isinstance(suggestion.get("shared_fields"), list) else []:
+        if not isinstance(item, dict):
+            continue
+        fields.add(str(item.get("field", "")))
+        aliases = item.get("aliases", [])
+        if isinstance(aliases, list):
+            fields.update(str(alias) for alias in aliases if alias)
+    selected = discovery.get("selected_fields", {}) if isinstance(discovery.get("selected_fields"), dict) else {}
+    for value in selected.values():
+        if isinstance(value, list):
+            fields.update(str(field) for field in value if field)
     for key in ("entity_fields", "categorical_fields", "numeric_fields"):
         value = profile.get(key, [])
         if isinstance(value, list):
@@ -490,6 +554,19 @@ def apply_profile_advice(suggestions: list[dict[str, object]], advice: list[dict
             timestamp = str(heuristic.get("timestamp_field", "timestamp"))
         categorical = valid_fields("categorical_fields", list(heuristic.get("categorical_fields", [])) if isinstance(heuristic.get("categorical_fields"), list) else [])
         numeric = valid_fields("numeric_fields", list(heuristic.get("numeric_fields", [])) if isinstance(heuristic.get("numeric_fields"), list) else [])
+        relationships = []
+        raw_relationships = advisor.get("relationship_fields", [])
+        if isinstance(raw_relationships, list):
+            for relation in raw_relationships:
+                if not isinstance(relation, dict):
+                    continue
+                left = str(relation.get("left", "")).strip()
+                right = str(relation.get("right", "")).strip()
+                if left not in allowed_fields or right not in allowed_fields or left == right:
+                    continue
+                relationships.append({"left": left, "right": right, "name": str(relation.get("name", "") or f"{left}->{right}")})
+        if not relationships and isinstance(heuristic.get("relationship_fields"), list):
+            relationships = [item for item in heuristic.get("relationship_fields", []) if isinstance(item, dict)]
         detectors_raw = advisor.get("detectors", {})
         detectors = {}
         if isinstance(detectors_raw, dict):
@@ -503,6 +580,23 @@ def apply_profile_advice(suggestions: list[dict[str, object]], advice: list[dict
         if not detectors:
             detectors = dict(heuristic.get("detectors", {})) if isinstance(heuristic.get("detectors"), dict) else {}
         if entity_fields:
+            raw_field_roles = advisor.get("field_roles", {})
+            field_roles = {}
+            if isinstance(raw_field_roles, dict):
+                field_roles = {
+                    str(field): str(role)
+                    for field, role in raw_field_roles.items()
+                    if str(field) in allowed_fields and str(role)
+                }
+            raw_correlation_roles = advisor.get("correlation_roles", {})
+            correlation_roles = {}
+            if isinstance(raw_correlation_roles, dict):
+                for role, fields in raw_correlation_roles.items():
+                    if not isinstance(fields, list):
+                        continue
+                    valid = [str(field) for field in fields if str(field) in allowed_fields]
+                    if valid:
+                        correlation_roles[str(role)] = list(dict.fromkeys(valid))
             profile = {
                 **heuristic,
                 "entity_field": entity_fields[0],
@@ -511,9 +605,16 @@ def apply_profile_advice(suggestions: list[dict[str, object]], advice: list[dict
                 "categorical_fields": categorical,
                 "numeric_fields": numeric,
                 "detectors": detectors,
+                "relationship_fields": relationships,
+            }
+            item.update({"profile": profile, "entity_fields": entity_fields, "timestamp_field": timestamp, "categorical_fields": categorical, "numeric_fields": numeric, "detectors": detectors, "relationship_fields": relationships})
+            item["profile_advisor"] = {
+                "status": "ok",
+                "model": str(advisor.get("model", "")),
+                "reason": str(advisor.get("reason", "")),
+                "field_roles": field_roles,
+                "correlation_roles": correlation_roles,
             }
-            item.update({"profile": profile, "entity_fields": entity_fields, "timestamp_field": timestamp, "categorical_fields": categorical, "numeric_fields": numeric, "detectors": detectors})
-            item["profile_advisor"] = {"status": "ok", "model": str(advisor.get("model", "")), "reason": str(advisor.get("reason", ""))}
         else:
             item["profile_advisor"] = {"status": "invalid", "reason": "advisor returned no valid entity fields"}
         output.append(item)
@@ -539,10 +640,11 @@ def suggest_stream_profiles(events: list[LogEvent], *, existing_profiles: dict[s
             if field not in IGNORED_DISCOVERY_FIELDS and _has_discovery_value(value)
         }
         stream_field_counts.update(fields)
-        semantic_groups = {_semantic_field_group(field) for field in fields}
+        semantic_groups = {group for field in fields for group in _semantic_field_groups(field)}
         stream_semantic_counts.update(semantic_groups)
         for field in fields:
-            stream_semantic_fields[_semantic_field_group(field)].add(field)
+            for group in _semantic_field_groups(field):
+                stream_semantic_fields[group].add(field)
 
     suggestions: list[dict[str, object]] = []
     for stream_id, stream_events in grouped.items():
@@ -673,6 +775,7 @@ def suggest_stream_profiles(events: list[LogEvent], *, existing_profiles: dict[s
                 "numeric_fields": numeric,
                 "detectors": detectors,
                 "field_weights": {},
+                "relationship_fields": [],
             },
             "reason": "selected common entity, time, categorical, and numeric fields from observed events; non-priority fields are included when coverage and cardinality look useful",
         })

src/fgai/stream_profiles.py

@@ -3,6 +3,13 @@ from __future__ import annotations
 from dataclasses import dataclass, field
 
 
+@dataclass(frozen=True)
+class RelationshipField:
+    left: str
+    right: str
+    name: str = ""
+
+
 @dataclass(frozen=True)
 class StreamProfile:
     stream_id: str
@@ -14,6 +21,7 @@ class StreamProfile:
     numeric_fields: tuple[str, ...] = ()
     detectors: dict[str, dict[str, object]] = field(default_factory=dict)
     field_weights: dict[str, object] = field(default_factory=dict)
+    relationship_fields: tuple[RelationshipField, ...] = ()
 
 
 def _detectors(value: object) -> dict[str, dict[str, object]]:
@@ -54,6 +62,25 @@ def _field_weights(value: object) -> dict[str, object]:
     return output
 
 
+def _relationship_fields(value: object) -> tuple[RelationshipField, ...]:
+    output: list[RelationshipField] = []
+    seen: set[tuple[str, str]] = set()
+    for item in value if isinstance(value, list) else []:
+        if not isinstance(item, dict):
+            continue
+        left = str(item.get("left") or item.get("entity") or item.get("from") or "").strip()
+        right = str(item.get("right") or item.get("value") or item.get("to") or "").strip()
+        if not left or not right or left == right:
+            continue
+        key = (left.lower(), right.lower())
+        if key in seen:
+            continue
+        seen.add(key)
+        name = str(item.get("name") or f"{left}->{right}").strip()
+        output.append(RelationshipField(left=left, right=right, name=name))
+    return tuple(output)
+
+
 def parse_profiles(value: object) -> dict[str, StreamProfile]:
     profiles: dict[str, StreamProfile] = {}
     for item in value if isinstance(value, list) else []:
@@ -69,11 +96,13 @@ def parse_profiles(value: object) -> dict[str, StreamProfile]:
             continue
         detectors = _detectors(item.get("detectors", {}))
         field_weights = _field_weights(item.get("field_weights", {}))
+        relationship_fields = _relationship_fields(item.get("relationship_fields", []))
         profiles[stream_id] = StreamProfile(
             stream_id, str(item.get("name", "")).strip() or stream_id, entity, timestamp, entity_fields,
             tuple(str(field) for field in item.get("categorical_fields", []) if field),
             tuple(str(field) for field in item.get("numeric_fields", []) if field),
             detectors,
             field_weights,
+            relationship_fields,
         )
     return profiles

tests/test_baseline.py

@@ -147,3 +147,24 @@ class BaselineTests(unittest.TestCase):
 
             self.assertNotIn("alice", store.profile_deviations(burst, profiles, min_training_days=7))
             self.assertIn("alice", store.profile_deviations(burst, profiles, min_training_days=0))
+
+    def test_custom_relationship_flags_new_user_source_ip(self):
+        with tempfile.TemporaryDirectory() as directory:
+            store = BaselineStore(str(Path(directory) / "baseline.sqlite3"))
+            profiles = parse_profiles([{
+                "stream_id": "windows",
+                "entity_field": "username",
+                "categorical_fields": ["action"],
+                "relationship_fields": [{"left": "username", "right": "srcip", "name": "user source IP"}],
+            }])
+            for index in range(12):
+                event = parse_log_line(f"fgai_stream_id=windows username=peter srcip=10.0.0.10 action=success baseline={index}")
+                store.ingest_profile_fields([event], profiles, observed_at=1_700_000_000 + index * 300)
+
+            current = [parse_log_line("fgai_stream_id=windows username=peter srcip=10.0.0.99 action=success current=1")]
+            deviations = store.profile_deviations(current, profiles)["peter"]
+
+            relation = next(item for item in deviations if item["detector"] == "new_relationship")
+            self.assertEqual(relation["field"], "relationship:username->srcip")
+            self.assertEqual(relation["value"], "10.0.0.99")
+            self.assertIn("new srcip value for username=peter", relation["reason"])

tests/test_monitor.py

@@ -5,6 +5,7 @@ from pathlib import Path
 from unittest.mock import patch
 
 from fgai.history import StatusSnapshotStore
+from fgai.history import FieldDiscoveryStore
 from fgai.monitor import add_llm_assessment, build_status, cached_status_with_error, write_refreshing_status, write_status
 
 
@@ -192,6 +193,24 @@ class MonitorTests(unittest.TestCase):
             self.assertIn("lcs_result", suggestion["categorical_fields"])
             self.assertGreater(status["baseline"]["discovery_cache_events"], 0)
 
+    def test_field_catalog_creates_discovery_events_without_raw_values(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            store = FieldDiscoveryStore(str(Path(tmp) / "history.sqlite3"))
+            store.ingest_catalog(
+                "app",
+                "App",
+                [
+                    {"name": "lcs_customer_id", "type": {"type": "string", "properties": ["enumerable"]}},
+                    {"name": "lcs_duration_ms", "type": {"type": "long", "properties": ["numeric"]}},
+                ],
+            )
+
+            events = store.synthetic_events()
+
+            fields = {field for event in events for field in event.fields}
+            self.assertIn("lcs_customer_id", fields)
+            self.assertIn("lcs_duration_ms", fields)
+
     def test_cached_status_with_error_keeps_last_good_dashboard_data(self):
         with tempfile.TemporaryDirectory() as tmp:
             cache_path = str(Path(tmp) / "status-cache.sqlite3")

tests/test_profile_suggestions.py

@@ -202,6 +202,25 @@ class ProfileSuggestionTests(unittest.TestCase):
             self.assertIn("source.ip", shared["srcip"]["aliases"])
             self.assertIn("client_ip", shared["srcip"]["aliases"])
 
+    def test_custom_namespace_and_suffix_fields_are_shared_across_streams(self):
+        events = []
+        for stream_id, field in (("app1", "lcs_customer_id"), ("app2", "lcs_order_id")):
+            events.extend(
+                parse_log_line(
+                    f"fgai_stream_id={stream_id} fgai_stream={stream_id} {field}=id{index % 4} result_state=ok actor_name=user{index % 3}"
+                )
+                for index in range(1, 20)
+            )
+
+        suggestions = {item["stream_id"]: item for item in suggest_stream_profiles(events)}
+
+        for suggestion in suggestions.values():
+            shared = {item["field"]: item for item in suggestion["shared_fields"]}
+            self.assertIn("lcs_*", shared)
+            self.assertIn("*_id", shared)
+            self.assertIn("*_status", shared)
+            self.assertIn("*_name", shared)
+
     def test_applies_valid_llm_advice_and_rejects_unknown_fields(self):
         suggestion = suggest_stream_profiles([
             parse_log_line("fgai_stream_id=windows fgai_stream=Windows username=alice hostname=host01 eventid=4625 action=failure")
@@ -214,6 +233,9 @@ class ProfileSuggestionTests(unittest.TestCase):
             "categorical_fields": ["eventid", "full_message"],
             "numeric_fields": ["missing_number"],
             "detectors": {"auth_failure": {"enabled": True, "minimum": 3, "z_threshold": 2.5}, "made_up": {"enabled": True}},
+            "relationship_fields": [{"left": "username", "right": "eventid", "name": "user event"}, {"left": "username", "right": "not_a_field"}],
+            "field_roles": {"username": "identity", "not_a_field": "entity"},
+            "correlation_roles": {"identity": ["username", "not_a_field"], "asset": ["missing_host"]},
             "reason": "Windows auth fields",
         }])[0]
 
@@ -222,6 +244,9 @@ class ProfileSuggestionTests(unittest.TestCase):
         self.assertIn("eventid", advised["profile"]["categorical_fields"])
         self.assertNotIn("full_message", advised["profile"]["categorical_fields"])
         self.assertEqual(set(advised["profile"]["detectors"]), {"auth_failure"})
+        self.assertEqual(advised["profile"]["relationship_fields"], [{"left": "username", "right": "eventid", "name": "user event"}])
+        self.assertEqual(advised["profile_advisor"]["field_roles"], {"username": "identity"})
+        self.assertEqual(advised["profile_advisor"]["correlation_roles"], {"identity": ["username"]})
 
     def test_missing_llm_advice_keeps_heuristic_profile(self):
         suggestion = suggest_stream_profiles([

tests/test_stream_profiles.py

@@ -25,3 +25,13 @@ class StreamProfileTests(unittest.TestCase):
         self.assertEqual(profiles["dns"].field_weights["qh"], 1.5)
         self.assertEqual(profiles["dns"].field_weights["query_domain"]["rare_value"], 2.0)
         self.assertNotIn("bad", profiles["dns"].field_weights)
+
+    def test_parses_relationship_fields(self):
+        profiles = parse_profiles([{"stream_id": "windows", "entity_field": "username", "relationship_fields": [{"left": "username", "right": "srcip", "name": "user source IP"}, {"left": "username", "right": "srcip"}]}])
+
+        relationships = profiles["windows"].relationship_fields
+
+        self.assertEqual(len(relationships), 1)
+        self.assertEqual(relationships[0].left, "username")
+        self.assertEqual(relationships[0].right, "srcip")
+        self.assertEqual(relationships[0].name, "user source IP")